Betting Against Beta in India

1from AlgorithmImports import *
2import numpy as np
3from scipy import stats
4import data_tools
5#endregion
6
7class BettingAgainstBetainIndia(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2005, 1, 1)
11self.SetCash(100000)
12
13self.data:dict = {}
14self.long_term_period:int = 5 * 12 * 21
15self.short_term_period:int = 1 * 12 * 21
16self.SetWarmUp(self.short_term_period, Resolution.Daily)
17
18self.quantile:int = 5
19self.max_missing_days:int = 5
20self.market = self.AddData(data_tools.BSE_200, 'BSE_200', Resolution.Daily).Symbol
21self.data[self.market] = data_tools.SymbolData(self.short_term_period)
22
23csv_string_file = self.Download('data.quantpedia.com/backtesting_data/equity/india_stocks/india_nifty_100_tickers.csv')
24line_split = csv_string_file.split(';')
25
26# NOTE: Download method is rate-limited to 100 calls (https://github.com/QuantConnect/Documentation/issues/345)
27self.ticker:list[str] = line_split[:99]
28
29for ticker in self.ticker:
30    security = self.AddData(data_tools.QuantpediaIndiaStocks, ticker, Resolution.Daily)
31    security.SetFeeModel(data_tools.CustomFeeModel())
32    security.SetLeverage(5)
33
34    self.data[ticker] = data_tools.SymbolData(self.short_term_period)
35
36self.recent_month:int = -1
37
38def OnData(self, data):
39rebalance_flag:bool = False
40
41# rebalance once a month
42if not self.IsWarmingUp and self.Time.month != self.recent_month:
43    rebalance_flag = True
44    self.recent_month = self.Time.month
45
46beta:dict[str, float] = {}
47
48# store daily price data
49if self.market in data and data[self.market]:
50    # market price data
51    self.data[self.market].update(data[self.market].Value)
52    
53    # stock price data
54    for ticker in self.ticker:
55        if ticker in data and data[ticker]:
56            self.data[ticker].update(data[ticker].Value)
57        else:
58            if self.data[ticker].closes.Count != 0:
59                self.data[ticker].update(self.data[ticker].closes[0])
60
61        if self.IsWarmingUp: continue
62        if rebalance_flag:
63            if self.data[ticker].is_ready() and self.data[self.market].is_ready():
64                # stock price data is still comming in
65                if self.Securities[ticker].GetLastData() and (self.Time.date() - self.Securities[ticker].GetLastData().Time.date()).days < self.max_missing_days:
66                    # market and stock returns
67                    market_prices:np.ndarray = np.array([x for x in self.data[self.market].closes])
68                    market_returns:np.ndarray = market_prices[:-1] / market_prices[1:] - 1
69                    market_returns = np.nan_to_num(market_returns, nan=0.)
70
71                    stock_prices:np.ndarray = np.array([x for x in self.data[ticker].closes])
72                    stock_returns:np.ndarray = stock_prices[:-1] / stock_prices[1:] - 1
73                    stock_returns = np.nan_to_num(stock_returns, nan=0.)
74
75                    # NOTE source paper version of beta estimate
76                    slope, intercept, r_value, p_value, std_err = stats.linregress(market_returns, stock_returns)
77                    beta[ticker] = slope
78
79if rebalance_flag:
80    # z-score normalization
81    beta_values:list[float] = [beta for ticker, beta in beta.items()]
82    beta_std:float = np.std(beta_values)
83    beta_mean:float = np.mean(beta_values)
84    z_score:dict = { ticker: (beta - beta_mean) / beta_std for ticker, beta in beta.items()}
85    z_transform:dict = { ticker : (1+z) if z>= 0 else (1/(1-z)) for ticker, z in z_score.items()}
86
87    long:list[Symbol] = []
88    short:list[Symbol] = []
89
90    if len(z_transform) >= self.quantile:
91        quantile:int = int(len(z_transform) / self.quantile)
92        sorted_by_zscore = sorted(z_transform.items(), key=lambda item: item[1], reverse=True)
93
94        long = [x[0] for x in sorted_by_zscore[:quantile]]
95        short = [x[0] for x in sorted_by_zscore[-quantile:]]
96        
97    # trade execution
98    long_count:int = len(long)
99    short_count:int = len(short)
100
101    stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
102    for symbol in stocks_invested:
103        if symbol not in long + short:
104            self.Liquidate(symbol)
105
106    for symbol in long:
107        self.SetHoldings(symbol, 1 / long_count)
108    for symbol in short:
109        self.SetHoldings(symbol, -1 / short_count)