Combining Fundamental and Transitory Component of Value Strategy

1from AlgorithmImports import *
2import numpy as np
3import statsmodels.api as sm
4from typing import List, Dict
5from numpy import isnan
6#endregion
7 
8class CombiningFundamentalAndTransitoryComponentOfValueStrategy(QCAlgorithm):
9def Initialize(self):
10self.SetStartDate(2000, 1, 1)
11self.SetCash(100000)
12
13self.exchange_codes: List[str] = ['NYS', 'NAS', 'ASE']
14self.tickers_to_ignore: List[str] = ['SGA']
15self.period: int = 2 # need n values for regression
16
17self.data: Dict[Symbol, SymbolData] = {}
18self.quantities: Dict[Symbol, int] = {}
19self.last_selection: List[Symbol] = []
20
21self.min_share_price: int = 5
22self.leverage: int = 5
23self.quantile: int = 5
24self.month_counter: int = 0
25self.fundamental_count: int = 3000
26self.fundamental_sorting_key = lambda x: x.MarketCap
27self.selection_flag: bool = False
28self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
29
30self.settings.daily_precise_end_time = False
31self.Settings.MinimumOrderMarginPortfolioPercentage = 0
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34
35self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol, 0), self.Selection)
36def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
37for security in changes.AddedSecurities:
38    security.SetFeeModel(CustomFeeModel())
39    security.SetLeverage(self.leverage)
40def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List:
41# selection yearly
42if not self.selection_flag:
43    return Universe.Unchanged
44
45# filter top n stocks by dollar volume
46selected: List[Fundamental] = [
47    x for x in fundamental \
48    if x.HasFundamentalData and \
49    x.Market == 'usa' and \
50    x.Price > self.min_share_price and not \
51    isnan(x.ValuationRatios.PBRatio) and x.ValuationRatios.PBRatio != 0 and not\
52    isnan(x.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths) and x.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths != 0 and not\
53    isnan(x.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths) and x.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths != 0 and \
54    x.MarketCap != 0 and \
55    x.SecurityReference.ExchangeId in self.exchange_codes and \
56    x.Symbol.Value not in self.tickers_to_ignore
57]
58if len(selected) > self.fundamental_count:
59            selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
60                
61HMLFundamental: Dict[Fundamental, float] = {} 
62HMLTransitory: Dict[Fundamental, float] = {}
63
64# store current stocks prices for trenching
65for stock in selected:
66    symbol: Symbol = stock.Symbol
67    
68    if symbol not in self.data:
69        self.data[symbol] = SymbolData(self.period)
70    
71    # update price
72    self.data[symbol].update_price(stock.AdjustedPrice)
73   
74    # symbol = stock.Symbol
75    pb_ratio: float = stock.ValuationRatios.PBRatio
76    total_assets: float = stock.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths
77    gross_profit: float = stock.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths
78    
79    # make sure data are consecutive
80    if symbol not in self.last_selection:
81        self.data[symbol] = SymbolData(self.period)
82        
83    self.data[symbol].update_regression_data(pb_ratio, total_assets, gross_profit)
84        
85    # make sure data for regression are ready
86    if not self.data[symbol].is_data_ready():
87        continue
88    
89    # calculate stock's Y and Xs for regression
90    regression_y: List[float] = self.data[symbol].get_regression_y()
91    regression_x: List[List[float]] = self.data[symbol].get_regression_x()
92    
93    regression_model: RegressionResultWrapper = self.MultipleLinearRegression(regression_x, regression_y)
94    
95    # calculate fit value
96    fit_value: float = regression_model.params[0]
97    # iterate through betas - handles missing beta value if profit growth value is 0
98    for i, beta in enumerate(regression_model.params[1:]):
99        corresponding_x: float = regression_x[i][1]
100        fit_value += corresponding_x * beta
101    
102    # store fit value keyed by stock
103    HMLFundamental[stock] = fit_value
104    
105    # last residual from regression is needed for HMLTransitory strategy    
106    last_residual: float = regression_model.resid[-1]
107    # store last residual keyed by stock
108    HMLTransitory[stock] = last_residual
109    
110# change last selection, to make data consecutive
111self.last_selection = [x.Symbol for x in selected]
112
113# there has to be enough stocks for quintile selections    
114if len(HMLFundamental) < self.quantile or len(HMLTransitory) < self.quantile:
115    return Universe.Unchanged
116    
117quantile: int = int(len(HMLFundamental) / self.quantile)
118sorted_by_fundamental: List[Fundamental] = [x[0] for x in sorted(HMLFundamental.items(), key=lambda item: item[1])]
119sorted_by_transitory: List[Fundamental] = [x[0] for x in sorted(HMLTransitory.items(), key=lambda item: item[1])]
120
121# select long and short
122fundamental_long_stocks: List[Fundamental] = sorted_by_fundamental[:quantile]
123fundamental_short_stocks: List[Fundamental] = sorted_by_fundamental[-quantile:]
124
125transitory_long_stocks: List[Fundamental] = sorted_by_transitory[:quantile]
126transitory_short_stocks: List[Fundamental] = sorted_by_transitory[-quantile:]
127
128# perform trenching
129# have to divide weight by 2, because there are 2 different strategies in portfolio
130weight: float = self.Portfolio.TotalPortfolioValue / 2
131# NOTE self.quantities is modified bellow
132# calculate quantities for long parts
133self.CalculateQuantities(fundamental_long_stocks, weight, True)
134self.CalculateQuantities(transitory_long_stocks, weight, True)
135
136# calculate quantities for short parts
137self.CalculateQuantities(fundamental_short_stocks, weight, False)
138self.CalculateQuantities(transitory_short_stocks, weight, False)
139
140return list(self.quantities.keys())
141
142def OnData(self, data: Slice) -> None:
143# rebalance yearly
144if not self.selection_flag:
145    return
146self.selection_flag = False
147
148# trade execution
149self.Liquidate()
150        
151for symbol, quantity in self.quantities.items():
152    if symbol in data and data[symbol]:
153        self.MarketOrder(symbol, quantity)
154    
155self.quantities.clear()
156
157def MultipleLinearRegression(self, x: List[List[float]], y: List[float]):
158x: np.ndarray = np.array(x).T
159x = sm.add_constant(x)
160result: RegressionResultWrapper = sm.OLS(endog=y, exog=x).fit()
161return result
162
163def CalculateQuantities(self, stock_list: List[Fundamental], weight: float, long_flag: bool) -> None:
164total_cap: float = sum([stock.MarketCap for stock in stock_list])
165
166for stock in stock_list:
167    price: float = self.data[stock.Symbol].price
168    market_cap: float = stock.MarketCap
169    
170    # calculate quantity
171    quantity: int = np.floor((weight * (market_cap / total_cap)) / price) 
172    
173    # stock goes short
174    if not long_flag:
175        quantity = -1 * quantity
176    
177    self.quantities[stock.Symbol] = quantity
178
179def Selection(self) -> None:
180# rebalance yearly
181if self.month_counter % 12 == 0:
182    self.selection_flag = True
183self.month_counter += 1
184    
185class SymbolData():
186def __init__(self, period: int) -> None:
187self.pb_ratio: RollingWindow = RollingWindow[float](period)
188self.gross_profit: RollingWindow = RollingWindow[float](period + 1) 
189self.total_assets: RollingWindow = RollingWindow[float](period + 1)
190self.price: Union[None, float] = None
191
192def update_price(self, price: float) -> None:
193self.price = price
194
195def update_regression_data(self, pb_ratio: float, total_assets: float, gross_profit: float) -> None:
196self.pb_ratio.Add(pb_ratio)
197self.total_assets.Add(total_assets)
198self.gross_profit.Add(gross_profit)
199
200def is_data_ready(self) -> bool:
201# return self.pb_ratio.IsReady and self.gross_profit.IsReady and \
202#         self.total_assets.IsReady and self.market_cap != None and self.price != None
203return self.pb_ratio.IsReady and self.gross_profit.IsReady and self.total_assets.IsReady
204
205def get_regression_y(self) -> List[float]:
206return [x for x in self.pb_ratio][::-1]
207        
208def get_regression_x(self) -> List[List[float]]:
209gross_profit_values: np.ndarray = np.array([x for x in self.gross_profit])
210total_assets_values: np.ndarray = np.array([x for x in self.total_assets])
211
212x1: List[float] = [gpv / tav for gpv, tav in zip(gross_profit_values[:-1], total_assets_values[:-1])]
213x2: List[float] = (gross_profit_values[:-1] / gross_profit_values[1:] - 1) / total_assets_values[1:]     # profit growth / total assets from previous year
214
215return [x1[::-1], x2[::-1]]
216
217# Custom fee model
218class CustomFeeModel(FeeModel):
219def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
220fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
221return OrderFee(CashAmount(fee, "USD"))