Hedging Momentum Crashes

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict
4from pandas.core.frame import DataFrame
5# endregion
6
7class HedgingMomentumCrashes(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2000, 1, 1)
11self.SetCash(100000)
12
13self.treasury:Symbol = self.AddEquity('BIL', Resolution.Daily).Symbol
14
15self.weight:Dict[Symbol, float] = {}
16self.data:Dict[Symbol, SymbolData] = {}
17self.volatilities:List[float] = []
18self.short:List[Fundamental] = []
19
20self.volatility_period:int = 126
21self.min_period:int = 12
22self.annual_period:int = self.min_period * 21
23self.quantile:int = 10
24self.leverage:int = 5
25
26self.fundamental_count:int = 500
27self.fundamental_sorting_key = lambda x:x.DollarVolume
28
29self.selection_flag:bool = False
30self.UniverseSettings.Resolution = Resolution.Daily
31self.AddUniverse(self.FundamentalSelectionFunction)
32self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
33self.Schedule.On(self.DateRules.MonthStart(self.treasury), self.TimeRules.AfterMarketOpen(self.treasury), self.Selection)
34
35def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
36for security in changes.AddedSecurities:
37    security.SetFeeModel(data_tools.CustomFeeModel())
38    security.SetLeverage(self.leverage)
39
40def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
41# store daily stock prices
42for stock in fundamental:
43    symbol:Symbol = stock.Symbol
44
45    if symbol in self.data:
46        self.data[symbol].update_daily_return(stock.AdjustedPrice)
47        
48        if self.selection_flag:
49            self.data[symbol].update_monthly_data(stock.AdjustedPrice)
50
51if self.selection_flag:
52    short_count:int = len(self.short)
53    if short_count != 0:
54        short_portfolio_returns:np.ndarray = [self.data[stock.Symbol].get_daily_returns(self.volatility_period) for stock in self.short]
55        weights:np.ndarray = np.ones(short_count) / short_count
56        self.volatilities.append(self.annual_port_vol(weights, short_portfolio_returns))
57
58# monthly selection
59if not self.selection_flag:
60    return Universe.Unchanged
61
62selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.MarketCap != 0 and x.Market == 'usa' and \
63                                ((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE"))]
64
65if len(selected) > self.fundamental_count:
66    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
67
68perf:Dict[Symbol, float] = {}
69
70# price warmup
71for stock in selected:
72    symbol:Symbol = stock.Symbol
73
74    if symbol not in self.data:
75        self.data[symbol] = data_tools.SymbolData(self.annual_period, self.min_period)
76        history:DataFrame = self.History(symbol, self.annual_period, Resolution.Daily)
77        if history.empty:
78            self.Log(f"Not enough data for {symbol} yet.")
79            continue
80        data:pd.DataFrame = history.loc[symbol]
81        monthly_data = data.groupby(pd.Grouper(freq='MS')).last()
82        for time, row in data.iterrows():
83            self.data[symbol].update_daily_return(row.close)
84        for time, row in monthly_data.iterrows():
85            self.data[symbol].update_monthly_data(row.close)
86    
87    # calculate momentum
88    if self.data[symbol].is_ready():
89        if self.data[symbol].get_momentum() == 0:
90            continue
91        perf[stock] = self.data[symbol].get_momentum()
92
93if len(perf) >= self.quantile:
94    sorted_by_perf:List = sorted(perf.items(), key = lambda x:x[1], reverse=True)
95    quantile:int = int(len(sorted_by_perf) / self.quantile)
96    long:List[Fundamental] = [x[0] for x in sorted_by_perf[:quantile]]
97    self.short:List[Fundamental] = [x[0] for x in sorted_by_perf[-quantile:]]
98
99    if len(self.volatilities) >= self.min_period:
100        trade_direction:bool = True if self.volatilities[-1] < np.median(self.volatilities[:-1]) else False
101
102        if trade_direction:
103            for i, portfolio in enumerate([long, self.short]):
104                mc_sum:float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
105                for stock in portfolio:
106                    self.weight[stock.Symbol] = ((-1)**i) * stock.MarketCap / mc_sum
107        else:
108            # trade only treasury bills
109            self.weight[self.treasury] = 1
110
111return list(self.weight.keys())
112
113def OnData(self, data: Slice) -> None:
114if not self.selection_flag:
115    return
116self.selection_flag = False
117
118# trade execution
119portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
120self.SetHoldings(portfolio, True)
121
122self.weight.clear()
123
124def Selection(self) -> None:
125self.selection_flag = True
126
127def annual_port_vol(self, weights: np.ndarray, daily_returns: np.ndarray) -> float:
128# calculate the annual volatility of portfolio
129returns_array:np.ndarray = np.column_stack(daily_returns)
130covariance_matrix:np.ndarray = np.cov(returns_array, rowvar=False) * 252
131result:float = np.sqrt(np.dot(weights.T, np.dot(covariance_matrix, weights)))
132return result