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