Quality at Reasonable Price

1from AlgorithmImports import *
2from typing import Dict, List
3from data_tools import SymbolData, CustomFeeModel
4import statsmodels.api as sm
5# endregion
6
7class QualityAtReasonablePrice(QCAlgorithm):
8
9def Initialize(self):
10self.SetStartDate(2010, 1, 1)
11self.SetCash(100000)
12
13self.leverage:int = 5
14self.quantile:int = 10
15
16self.prices_period:int = 12 * 21                    # n daily prices
17self.volumes_period:int = self.prices_period        # n daily volumes
18self.max_missing_days_fundamentals:int = 356 + 10
19self.max_missing_days_daily_data:int = 10           
20
21self.market_cap_threshold:int = 1e09
22
23self.profitability_metrics:List[str] = ['gp_to_ta', 'roe', 'fcfo_to_ta', 'gpm']
24self.growth_metrics:List[str] = ['sales', 'gp', 'EBITDA', 'EBIT']
25self.rischio_metrics:List[str] = ['dept_to_ta', 'market_beta', 'inverse_QR']
26self.liquidity_metrics:List[str] = ['turnover']
27
28self.data:Dict[Symbol, SymbolData] = {}
29self.weights:Dict[Symbol, float] = {}
30
31self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
32self.market_prices:RollingWindow = RollingWindow[float](self.prices_period)
33history = self.History(self.market, self.prices_period, Resolution.Daily)
34if not history.empty:
35    closes = history.loc[self.market].close
36    
37    for (_, close) in closes.iteritems():
38        self.market_prices.Add(close)
39
40self.selection_month:int = 6
41self.selection_flag:bool = False
42self.UniverseSettings.Resolution = Resolution.Daily
43self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
44
45self.Schedule.On(self.DateRules.MonthEnd(self.market), self.TimeRules.BeforeMarketClose(self.market, 0), self.Selection)
46
47def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
48for security in changes.AddedSecurities:
49    security.SetFeeModel(CustomFeeModel())
50    security.SetLeverage(self.leverage)
51
52def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
53for equity in coarse:
54    symbol:Symbol = equity.Symbol
55
56    if symbol in self.data:
57        self.data[symbol].update_prices(equity.AdjustedPrice)
58
59if not self.selection_flag:
60    return Universe.Unchanged
61
62selected_symbols:List[CoarseFundamental] = [x.Symbol for x in coarse if x.HasFundamentalData]
63
64for symbol in selected_symbols:
65    if symbol in self.data:
66        continue
67
68    self.data[symbol] = SymbolData(self.prices_period, self.volumes_period)
69    history = self.History(symbol, self.prices_period, Resolution.Daily)
70    if not history.empty and all(column in history.columns for column in ['close', 'volume']):
71        closes = history.loc[symbol].close
72        volumes = history.loc[symbol].volume
73        
74        for (time, close), (_, volume) in zip(closes.iteritems(), volumes.iteritems()):
75            self.data[symbol].update_prices(close)
76            self.data[symbol].update_volumes(volume)
77            self.data[symbol].set_last_daily_data_update(time.date())
78
79return selected_symbols
80
81def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
82fine:List[FineFundamental] = [x for x in fine if x.MarketCap != 0 and x.MarketCap >= self.market_cap_threshold \
83    and x.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths != 0 and x.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths != 0 \
84    and x.OperationRatios.ROE.OneYear != 0 and x.FinancialStatements.CashFlowStatement.OperatingCashFlow.TwelveMonths != 0 \
85    and x.FinancialStatements.BalanceSheet.TotalLiabilitiesAsReported.TwelveMonths != 0 \
86    and x.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths != 0 and x.FinancialStatements.IncomeStatement.OperatingExpenseAsReported.TwelveMonths != 0 \
87    and x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths != 0 \
88    and x.OperationRatios.QuickRatio.OneYear != 0 and x.ValuationRatios.PBRatio != 0
89]
90
91stocks_with_metrics_counter:int = 0
92curr_date:datetime.date = self.Time.date()
93
94price_to_book:Dict[Symbol, float] = {}
95quality_metrics:Dict[Dict[str, Dict[Symbol, float]]] = {
96    'Profitability': { minor_metric: {} for minor_metric in self.profitability_metrics },
97    'Growth': { minor_metric: {} for minor_metric in self.growth_metrics },
98    'Rischio': { minor_metric: {} for minor_metric in self.rischio_metrics },
99    'Liquidity': { minor_metric: {} for minor_metric in self.liquidity_metrics }
100}
101
102market_daily_returns:List[float]|None = self.GetMarketDailyReturns() if self.market_prices.IsReady else None
103selected_symbols:List[Symbol] = []
104for stock in fine:
105    symbol:Symbol = stock.Symbol
106
107    gross_profit:float = stock.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths
108    total_revenue:float = stock.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths
109    operating_expenses:float = stock.FinancialStatements.IncomeStatement.OperatingExpenseAsReported.TwelveMonths
110    deprecation_amortization:float = stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.TwelveMonths
111
112    EBIT:float = total_revenue - operating_expenses
113    EBITDA:float = EBIT + deprecation_amortization
114
115    symbol_data:SymbolData = self.data[symbol]
116
117    # make sure fundamental data are consecutive, because of delta calculation in growth ratio
118    if not symbol_data.fundamentals_still_coming(curr_date, self.max_missing_days_fundamentals):
119        symbol_data.reset_fundamentals()
120
121    # make sure daily data (prices, volumes) are still coming
122    if not symbol_data.daily_data_still_coming(curr_date, self.max_missing_days_daily_data):
123        symbol_data.reset_daily_data()
124
125    # check if required data for metrics calculations are ready
126    if symbol_data.fundamentals_ready() and symbol_data.prices_ready() and symbol_data.volumes_ready() \
127        and market_daily_returns != None: # and symbol_data.spread_ready():
128
129        # calculate minor metrics values
130        market_cap:float = stock.MarketCap
131        total_assets:float = stock.FinancialStatements.BalanceSheet.TotalAssets.TwelveMonths
132        gross_profit:float = stock.FinancialStatements.IncomeStatement.GrossProfit.TwelveMonths
133        roe:float = stock.OperationRatios.ROE.OneYear
134        operating_cash_flow:float = stock.FinancialStatements.CashFlowStatement.OperatingCashFlow.TwelveMonths
135        total_liabilities:float = stock.FinancialStatements.BalanceSheet.TotalLiabilitiesAsReported.TwelveMonths
136        total_revenue:float = stock.FinancialStatements.IncomeStatement.TotalRevenue.TwelveMonths
137        quick_ratio:float = stock.OperationRatios.QuickRatio.OneYear
138
139        # profitability
140        quality_metrics['Profitability']['gp_to_ta'][symbol] = gross_profit / total_assets
141        quality_metrics['Profitability']['roe'][symbol] = roe
142        quality_metrics['Profitability']['fcfo_to_ta'][symbol] = operating_cash_flow / total_assets
143        quality_metrics['Profitability']['gpm'][symbol] = gross_profit
144
145        # Growth
146        quality_metrics['Growth']['sales'][symbol] = symbol_data.get_sales_change(total_revenue)
147        quality_metrics['Growth']['gp'][symbol] = symbol_data.get_gross_profit_change(gross_profit)
148        quality_metrics['Growth']['EBITDA'][symbol] = symbol_data.get_EBITDA_change(EBITDA)
149        quality_metrics['Growth']['EBIT'][symbol] = symbol_data.get_EBIT_change(EBIT)
150
151        # Rischio
152        quality_metrics['Rischio']['dept_to_ta'][symbol] = total_liabilities / total_assets
153        daily_returns:np.array = symbol_data.get_daily_returns()
154        regression_model = self.MultipleLinearRegression(market_daily_returns, daily_returns)
155        quality_metrics['Rischio']['market_beta'][symbol] = regression_model.params[-1]
156        quality_metrics['Rischio']['inverse_QR'][symbol] = 1 / quick_ratio
157        
158        # Liquidity
159        # inverse_spread:float = 1 / symbol_data.get_spread()
160        turnover:float = symbol_data.get_volumes_mean() / market_cap
161        quality_metrics['Liquidity']['turnover'][symbol] = turnover
162
163        price_to_book[symbol] = stock.ValuationRatios.PBRatio
164
165    # update stock's fundamentals
166    symbol_data.set_total_revenue(total_revenue)
167    symbol_data.set_EBIT(EBIT)
168    symbol_data.set_EBITDA(EBITDA)
169    symbol_data.set_gross_profit(gross_profit)
170    symbol_data.set_last_fundamentals_update(curr_date)
171
172    selected_symbols.append(stock.Symbol)
173
174# make sure there are enough stocks for selection
175if len(price_to_book) >= self.quantile:
176    quality_values:Dict[Symbol, float] = {}
177
178    for main_metric, minor_metrics_data in quality_metrics.items():
179        main_metric_values:Dict[Symbol, float] = {}
180
181        for minor_metric, values_by_symbol in minor_metrics_data.items():
182            # sort stocks by their minor metric values in descending order
183            sorted_by_value:List[Symbol] = [x[0] for x in sorted(values_by_symbol.items(), key=lambda item: item[1], reverse=False)]
184
185            # create array of stocks ranks and calculate it's mean and std
186            arranged_array:np.array = np.arange(1, len(sorted_by_value) + 1)
187            minor_metric_mean:float = np.mean(arranged_array)
188            minor_metric_std:float = np.std(arranged_array)
189
190            # calculate stock's minor metric z-score based on stock's rank
191            for i, symbol in enumerate(sorted_by_value):
192                minor_metric_z_score:float = ((i + 1) - minor_metric_mean) / minor_metric_std
193
194                if symbol not in main_metric_values:
195                    main_metric_values[symbol] = 0
196                # sum z-scores of minor metrics to get the main metric value
197                main_metric_values[symbol] += minor_metric_z_score
198        
199        main_metric_mean:float = np.mean(list(main_metric_values.values()))
200        main_metric_std:float = np.std(list(main_metric_values.values()))
201
202        # calculate z-scores of main metrics
203        for symbol, value in main_metric_values.items():
204            main_metric_z_score:float = (value - main_metric_mean) / main_metric_std
205            main_metric_z_score = -main_metric_z_score if main_metric == 'Rischio' else main_metric_z_score
206
207            if symbol not in quality_values:
208                quality_values[symbol] = 0
209            # summ z-scores of main metrics to get the quality values
210            quality_values[symbol] += main_metric_z_score
211
212    # calculate quality score
213    quality_values_mean:float = np.mean(list(quality_values.values()))
214    quality_values_std:float = np.std(list(quality_values.values()))
215    quality_scores:Dict[Symbol] = { symbol: (quality_value - quality_values_mean) / quality_values_std \
216        for symbol, quality_value in quality_values.items() }
217
218    # calculate price score
219    sorted_by_pb:List[Symbol] = [x[0] for x in sorted(price_to_book.items(), key=lambda item: item[1], reverse=False)]
220    
221    # create array of stocks ranks and calculate it's mean and std
222    arranged_array:np.array = np.arange(1, len(sorted_by_pb) + 1)
223    price_to_book_score_mean:float = np.mean(arranged_array)
224    price_to_book_score_std:float = np.mean(arranged_array)
225    price_scores:Dict[Symbol] = { symbol: ((i + 1) - price_to_book_score_mean) / price_to_book_score_std \
226        for i, symbol in enumerate(sorted_by_pb) }
227    
228    # calculate final score as quality score minus price score
229    final_score:Dict[Symbol, float] = { symbol: quality_scores[symbol] - price_scores[symbol] \
230        for symbol in price_scores }
231
232    quantile:int = int(len(final_score) / self.quantile)
233    sorted_by_final_score:List[Symbol] = [x[0] for x in sorted(final_score.items(), key=lambda item: item[1], reverse=True)]
234
235    # go long on the first decile and short on the last decile
236    long_leg:List[Symbol] = sorted_by_final_score[:quantile]
237    short_leg:List[Symbol] = sorted_by_final_score[-quantile:]
238
239    for symbol in long_leg:
240        self.weights[symbol] = 1 / quantile
241
242    for symbol in short_leg:
243        self.weights[symbol] = -1 / quantile
244
245return selected_symbols
246
247def OnData(self, data:Slice) -> None:
248curr_date:datetime.date = self.Time.date()
249
250for symbol, symbol_data in self.data.items():
251    if symbol in data and data[symbol]:
252        spread:float = self.Securities[symbol].BidPrice - self.Securities[symbol].AskPrice
253        price:float = data[symbol].Value
254        volume:float = data[symbol].Volume
255
256        self.data[symbol].update_prices(price)
257        self.data[symbol].update_volumes(volume)
258        self.data[symbol].set_spread(spread)
259
260        self.data[symbol].set_last_daily_data_update(curr_date)
261
262if self.market in data and data[self.market]:
263    price:float = data[self.market].Value
264    self.market_prices.Add(price)
265
266if not self.selection_flag:
267    return
268self.selection_flag = False
269
270# trade execution
271invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
272for symbol in invested:
273    if symbol not in self.weights:
274        self.Liquidate(symbol)
275        
276for symbol, w in self.weights.items():
277    self.SetHoldings(symbol, w)
278        
279self.weights.clear()
280
281def GetMarketDailyReturns(self) -> List[float]:
282market_daily_prices:np.array= np.array(list(self.market_prices))
283return list((market_daily_prices[:-1] - market_daily_prices[1:]) / market_daily_prices[1:])
284
285def MultipleLinearRegression(self, x:np.array, y:np.array):
286x:np.array = np.array(x).T
287x = sm.add_constant(x)
288result = sm.OLS(endog=y, exog=x).fit()
289return result
290
291def Selection(self):
292# rebalance at the end of selection month
293if self.selection_month == self.Time.month:
294    self.selection_flag = True