Keller’s & Keunig’s Defensive Asset Allocation

1from AlgorithmImports import *
2from typing import List, Dict
3import numpy as np
4# endregion
5
6class DefensiveAssetAllocation(QCAlgorithm):
7
8def Initialize(self):
9self.SetCash(100000)
10self.SetStartDate(2008, 1, 1)
11
12self.risky_topT:int = 6
13self.half_risky_topT:int = 3
14
15self.risky_universe:List[str] = [
16    "SPY", "IWM",
17    "QQQ", "VGK",
18    "EWJ", "VWO",
19    "VNQ", "GSG",
20    "GLD", "TLT",
21    "HYG", "LQD",
22]
23
24self.cash_universe:List[str] = ["BIL", "IEF", "LQD"]
25self.canary_universe:List[str] = ["VWO", "BND"]
26self.all_tickers:List[str] = self.risky_universe + self.cash_universe + self.canary_universe
27
28self.period_list:List[int] = [21, 63, 126, 252]
29self.objective_score_weights:np.ndarray = [12., 4., 2., 1.]
30
31self.SetWarmUp(max(self.period_list), Resolution.Daily)
32
33self.momp_data_by_ticker:Dict[str, List[MomentumPercent]] = {}
34
35# subscribe data
36for ticker in self.all_tickers:
37    self.AddEquity(ticker, Resolution.Daily)
38    self.momp_data_by_ticker[ticker] = [self.MOMP(ticker, period, Resolution.Daily) for period in self.period_list]
39
40self.recent_month:int = -1
41
42def OnData(self, data:Slice) -> None:
43if self.IsWarmingUp:
44    return
45
46if self.recent_month == self.Time.month:
47    return
48self.recent_month = self.Time.month
49
50# rank all the risky and cash symbol groups by momentum score
51sorted_risky:List = sorted([item for item in self.momp_data_by_ticker.items() if item[0] in self.risky_universe and
52                            all(indicator.IsReady for indicator in item[1])],   # all indicators for ticker are ready
53                            key=lambda x: np.dot(np.array([momentum.Current.Value for momentum in x[1]]), self.objective_score_weights),
54                            reverse=True)
55
56best_growth:List[str] = [x[0] for x in sorted_risky[:self.risky_topT]]
57second_best_growth:List[str] = best_growth[:self.half_risky_topT]
58
59sorted_cash:List = sorted([item for item in self.momp_data_by_ticker.items() if item[0] in self.cash_universe and
60                            all(indicator.IsReady for indicator in item[1])],   # all indicators for ticker are ready
61                            key=lambda x: np.dot(np.array([momentum.Current.Value for momentum in x[1]]), self.objective_score_weights),
62                            reverse=True)
63if len(sorted_cash) < 1:
64    self.Liquidate()
65    return
66
67best_cash:str = sorted_cash[0][0]
68
69# calculate the momentum score of the canary symbols
70canary_scores:List[float] = [np.dot(np.array([momentum.Current.Value for momentum in item[1]]), self.objective_score_weights) \
71                for item in self.momp_data_by_ticker.items() if item[0] in self.canary_universe and \
72                all(indicator.IsReady for indicator in item[1])]
73
74if len(canary_scores) != 2:
75    self.Liquidate()
76    return
77
78weight:Dict[str, float] = {}
79if all(x < 0 for x in canary_scores):
80    weight[best_cash] = 1.
81
82elif all(x > 0 for x in canary_scores):
83    weight = { x : 1. / float(len(best_growth)) for x in best_growth}
84
85elif any(x < 0 for x in canary_scores):
86    weight[best_cash] = .5
87    for x in second_best_growth:
88        weight[x] = .5 / float(len(second_best_growth))
89
90# liquidate
91invested:List[str] = [x.Symbol.Value for x in self.Portfolio.Values if x.Invested]
92for ticker in invested:
93    if ticker not in weight:
94        self.Liquidate(ticker)
95
96# new trade execution / rebalance
97for ticker, w in weight.items():
98    if ticker in data and data[ticker]:
99        self.SetHoldings(ticker, w)