Keller’s & van Putten’s Generalized Momentum and Flexible Asset Allocation

1from AlgorithmImports import *
2from pandas.core.frame import DataFrame
3from typing import List, Dict, Tuple
4# endregion
5
6class GeneralizedMomentumandFlexibleAssetAllocation(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2005, 1, 1)
10self.SetCash(100000)
11
12self.period:int = 4 * 21
13self.top_equities:int = 3
14self.leverage:int = 2
15self.normalization_weights:np.ndarray = np.array([1., .5, .5])
16self.momentum_threshold:float = 0.
17
18self.investment_universe:List[str] = [
19    "VTI", "VEA",
20    "VWO", "SHY",
21    "BND", "GSG",
22    "VNQ",
23]
24
25# equity subscription
26for ticker in self.investment_universe:
27    self.AddEquity(ticker, Resolution.Daily, leverage=self.leverage)
28self.cash:Symbol = self.AddEquity("BIL", Resolution.Daily, leverage=self.leverage).Symbol
29
30self.recent_month:int = -1
31
32def OnData(self, data:Slice) -> None:
33# monthly rebalance
34if self.Time.month == self.recent_month:
35    return
36self.recent_month = self.Time.month
37
38if not (data.ContainsKey(self.cash) and data[self.cash]):
39    return
40
41period:int = self.period + 1
42closes:DataFrame = self.History(self.investment_universe, period, Resolution.Daily).unstack(level=0)['close']
43
44if len(closes) == period and len(closes.columns) >= len(self.investment_universe):
45    # momentum, volatility and correlation calculation
46    returns:DataFrame = closes.pct_change().iloc[1:]
47    momentum:pd.Series = closes.iloc[0] / closes.iloc[-1] - 1
48    volatility:pd.Series = returns.std(axis=0) * np.sqrt(252)
49    corr:DataFrame = returns.corr(method='pearson')
50    mean_corr:pd.Series = abs((corr.sum(axis=0) - 1) / len(returns.columns)) # -1 for diagonal value we don't want to count in
51
52    sorted_momentum:pd.Series = momentum.sort_index().sort_values(axis=0, ascending=False)
53    sorted_volatility:pd.Series = volatility.sort_index().sort_values(axis=0, ascending=True)
54    sorted_corr:pd.Series = mean_corr.sort_index().sort_values(axis=0, ascending=True)
55
56    L:Dict[str, float] = {asset: np.dot(self.normalization_weights, np.array([list(sorted_momentum.index).index(asset), list(sorted_volatility.index).index(asset), list(sorted_corr.index).index(asset)])) for asset in list(returns.columns)}
57    top_by_L:List[str] = sorted(L, key=L.get, reverse=True)[:self.top_equities]
58    
59    # equity allocation
60    weight:Dict[Symbol, float] = {}
61    for asset in top_by_L:
62        if momentum.loc[asset] > self.momentum_threshold:
63            weight[self.Symbol(asset)] = 1. / float(self.top_equities)
64    
65    # cash allocation
66    if len(weight) != self.top_equities:
67        cash_fraction_count:int = self.top_equities - len(weight)
68        weight[self.cash] = float(cash_fraction_count) / float(self.top_equities)
69
70    # trade execution
71    invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
72    for symbol in invested:
73        if symbol not in weight:
74            self.Liquidate(symbol)
75
76    for symbol, w in weight.items():
77        self.SetHoldings(symbol, w)
78else:
79    if self.Portfolio.Invested:
80        self.Liquidate()