Accruals Effect Combined with Price Momentum

1from numpy import floor, isnan
2from AlgorithmImports import *
3from typing import List, Dict
4import data_tools
5class AccrualsEffectPriceMomentum(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9
10self.exchange_codes:List[str] = ['NYS', 'ASE']	
11self.long:List[Symbol] = []
12self.short:List[Symbol] = []
13
14self.quantile:int = 5
15self.leverage:int = 5
16self.min_share_price:int = 5
17self.months:int = 0
18self.period:int = 6 * 21
19self.holding_period:int = 6
20
21self.data:Dict[Symbol, data_tools.SymbolData] = {}
22self.managed_queue:List[data_tools.RebalanceQueueItem] = []
23
24# Latest accurals data
25self.accural_data:Dict[Symbol, data_tools.AccuralsData] = {}
26
27self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
28
29self.fundamental_count:int = 1000
30self.fundamental_sorting_key = lambda x: x.MarketCap
31self.selection_flag:bool = False
32self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
33self.UniverseSettings.Resolution = Resolution.Daily
34self.AddUniverse(self.FundamentalSelectionFunction)
35self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
36self.settings.daily_precise_end_time = False
37def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
38for security in changes.AddedSecurities:
39    security.SetFeeModel(data_tools.CustomFeeModel())
40    security.SetLeverage(self.leverage)
41
42def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
43# Update the rolling window every day.
44for stock in fundamental:
45    symbol:Symbol = stock.Symbol
46    # Store monthly price.
47    if symbol in self.data:
48        self.data[symbol].update(stock.AdjustedPrice)
49if not self.selection_flag:
50    return Universe.Unchanged
51selected:List[Fundamental] = [
52    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' and x.MarketCap != 0
53    and not isnan(x.FinancialStatements.BalanceSheet.CurrentAssets.ThreeMonths) and (x.FinancialStatements.BalanceSheet.CurrentAssets.ThreeMonths > 0) \
54    and not isnan(x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths) and (x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths) > 0 \
55    and not isnan(x.FinancialStatements.BalanceSheet.CurrentLiabilities.ThreeMonths) and (x.FinancialStatements.BalanceSheet.CurrentLiabilities.ThreeMonths) > 0 \
56    and not isnan(x.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths) and (x.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths) > 0 \
57    and not isnan(x.FinancialStatements.BalanceSheet.IncomeTaxPayable.ThreeMonths) and (x.FinancialStatements.BalanceSheet.IncomeTaxPayable.ThreeMonths) > 0 \
58    and not isnan(x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths) and (x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths) > 0 \
59    and x.SecurityReference.ExchangeId in self.exchange_codes
60]
61if len(selected) > self.fundamental_count:
62    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
63# Warmup price rolling windows.
64for stock in selected:
65    symbol:Symbol = stock.Symbol
66    if symbol in self.data:
67        continue
68    self.data[symbol] = data_tools.SymbolData(self.period)
69    history = self.History(symbol, self.period, Resolution.Daily)
70    if history.empty:
71        self.Log(f"Not enough data for {symbol} yet")
72        continue
73    closes = history.loc[symbol].close
74    for time, close in closes.items():
75        self.data[symbol].update(close)
76        
77bs_acc:Dict[Symbol, float] = {}
78momentum:Dict[Symbol, float] = {} 
79current_accurals_data:Dict[Symbol, data_tools.AccuralsData] = {}
80
81for stock in selected:
82    symbol = stock.Symbol
83    
84    if not self.data[symbol].is_ready():
85        continue
86    momentum[symbol] = self.data[symbol].performance()
87    
88    # Accural calc
89    current_accurals_data[symbol] = data_tools.AccuralsData(stock.FinancialStatements.BalanceSheet.CurrentAssets.ThreeMonths, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.ThreeMonths,
90                                                stock.FinancialStatements.BalanceSheet.CurrentLiabilities.ThreeMonths, stock.FinancialStatements.BalanceSheet.CurrentDebt.ThreeMonths, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.ThreeMonths,
91                                                stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.ThreeMonths, stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths)
92
93    if symbol in self.accural_data:
94        bs_acc[symbol] = self.CalculateAccurals(current_accurals_data[symbol], self.accural_data[symbol])
95
96# Clear old accruals and set new ones 
97self.accural_data.clear()
98for symbol in current_accurals_data:
99    self.accural_data[symbol] = current_accurals_data[symbol]
100
101long:List[Symbol] = []
102short:List[Symbol] = []
103    
104if len(momentum) != 0 and len(bs_acc) != 0:
105    # Momentum sorting
106    sorted_by_mom:List[Tuple[Symbol, float]] = sorted(momentum.items(), key = lambda x: x[1], reverse = True)
107    quintile:int = int(len(sorted_by_mom) / self.quantile)
108    top_by_mom:List[Symbol] = [x[0] for x in sorted_by_mom[:quintile]]
109    low_by_mom:List[Symbol] = [x[0] for x in sorted_by_mom[-quintile:]]
110
111    # Accural sorting
112    sorted_by_acc:List[Tuple[Symbol, float]] = sorted(bs_acc.items(), key = lambda x: x[1], reverse = True)
113    quintile:int = int(len(sorted_by_acc) / self.quantile)
114    top_by_acc:List[Symbol] = [x[0] for x in sorted_by_acc[:quintile]]
115    low_by_acc:List[Symbol] = [x[0] for x in sorted_by_acc[-quintile:]]
116    
117    long = [x for x in top_by_mom if x in low_by_acc]
118    short = [x for x in low_by_mom if x in top_by_acc]
119    
120    if len(long) != 0:
121        long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
122        # symbol/quantity collection
123        long_symbol_q:List[Tuple[Symbol, int]] = [(x, floor(long_w / self.data[x].LastPrice)) for x in long]
124    else:
125        long_symbol_q = []
126
127    if len(short) != 0:
128        short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
129        # symbol/quantity collection
130        short_symbol_q:List[Tuple[Symbol, int]] = [(x, floor(short_w / self.data[x].LastPrice)) for x in short]
131    else:
132        short_symbol_q = []
133        
134    self.managed_queue.append(data_tools.RebalanceQueueItem(long_symbol_q, short_symbol_q))
135
136return long + short
137
138def OnData(self, data: Slice) -> None:
139if not self.selection_flag:
140    return
141self.selection_flag = False
142
143remove_item = None
144
145# Rebalance portfolio
146for item in self.managed_queue:
147    if item.holding_period == self.holding_period + 1: # All portfolios are held for six months (month 2 to 7)
148        
149        # Selling long on Liquidate
150        for symbol, quantity in item.long_symbol_q:
151            self.MarketOrder(symbol, -quantity)
152                    
153        # Buying short on Liquidate
154        for symbol, quantity in item.short_symbol_q:
155            self.MarketOrder(symbol, quantity)
156        
157        remove_item = item
158    
159    # Trade execution    
160    if item.holding_period == 1: # All portfolios are held for six months (month 2 to 7)
161        open_long_symbol_q:List[Tuple[Symbol, int]] = []
162        open_short_symbol_q:List[Tuple[Symbol, int]] = []
163        
164        for symbol, quantity in item.long_symbol_q:
165            if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
166                self.MarketOrder(symbol, quantity)
167                open_long_symbol_q.append((symbol, quantity))
168                    
169        for symbol, quantity in item.short_symbol_q:
170            if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
171                self.MarketOrder(symbol, -quantity)
172                open_short_symbol_q.append((symbol, quantity))
173        
174        # Only opened orders will be closed        
175        item.long_symbol_q = open_long_symbol_q
176        item.short_symbol_q = open_short_symbol_q
177        
178    item.holding_period += 1
179    
180# We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue.
181if remove_item:
182    self.managed_queue.remove(remove_item)
183
184def Selection(self) -> None:
185self.selection_flag = True
186def CalculateAccurals(self, current_accural_data, prev_accural_data):
187delta_assets:float = current_accural_data.CurrentAssets - prev_accural_data.CurrentAssets
188delta_cash:float = current_accural_data.CashAndCashEquivalents - prev_accural_data.CashAndCashEquivalents
189delta_liabilities:float = current_accural_data.CurrentLiabilities - prev_accural_data.CurrentLiabilities
190delta_debt:float = current_accural_data.CurrentDebt - prev_accural_data.CurrentDebt
191delta_tax:float = current_accural_data.IncomeTaxPayable - prev_accural_data.IncomeTaxPayable
192dep:float = current_accural_data.DepreciationAndAmortization
193avg_total:float = (current_accural_data.TotalAssets + prev_accural_data.TotalAssets) / 2
194
195bs_acc:float = ((delta_assets - delta_cash) - (delta_liabilities - delta_debt-delta_tax) - dep) / avg_total
196return bs_acc