Resilient Asset Allocation

1from AlgorithmImports import *
2from typing import List
3from dateutil.relativedelta import relativedelta, FR
4# endregion
5
6class ResilientAssetAllocation(QCAlgorithm):
7
8def Initialize(self) -> None:
9 self.SetStartDate(2010, 1, 1)
10 self.SetCash(100000)
11 
12 # set assets variables
13 self.static_universe:List[str] = ['QQQ', 'IWN', 'IEF', 'TLT', 'GLD']
14 self.cash_universe:List[str] = ['IEF', 'TLT']
15 self.canary_universe:List[str] = ['VWO', 'BND']
16
17 self.ue_period:int = 12
18
19 momentum_periods:List[int] = [21, 63, 126, 252]
20 self.score_weights:np.ndarray = np.array([12, 4, 2, 1])
21
22 self.symbol_data:Dict[List] = {}
23
24 # warm up of indicators
25 self.SetWarmup(self.ue_period * 31, Resolution.Daily)
26
27 for ticker in self.static_universe + self.cash_universe:
28     self.AddEquity(ticker, Resolution.Daily)
29
30 for ticker in self.canary_universe:
31     self.AddEquity(ticker, Resolution.Daily)            
32     self.symbol_data[ticker] = [self.MOMP(ticker, period, Resolution.Daily) for period in momentum_periods]
33
34 # SMA assets
35 self.ue:Symbol = self.AddData(UnemploymentRate, 'UE', Resolution.Daily).Symbol
36
37 self.ue_window = RollingWindow[float](12)
38
39def OnData(self, data: Slice) -> None:
40 if self.IsWarmingUp:
41     return
42
43 # calculate the momentum scores of the canary symbols
44 canary_score:List[float] = [np.dot(np.array([momentum.Current.Value for momentum in item[1]]), self.score_weights) \
45                 for item in self.symbol_data.items() if item[0] in self.canary_universe and \
46                 all(indicator.IsReady for indicator in item[1])]
47  
48 # wait until all the assets' indicators are warmed-up
49 if len(canary_score) != len(self.canary_universe):
50     return
51
52 # rebalance when 'UE' data are in - monthly
53 if data.ContainsKey(self.ue) and data[self.ue]:
54 
55     self.ue_window.Add(data[self.ue].Value)
56
57     if not self.ue_window.IsReady:
58         return
59
60     if self.ue_window.IsReady:
61         invested_tickers:List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
62         traded_universe:List[str] = self.static_universe if not self.Portfolio.Invested else invested_tickers
63
64         # change allocation of assets
65         if not self.Portfolio.Invested:
66             # both trends are positive
67             if data[self.ue].Value > self.ue_window[self.ue_window.Count-1] and canary_score[0] < 0 and canary_score[1] < 0:
68                 traded_universe = self.cash_universe
69             else:
70                 traded_universe = self.static_universe
71        
72         # firstly, liquidate symbols that should not be held
73         for ticker in invested_tickers:
74             if ticker not in traded_universe:
75                 self.Liquidate(ticker)
76         
77         # rebalance new portfolio
78         for ticker in traded_universe:
79             if ticker in data and data[ticker]:
80                 self.SetHoldings(ticker, 1 / len(traded_universe))
81 else:
82     if self.Portfolio.Invested:
83         last_update_date:datetime.date = UnemploymentRate.get_last_update_date()
84         
85         # custom data stopped comming in
86         if self.Securities[self.ue].GetLastData() and self.Time.date() > last_update_date:
87             self.Liquidate()
88
89class UnemploymentRate(PythonData):
90def GetSource(self, config: SubscriptionDataConfig, date: datetime, isLiveMode: bool) -> SubscriptionDataSource:
91 return SubscriptionDataSource('data.quantpedia.com/backtesting_data/economic/UNEMPLOYMENT_RATE.csv', SubscriptionTransportMedium.RemoteFile, FileFormat.Csv)
92
93_last_update_date:datetime.date = datetime(1,1,1).date()
94
95@staticmethod
96def get_last_update_date() -> datetime.date:
97return UnemploymentRate._last_update_date
98
99def Reader(self, config: SubscriptionDataConfig, line: str, date: datetime, isLiveMode: bool) -> BaseData:
100 data = UnemploymentRate()
101 data.Symbol = config.Symbol
102
103 if not line[0].isdigit(): return None
104 split = line.split(';')
105 
106 # Parse the CSV file's columns into the custom data class - first Friday of the month
107 data.Time = (datetime.strptime(split[0], '%Y-%m-%d').date() + relativedelta(weekday=FR(1))) + timedelta(days=1)
108
109 if data.Time.date() > UnemploymentRate._last_update_date:
110     UnemploymentRate._last_update_date = data.Time.date()
111
112 data.Value = float(split[1])
113 
114 return data