Low Value Factor in India

1from AlgorithmImports import *
2from datetime import datetime
3import data_tools
4# endregion
5
6class LowValueFactorInIndia(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2006, 1, 1)
10self.SetCash(100000)
11
12self.leverage:int = 5
13self.quantile:int = 10
14
15self.max_missing_days:int = 5
16
17self.PB_data:dict = {}
18self.data:dict[str, data_tools.SymbolData] = {}
19
20self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
21
22# load price to book values
23csv_string_file:str = self.Download('data.quantpedia.com/backtesting_data/equity/india_stocks/price_to_book.csv')
24lines:list = csv_string_file.split('\r\n')
25tickers_with_PB_values:list = lines[0].split(';')[1:]
26
27for line in lines[1:]: # skip header
28    if line == '':
29        continue
30
31    line_split:list = line.split(';')
32
33    date:datetime.date = datetime.strptime(line_split[0], '%Y-%m-%d').date()
34    # init dictionary for this date
35    self.PB_data[date] = {}
36
37    length = len(line_split[1:])
38    for i in range(length):
39        PB_value:float = float(line_split[i + 1])
40
41        if PB_value != 0:
42            ticker:str = tickers_with_PB_values[i]
43
44            self.PB_data[date][ticker] = PB_value
45
46# subscribe india stocks
47csv_string_file:str = self.Download('data.quantpedia.com/backtesting_data/equity/india_stocks/india_nifty_100_tickers.csv')
48lines:list = csv_string_file.split('\r\n')
49
50for line in lines[:99]:
51    line_split:list = line.split(';')
52
53    for ticker in line_split:
54        # subscribe india stock
55        security:Security = self.AddData(data_tools.QuantpediaIndiaStocks, ticker, Resolution.Daily)
56        security.SetFeeModel(data_tools.CustomFeeModel())
57        security.SetLeverage(self.leverage)
58
59        india_stock_symbol:Symbol = security.Symbol
60
61        self.data[ticker] = data_tools.SymbolData(india_stock_symbol)
62
63self.selection_flag = False
64self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)
65
66def OnData(self, data: Slice):
67curr_date:datetime.date = self.Time.date()
68
69if curr_date in self.PB_data:
70    PB_values:dict[ticker, float] = self.PB_data[curr_date] 
71    
72    for ticker, PB_value in PB_values.items():
73        self.data[ticker].update_PB_value(PB_value)
74
75# rebalance monthly
76if not self.selection_flag:
77    return
78self.selection_flag = False
79
80book_to_market:dict[Symbol, float] = {}
81all_book_to_markets:list = []
82
83for _, symbol_obj in self.data.items():
84    if not symbol_obj.is_ready():
85        continue
86    
87    india_stock_symbol:Symbol = symbol_obj.india_stock_symbol
88    if self.Securities[india_stock_symbol].GetLastData() and (self.Time.date() - \
89        self.Securities[india_stock_symbol].GetLastData().Time.date()).days < self.max_missing_days:
90        curr_PB_value:float = symbol_obj.PB_value
91        book_to_market_value:float = 1 / curr_PB_value
92        all_book_to_markets.append(book_to_market_value)
93        book_to_market[india_stock_symbol] = book_to_market_value
94
95    # clear stocks PB_value for last month
96    symbol_obj.clear_PB_value()
97
98# make sure there are enough stocks for selection
99if len(book_to_market) <= self.quantile:
100    self.Liquidate()
101    return
102
103# z-score calculation
104z_score:dict[Symbol, float] = {}
105mean_book_to_markets:float = np.mean(all_book_to_markets)
106std_book_to_markets:float = np.std(all_book_to_markets)
107
108# z-score transform
109z_score:dict[Symbol, float] = {symbol:((value - mean_book_to_markets) / std_book_to_markets) for symbol, value in book_to_market.items()}
110z_score_transform:dict[Symbol, float] = {symbol : (1+z) if z >= 0 else (1/ (1-z)) for symbol, z in z_score.items()}
111
112quantile:int = int(len(z_score_transform) / self.quantile)
113sorted_by_z_score:list[Symbol] = [x[0] for x in sorted(z_score_transform.items(), key=lambda item: item[1])]
114
115# Buy the bottom decile (stocks with the lowest book-to-market ratios)
116long_part:list[Symbol] = sorted_by_z_score[:quantile]
117long_part_length:int = len(long_part)
118
119# trade execution
120invested:list[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
121for symbol in invested:
122    if symbol not in long_part:
123        self.Liquidate(symbol)
124
125for symbol in long_part:
126    self.SetHoldings(symbol, 1 / long_part_length)
127
128def Selection(self):
129self.selection_flag = True