Factor Momentum in the Chinese Stock Market

1from AlgorithmImports import *
2from data_tools import ChineseStocks, CustomFeeModel, ChineseBalanceSheet, \
3ChineseIncomeStatement, ChineseCashflowStatement, SymbolData, FactorData
4# endregion
5
6class FactorMomentumInTheChineseStockMarket(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2010, 1, 1)
10self.SetCash(100000)
11
12self.top_size_symbol_count:int = 200
13ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/chinese_stocks/large_cap_500.csv')
14self.tickers:List[str] = ticker_file_str.split('\r\n')[:self.top_size_symbol_count]
15
16self.quantile:int = 3
17self.leverage:int = 5
18
19self.period:int = 3 * 21
20self.monthly_returns_period:int = 1
21
22self.min_prices:int = 15
23self.min_volumes:int = 15
24self.max_missing_statement_days:int = 3 * 30
25
26self.data:dict[Symbol, SymbolData] = {}
27
28self.factors_identificators:list[str] = ['SIZE', 'BM', 'ILLIQUIDITY','EARNINGS-TO-PRICE','CF-TO-PRICE','TURNOVER']
29self.factors_data:dict[str, FactorData] = { identificator: FactorData(self.monthly_returns_period) \
30    for identificator in self.factors_identificators }
31
32for t in self.tickers:
33    # price data
34    data = self.AddData(ChineseStocks, t, Resolution.Daily)
35    data.SetFeeModel(CustomFeeModel())
36    data.SetLeverage(self.leverage)
37
38    china_stock_symbol:Symbol = data.Symbol
39    income_symbol:Symbol = self.AddData(ChineseIncomeStatement, t, Resolution.Daily).Symbol
40    balance_symbol:Symbol = self.AddData(ChineseBalanceSheet, t, Resolution.Daily).Symbol
41    cashflow_symbol:Symbol = self.AddData(ChineseCashflowStatement, t, Resolution.Daily).Symbol
42
43    self.data[china_stock_symbol] = SymbolData(self.period, income_symbol, balance_symbol, cashflow_symbol)
44
45self.recent_month:int = -1
46self.spy:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
47self.Schedule.On(self.DateRules.MonthEnd(self.spy), self.TimeRules.BeforeMarketClose(self.spy, 0), self.UpdateFactorsPerformances)
48
49def OnData(self, data: Slice):
50factors:dict[str, dict] = { identificator: {} for identificator in self.factors_identificators }
51curr_date:datetime.date = self.Time.date()
52
53for symbol, symbol_data in self.data.items():
54    income_symbol:Symbol = symbol_data.income_symbol
55    if income_symbol in data and data[income_symbol] and data[income_symbol].GetProperty('statement'):
56        symbol_data.update_income_data(curr_date, data[income_symbol].GetProperty('statement'))
57
58    balance_symbol:Symbol = symbol_data.balance_symbol
59    if balance_symbol in data and data[balance_symbol] and data[balance_symbol].GetProperty('statement'):
60        symbol_data.update_balance_data(curr_date, data[balance_symbol].GetProperty('statement'))
61
62    cashflow_symbol:Symbol = symbol_data.cashflow_symbol
63    if cashflow_symbol in data and data[cashflow_symbol] and data[cashflow_symbol].GetProperty('statement'):
64        symbol_data.update_cashflow_data(curr_date, data[cashflow_symbol].GetProperty('statement'))
65
66    if symbol in data and data[symbol] and data[symbol].Value and data[symbol].GetProperty('price_data'):
67        price_data:dict = data[symbol].GetProperty('price_data')
68    
69        price:float = data[symbol].Value 
70        volume:float = price_data['turnoverVol']
71        symbol_data.update_prices_volumes_last_update(price, volume, curr_date)
72
73        if self.recent_month != self.Time.month \
74            and symbol_data.statements_still_coming(curr_date, self.max_missing_statement_days):
75
76            balance_data:dict[str, str] = symbol_data.balance_data
77            income_data:dict[str, str] = symbol_data.income_data
78            cashflow_data:dict[str, str] = symbol_data.cashflow_data
79
80            market_cap:float = float(price_data['marketValue'])
81            total_assets:float = float(balance_data['TAssets'])
82
83            if market_cap != 0 and total_assets != 0 and symbol_data.turnover_volumes_ready() \
84                and symbol_data.prices_ready(self.min_prices) and symbol_data.volumes_ready(self.min_volumes):
85                
86                factors['SIZE'][symbol] = market_cap
87
88                book_value:float = total_assets - float(balance_data['TLiab'])
89                book_to_market:float = book_value / market_cap
90                factors['BM'][symbol] = book_to_market
91
92                factors['ILLIQUIDITY'][symbol] = symbol_data.get_illiquidity()
93
94                net_income:float = float(income_data['NIncome'])
95                earnings_to_price:float = net_income / market_cap
96                factors['EARNINGS-TO-PRICE'][symbol] = earnings_to_price
97
98                shares_oustanding:float = market_cap / price
99                avg_vol_for_turnover:float = symbol_data.get_avg_vol_for_turnover()
100                factors['TURNOVER'][symbol] = avg_vol_for_turnover / shares_oustanding
101
102                operating_cashflow:float = float(cashflow_data['NCFOperateA'])
103                factors['CF-TO-PRICE'][symbol] = operating_cashflow / market_cap
104
105            symbol_data.reset_prices_and_volumes()
106
107if self.recent_month != self.Time.month:
108    self.Liquidate()
109    self.recent_month = self.Time.month
110
111    if len(factors[self.factors_identificators[0]]) >= self.quantile:
112        for factor_identificator, factor_value_by_symbol in factors.items():
113            quantile:int = int(len(factor_value_by_symbol) / self.quantile)
114            sorted_by_factor_value:list[Symbol] = [x[0] for x in sorted(factor_value_by_symbol.items(), key=lambda item: item[1])]
115
116            highest_quantile:list[Symbol] = sorted_by_factor_value[-quantile:]
117            lowest_quantile:list[Symbol] = sorted_by_factor_value[:quantile]
118
119            self.factors_data[factor_identificator].set_factor_quantiles(highest_quantile, lowest_quantile)
120
121        factor_cumulative_perf:dict[str, float] = { identificator: factor_data.get_cumulative_perf() \
122            for identificator, factor_data in self.factors_data.items() if factor_data.factor_monthly_perfs_ready() }
123        
124        factor_perf_median:float = np.median(list(factor_cumulative_perf.values()))
125
126        long_leg:list[Symbol] = []
127        short_leg:list[Symbol] = []
128
129        for identificator, cumulative_perf in factor_cumulative_perf.items():
130            if cumulative_perf > factor_perf_median:
131                long_leg += list(factors[identificator].keys())
132            else:
133                short_leg += list(factors[identificator].keys())
134
135        long_len:int = len(long_leg)
136        short_len:int = len(short_leg)
137
138        for symbol in long_leg:
139            self.SetHoldings(symbol, 1 / long_len)
140
141        for symbol in short_leg:
142            self.SetHoldings(symbol, -1 / short_len)
143
144def UpdateFactorsPerformances(self) -> None:
145for identificator, factor_data in self.factors_data.items():
146    if factor_data.factor_quantiles_ready():
147        highest_quantile_perf:float = sum([self.data[symbol].get_performance() for symbol in factor_data.highest_quantile])
148        lowest_quantile_perf:float = -sum([self.data[symbol].get_performance() for symbol in factor_data.lowest_quantile])
149
150        factor_perf:float = highest_quantile_perf + lowest_quantile_perf
151
152        factor_data.update_factor_monthly_perfs(factor_perf)
153
154    else:
155        factor_data.reset_monthly_perfs()
156
157    factor_data.reset_factor_quantiles()
158
159def AreDataStillComing(self, symbol:Symbol, max_missing_days:int) -> bool:
160return (self.Time.date() - self.Securities[symbol].GetLastData().Time.date()).days <= max_missing_days