Daily Momentum in Chinese Equities

1from AlgorithmImports import *
2from typing import Dict, List
3import data_tools
4# endregion
5
6class DailyMomentuminChineseEquities(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2015, 1, 1)   # Chinese data starts in 2015
10self.SetCash(100000)
11
12self.leverage:int = 5
13self.quantile:int = 10
14self.period:int = 2 + 1
15
16self.data:Dict[Symbol, SymbolData] = {}
17
18self.top_size_symbol_count:int = 300
19ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/chinese_stocks/large_cap_500.csv')
20tickers:List[str] = ticker_file_str.split('\r\n')[:self.top_size_symbol_count]
21
22for t in tickers:
23    data = self.AddData(data_tools.ChineseStocks, t, Resolution.Daily)
24    data.SetFeeModel(data_tools.CustomFeeModel())
25    data.SetLeverage(self.leverage)
26
27    stock_symbol:Symbol = data.Symbol
28
29    self.data[stock_symbol] = data_tools.SymbolData(data.Symbol, self.period)
30
31self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
32
33def OnData(self, data: Slice) -> None:
34price_last_update_date:Dict[Symbol, datetime.date] = data_tools.ChineseStocks.get_last_update_date()
35asset_returns:Dict[Symbol, float] = {}
36
37# store daily data
38for symbol, symbol_data in self.data.items():
39    if symbol in data and data[symbol]:
40        price_data:dict[str, str] = data[symbol].GetProperty('price_data')
41        # valid price data
42        if data[symbol].Value != 0 and price_data:
43            close_price:float = data[symbol].Value
44            symbol_data.update_price(close_price)
45
46            mc:float = float(price_data['marketValue'])
47            symbol_data.update_market_cap(mc)
48    
49    # calculate momentum
50    if symbol_data.is_ready() and self.Time.date() < price_last_update_date[symbol]:
51        asset_returns[symbol] = symbol_data.get_momentum()
52
53weight:Dict[Symbol, float] = {}
54
55# sort and divide to quantiles
56if len(asset_returns) >= self.quantile:
57    sorted_stocks:List[Symbol] = sorted(asset_returns, key=asset_returns.get , reverse=True)
58    quantile:int = int(len(sorted_stocks) / self.quantile)
59    long:List[Symbol] = sorted_stocks[:quantile]
60    short:List[Symbol] = sorted_stocks[-quantile:]
61
62    # calculate weights based on market cap
63    for i, portfolio in enumerate([long, short]):
64        mc_sum:float = sum(list(map(lambda symbol: self.data[symbol].get_market_cap(), portfolio)))
65        for symbol in portfolio:
66            weight[symbol] = ((-1)**i) * self.data[symbol].get_market_cap() / mc_sum
67
68# trade execution
69invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
70for symbol in invested:
71    if symbol not in weight:
72        self.Liquidate(symbol)
73
74for price_symbol, weight in weight.items():
75    if price_symbol in data and data[price_symbol]:
76        self.SetHoldings(price_symbol, weight)