Industry-Relative Stock Beta in Chinese Equities

1from AlgorithmImports import *
2from dateutil.relativedelta import relativedelta
3from pandas.core.frame import DataFrame
4from typing import Dict, List
5import statsmodels.api as sm
6# endregion
7
8class IndustryRelativeStockBetainChineseEquities(QCAlgorithm):
9
10def Initialize(self):
11self.SetStartDate(2000, 1, 1)
12self.SetCash(100000)
13
14self.weight:dict[Symbol, float] = {}
15self.leverage:float = 10
16self.industry_beta:Dict[str, RollingWindow] = {}
17self.stock_beta:Dict[Symbol, float] = {}
18
19self.period:int = 15
20self.rolling_period:int = 2
21self.traded_percentage:float = .2
22
23self.market:Symbol = self.AddEquity("FXI", Resolution.Daily).Symbol
24
25self.quantile:int = 10
26self.count:int = 100
27self.selection_flag:bool = False
28self.UniverseSettings.Resolution = Resolution.Daily
29self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
30self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
31
32def OnSecuritiesChanged(self, changes:SecurityChanges):
33for security in changes.AddedSecurities:
34    security.SetFeeModel(CustomFeeModel())
35    security.SetLeverage(self.leverage)
36    
37def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
38# monthly selection
39if not self.selection_flag:
40    return Universe.Unchanged
41# self.selection_flag = False
42
43selected:List[Symbol] = [x.Symbol
44    for x in sorted([x for x in coarse if x.HasFundamentalData], 
45    key = lambda x: x.DollarVolume, reverse = True)]
46
47return selected
48
49def FineSelectionFunction(self, fine: List[FineFundamental]) -> List[Symbol]:
50# filter chinese stocks by BusinessCountryID 
51fine:List[FineFundamental] = [x for x in fine if x.AssetClassification.MorningstarIndustryGroupCode != 0 and x.MarketCap != 0 and x.CompanyReference.BusinessCountryID == 'CHN']
52fine = sorted(fine, key=lambda x: x.MarketCap, reverse=True)[:self.count]
53
54# call history on fine and market
55history:DataFrame = self.History(list(map(lambda x:x.Symbol,fine)) + [self.market], start=self.Time.date() - relativedelta(months=self.period), end=self.Time.date())['close'].unstack(level=0)
56history = history.groupby(pd.Grouper(freq='M')).last()
57
58# sort stocks on industry numbers
59grouped_industries:Dict[MorningstarIndustryGroupCode, List[Symbol]] = {}
60
61for stock in fine:
62    symbol:Symbol = stock.Symbol
63    
64    industry_group_code:str = str(stock.AssetClassification.MorningstarIndustryGroupCode)
65
66    if not industry_group_code in grouped_industries:
67        grouped_industries[industry_group_code] = []
68    grouped_industries[industry_group_code].append(symbol)
69
70if len(history) >= self.period and self.market in history.columns and not history[self.market].isnull().values.any():
71    history = history.iloc[-self.period:]
72    history.index = history.index.to_pydatetime()
73
74    asset_returns:DataFrame = history.pct_change().iloc[1:]
75
76    stock_returns = asset_returns.loc[:, asset_returns.columns != self.market].tail(-1)
77    asset_returns['market_lag_t-1'] = asset_returns[self.market].shift(1)
78    asset_returns['market_lag_t+1'] = asset_returns[self.market].shift(-1)
79    asset_returns = asset_returns.iloc[1:-1]
80    stock_returns = stock_returns.iloc[:-1]
81
82    industry_df:DataFrame = pd.DataFrame(index=asset_returns.index)
83
84    for industry_code in grouped_industries:
85        if len(grouped_industries[industry_code]) > 1:
86            ind_symbols:List[Symbol] = grouped_industries[industry_code]
87            if all(symbol in asset_returns.columns for symbol in ind_symbols):
88                industry_df[industry_code] = asset_returns[ind_symbols].mean(axis=1)
89        else:
90            stock_returns = stock_returns.drop(stock_returns[grouped_industries[industry_code]], axis=1)
91
92    # run industry regression
93    x:np.ndarray = asset_returns[[self.market, 'market_lag_t-1', 'market_lag_t+1']].values
94    y:np.ndarray = industry_df.values
95    model = self.multiple_linear_regression(x, y)
96    beta_values_industry:np.ndarray = sum(model.params[1:])
97
98    for i, industry in enumerate(list(industry_df.columns)):
99        if industry not in self.industry_beta:
100            self.industry_beta[industry] = RollingWindow[float](self.rolling_period)
101        if not np.isnan(beta_values_industry[i]):
102            self.industry_beta[industry].Add(beta_values_industry[i])
103
104    # run stock regression
105    x:np.ndarray = asset_returns[[self.market, 'market_lag_t-1', 'market_lag_t+1']].values
106    y:np.ndarray = stock_returns.values
107    model = self.multiple_linear_regression(x, y)
108    beta_values_stocks:np.ndarray = sum(model.params[1:])
109
110    for i, asset in enumerate(list(stock_returns.columns)):
111        asset_s:Symbol = self.Symbol(asset)
112        if not np.isnan(beta_values_stocks[i]):
113            self.stock_beta[asset_s] = beta_values_stocks[i]
114
115irsb_values:Dict[FineFundamental, float] = {}
116
117# IRSB calculation
118for stock in fine:
119    symbol:Symbol = stock.Symbol
120    if symbol in self.stock_beta:
121        for industry in grouped_industries:
122            if industry in self.industry_beta:
123                if symbol in grouped_industries[industry] and self.industry_beta[industry].IsReady:
124                    irsb_values[stock] = (self.stock_beta[symbol] - self.industry_beta[industry][1]) / self.industry_beta[industry][1]
125
126# sorting by IRSB values
127if len(irsb_values) >= self.quantile:
128    sorted_by_irsb:List[FineFundamental] = sorted(irsb_values, key=irsb_values.get, reverse=True)
129    quantile:int = len(sorted_by_irsb) // self.quantile
130    
131    # portfolios
132    long:List[FineFundamental] = sorted_by_irsb[:quantile]
133    short:List[FineFundamental] = sorted_by_irsb[-quantile:]
134
135    # calculate weights based on market cap
136    sum_long:float = sum([x.MarketCap for x in long])
137    for stock in long:
138        self.weight[stock.Symbol] = (stock.MarketCap / sum_long) * self.traded_percentage 
139
140    sum_short:float = sum([x.MarketCap for x in short])
141    for stock in short:
142        self.weight[stock.Symbol] = (-stock.MarketCap / sum_short) * self.traded_percentage 
143
144return list(self.weight.keys())
145
146def OnData(self, data):
147if not self.selection_flag:
148    return
149self.selection_flag = False
150
151# trade execution
152stocks_invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
153for symbol in stocks_invested:
154    if symbol not in self.weight:
155        self.Liquidate(symbol)
156        
157for symbol, w in self.weight.items():
158    if symbol in data and data[symbol]:
159        self.SetHoldings(symbol, w)
160    
161self.weight.clear()
162
163def Selection(self):
164self.selection_flag = True
165
166def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
167# x:np.ndarray = np.array(x).T
168x = sm.add_constant(x, prepend=True)
169result = sm.OLS(endog=y, exog=x).fit()
170return result
171
172# Custom fee model
173class CustomFeeModel(FeeModel):
174def GetOrderFee(self, parameters):
175fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
176return OrderFee(CashAmount(fee, "USD"))