Profitability Factor in Indian Stocks

1from AlgorithmImports import *
2import data_tools
3from typing import List, Dict, Tuple
4from datetime import datetime
5# endregion
6
7class ProfitabilityFactorInIndianStocks(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2010, 1, 1)
11self.SetCash(1000000000) # 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    income_statement = self.AddData(data_tools.IndiaIncomeStatement, t, Resolution.Daily).Symbol
33
34    self.data[stock_symbol] = data_tools.SymbolData(stock_symbol, balance_sheet, income_statement, self.period)
35
36self.rebalance_month:int = 10
37self.rebalance_day:int = 1
38
39def OnData(self, data: Slice):
40rebalance_flag:bool = False
41metric_by_symbol:Dict[Symbol, Tuple(float, float)] = {}
42
43price_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaStocks.get_last_update_date()
44bs_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaBalanceSheet.get_last_update_date()
45is_last_update_date:Dict[Symbol, datetime.date] = data_tools.IndiaIncomeStatement.get_last_update_date()
46
47for symbol, symbol_data in self.data.items():
48    # store price
49    if data.ContainsKey(symbol) and data[symbol] and data[symbol].Value != 0:
50        price:float = data[symbol].Value
51        self.data[symbol].update_price(price)
52    bs_symbol:Symbol = symbol_data._balance_sheet_symbol
53    is_symbol:Symbol = symbol_data._income_statement_symbol
54
55    # check if BS an IS statement is present
56    if bs_symbol in data and data[bs_symbol] and is_symbol in data and data[is_symbol]:
57        bs_statement:Dict = data[bs_symbol].Statement
58        is_statement:Dict = data[is_symbol].Statement
59
60        revenue_field:str = 'totalRevenue'
61        cost_of_revenue_field:str = 'costOfRevenue'
62        assets_field:str = 'totalAssets'
63        liab_field:str = 'totalLiab'
64        shares_field:str = 'commonStockSharesOutstanding'
65
66        if revenue_field in is_statement and is_statement[revenue_field] is not None \
67            and cost_of_revenue_field in is_statement and is_statement[cost_of_revenue_field] is not None \
68            and assets_field in bs_statement and bs_statement[assets_field] is not None \
69            and liab_field in bs_statement and bs_statement[liab_field] is not None \
70            and shares_field in bs_statement and bs_statement[shares_field] is not None:
71            date:datetime.date = self.Time
72            revenue:float = float(is_statement[revenue_field])
73            cost_of_revenue:float = float(is_statement[cost_of_revenue_field])
74            assets:float = float(bs_statement[assets_field])
75            liab:float = float(bs_statement[liab_field])
76            shares:float = float(bs_statement[shares_field])
77            # store fundamentals
78            symbol_data.update_fundamentals(date, revenue, cost_of_revenue, assets, liab, shares)
79    
80    if self.IsWarmingUp: 
81        continue
82
83    # rebalance on first of October
84    if self.Time.month == self.rebalance_month and self.Time.day == self.rebalance_day:
85        rebalance_flag = True
86
87        # fundamental data are ready and still arriving
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[is_symbol].GetLastData() and is_symbol in is_last_update_date and self.Time.date() <= is_last_update_date[is_symbol]:
90            
91            market_cap:float = symbol_data.get_marketcap()
92            if market_cap != 0:
93                revenue:Tuple[datetime.date, float, float] = symbol_data.get_revenue()
94                total_assets_liab:Tuple[datetime.date, float, float] = symbol_data.get_total_assets_liab()
95                total_revenue:List[float] = [x[1] for x in revenue if x[0].year == self.Time.year]
96                cost_revenue:List[float] = [x[2] for x in revenue if x[0].year == self.Time.year]
97                total_assets:List[float] = [x[1] for x in total_assets_liab if x[0].year == self.Time.year - 1]
98                total_liab:List[float] = [x[2] for x in total_assets_liab if x[0].year == self.Time.year - 1]
99                
100                fundamentals:List[float] = [total_revenue, cost_revenue, total_assets, total_liab]
101
102                if all(len(fundamental) > 0 for fundamental in fundamentals):
103                    if (total_assets[0] - total_liab[0]) > 0:
104                        change:float = (sum(total_revenue) - sum(cost_revenue)) / (total_assets[0] - total_liab[0])
105                        metric_by_symbol[symbol] = (change, market_cap)
106
107if rebalance_flag:
108    weights:Dict[Symbol, float] = {}
109
110    if len(metric_by_symbol) >= self.quantile:
111        # sort by profitability factor
112        sorted_changes:List = sorted(metric_by_symbol.items(), key=lambda x: x[1][0], reverse=True)
113        quantile: int = int(len(sorted_changes) / self.quantile)
114
115        # get top and bottom tercile
116        long_tercile:List[Symbol] = [x[0] for x in sorted_changes][:quantile]
117        short_tercile:List[Symbol] = [x[0] for x in sorted_changes][-quantile:]
118
119        # calculate weights based on marketcap
120        sum_long = sum([metric_by_symbol[i][1] for i in long_tercile])
121        for asset in long_tercile:
122            weights[asset] = metric_by_symbol[asset][1] / sum_long
123
124        sum_short = sum([metric_by_symbol[i][1] for i in short_tercile])
125        for asset in short_tercile:
126            weights[asset] = -metric_by_symbol[asset][1] / sum_short
127    
128    # liquidate and rebalance
129    invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
130    for symbol in invested:
131        if symbol not in weights:
132            self.Liquidate(symbol)
133    
134    for symbol, weight in weights.items():
135        if symbol in data and data[symbol]:
136            self.SetHoldings(symbol, weight)