Conservative Formula in India

1from AlgorithmImports import *
2from typing import Dict, Tuple, List
3import data_tools
4from datetime import datetime
5# endregion
6
7class ConservativeFormulaInIndia(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2010, 1, 1)
11self.SetCash(10000000) # INR
12
13self.price_period:int = 12 * 21
14self.price_skip_period:int = 21
15self.q_fundamental_period:int = 8 # 2 years of quarters
16
17self.data:Dict[Symbol, data_tools.SymbolData] = {}
18
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
23self.quantile:int = 10
24self.leverage:int = 5
25self.rebalance_every_n_months:int = 3
26
27for t in tickers:
28    # price data subscription
29    data:Security = self.AddData(data_tools.IndiaStocks, t, Resolution.Daily)
30    data.SetFeeModel(data_tools.CustomFeeModel())
31    data.SetLeverage(self.leverage)
32    stock_symbol:Symbol = data.Symbol
33    
34    # fundamental data subscription
35    balance_sheet_symbol:Symbol = self.AddData(data_tools.IndiaBalanceSheetStatement, t, Resolution.Daily).Symbol
36    cashflow_symbol:Symbol = self.AddData(data_tools.IndiaCashflowStatement, t, Resolution.Daily).Symbol
37
38    self.data[stock_symbol] = data_tools.SymbolData(stock_symbol, balance_sheet_symbol, cashflow_symbol, self.price_period, self.q_fundamental_period)
39
40# BSE index hedge
41self.hedge_with_index:bool = False
42self.bse_index_data:Security = self.AddData(data_tools.BSEIndex, 'BSE_100', Resolution.Daily)
43self.bse_index_data.SetFeeModel(data_tools.CustomFeeModel())
44self.bse_index_data.SetLeverage(self.leverage)
45self.bse_index:Symbol = self.bse_index_data.Symbol
46
47self.recent_month:int = -1
48
49def OnData(self, data: Slice) -> None:
50rebalance_flag:bool = False
51metrics_by_symbol:Dict[Symbol, Tuple[float, float]] = {}
52
53price_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaStocks.get_last_update_date()
54bs_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaBalanceSheetStatement.get_last_update_date()
55cf_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaCashflowStatement.get_last_update_date()
56
57for price_symbol, symbol_data in self.data.items():
58    # store price data
59    if price_symbol in data and data[price_symbol] and data[price_symbol].Value != 0:
60        price:float = data[price_symbol].Value
61        self.data[price_symbol].update_price(price)
62    
63    bs_symbol:Symbol = symbol_data._balance_sheet_symbol
64    cf_symbol:Symbol = symbol_data._cashflow_symbol
65
66    # both CF and BS statement data is present at the same time
67    if bs_symbol in data and data[bs_symbol] and cf_symbol in data and data[cf_symbol]:
68        bs_statement:Dict = data[bs_symbol].Statement
69        cf_statement:Dict = data[cf_symbol].Statement
70
71        shares_field:str = 'commonStockSharesOutstanding'
72        dividends_field:str = 'dividendsPaid'
73        if shares_field in bs_statement and bs_statement[shares_field] is not None and \
74            dividends_field in cf_statement and cf_statement[dividends_field] is not None:
75            shares_outstanding:float = float(bs_statement[shares_field])
76            dividend:float = float(cf_statement[dividends_field])
77
78            # store fundamentals
79            symbol_data.update_fundamentals(shares_outstanding, dividend)
80
81    if self.IsWarmingUp: continue
82
83    if (self.recent_month != self.Time.month and self.Time.month % self.rebalance_every_n_months == 0) or rebalance_flag:
84        self.recent_month = self.Time.month
85        rebalance_flag = True
86
87        if self.Securities[price_symbol].GetLastData() and price_symbol in price_last_update_date and self.Time.date() <= price_last_update_date[price_symbol]:
88            if self.Securities[bs_symbol].GetLastData() and bs_symbol in bs_last_update_date and self.Time.date() <= bs_last_update_date[bs_symbol] and \
89                self.Securities[cf_symbol].GetLastData() and cf_symbol in cf_last_update_date and self.Time.date() <= cf_last_update_date[cf_symbol]:
90                # momentum and fundamental data are ready and still arriving
91                if symbol_data.momentum_ready() and symbol_data.fundamentals_ready():
92                    shares_outstanding, dividend = symbol_data.get_recent_fundamentals()
93                    dps:float = dividend / shares_outstanding
94                    price:float = symbol_data.get_recent_price()
95                    
96                    if price != 0.:
97                        dividend_yield:float = dps / price
98                        buyback_yield:float = shares_outstanding / symbol_data.get_avg_so()
99
100                        # net payout yield
101                        NPY:float = dividend_yield + buyback_yield
102                        momentum:float = symbol_data.get_momentum(self.price_skip_period)
103                        metrics_by_symbol[price_symbol] = (NPY, momentum)
104
105# rebalance once a quarter
106if rebalance_flag:
107    long:List[Symbol] = []
108
109    # sorting
110    if len(metrics_by_symbol) >= self.quantile:
111        # calculate aggregate rank from the momentum and NPY ranks
112        sorted_by_npy:List = sorted(metrics_by_symbol.items(), key=lambda x: x[1][0])
113        sorted_by_momentum:List = sorted(metrics_by_symbol.items(), key=lambda x: x[1][1])
114        rank:Dict[Symbol, float] = { data[0] : np.mean([sorted_by_npy.index(data), sorted_by_momentum.index(data)]) for data in sorted_by_npy}
115
116        # portfolio consists of the top ranked stocks
117        sorted_by_rank:List = sorted(rank.items(), key=lambda x: x[1], reverse=True)
118        quantile:int = int(len(sorted_by_rank) / self.quantile)
119        long = [x[0] for x in sorted_by_rank[:quantile]]
120
121    # liquidate and rebalance
122    invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
123    for price_symbol in invested:
124        if price_symbol not in long:# + [self.bse_index] if self.hedge_with_index else []:
125            self.Liquidate(price_symbol)
126
127    long_count:float = float(len(long))
128    for price_symbol in long:
129        if price_symbol in data and data[price_symbol]:
130            self.SetHoldings(price_symbol, 1. / long_count)
131    
132    # hedge
133    if self.hedge_with_index:
134        if long_count != 0:
135            self.SetHoldings(self.bse_index, -1)
136        else:
137            if self.Portfolio[self.bse_index].Invested:
138                self.Liquidate(self.bse_index)
139else:
140    if self.Portfolio.Invested:
141        # BSE index ended
142        bse_last_udpate_date:datetime.date = data_tools.BSEIndex.get_last_update_date()
143        if self.Securities[self.bse_index].GetLastData() and self.Time.date() > bse_last_udpate_date:
144            self.Liquidate()