Institutional Equity Momentum in China

1from AlgorithmImports import *
2from data_tools import CustomFeeModel, SymbolData, ChineseStocks
3# endregion
4
5class InstitutionalEquityMomentumInChina(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2005, 1, 1)
9self.SetCash(100000)
10
11self.perf_quantile:int = 5
12self.price_quantile:int = 10
13
14self.leverage:int = 5
15
16self.period:int = 21 * 12
17self.max_missing_days:int = 5
18
19self.data:dict[Symbol, SymbolData] = {}
20
21top_size_symbol_count:int = 400
22ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/chinese_stocks/large_cap_500.csv')
23tickers:List[str] = ticker_file_str.split('\r\n')[:top_size_symbol_count]
24
25for t in tickers:
26    # price data
27    data = self.AddData(ChineseStocks, t, Resolution.Daily)
28    data.SetFeeModel(CustomFeeModel())
29    data.SetLeverage(self.leverage)
30
31    self.data[data.Symbol] = SymbolData(self.period)
32
33self.recent_month:int = -1
34self.exclusion_flag:bool = False
35
36def OnData(self, data):
37curr_date:datetime.date = self.Time.date()
38
39for symbol, symbol_data in self.data.items():
40    if symbol in data and data[symbol] and data[symbol].Value and data[symbol].GetProperty('price_data'):
41        price:float = data[symbol].Value 
42        symbol_data.update_prices(price)
43        symbol_data.set_last_update(curr_date)
44
45        price_data:dict = data[symbol].GetProperty('price_data')
46        if self.exclusion_flag and 'marketValue' in price_data and price_data['marketValue'] != 0:
47            market_cap:float = float(price_data['marketValue'])
48            symbol_data.set_market_cap(market_cap)
49
50if self.recent_month == self.Time.month:
51    return
52self.recent_month = self.Time.month
53
54if self.exclusion_flag:
55    market_caps:dict[Symbol, float] = { sym: sym_data.get_market_cap() for sym, sym_data in self.data.items() if sym_data.market_cap_ready() }
56    sorted_by_cap:list[Symbol] = [x[0] for x in sorted(market_caps.items(), key=lambda item: item[1])]
57    # exclude lowest 30%
58    active_universe:list[Symbol] = sorted_by_cap[int(len(sorted_by_cap) * 0.3):]
59else:
60    active_universe:list[Symbol] = list(self.data.keys())
61
62performances:dict[Symbol, float] = {}
63prices:dict[Symbol, float] = {}
64
65for symbol in active_universe:
66    symbol_data:SymbolData = self.data[symbol]
67
68    if not symbol_data.data_still_coming(curr_date, self.max_missing_days):
69        symbol_data.reset_data()
70
71    if symbol_data.prices_ready():
72        performances[symbol] = symbol_data.get_performance(self.period)
73        prices[symbol] = symbol_data.get_last_price()
74
75if len(performances) < self.perf_quantile or len(prices) < self.price_quantile:
76    self.Liquidate()
77    return
78
79quantile:int = int(len(performances) / self.perf_quantile)
80sorted_by_perf:list[Symbol] = [x[0] for x in sorted(performances.items(), key=lambda item: item[1])]
81winners:list[Symbol] = sorted_by_perf[-quantile:]
82losers:list[Symbol] = sorted_by_perf[:quantile]
83
84quantile:int = int(len(prices) / self.price_quantile)
85sorted_by_price:list[Symbol] = [x[0] for x in sorted(prices.items(), key=lambda item: item[1])]
86top_price:list[Symbol] = sorted_by_price[-quantile:]
87
88long_leg:list[Symbol] = [symbol for symbol in winners if symbol in top_price]
89short_leg:list[Symbol] = [symbol for symbol in losers if symbol in top_price]
90
91long_len:int = len(long_leg)
92short_len:int = len(short_leg)
93
94# Trade Execution
95stocks_invested:list[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
96for symbol in stocks_invested:
97    if symbol not in long_leg + short_leg:
98        self.Liquidate(symbol)
99
100for symbol in long_leg:
101    self.SetHoldings(symbol, 1 / long_len)
102
103for symbol in short_leg:
104    self.SetHoldings(symbol, -1 / short_len)