Keller’s & Keunig’s Vigilant Asset Allocation

1from AlgorithmImports import *
2from typing import List, Dict
3#endregion
4
5class VigilantAssetAllocation(QCAlgorithm):
6
7def Initialize(self) -> None:
8self.SetCash(100000)
9self.SetStartDate(2008, 1, 1)
10
11# VAA-G12 risk investment universe assets
12self.growth_symbols:List[str] = [
13    "SPY", "IWM", "QQQ", "VGK", 
14    "EWJ", "EEM", "EFA", "ACWX", 
15    "IYR", "GSG", "GLD", "SHY", 
16    "IEF", "TLT", "LQD", "HYG", "AGG"
17    ]
18
19# cash universe assets
20self.safety_symbols:List[str] = ["LQD", "IEF", "SHY"]
21
22momentum_periods:List[int] = [21, 63, 126, 252]
23self.score_weights:np.ndarray = np.array([12, 4, 2, 1])
24
25self.SetWarmUp(momentum_periods[-1], Resolution.Daily)
26
27self.symbol_data:Dict[List] = {}
28self.B:int = 4
29self.T:int = 2
30self.leverage:int = 3
31
32for ticker in self.growth_symbols + self.safety_symbols:
33    data:Equity = self.AddEquity(ticker, Resolution.Minute)
34    data.SetLeverage(self.leverage)
35    self.symbol_data[ticker] = [self.MOMP(ticker, period, Resolution.Daily) for period in momentum_periods]
36
37self.recent_month:int = -1
38
39def OnData(self, data:Slice) -> None:
40# monthly rebalance
41if self.recent_month == self.Time.month:
42    return
43self.recent_month = self.Time.month
44
45if self.IsWarmingUp: return
46
47# this approach overweights the front month momentum value and progressively underweights older momentum values
48growth_score:Dict[str, float] = { ticker : sum(np.array([x.Current.Value for x in momentum_indicators]) * self.score_weights) for ticker, momentum_indicators in self.symbol_data.items() 
49                                if ticker in self.growth_symbols if all(ind.IsReady for ind in momentum_indicators) }
50
51safe_score:Dict[str, float] = { ticker : sum(np.array([x.Current.Value for x in momentum_indicators]) * self.score_weights) for ticker, momentum_indicators in self.symbol_data.items() 
52                                if ticker in self.safety_symbols if all(ind.IsReady for ind in momentum_indicators) }
53
54# wait until all the assets' indicators are warmed-up
55if len(growth_score) != len(self.growth_symbols) or len(safe_score) != len(self.safety_symbols):
56    return
57
58sorted_growth:List = sorted(growth_score.items(), key=lambda x: x[1], reverse=True)
59sorted_safe:List = sorted(safe_score.items(), key=lambda x: x[1], reverse=True)
60
61# count the number of risky assets with negative momentum scores and store in b
62b:int = sum(1 for x in sorted_growth if x[1] < 0)
63CF:float = float(b) / float(self.B)
64        
65# select two offensive asset with the highest score and allocate 25% of the portfolio to that asset at the close
66# pick two best from risky and one risk-free symbols
67best_growth:List[str] = [x[0] for x in sorted_growth][:self.T]
68best_safe:str = sorted_safe[0][0]
69
70# if more than four risky assets exhibit negative momentum scores, select the risk-free asset (LQD, IEF or SHY) with the highest score
71if b >= self.B:
72    # liquidate
73    invested:List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
74    for ticker in invested:
75        if ticker != best_safe:
76            self.Liquidate(ticker)
77
78    self.SetHoldings(best_safe, 1)
79
80# if none of the risky assets come back with negative momentum scores, allocation 100% to the 2 best scoring risky assets
81else:
82    # liquidate
83    invested:List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
84    for ticker in invested:
85        if ticker not in best_growth + [best_safe]:
86            self.Liquidate(ticker)
87
88    for ticker in best_growth:
89        self.SetHoldings(ticker, (1-CF) / 2) 
90    
91    self.SetHoldings(best_safe, CF)