Skewness and 52-Week Highs in China

1from AlgorithmImports import *
2from data_tools import CustomFeeModel, SymbolData, ChineseStocks
3# endregion
4
5class SkewnessAnd52WeekHighsInChina(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2015, 1, 1)
9self.SetCash(100000)
10
11# chinese stock universe
12self.top_size_symbol_count:int = 300
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 = 10
17self.leverage:int = 5
18self.max_missing_days:int = 5
19self.period:int = 52 * 5                  # daily period
20self.SetWarmUp(self.period, Resolution.Daily)
21
22self.data:dict[str, SymbolData] = {}
23
24for t in self.tickers:
25    data = self.AddData(ChineseStocks, t, Resolution.Daily)
26    data.SetFeeModel(CustomFeeModel())
27    data.SetLeverage(self.leverage)
28
29    self.data[data.Symbol] = SymbolData(self.period)
30
31self.recent_month:int = -1
32
33def OnData(self, data: Slice):
34curr_date:datetime.date = self.Time.date()
35
36# store daily data
37for symbol, symbol_data in self.data.items():
38    if data.ContainsKey(symbol):
39        price_data:dict[str, str] = data[symbol].GetProperty('price_data')
40        # valid price data
41        if data[symbol].Value != 0. and price_data:
42            # update price and market cap
43            close:float = float(data[symbol].Value)
44            symbol_data.update(curr_date, close)
45
46            mc:float = float(price_data['marketValue'])
47            symbol_data.update_market_cap(mc)
48
49if self.IsWarmingUp: return
50
51if self.recent_month == self.Time.month:
52    return
53self.recent_month = self.Time.month
54
55W52:dict[Symbol, float] = {}
56SKEWNESS:dict[Symbol, float] = {}
57
58for symbol, symbol_data in self.data.items():
59    if not symbol_data.data_still_coming(curr_date, self.max_missing_days):
60        symbol_data.reset_data()
61        continue
62
63    if symbol_data.is_ready():
64        W52[symbol] = symbol_data.get_W52_value()
65        SKEWNESS[symbol] = symbol_data.get_SKEWNESS_value()
66
67    symbol_data.reset_monthly_closes()
68
69if len(W52) < self.quantile:
70    self.Liquidate()
71    return
72
73# sort by skew and nearness
74quantile:int = int(len(W52) / self.quantile)
75sorted_W52:list[Symbol] = [x[0] for x in sorted(W52.items(), key=lambda item: item[1])]
76sorted_SKEWNESS:list[Symbol] = [x[0] for x in sorted(SKEWNESS.items(), key=lambda item: item[1])]
77
78farthest_W52:list[Symbol] = sorted_W52[:quantile]
79lowest_SKEWNESS:list[Symbol] = sorted_SKEWNESS[:quantile]
80highest_SKEWNESS:list[Symbol] = sorted_SKEWNESS[-quantile:]
81
82long_leg:list[Symbol] = [symbol for symbol in farthest_W52 if symbol in lowest_SKEWNESS]
83short_leg:list[Symbol] = [symbol for symbol in farthest_W52 if symbol in highest_SKEWNESS]
84
85# trade execution
86invested:list[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
87for symbol in invested:
88    if symbol not in long_leg + short_leg:
89        self.Liquidate(symbol)
90
91total_long_cap:float = sum(list(map(lambda symbol: self.data[symbol].get_market_cap(), long_leg)))
92for symbol in long_leg:
93    self.SetHoldings(symbol, self.data[symbol].get_market_cap() / total_long_cap)
94
95total_short_cap:float = sum(list(map(lambda symbol: self.data[symbol].get_market_cap(), short_leg)))
96for symbol in short_leg:
97    self.SetHoldings(symbol, -self.data[symbol].get_market_cap() / total_short_cap)