Coreversal in Chinese Equities

1from AlgorithmImports import *
2from data_tools import SymbolData, CustomFeeModel, ChineseStocks
3import numpy as np
4from scipy import stats
5# endregion
6
7class CoreversalinChineseEquities(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2015, 1, 1)
11self.SetCash(100000)
12
13# chinese stock universe
14self.top_size_symbol_count:int = 500
15ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/chinese_stocks/large_cap_500.csv')
16self.tickers:List[str] = ticker_file_str.split('\r\n')[:self.top_size_symbol_count]
17
18# trenching
19self.managed_queue:List[RebalanceQueueItem] = []
20self.holding_period:int = 12            # months
21
22# CoREV
23self.CoREV_period:int = 12
24self.CoREV_values:List[float] = []
25self.CoREV_quantile:int = 5
26self.quantile:int = 10
27
28self.period = 52 * 5                    # daily period
29self.data:dict[str, SymbolData] = {}    # symbol data
30self.value_weighted:bool = False        # True - value weighted; False - equally weighted
31self.leverage:int = 5
32
33self.SetWarmUp(self.period, Resolution.Daily)
34
35for t in self.tickers:
36    data = self.AddData(ChineseStocks, t, Resolution.Daily)
37    data.SetFeeModel(CustomFeeModel())
38    data.SetLeverage(self.leverage)
39
40    self.data[data.Symbol] = SymbolData(self.period)
41
42self.recent_month:int = -1
43
44def OnData(self, data: Slice):
45performance:dict[Symbol, bool] = {}
46
47# store daily data
48for symbol, symbol_data in self.data.items():
49    if data.ContainsKey(symbol):
50        price_data:dict[str, str] = data[symbol].GetProperty('price_data')
51        # valid price data
52        if data[symbol].Value != 0. and price_data:
53            # update price and market cap
54            close:float = float(data[symbol].Value)
55            symbol_data.update_price(close)
56
57            mc:float = float(price_data['marketValue'])
58            symbol_data.update_market_cap(mc)
59
60            if symbol_data.is_ready():
61                if self.recent_month != self.Time.month and not self.IsWarmingUp:
62                    perf:float = symbol_data.performance()
63                    performance[symbol] = perf
64
65# rebalance monthly
66if self.recent_month == self.Time.month:
67    return
68self.recent_month = self.Time.month
69
70if self.IsWarmingUp:
71    return
72
73long:List[Symbol] = []
74short:List[Symbol] = []
75
76if len(performance) >= self.quantile:
77    # sort by performance
78    sorted_by_performance:List = sorted(performance.items(), key=lambda x: x[1], reverse=True)
79    quantile:int = int(len(sorted_by_performance) / self.quantile)
80    top_quantile:List[Symbol] = [x[0] for x in sorted_by_performance[:quantile]]
81    bottom_quantile:List[Symbol] = [x[0] for x in sorted_by_performance[-quantile:]]
82
83    total_pairwise_abnormal_correlations:float = 0
84    for quantile in [top_quantile, bottom_quantile]:
85        total_quantile_correlation:float = 0
86
87        for symbol in quantile:
88            # excess return of each stock
89            stock_weekly_returns:np.ndarray = self.data[symbol].returns_for_period_n()
90
91            # equal-weighted average excess return of the bottom (top) decile excluding stock i
92            relevant_symbols:List[Symbol] = [x for x in quantile if x != symbol]
93            if len(relevant_symbols) != 0:
94                quantile_ew_perf:float = np.mean(np.array([self.data[x].returns_for_period_n() for x in relevant_symbols]), axis=0)
95
96            symbol_corr:float = np.corrcoef(stock_weekly_returns, quantile_ew_perf)[0, 1]
97
98            total_quantile_correlation += symbol_corr
99        
100        # take the average of this sum
101        total_quantile_correlation /= len(quantile)
102        total_pairwise_abnormal_correlations += total_quantile_correlation
103    
104    # simple average of the average pairwise abnormal correlations in the loser decile and winner decile
105    CoREV:float = total_pairwise_abnormal_correlations / 2
106    if len(self.CoREV_values) >= self.CoREV_period:
107        # if CoREV is at the month-end within its top quintile, buy the standard long-short reversal portfolio over the next year
108        percentile:float = stats.percentileofscore(self.CoREV_values, CoREV) / 100
109        top_percentile:float = 1 - (1 / self.CoREV_quantile)
110        if percentile >= top_percentile:
111            long = bottom_quantile
112            short = top_quantile
113
114    # append this month's corev value
115    if CoREV != 0:
116        # append this month's corev value
117        self.CoREV_values.append(CoREV)
118
119if long and short:
120    # calculate quantities for long and short trenche
121    if self.value_weighted:
122        total_market_cap_long:float = sum([self.data[x].recent_market_cap() for x in long])
123        total_market_cap_short:float = sum([self.data[x].recent_market_cap() for x in short])
124        
125        long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period
126        short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period
127        
128        long_symbol_q:List[tuple[Symbol, float]] = [(x, np.floor(long_w * (self.data[x].recent_market_cap() / total_market_cap_long) / data[symbol].Value)) for x in long]
129        short_symbol_q:List[tuple[Symbol, float]] = [(x, -np.floor(short_w * (self.data[x].recent_market_cap() / total_market_cap_short) / data[symbol].Value)) for x in short]
130        
131        self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
132    else:
133        long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
134        short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
135        
136        long_symbol_q:List[tuple[Symbol, float]] = [(x, np.floor(long_w / data[x].Value)) for x in long]
137        short_symbol_q:List[tuple[Symbol, float]] = [(x, -np.floor(short_w / data[x].Value)) for x in short]
138        
139        self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
140    
141# trade execution - rebalance portfolio
142remove_item:RebalanceQueueItem|None = None
143
144for item in self.managed_queue:
145    # liquidate
146    if item.holding_period == self.holding_period: # all portfolio parts are held for n months
147        for symbol, quantity in item.opened_symbol_q:
148            self.MarketOrder(symbol, -quantity)
149                    
150        remove_item = item
151    
152    # trade execution    
153    if item.holding_period == 0: # all portfolio parts are held for n months
154        opened_symbol_q:List[tuple[Symbol, float]] = []
155        
156        for symbol, quantity in item.opened_symbol_q:
157            self.MarketOrder(symbol, quantity)
158            opened_symbol_q.append((symbol, quantity))
159                    
160        # only opened orders will be closed        
161        item.opened_symbol_q = opened_symbol_q
162        
163    item.holding_period += 1
164    
165# need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue
166if remove_item:
167    self.managed_queue.remove(remove_item)
168
169class RebalanceQueueItem():
170def __init__(self, symbol_q:List):
171# symbol/quantity collections
172self.opened_symbol_q:List[tuple[Symbol, float]] = symbol_q  
173self.holding_period:int = 0