Monthly Reversal in Chinese Equities

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