Absolute Delta Beta Strategy in Chinese Equities

1from AlgorithmImports import *
2from data_tools import QuantpediaCSI300, CustomFeeModel, SymbolData, MultipleLinearRegression
3# endregion
4
5class AbsoluteDeltaBetaStrategyInChineseEquities(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2010, 1, 1)   # CSI 300 data starts in 2005
9self.SetCash(100000)
10
11self.leverage:int = 10
12self.quantile:int = 10
13
14self.min_prices:int = 15
15
16self.weights:dict[Symbol, float] = {}
17self.data:dict[Symbol, SymbolData] = {}
18
19self.benchmark:Symbol = self.AddData(QuantpediaCSI300, 'CSI_300', Resolution.Daily).Symbol
20self.benchmark_data = SymbolData()
21
22self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
23
24self.selection_flag = False
25self.UniverseSettings.Resolution = Resolution.Daily
26self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
27self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.BeforeMarketClose(self.market, 0), self.Selection)
28
29def OnSecuritiesChanged(self, changes):
30for security in changes.AddedSecurities:
31    security.SetFeeModel(CustomFeeModel())
32    security.SetLeverage(self.leverage)
33
34def CoarseSelectionFunction(self, coarse):
35curr_date:datetime.date = self.Time.date()
36
37for stock in coarse:
38    symbol:Symbol = stock.Symbol
39    
40    if symbol in self.data:
41        self.data[symbol].update_price_with_date(stock.AdjustedPrice, curr_date)
42
43# monthly rebalance
44if not self.selection_flag:
45    return Universe.Unchanged
46
47selected_symbols = [x.Symbol for x in coarse if x.HasFundamentalData]
48
49return selected_symbols
50
51def FineSelectionFunction(self, fine):
52csi_rets_with_dates:list[datetime.date, float]|None = self.benchmark_data.get_rets_with_dates() \
53    if self.benchmark_data.is_ready(self.min_prices) else None
54
55if csi_rets_with_dates == None:
56    return Universe.Unchanged
57
58fine = list(filter(lambda stock: stock.MarketCap != 0 and stock.CompanyReference.BusinessCountryID == 'CHN', fine))
59
60# Exclude 30% of lowest stocks by MarketCap
61sorted_by_market_cap = sorted(fine, key = lambda x: x.MarketCap)
62fine = sorted_by_market_cap[int(len(sorted_by_market_cap) * 0.3):]
63
64ADB:dict[Symbol, float] = {}
65
66for stock in fine:
67    symbol:Symbol = stock.Symbol
68    market_cap:float = stock.MarketCap
69
70    if symbol not in self.data:
71        self.data[symbol] = SymbolData()
72
73    if self.data[symbol].is_ready(self.min_prices):
74        stock_rets_with_dates:list[datetime.date, float] = self.data[symbol].get_rets_with_dates()
75        rising_x, rising_y, declining_x, declining_y = ([] for i in range(4))
76
77        for (csi_ret, csi_date), (stock_ret, stock_date) in zip(csi_rets_with_dates, stock_rets_with_dates):
78            if csi_date == stock_date:
79                if stock_ret > 0:
80                    rising_x.append(csi_ret)
81                    rising_y.append(stock_ret)
82                else:
83                    declining_x.append(csi_ret)
84                    declining_y.append(stock_ret)
85
86        if len(rising_x) != 0 and len(declining_x) != 0:
87            regression_model = MultipleLinearRegression(rising_x, rising_y)
88            rising_market_beta:float = regression_model.params[0]
89
90            regression_model = MultipleLinearRegression(declining_x, declining_y)
91            declining_market_beta:float = regression_model.params[0]
92
93            ADB[stock] = abs(rising_market_beta - declining_market_beta)
94    
95    self.data[symbol].reset_data()
96
97if len(ADB) < self.quantile:
98    return Universe.Unchanged
99
100quantile:int = int(len(ADB) / self.quantile)
101sorted_by_ADB:list[Symbol] = [x[0] for x in sorted(ADB.items(), key=lambda item: item[1])]
102
103long_leg:list[Symbol] = sorted_by_ADB[:quantile]
104short_leg:list[Symbol] = sorted_by_ADB[-quantile:]
105
106total_long_cap:float = sum(list(map(lambda stock: stock.MarketCap, long_leg)))
107for stock in long_leg:
108    self.weights[stock.Symbol] = stock.MarketCap / total_long_cap
109
110total_short_cap:float = sum(list(map(lambda stock: stock.MarketCap, short_leg)))
111for stock in short_leg:
112    self.weights[stock.Symbol] = -stock.MarketCap / total_short_cap
113
114return list(self.weights.keys())
115
116def OnData(self, data: Slice):
117curr_date:datetime.date = self.Time.date()
118
119if self.benchmark in data and data[self.benchmark] and data[self.benchmark].Value != 0:
120    self.benchmark_data.update_price_with_date(data[self.benchmark].Value, curr_date)
121
122if not self.selection_flag:
123    return
124self.selection_flag = False
125
126self.benchmark_data.reset_data()
127
128# trade execution
129invested:list[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
130for symbol in invested:
131    if symbol not in self.weights:
132        self.Liquidate(symbol)
133
134for symbol, w in self.weights.items():
135    if symbol in data and data[symbol]:
136        self.SetHoldings(symbol, w)
137
138self.weights.clear()
139
140def Selection(self):
141self.selection_flag = True