Stock Splits Strategy Based on Earnings Management

1from AlgorithmImports import *
2import numpy as np
3import statsmodels.api as sm
4from typing import List, Dict
5from numpy import isnan
6#endregion
7class StockSplitsStrategyBasedOnEarningsManagement(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100000)
11self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']
12  
13self.quantile: int = 3
14self.leverage: int = 5
15self.holding_period: int = 12
16self.lookback_period: int = 30
17self.fundamental_count: int = 500
18self.fundamental_sorting_key = lambda x: x.DollarVolume
19self.selection_flag: bool = False
20
21self.invested_symbols: Dict[Symbol, float] = {}
22self.accruals_data: Dict[Symbol, StockData] = {}
23self.data: Dict[Symbol, SymbolData] = {}
24self.splits_data: Dict[Symbol, datetime.date] = {}
25self.last_year_revenue: Dict[Symbol, float] = {}
26self.last_year_receivables: Dict[Symbol, float] = {}
27self.long: List[Symbol] = []
28self.short: List[Symbol] = []
29
30csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/splits.csv')
31lines: List[str] = csv_string_file.split('\r\n')
32for line in lines:
33    line_split: List[str] = line.split(';')
34    symbol: str = line_split[0]
35    
36    self.splits_data[symbol] = []
37    
38    for i in range(1, len(line_split)):
39        if line_split[i] is not '':
40            date: datetime.date = datetime.strptime(line_split[i], '%m/%d/%Y').date()
41            self.splits_data[symbol].append(date)
42
43self.settings.daily_precise_end_time = False
44self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
45self.UniverseSettings.Resolution = Resolution.Daily
46self.AddUniverse(self.FundamentalSelectionFunction)
47symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
48self.Schedule.On(self.DateRules.MonthEnd(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
49
50def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
51for security in changes.AddedSecurities:
52    security.SetFeeModel(CustomFeeModel())
53    security.SetLeverage(self.leverage)
54def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
55if not self.selection_flag:
56    return Universe.Unchanged
57
58# x.ValuationRatios.CFOPerShare and x.ValuationRatios.TotalAssetPerShare error:
59# Runtime Error: N7parquet38ParquetInvalidOrCorruptedFileExceptionE (message: 'Invalid: Parquet magic bytes not found in footer. 
60# Either the file is corrupted or this is not a parquet file.')
61selected: List[Fudnamental] = [
62    x for x in fundamental 
63    if x.HasFundamentalData 
64    and x.Market == 'usa'
65    and not isnan(x.FinancialStatements.BalanceSheet.CurrentAssets.Value) and x.FinancialStatements.BalanceSheet.CurrentAssets.Value > 0
66    and not isnan(x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value) and x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value > 0
67    and not isnan(x.FinancialStatements.BalanceSheet.CurrentLiabilities.Value) and x.FinancialStatements.BalanceSheet.CurrentLiabilities.Value > 0
68    and not isnan(x.FinancialStatements.BalanceSheet.CurrentDebt.Value) and x.FinancialStatements.BalanceSheet.CurrentDebt.Value > 0
69    and not isnan(x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value) and x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value > 0
70    and not isnan(x.FinancialStatements.BalanceSheet.GrossPPE.Value) and x.FinancialStatements.BalanceSheet.GrossPPE.Value > 0
71    and not isnan(x.ValuationRatios.CFOPerShare) and x.ValuationRatios.CFOPerShare > 0
72    and not isnan(x.ValuationRatios.TotalAssetPerShare) and x.ValuationRatios.TotalAssetPerShare > 0
73    and not isnan(x.FinancialStatements.BalanceSheet.AccountsReceivable.Value) and x.FinancialStatements.BalanceSheet.AccountsReceivable.Value > 0
74    and not isnan(x.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value) and x.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value > 0
75    and x.SecurityReference.ExchangeId in self.exchange_codes
76]
77if len(selected) > self.fundamental_count:
78    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
79
80market_cap: Dict[Symbol, float] = {}
81ram_residual: Dict[Symbol, float] = {}
82accruals_residual: Dict[Symbol, float] = {}
83
84current_date: datetime.date = self.Time.date()
85# We have only stocks with needed values for linear regressions, because fine was filtered
86for stock in selected:
87    symbol: Symbol = stock.Symbol
88
89    market_cap[symbol] = stock.MarketCap
90    if symbol not in self.data:
91        # Data for linear regressions 
92        self.data[symbol] = SymbolData()
93    
94    if symbol not in self.accruals_data:
95        # Data for previous year.
96        self.accruals_data[symbol] = None
97                
98    # Accrual calc.
99    current_accruals_data: StockData = StockData(
100        stock.FinancialStatements.BalanceSheet.CurrentAssets.Value, 
101        stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value,
102        stock.FinancialStatements.BalanceSheet.CurrentLiabilities.Value, 
103        stock.FinancialStatements.BalanceSheet.CurrentDebt.Value, 
104        stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.Value,
105        stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value, 
106        stock.FinancialStatements.BalanceSheet.TotalAssets.Value,
107        stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value
108        )
109    
110    current_year_receivables: float = stock.FinancialStatements.BalanceSheet.AccountsReceivable.Value
111    current_year_revenue: float = stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value
112    
113    if symbol not in self.last_year_receivables: # Need for change calculation of receivables
114        self.last_year_receivables[symbol] = None
115        
116    if symbol not in self.last_year_revenue: # Need for change calculation of revenue
117        self.last_year_revenue[symbol] = None
118    
119    # There is not previous accruals data.
120    if not self.accruals_data[symbol]:
121        self.accruals_data[symbol] = current_accruals_data
122        # Store current year values, which will figure as last year values next year
123        self.last_year_revenue[symbol] = current_year_revenue
124        self.last_year_receivables[symbol] = current_year_receivables
125        continue
126    
127    # Calculate change
128    receivables_change: float = current_year_receivables - self.last_year_receivables[symbol]
129    revenue_change: float = current_year_revenue - self.last_year_revenue[symbol]
130    
131    # Assign current values, which will figure as last year values next year
132    self.last_year_receivables[symbol] = current_year_receivables
133    self.last_year_revenue[symbol] = current_year_revenue
134    
135    # This is Y in accruals linear regression    
136    current_accruals: float = self.CalculateAccruals(current_accruals_data, self.accruals_data[symbol])
137    self.data[symbol].accruals.append(current_accruals)
138    
139    # This is Y in RAM linear regression
140    self.data[symbol].cfo_per_share.append(stock.ValuationRatios.CFOPerShare)
141    
142    # These are x's for linear regressions 
143    self.data[symbol].total_revenue.append(stock.FinancialStatements.IncomeStatement.TotalRevenueAsReported.Value) 
144    self.data[symbol].gross_ppe.append(stock.FinancialStatements.BalanceSheet.GrossPPE.Value) 
145    self.data[symbol].total_asset_per_share.append(stock.ValuationRatios.TotalAssetPerShare) 
146    
147    # These are x's for linear regressions, which has to be calculated as a change
148    self.data[symbol].total_revenue_change.append(revenue_change)
149    self.data[symbol].accounts_receivable_change.append(receivables_change) 
150    
151    # This substraction is needed for Accruals regression
152    revenues_minus_receivables: List[float] = []
153    for x, y in zip(self.data[symbol].total_revenue_change, self.data[symbol].accounts_receivable_change):
154        revenues_minus_receivables.append(x - y)
155    
156    # Accruals regression
157    regression_x: List[List[float]] = [
158        [1 / x for x in self.data[symbol].total_asset_per_share], # 1 / TA
159        [x for x in revenues_minus_receivables],                  # delta SALES - delta REC
160        [x for x in self.data[symbol].gross_ppe]                  # PPE
161        ] 
162    
163    regression_y: List[float] = [x for x in self.data[symbol].accruals] 
164         
165    regression_model: RegressionResultWrapper = self.MultipleLinearRegression(regression_x, regression_y, True)
166    
167    accruals_residual[symbol] = regression_model.resid[-1]
168    
169    # RAM regression
170    regression_x: List[List[float]] = [
171        [1 / x for x in self.data[symbol].total_asset_per_share], # 1 / TA
172        [x for x in self.data[symbol].total_revenue],             # SALES
173        [x for x in self.data[symbol].total_revenue_change]       # delta SALES
174        ] 
175                    
176    regression_y: List[float] = [x for x in self.data[symbol].cfo_per_share]
177    
178    regression_model: RegressionResultWrapper = self.MultipleLinearRegression(regression_x, regression_y, False)
179    
180    ram_residual[symbol] = regression_model.resid[-1]
181
182long: List[Symbol] = []
183short: List[Symbol] = []
184# Sort residuals from regressions into terciles
185if len(ram_residual) >= self.quantile * 2:
186    sorted_by_ram: List[Symbol] = [x[0] for x in sorted(ram_residual.items(), key=lambda item: item[1])]
187    sorted_by_accruals: List[Symbol] = [x[0] for x in sorted(accruals_residual.items(), key=lambda item: item[1])]
188    
189    quantile: int = int(len(sorted_by_ram) / self.quantile)
190    
191    long: List[Symbol] = [x for x in sorted_by_ram[:quantile] if x in sorted_by_accruals[-quantile:]] # Low ram and high accruals
192    short: List[Symbol] = [x for x in sorted_by_ram[-quantile:] if x in sorted_by_accruals[:quantile]] # High ram and low accruals
193
194for symbol in long:
195    if symbol.Value in self.splits_data and symbol not in self.invested_symbols:
196        if self.CheckStockSplitDate(symbol.Value, current_date-timedelta(days=self.lookback_period), current_date):
197            self.long.append(symbol)
198
199for symbol in short:
200    if symbol.Value in self.splits_data and symbol not in self.invested_symbols:
201        if self.CheckStockSplitDate(symbol.Value, current_date-timedelta(days=self.lookback_period), current_date):
202            self.short.append(symbol)
203            
204return self.long + self.short
205def OnData(self, data: Slice) -> None:
206if not self.selection_flag:
207    return
208self.selection_flag = False
209
210# Trade execution.
211for symbol in self.long + self.short:
212    if symbol not in self.invested_symbols:
213        self.invested_symbols[symbol] = 0
214    else:
215        self.invested_symbols[symbol] += 1
216        # Remove stock, because we held it for 12 months
217        if self.invested_symbols[symbol] == self.holding_period:
218            del self.invested_symbols[symbol]
219            self.Liquidate(symbol)
220            
221            if symbol in self.long:
222                self.long.remove(symbol)
223            else:
224                self.short.remove(symbol)
225        
226# trade execution
227targets: List[PortfolioTarget] = []
228for i, portfolio in enumerate([self.long, self.short]):
229    for symbol in portfolio:
230        if symbol in data and data[symbol]:
231            targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
232self.SetHoldings(targets, True)
233        
234def CheckStockSplitDate(self, 
235                    ticker: str, 
236                    date_from: datetime.date, 
237                    date_to: datetime.date) -> bool:
238if len(self.splits_data[ticker]) > 0:
239    for split_date in self.splits_data[ticker]:
240        if date_from <= split_date <= date_to:
241            return True
242return False
243
244def Selection(self) -> None:
245self.selection_flag = True
246
247# Source: https://papers.ssrn.com/sol3/papers.cfm?abstract_id=3188172                
248def CalculateAccruals(self, current_accrual_data, prev_accrual_data) -> float:
249delta_assets: float = current_accrual_data.CurrentAssets - prev_accrual_data.CurrentAssets
250delta_cash: float = current_accrual_data.CashAndCashEquivalents - prev_accrual_data.CashAndCashEquivalents
251delta_liabilities: float = current_accrual_data.CurrentLiabilities - prev_accrual_data.CurrentLiabilities
252delta_debt: float = current_accrual_data.CurrentDebt - prev_accrual_data.CurrentDebt
253dep: float = current_accrual_data.DepreciationAndAmortization
254total_assets_prev_year: float = prev_accrual_data.TotalAssets
255
256acc: float = (delta_assets - delta_liabilities - delta_cash + delta_debt - dep) / total_assets_prev_year
257return acc
258
259def MultipleLinearRegression(self, x: List[List[float]], y: List[float], add_intercept: bool):
260x: np.ndarray = np.array(x).T
261if add_intercept: # One regression has intercept, the other one hasn't
262    x = sm.add_constant(x)
263result: RegressionResultWrapper = sm.OLS(endog=y, exog=x).fit()
264return result        
265
266# Custom fee model
267class CustomFeeModel(FeeModel):
268def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
269fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
270return OrderFee(CashAmount(fee, "USD"))
271class StockData():
272def __init__(self, 
273        current_assets: float, 
274        cash_and_cash_equivalents: float, 
275        current_liabilities: float, 
276        current_debt: float, 
277        income_tax_payable: float, 
278        depreciation_and_amortization: float, 
279        total_assets: float, 
280        sales: float) -> None:
281self.CurrentAssets: float = current_assets
282self.CashAndCashEquivalents: float = cash_and_cash_equivalents
283self.CurrentLiabilities: float = current_liabilities
284self.CurrentDebt: float = current_debt
285self.IncomeTaxPayable: float = income_tax_payable
286self.DepreciationAndAmortization: float = depreciation_and_amortization
287self.TotalAssets: float = total_assets
288self.Sales: float = sales
289
290class SymbolData():
291def __init__(self) -> None:
292self.cfo_per_share: List[float] = []
293self.accruals: List[float] = []
294self.total_revenue: List[float] = []
295self.total_revenue_change: List[float] = []
296self.total_asset_per_share: List[float] = []
297self.accounts_receivable_change: List[float] = []
298self.gross_ppe: List[float] = []