Disposition Effect in China

1from AlgorithmImports import *
2from data_tools import SymbolData, CustomFeeModel, ChineseStocks, MultipleLinearRegression
3import numpy as np
4# endregion
5
6class DispositionEffectinChina(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2015, 1, 1)
10self.SetCash(100000)
11
12# chinese stock universe
13self.top_size_symbol_count:int = 300
14ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/chinese_stocks/large_cap_500.csv')
15self.tickers:List[str] = ticker_file_str.split('\r\n')[:self.top_size_symbol_count]
16
17self.period = 365                       # daily period
18self.data:dict[str, SymbolData] = {}    # symbol data
19self.max_missing_days:int = 5           # max missing n of price data entries in a row for custom data
20self.value_weighted:bool = True         # True - value weighted; False - equally weighted
21self.min_symbols:int = 2
22self.leverage:int = 5
23self.rebalance_month:int = 5            # May rebalance
24
25self.SetWarmUp(self.period, Resolution.Daily)
26
27for t in self.tickers:
28    data = self.AddData(ChineseStocks, t, Resolution.Daily)
29    data.SetFeeModel(CustomFeeModel())
30    data.SetLeverage(self.leverage)
31
32    self.data[data.Symbol] = SymbolData(self.period)
33
34self.recent_month:int = -1
35
36def OnData(self, data: Slice):
37total_trading_volume:float = 0
38included_symbols:List[Symbol] = []
39disposition:dict[Symbol, bool] = {}
40
41# store daily data
42for symbol, symbol_data in self.data.items():
43    if data.ContainsKey(symbol):
44        price_data:dict[str, str] = data[symbol].GetProperty('price_data')
45        # valid price data
46        if data[symbol].Value != 0. and price_data:
47            # update price and market cap
48            close:float = float(data[symbol].Value)
49            volume:float = float(price_data['turnoverVol'])
50            
51            included_symbols.append(symbol)
52            total_trading_volume += volume
53            
54            symbol_data.update_price(close)
55            symbol_data.update_volume(volume)
56
57            mc:float = float(price_data['marketValue'])
58            symbol_data.update_market_cap(mc)
59
60            # update second regression variable
61            if symbol_data.is_ready():
62                symbol_data.update_disposition_regression_x()
63
64                if self.recent_month != self.Time.month and self.Time.month == self.rebalance_month and not self.IsWarmingUp:
65                    if symbol_data.cumulative_raw_volume_is_ready():
66                        # first regression - abnormal trading volume
67                        regr_data:np.ndarray = symbol_data.get_abnormal_trading_volume_regression_data()
68                        ATV_regression_model = MultipleLinearRegression(regr_data[0], regr_data[1])
69                        abnormal_trading_volume:np.ndarray = ATV_regression_model.resid
70
71                        # second regression - disposition
72                        regr_data:np.ndarray = symbol_data.get_disposition_regression_data()
73                        disposition_regression_model = MultipleLinearRegression(regr_data, abnormal_trading_volume.T)
74
75                        # negative beta_1 or positive beta_2 indicates a disposition effect
76                        if disposition_regression_model.params[1] < 0 or disposition_regression_model.params[2] > 0:
77                            disposition[symbol] = True
78                        else:
79                            disposition[symbol] = False
80
81# update total market volume for included stocks in todays data structure
82for symbol in included_symbols:
83    self.data[symbol].update_total_market_volume(total_trading_volume)
84
85# rebalance yearly
86if self.recent_month == self.Time.month:
87    return
88self.recent_month = self.Time.month
89if self.Time.month != self.rebalance_month:
90    return
91
92if self.IsWarmingUp:
93    return
94
95long:List[str] = []
96short:List[str] = []
97
98if len(disposition) >= self.min_symbols:
99    # according to the paper, long the no disposition small and big size portfolios and short the disposition small and big size portfolios
100    long = [symbol for symbol, disp in disposition.items() if disp == False]
101    short = [symbol for symbol, disp in disposition.items() if disp == True]
102
103# weight calculation
104weights_to_trade = {}
105
106if self.value_weighted:
107    total_market_cap_long:float = sum([self.data[x].recent_market_cap() for x in long])
108    total_market_cap_short:float = sum([self.data[x].recent_market_cap() for x in short])
109
110    for symbol in long:
111        if symbol in data and data[symbol]:
112            weights_to_trade[symbol] = self.data[symbol].recent_market_cap() / total_market_cap_long
113    for symbol in short:
114        if symbol in data and data[symbol]:
115            weights_to_trade[symbol] = -self.data[symbol].recent_market_cap() / total_market_cap_short
116else:
117    long_c:int = len(long)
118    short_c:int = len(short)
119
120    for symbol in long:
121        if symbol in data and data[symbol]:
122            weights_to_trade[symbol] = 1 / long_c
123    for symbol in short:
124        if symbol in data and data[symbol]:
125            weights_to_trade[symbol] = -1 / short_c
126
127# trade execution
128invested = [x.Key for x in self.Portfolio if x.Value.Invested]
129for symbol in invested:
130    if symbol not in weights_to_trade:
131        self.Liquidate(symbol)
132
133for symbol, w in weights_to_trade.items():
134    self.SetHoldings(symbol, w)