Accruals Momentum

1from AlgorithmImports import *
2from collections import deque
3import numpy as np
4import statsmodels.api as sm
5from typing import List, Dict, Deque, Tuple
6from numpy import isnan
7class AccrualsMomentum(QCAlgorithm):
8def Initialize(self):
9 self.SetStartDate(2000, 1, 1)
10 self.SetCash(100000)
11 self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']	
12 # Latest accruals data.
13 self.accruals_data:Dict[Symbol, StockData] = {}
14 self.min_share_price:int = 5
15 self.period:int = 5
16 self.leverage:int = 15
17 self.quantile:int = 4
18 
19 # Accruals value for last year.
20 self.latest_accruals:Dict[Symbol, StockData] = {}
21 
22 self.fundamental_count:int = 1000
23 self.fundamental_sorting_key = lambda x: x.DollarVolume
24 
25 self.long:List[Symbol] = []
26 self.short:List[Symbol] = []
27 
28 self.consecutive_period:int = 4
29 self.low_residual_portfolios:Deque = deque(maxlen = self.consecutive_period)
30 self.high_residual_portfolios:Deque = deque(maxlen = self.consecutive_period)
31 
32 self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
33 self.regression_period:int = 5
34 self.regression_data:Dict[Symbol, Tuple[float]] = {}
35 
36 self.months:int = 12
37 self.selection_flag:bool = False
38 self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
39 self.UniverseSettings.Resolution = Resolution.Daily
40 self.AddUniverse(self.FundamentalSelectionFunction)
41 self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
42 self.settings.daily_precise_end_time = False
43def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
44 for security in changes.AddedSecurities:
45     symbol = security.Symbol
46     
47     security.SetFeeModel(CustomFeeModel())
48     security.SetLeverage(self.leverage)
49     
50     self.regression_data[symbol] = deque(maxlen = self.regression_period)
51         
52def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
53 if not self.selection_flag:
54     return Universe.Unchanged
55 
56 selected:List[Fundamental] = [
57     x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price >= self.min_share_price \
58     and not isnan(x.FinancialStatements.BalanceSheet.CurrentAssets.Value) and x.FinancialStatements.BalanceSheet.CurrentAssets.Value > 0 \
59     and not isnan(x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value) and x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value > 0 \
60     and not isnan(x.FinancialStatements.BalanceSheet.CurrentLiabilities.Value) and x.FinancialStatements.BalanceSheet.CurrentLiabilities.Value > 0 \
61     and not isnan(x.FinancialStatements.BalanceSheet.CurrentDebt.Value) and x.FinancialStatements.BalanceSheet.CurrentDebt.Value > 0 \
62     and not isnan(x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value) and x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value > 0 \
63     and not isnan(x.FinancialStatements.BalanceSheet.GrossPPE.Value) and x.FinancialStatements.BalanceSheet.GrossPPE.Value > 0 \
64     and not isnan(x.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value) and x.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value > 0 \
65     and x.SecurityReference.ExchangeId in self.exchange_codes
66 ]
67 
68 if len(selected) > self.fundamental_count:
69     selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
70 updated:List[Symbol] = []
71 residual:Dict[Symbol, float] = {}
72 
73 for stock in selected:
74     symbol:Symbol = stock.Symbol
75     if symbol not in self.accruals_data:
76         # Data for previous year.
77         self.accruals_data[symbol] = None
78         
79     if symbol not in self.latest_accruals:
80         # Previous year's accruals.
81         self.latest_accruals[symbol] = None
82                 
83     # Accrual calc.
84     current_accruals_data:StockData = StockData(stock.FinancialStatements.BalanceSheet.CurrentAssets.Value, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value,
85                                         stock.FinancialStatements.BalanceSheet.CurrentLiabilities.Value, stock.FinancialStatements.BalanceSheet.CurrentDebt.Value, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.Value,
86                                         stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value, stock.FinancialStatements.BalanceSheet.TotalAssets.Value,
87                                         stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value)
88     
89     # There is not previous accruals data.
90     if not self.accruals_data[symbol]:
91         self.accruals_data[symbol] = current_accruals_data
92         updated.append(symbol)
93         continue
94         
95     current_accruals:float = self.CalculateAccruals(current_accruals_data, self.accruals_data[symbol])
96     delta_sales:float = (current_accruals_data.Sales - self.accruals_data[symbol].Sales) / self.accruals_data[symbol].TotalAssets
97     ppe:float = stock.FinancialStatements.BalanceSheet.GrossPPE.Value / self.accruals_data[symbol].TotalAssets
98     # There is not previous accruals value.
99     if not self.latest_accruals[symbol]:
100         self.latest_accruals[symbol] = current_accruals
101         updated.append(symbol)
102         continue                
103     
104     # Regression data.
105     accruals_factor:float = 1 / self.latest_accruals[symbol]
106     reg_data:Tuple[float] = (current_accruals, accruals_factor, delta_sales, ppe)
107     
108     if symbol not in self.regression_data:
109         self.regression_data[symbol] = deque(maxlen = self.regression_period)
110     self.regression_data[symbol].append(reg_data)
111     
112     if len(self.regression_data[symbol]) == self.regression_data[symbol].maxlen:
113         total_accruals:List[float] = [x[0] for x in self.regression_data[symbol]]
114         accruals_factors:List[float] = [x[1] for x in self.regression_data[symbol]]
115         sales_deltas:List[float] = [x[2] for x in self.regression_data[symbol]]
116         ppes:List[float] = [x[3] for x in self.regression_data[symbol]]
117         # Regression.
118         x:List[List[float]] = [accruals_factors, sales_deltas, ppes]
119         regression_model = MultipleLinearRegression(x, total_accruals)
120         # x = [accruals_factors[:-1], sales_deltas[:-1], ppes[:-1]]
121         # regression_model = MultipleLinearRegression(x, total_accruals[1:])
122         alpha:float = regression_model.params[0]
123         residual[symbol] = alpha
124         
125     # Update accruals data and value.
126     self.accruals_data[symbol] = current_accruals_data
127     self.latest_accruals[symbol] = current_accruals
128     updated.append(symbol)
129 
130 # Make sure we ahve consecutive accruals data. 
131 symbols_to_remove:List[Symbol] = []
132 for symbol in self.accruals_data:
133     if symbol not in updated:
134         symbols_to_remove.append(symbol)
135 for symbol in symbols_to_remove:
136     del self.accruals_data[symbol]
137     del self.latest_accruals[symbol]
138 
139 sorted_by_residual:List[Tuple[Symbol, float]] = sorted(residual.items(), key = lambda x : x[1], reverse = True)
140 quartile:int = int(len(sorted_by_residual) / self.quantile)
141 high_by_residual:List[Symbol] = [x[0] for x in sorted_by_residual[:quartile]]
142 low_by_residual:List[Symbol] = [x[0] for x in sorted_by_residual[-quartile:]]
143 
144 self.high_residual_portfolios.append(high_by_residual)
145 self.low_residual_portfolios.append(low_by_residual)
146 
147 if len(self.high_residual_portfolios) == self.high_residual_portfolios.maxlen and len(self.low_residual_portfolios) == self.low_residual_portfolios.maxlen:
148         
149     self.long = [x[0] for x in residual.items() if x[0] in self.high_residual_portfolios[0] and x[0] in self.high_residual_portfolios[1]
150                                                 and x[0] in self.high_residual_portfolios[2] and x[0] in self.high_residual_portfolios[3]]
151     self.short = [x[0] for x in residual.items() if x[0] in self.low_residual_portfolios[0] and x[0] in self.low_residual_portfolios[1]
152                                                 and x[0] in self.low_residual_portfolios[2] and x[0] in self.low_residual_portfolios[3]]
153 
154 return self.long + self.short
155# Source: https://papers.ssrn.com/sol3/papers.cfm?abstract_id=3188172                
156def CalculateAccruals(self, current_accrual_data, prev_accrual_data) -> float:
157 delta_assets:float = current_accrual_data.CurrentAssets - prev_accrual_data.CurrentAssets
158 delta_cash:float = current_accrual_data.CashAndCashEquivalents - prev_accrual_data.CashAndCashEquivalents
159 delta_liabilities:float = current_accrual_data.CurrentLiabilities - prev_accrual_data.CurrentLiabilities
160 delta_debt:float = current_accrual_data.CurrentDebt - prev_accrual_data.CurrentDebt
161 dep:float = current_accrual_data.DepreciationAndAmortization
162 total_assets_prev_year:float = prev_accrual_data.TotalAssets
163 
164 acc:float = (delta_assets - delta_liabilities - delta_cash + delta_debt - dep) / total_assets_prev_year
165 return acc
166def OnData(self, data: Slice) -> None:
167 if not self.selection_flag:
168     return
169 self.selection_flag = False
170     
171# order execution
172 targets:List[PortfolioTarget] = []
173 for i, portfolio in enumerate([self.long, self.short]):
174     for symbol in portfolio:
175         if symbol in data and data[symbol]:
176             targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
177 
178 self.SetHoldings(targets, True)
179 self.long.clear()
180 self.short.clear()
181def Selection(self) -> None:
182 self.selection_flag = True
183class StockData():
184def __init__(self, current_assets:float, cash_and_cash_equivalents:float, current_liabilities:float, current_debt:float, income_tax_payable:float, 
185                 depreciation_and_amortization:float, total_assets:float, sales:float):
186 self.CurrentAssets:float = current_assets
187 self.CashAndCashEquivalents:float = cash_and_cash_equivalents
188 self.CurrentLiabilities:float = current_liabilities
189 self.CurrentDebt:float = current_debt
190 self.IncomeTaxPayable:float = income_tax_payable
191 self.DepreciationAndAmortization:float = depreciation_and_amortization
192 self.TotalAssets:float = total_assets
193 
194 self.Sales:float = sales
195def MultipleLinearRegression(x, y):
196x = np.array(x).T
197x = sm.add_constant(x)
198result = sm.OLS(endog=y, exog=x).fit()
199return result
200# Custom fee model
201class CustomFeeModel(FeeModel):
202def GetOrderFee(self, parameters):
203 fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
204 return OrderFee(CashAmount(fee, "USD"))