Investment Factor in Indian Stocks

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict, Tuple
4from datetime import datetime
5# endregion
6
7class InvestmentFactorInIndianStocks(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2010, 1, 1)
11self.SetCash(1e9) # INR
12
13self.quantile:int = 3
14self.leverage:int = 20
15self.period:int = 12 # 3 years of quarters
16self.data:Dict[Symbol, data_tools.SymbolData] = {}
17
18# download tickers
19ticker_file_str:str = self.Download('data.quantpedia.com/backtesting_data/equity/india_stocks/nse_500_tickers.csv')
20ticker_lines:List[str] = ticker_file_str.split('\r\n')
21tickers:List[str] = [ ticker_line.split(',')[0] for ticker_line in ticker_lines[1:] ]
22
23for t in tickers:
24    # price data subscription
25    data = self.AddData(data_tools.IndiaStocks, t, Resolution.Daily)
26    data.SetFeeModel(data_tools.CustomFeeModel())
27    data.SetLeverage(self.leverage)
28    stock_symbol:Symbol = data.Symbol
29
30    # fundamental data subscription
31    balance_sheet = self.AddData(data_tools.IndiaBalanceSheet, t, Resolution.Daily).Symbol 
32
33    self.data[stock_symbol] = data_tools.SymbolData(stock_symbol, balance_sheet, self.period)
34
35self.rebalance_month:int = 10
36self.rebalance_day:int = 1
37
38def OnData(self, data: Slice) -> None:
39rebalance_flag:bool = False
40metric_by_symbol:Dict[Symbol, Tuple(float, float)] = {}
41
42price_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaStocks.get_last_update_date()
43bs_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaBalanceSheet.get_last_update_date()
44
45for symbol, symbol_data in self.data.items():
46    # store price data
47    if data.ContainsKey(symbol) and data[symbol] and data[symbol].Value != 0:
48        price:float = data[symbol].Value
49        self.data[symbol].update_price(price)
50
51    bs_symbol:Symbol = symbol_data._balance_sheet_symbol
52
53    # check if BS statement is present
54    if bs_symbol in data and data[bs_symbol]:
55        bs_statement:Dict = data[bs_symbol].Statement
56
57        assets_field:str = 'totalAssets'
58        shares_field:str = 'commonStockSharesOutstanding'
59        if assets_field in bs_statement and bs_statement[assets_field] is not None \
60            and shares_field in bs_statement and bs_statement[shares_field] is not None:
61            date:datetime.date = self.Time
62            total_assets:float = float(bs_statement[assets_field])
63            shares:float = float(bs_statement[shares_field])
64            # store fundamentals
65            symbol_data.update_fundamentals(date, total_assets, shares)
66    
67    if self.IsWarmingUp: 
68        continue
69    
70    # rebalance on first of October
71    if self.Time.month == self.rebalance_month and self.Time.day == self.rebalance_day:
72        rebalance_flag = True
73
74        # fundamental data are ready and still arriving
75        if self.Securities[bs_symbol].GetLastData() and bs_symbol in bs_last_update_date and self.Time.date() <= bs_last_update_date[bs_symbol]:
76            total_assets:Tuple[datetime.date, float] = symbol_data.get_total_assets()
77            market_cap:float = symbol_data.get_marketcap()
78            if market_cap != 0:
79                total_assets_t2:List[float] = [x[1] for x in total_assets if x[0].year == self.Time.year - 2]
80                total_assets_t1:List[float] = [x[1] for x in total_assets if x[0].year == self.Time.year - 1]
81                
82                if len(total_assets_t2) > 0 and len(total_assets_t1) > 0:
83                    change:float = (total_assets_t1[0] / total_assets_t2[0]) - 1
84                    metric_by_symbol[symbol] = (change, market_cap)
85
86if rebalance_flag:
87    weights:Dict[Symbol, float] = {}
88
89    if len(metric_by_symbol) >= self.quantile:
90        # sort by investment factor
91        sorted_changes:List = sorted(metric_by_symbol.items(), key=lambda x: x[1][0], reverse=True)
92        quantile:int = int(len(sorted_changes) / self.quantile)
93
94        # get top and bottom tercile
95        long_tercile:List[Symbol] = [x[0] for x in sorted_changes][:quantile]
96        short_tercile:List[Symbol] = [x[0] for x in sorted_changes][-quantile:]
97
98        # calculate weights based on marketcap
99        sum_long = sum([metric_by_symbol[i][1] for i in long_tercile])
100        for asset in long_tercile:
101            weights[asset] = metric_by_symbol[asset][1] / sum_long
102
103        sum_short = sum([metric_by_symbol[i][1] for i in short_tercile])
104        for asset in short_tercile:
105            weights[asset] = -metric_by_symbol[asset][1] / sum_short
106
107    # liquidate and rebalance
108    invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
109    for symbol in invested:
110        if symbol not in weights:
111            self.Liquidate(symbol)
112    
113    for symbol, weight in weights.items():
114        if symbol in data and data[symbol]:
115            self.SetHoldings(symbol, weight)