Reversal Effect in India

1from AlgorithmImports import *
2from data_tools import QuantpediaIndiaStocks, SymbolData, CustomFeeModel
3#endregion
4
5class ReversalEffectInIndia(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2000, 1, 1)
9self.SetCash(100000)
10
11self.data:dict = {}
12
13self.portfolio_percentage:float = 0.5
14self.days_in_month:int = 21
15self.period:int = 12 * self.days_in_month # 12 months of daily closes
16self.SetWarmUp(self.period, Resolution.Daily)
17
18self.quantile:int = 5
19self.leverage:int = 5
20self.perf_period:int = 12
21self.max_missing_days:int = 5
22self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
23
24csv_string_file = self.Download('data.quantpedia.com/backtesting_data/equity/india_stocks/india_nifty_500_tickers.csv')
25line_split = csv_string_file.split(';')
26
27# NOTE: Download method is rate-limited to 100 calls (https://github.com/QuantConnect/Documentation/issues/345)
28for ticker in line_split[:99]:
29    security = self.AddData(QuantpediaIndiaStocks, ticker, Resolution.Daily)
30    security.SetFeeModel(CustomFeeModel())
31    security.SetLeverage(self.leverage)
32
33    symbol:Symbol = security.Symbol
34    self.data[symbol] = SymbolData(self.period)
35
36self.recent_month:int = -1
37
38def OnData(self, data):
39# store daily prices
40for symbol in self.data:
41    if symbol in data and data[symbol]:
42        price:float = data[symbol].Value
43        if price != 0 and not np.isnan(price):
44            self.data[symbol].update(price)
45
46if self.IsWarmingUp: return
47
48# rebalance monthly
49if self.Time.month == self.recent_month:
50    return
51self.recent_month = self.Time.month
52
53performance:[Symbol, float] = {}
54
55for symbol, symbol_obj in self.data.items():
56    # prices are ready and data is still comming in
57    if symbol_obj.is_ready() and self.Securities[symbol].GetLastData() and \
58        (self.Time.date() - self.Securities[symbol].GetLastData().Time.date()).days <= self.max_missing_days:
59        perf:float = self.data[symbol].performance(self.days_in_month)
60        if perf != 0 and not np.isnan(perf):
61            performance[symbol] = perf
62
63long_part:list[Symbol] = []
64short_part:list[Symbol] = []
65
66if len(performance) >= self.quantile:
67    quantile:int = int(len(performance) / self.quantile)
68    sorted_by_performance = [x[0] for x in sorted(performance.items(), key=lambda item: item[1], reverse=True)]
69
70    # Long the loser portfolio and short the winner portfolio
71    # The winner portfolio consists of stocks with the prior 12-month CARs in the top 20%.
72    short_part = sorted_by_performance[-quantile:]
73    # The loser portfolio consists of stocks with the prior 12-month CARs in the bottom 20%.
74    long_part = sorted_by_performance[:quantile]
75        
76# trade execution
77long_count:int = len(long_part)
78short_count:int = len(short_part)
79
80stocks_invested = [x.Key for x in self.Portfolio if x.Value.Invested]
81for symbol in stocks_invested:
82    if symbol not in long_part + short_part:
83        self.Liquidate(symbol)
84
85for symbol in long_part:
86    self.SetHoldings(symbol, (1 / long_count) * self.portfolio_percentage)
87for symbol in short_part:
88    self.SetHoldings(symbol, (-1 / short_count) * self.portfolio_percentage)