Low-Risk Anomaly in India

1from AlgorithmImports import *
2from typing import List, Dict
3import data_tools
4import statsmodels.api as sm
5import numpy as np
6# endregion
7
8class LowRiskAnomalyinIndia(QCAlgorithm):
9
10def Initialize(self):
11self.SetStartDate(2010, 1, 1)
12self.SetCash(10000000) # INR
13
14self.price_period:int = 36 * 21
15
16self.data:Dict[Symbol, data_tools.SymbolData] = {}
17self.tickers_to_ignore:List[str] = ['TATAMTRDVR', 'LODHA']
18
19ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/india_stocks/nse_100_tickers.csv')
20ticker_lines:List[str] = ticker_file_str.split('\r\n')
21tickers = [ ticker_line.split(',')[0] for ticker_line in ticker_lines[1:] ]
22
23self.quantile:int = 3
24self.leverage:int = 5
25self.beta_p_target:float = 1.
26self.leverage_cap:float = 3.
27
28for t in tickers:
29    # price data subscription
30    if t in self.tickers_to_ignore:
31        continue
32    data:Security = self.AddData(data_tools.IndiaStocks, t, Resolution.Daily)
33    data.SetFeeModel(data_tools.CustomFeeModel())
34    data.SetLeverage(self.leverage)
35    stock_symbol:Symbol = data.Symbol
36
37    self.data[stock_symbol] = data_tools.SymbolData(stock_symbol, self.price_period)
38
39self.recent_month:int = -1
40
41def OnData(self, data: Slice) -> None:
42price_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaStocks.get_last_update_date()
43
44# custom data still comming in
45if all([self.Securities[x].GetLastData() for x in list(self.data.keys())]) and any([self.Time.date() >= price_last_update_date[x] for x in price_last_update_date]):
46    self.Liquidate()
47    return
48
49# store daily price data
50for price_symbol, symbol_data in self.data.items():
51    if price_symbol in data and data[price_symbol] and data[price_symbol].Value != 0:
52        price:float = data[price_symbol].Value
53        self.data[price_symbol].update_price(price)
54
55# montly rebalance
56if self.Time.month == self.recent_month:
57    return
58self.recent_month = self.Time.month
59
60volatility_by_symbol:Dict[Symbol, float] = {symbol: symbol_data.get_volatility() for symbol, symbol_data in self.data.items() if symbol_data.prices_ready()}
61
62asset_returns_dict:Dict[Symbol, np.ndarray] = {symbol : symbol_data.get_returns() for symbol, symbol_data in self.data.items() if symbol_data.prices_ready()}
63
64asset_returns:List[float] = list(zip(*[[i for i in x] for x in asset_returns_dict.values()]))
65market_returns:List[float] = [np.average(x) for x in asset_returns]
66
67if len(asset_returns) == 0:
68    return
69
70# run stock regression
71x:np.ndarray = np.array(market_returns)
72y:np.ndarray = np.array(asset_returns)
73model = self.multiple_linear_regression(x, y)
74beta_values:np.ndarray = model.params[1]
75
76# store betas
77beta_by_symbol:Dict[Symbol, float] = {sym : beta_values[n] for n, sym in enumerate(list(asset_returns_dict.keys()))}
78
79if len(volatility_by_symbol) < self.quantile:
80    return
81
82# sort by volatility and divide to quantiles
83sorted_volatility:List[Symbol] = sorted(volatility_by_symbol, key=volatility_by_symbol.get)
84quantile:int = len(volatility_by_symbol) // self.quantile
85long:List[Symbol] = sorted_volatility[:quantile]
86short:List[Symbol] = sorted_volatility[-quantile:]
87
88# beta neutral portfolio
89long_symbol_w:float = 1. / len(long)
90short_symbol_w:float = 1. / len(short)
91long_p_beta:float = sum([beta_by_symbol[x] * long_symbol_w for x in long if x in beta_by_symbol])
92short_p_beta:float = sum([beta_by_symbol[x] * short_symbol_w for x in short if x in beta_by_symbol])
93long_leverage:float = min(self.beta_p_target / long_p_beta, self.leverage_cap)
94short_leverage:float = min(self.beta_p_target / short_p_beta, self.leverage_cap)
95
96# trade execution
97invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
98for symbol in invested:
99    if symbol not in long + short:
100        self.Liquidate(symbol)
101
102for symbol in long:
103    if symbol in data and data[symbol]:
104        self.SetHoldings(symbol, long_symbol_w * long_leverage)
105
106for symbol in short:
107    if symbol in data and data[symbol]:
108        self.SetHoldings(symbol, -short_symbol_w * short_leverage)
109
110def multiple_linear_regression(self, x:np.ndarray, y:np.ndarray):
111x = sm.add_constant(x, has_constant='add')
112result = sm.OLS(endog=y, exog=x).fit()
113return result