Improved Cross-Asset Time-Series Momentum I-XTSM

1from AlgorithmImports import *
2# endregion
3
4class ImprovedCrossAssetTimeSeriesMomentumIXTSM(QCAlgorithm):
5
6def Initialize(self):
7 self.SetStartDate(2005, 1, 1) 
8 self.SetCash(100000)
9
10 self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
11 self.bil:Symbol = self.AddEquity('BIL', Resolution.Daily).Symbol
12 self.industrial_metal_index:Symbol = self.AddData(IndustrialMetalIndex, 'IMI', Resolution.Daily).Symbol
13 
14 self.leverage:int = 5
15 self.period:int = 12 * 21
16 self.selection_months:List[int] = [1, 4, 7, 10]
17
18 self.SetWarmup(self.period, Resolution.Daily)
19
20 for symbol in [self.market, self.bil]:
21     self.Securities[symbol.Value].SetLeverage(self.leverage)
22
23 self.market_mom:Momentum = self.MOM(self.market, self.period, Resolution.Daily)
24 self.industrial_metal_mom:Momentum = self.MOM(self.industrial_metal_index, self.period, Resolution.Daily)
25
26 self.rebalance_flag:bool = False
27 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
28 self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
29
30def OnData(self, data: Slice) -> None:
31 if self.IsWarmingUp:
32     return
33
34 # quarterly rebalance
35 if not self.rebalance_flag:
36     return
37 self.rebalance_flag = False
38
39 # check industrial metal index data arrival
40 industrial_metal_last_update_date:datetime.date = IndustrialMetalIndex.get_last_update_date()
41 if self.Securities[self.industrial_metal_index].GetLastData():
42     if self.Time.date() >= industrial_metal_last_update_date: 
43         self.Liquidate()
44         return
45
46 # compare momentums and define traded asset and trade direction
47 traded_asset:Symbol|None = None
48 trade_direction:int = 0
49
50 if all(x in data and data[x] for x in [self.market, self.industrial_metal_index]):
51     market_mom:float = self.market_mom.Current.Value
52     industrial_metal_mom:float = self.industrial_metal_mom.Current.Value
53
54     if market_mom >= 0:
55         traded_asset = self.market
56         trade_direction = 1
57     else:
58         if industrial_metal_mom < 0:
59             traded_asset = self.market
60             trade_direction = -1
61         elif industrial_metal_mom >= 0:
62             traded_asset = self.bil
63             trade_direction = 1
64 
65     # trade execution
66     if not self.Portfolio[traded_asset].Invested:
67         self.Liquidate()
68         self.SetHoldings(traded_asset, trade_direction)
69     else:
70         self.SetHoldings(traded_asset, trade_direction)
71
72def Selection(self) -> None:
73 if self.Time.month in self.selection_months:
74     self.rebalance_flag = True
75
76# Source: https://www.investing.com/indices/gsci-industrial-metals-historical-data
77class IndustrialMetalIndex(PythonData):
78def GetSource(self, config, date, isLiveMode):
79 return SubscriptionDataSource('data.quantpedia.com/backtesting_data/index/GSCI_Industrial_Metal_index.csv', SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
80
81_last_update_date:datetime.date = datetime(1,1,1).date()
82
83@staticmethod
84def get_last_update_date() -> datetime.date:
85return IndustrialMetalIndex._last_update_date
86
87def Reader(self, config, line, date, isLiveMode):
88 data = IndustrialMetalIndex()
89 data.Symbol = config.Symbol
90
91 if not line[0].isdigit(): return None
92 split = line.split(';')
93 
94 # Parse the CSV file's columns into the custom data class
95 data.Time = datetime.strptime(split[0], "%m/%d/%Y") + timedelta(days=1)
96 data.Value = float(split[1])
97
98 if data.Time.date() > IndustrialMetalIndex._last_update_date:
99     IndustrialMetalIndex._last_update_date = data.Time.date()
100 
101 return data