Bold Asset Allocation

1from AlgorithmImports import *
2import pandas as pd
3import numpy as np
4from typing import List, Dict
5from pandas.core.frame import DataFrame
6# endregion
7
8class BoldAssetAllocation(QCAlgorithm):
9
10def Initialize(self):
11self.SetCash(100000)
12self.SetStartDate(2008, 1, 1)
13
14# all assets
15self.offensive:List[str] = [
16    "SPY", "QQQ",
17    "IWM", "VGK",
18    "EWJ", "VWO",
19    "VNQ", "DBC",
20    "GLD", "TLT",
21    "HYG", "LQD",
22]
23
24self.protective:List[str] = ["SPY", "VWO", "VEA", "BND"]
25self.defensive:List[str] = ["TIP", "DBC", "BIL", "IEF", "TLT", "LQD", "BND"]
26self.safe:str = "BIL"
27
28# strategy parameters (our implementation)
29self.prds:List[int] = [1, 3, 6, 12]                     # fast momentum settings
30self.prdweights:np.ndarray = np.array([12, 4, 2, 1])    # momentum weights
31self.LO, self.LP, self.LD, self.B, self.TO, self.TD = [
32    len(self.offensive),
33    len(self.protective),
34    len(self.defensive),
35    1,
36    6,
37    3,
38]  # number of offensive, protective, defensive assets, threshold for "bad" assets, select top n of offensive and defensive assets
39
40self.hprd:int = (max(self.prds + [self.LO, self.LD]) * 21 + 50)  # momentum periods calculation
41
42# repeat safe asset so it can be selected multiple times
43self.all_defensive:List[str] = self.defensive + [self.safe] * max(
44    0, self.TD - sum([1 * (e == self.safe) for e in self.defensive])
45)
46
47self.equities:List[str] = list(
48    dict.fromkeys(self.protective + self.offensive + self.all_defensive)
49)
50
51leverage:int = 3
52for equity in self.equities:
53    data:Equity = self.AddEquity(equity, Resolution.Daily)
54    data.SetLeverage(leverage)
55
56self.recent_month:int = -1
57
58def OnData(self, data:Slice) -> None:
59if self.IsWarmingUp:
60    return
61
62# monthly rebalance
63if self.recent_month == self.Time.month:
64    return
65self.recent_month = self.Time.month
66
67# get price data and trading weights
68h:DataFrame = self.History(self.equities, self.hprd, Resolution.Daily)["close"].unstack(level=0)
69weights:pd.Series = self.trade_weights(h)
70
71# trade
72self.SetHoldings([PortfolioTarget(x, y) for x, y in zip(weights.index, weights.values) if x in data and data[x]])
73
74def trade_weights(self, hist:DataFrame) -> pd.Series:
75# initialize weights series
76weights:pd.Series = pd.Series(0, index=hist.columns)
77# end of month values
78h_eom:DataFrame = hist.loc[hist.groupby(hist.index.to_period("M")).apply(lambda x: x.index.max())].iloc[:-1, :]
79
80# Check if protective universe is triggered.
81
82# build dataframe of momentum values
83mom:DataFrame = (h_eom.iloc[-1, :].div(h_eom.iloc[[-p - 1 for p in self.prds], :], axis=0) - 1)
84mom = mom.loc[:, self.protective].T
85
86# determine number of protective securities with negative weighted momentum
87n_protective:float = np.sum(np.sum(mom.values * self.prdweights, axis=1) < 0)
88
89# % equity offensive
90pct_in:float = 1 - min(1, n_protective / self.B)
91
92# Get weights for offensive and defensive universes.
93
94# determine weights of offensive universe
95if pct_in > 0:
96    # price / SMA
97    mom_in = h_eom.iloc[-1, :].div(h_eom.iloc[[-t for t in range(1, self.LO + 1)]].mean(axis=0), axis=0)
98    mom_in = mom_in.loc[self.offensive].sort_values(ascending=False)
99    # equal weightings to top relative momentum securities
100    in_weights = pd.Series(pct_in / self.TO, index=mom_in.index[:self.TO])
101    weights = pd.concat([weights, in_weights])
102
103# determine weights of defensive universe
104if pct_in < 1:
105    # price / SMA
106    mom_out = h_eom.iloc[-1, :].div(h_eom.iloc[[-t for t in range(1, self.LD + 1)]].mean(axis=0), axis=0)
107    mom_out = mom_out.loc[self.all_defensive].sort_values(ascending=False)
108    # equal weightings to top relative momentum securities
109    out_weights = pd.Series((1 - pct_in) / self.TD, index=mom_out.index[:self.TD])
110    weights = pd.concat([weights, out_weights])
111
112weights:pd.Series = weights.groupby(weights.index).sum()
113
114return weights