VIX Beta Factor in Chinese Equities

1from AlgorithmImports import *
2import data_tools
3from statsmodels.tsa.arima.model import ARIMA
4import numpy as np
5from typing import List, Dict
6from pandas.core.frame import DataFrame as DF
7# endregion
8
9class VIXBetaFactorinChineseEquities(QCAlgorithm):
10
11def Initialize(self):
12self.SetStartDate(2015, 1, 1)
13self.SetCash(100000)
14
15self.quantile:int = 10
16self.leverage:int = 10
17self.period:int = 1 * 21
18self.max_missing_days:int = 5
19self.SetWarmUp(self.period, Resolution.Daily)
20
21# market cap filtering
22self.exclusion_flag:bool = True
23self.market_cap_cutoff:float = .5
24
25self.data:dict[str, data_tools.SymbolData] = {}
26self.weights:dict[Symbol, float] = {}
27
28# cboe data
29self.cboe_volatility:Symbol = self.AddData(data_tools.CBOEData, 'CBOE_china_etf_volatility', Resolution.Daily).Symbol
30self.cboe_volatility_data:RollingWindow = RollingWindow[float](self.period)
31
32self.spy:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
33
34self.selection_flag:bool = False
35self.UniverseSettings.Resolution = Resolution.Daily
36self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
37self.Schedule.On(self.DateRules.MonthStart(self.spy), self.TimeRules.AfterMarketOpen(self.spy), self.Selection)
38
39def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
40for security in changes.AddedSecurities:
41    security.SetFeeModel(data_tools.CustomFeeModel())
42    security.SetLeverage(self.leverage)
43
44def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
45if not self.selection_flag:
46    return Universe.Unchanged
47
48if not self.cboe_volatility_data.IsReady:
49    return Universe.Unchanged
50else:
51    # custom data is still comming in
52    if self.Securities[self.cboe_volatility].GetLastData() and (self.Time.date() - self.Securities[self.cboe_volatility].GetLastData().Time.date()).days > self.max_missing_days:
53        return Universe.Unchanged
54
55selected:List[Symbol] = [x.Symbol for x in coarse if x.HasFundamentalData and x.Price > 5]
56
57return selected
58
59def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
60# filter chinese stocks by BusinessCountryID 
61fine:List[FineFundamental] = list(filter(lambda stock: stock.MarketCap != 0 and stock.CompanyReference.BusinessCountryID == 'CHN', fine))
62
63if self.exclusion_flag:
64    # exclude 30% of lowest stocks by MarketCap
65    sorted_by_market_cap:List[FineFundamental] = sorted(fine, key = lambda x: x.MarketCap)
66    fine = sorted_by_market_cap[int(len(sorted_by_market_cap) * self.market_cap_cutoff):]
67
68# calculate vix beta
69vix_beta:Dict[FineFundamental, float] = {}
70
71# VIX innovations
72vix:np.ndarray = np.array(list(self.cboe_volatility_data))
73arma_model = ARIMA(vix, order=(1,0,1))
74model_fit = arma_model.fit()
75innovations:np.ndarray = model_fit.resid[::-1]
76
77for stock in fine:
78    symbol:Symbol = stock.Symbol
79    market_cap:float = stock.MarketCap
80
81    history:DF = self.History(symbol, self.period, Resolution.Daily)
82    if history.empty or history.loc[symbol].close.size != self.period:
83        continue
84
85    closes:np.ndarray = history['close'].values
86    stock_returns:np.array = (closes[1:] - closes[:-1]) / closes[:-1]
87
88    model_fit = data_tools.multiple_linear_regression(innovations[2:], stock_returns[:-1])
89    vix_beta[stock] = model_fit.params[1]
90
91if len(vix_beta) < self.quantile:
92    return Universe.Unchanged
93
94quantile:int = int(len(vix_beta) / self.quantile)
95sorted_vix_beta:List[FineFundamental] = [x[0] for x in sorted(vix_beta.items(), key=lambda item: item[1], reverse=True)]
96long:List[FineFundamental] = sorted_vix_beta[:quantile]
97short:List[FineFundamental] = sorted_vix_beta[-quantile:]
98
99total_long_cap:float = sum(list(map(lambda stock: stock.MarketCap, long)))
100for stock in long:
101    self.weights[stock.Symbol] = stock.MarketCap / total_long_cap
102
103total_short_cap:float = sum(list(map(lambda stock: stock.MarketCap, short)))
104for stock in short:
105    self.weights[stock.Symbol] = -stock.MarketCap / total_short_cap
106
107return list(self.weights.keys())
108
109def OnData(self, data: Slice) -> None:
110# store CBOE data
111if self.cboe_volatility in data and data[self.cboe_volatility]:
112    close:float = data[self.cboe_volatility].Value
113    self.cboe_volatility_data.Add(close)
114
115if self.IsWarmingUp:
116    return
117
118# monthly rebalance
119if not self.selection_flag:
120    return
121self.selection_flag = False
122
123# trade execution
124stocks_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
125for symbol in stocks_invested:
126    if symbol not in self.weights:
127        self.Liquidate(symbol)
128
129for symbol, w in self.weights.items():
130    if symbol in data and data[symbol]:
131        self.SetHoldings(symbol, w)
132
133self.weights.clear()
134
135def Selection(self):
136self.selection_flag = True