Market Timing with Aggregate Accruals

1from AlgorithmImports import *
2from collections import deque
3import numpy as np
4from scipy import stats
5from math import sqrt
6#endregion
7class MarketTimingAggregateAccruals(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100000)
11self.sp500_stocks:List = []
12
13# monthly market prices
14self.market_monthly_data = deque(maxlen = 12)
15# latest accruals data
16self.accrual_data:Dict[Symbol, float] = {}
17
18self.coarse_count:int = 500
19self.year_period:int = 10
20self.gamma:float = 5.
21self.leverage:int = 5
22self.leverage_cap:float = 3.
23
24self.aggregate_accruals = deque(maxlen = self.year_period + 1)
25
26# yearly market data -> price and variance pair
27self.market_yearly_data = deque(maxlen = self.year_period + 1)
28
29self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
30self.tbills:Symbol = self.AddEquity('BIL', Resolution.Daily).Symbol
31
32self.selection_flag:bool = False
33self.UniverseSettings.Resolution = Resolution.Daily
34self.Settings.MinimumOrderMarginPortfolioPercentage = 0
35self.AddUniverse(self.FundamentalSelectionFunction)
36self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
37self.settings.daily_precise_end_time = False
38def Selection(self) -> None:
39# store market price
40if self.Securities.ContainsKey(self.market) and self.Securities[self.market]:
41    self.market_monthly_data.append(self.Securities[self.market].Price)
42    
43if self.Time.month == 4:
44    self.selection_flag = True
45    if len(self.market_monthly_data) == self.market_monthly_data.maxlen:
46        # store yearly market data
47        yearly_volatility:float = (self.Volatility([x for x in self.market_monthly_data]) * sqrt(len(self.market_monthly_data)))
48        variance:float = yearly_volatility ** 2
49        self.market_yearly_data.append((self.Securities[self.market].Price, variance))
50
51def OnSecuritiesChanged(self, changes):
52for security in changes.AddedSecurities:
53    security.SetFeeModel(CustomFeeModel())
54    security.SetLeverage(self.leverage)
55for security in changes.RemovedSecurities:
56    if security.Symbol in self.accrual_data:
57        del self.accrual_data[security.Symbol]
58def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
59if not self.selection_flag:
60    return Universe.Unchanged
61# selected = [x.Symbol for x in coarse if x.HasFundamentalData and x.Market == 'usa']
62selected = [x for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
63                   and (x.FinancialStatements.BalanceSheet.CurrentAssets.HasValue and \
64                        x.FinancialStatements.BalanceSheet.CashAndCashEquivalents.HasValue and \
65                        x.FinancialStatements.BalanceSheet.CurrentLiabilities.HasValue and \
66                        x.FinancialStatements.BalanceSheet.CurrentDebt.HasValue and \
67                        x.FinancialStatements.BalanceSheet.IncomeTaxPayable.HasValue and \
68                        x.FinancialStatements.IncomeStatement.DepreciationAndAmortization.HasValue)],
69        key = lambda x: x.DollarVolume, reverse = True)[:self.coarse_count]]
70accruals_market_cap:dict[Symbol, Tuple] = {}
71for stock in selected:
72    symbol:Symbol = stock.Symbol
73    if symbol not in self.accrual_data:
74        self.accrual_data[symbol] = None
75        
76    # accrual calculation
77    current_accruals_data:AccrualsData = AccrualsData(stock.FinancialStatements.BalanceSheet.CurrentAssets.Value, stock.FinancialStatements.BalanceSheet.CashAndCashEquivalents.Value,
78                                                    stock.FinancialStatements.BalanceSheet.CurrentLiabilities.Value, stock.FinancialStatements.BalanceSheet.CurrentDebt.Value, stock.FinancialStatements.BalanceSheet.IncomeTaxPayable.Value,
79                                                    stock.FinancialStatements.IncomeStatement.DepreciationAndAmortization.Value, stock.FinancialStatements.BalanceSheet.TotalAssets.Value)
80    
81    # there is not previous accrual data
82    if not self.accrual_data[symbol]:
83        self.accrual_data[symbol] = current_accruals_data
84        continue
85    
86    # accruals and market cap calculation
87    accruals:float = self.CalculateAccruals(current_accruals_data, self.accrual_data[symbol])
88    market_cap:float = stock.MarketCap
89    accruals_market_cap[symbol] = (accruals, market_cap)
90
91    # update accruals data
92    self.accrual_data[symbol] = current_accruals_data
93if len(accruals_market_cap) == 0: return Universe.Unchanged
94
95# value weighted accruals calculation
96total_market_cap:float = sum([x[1][1] for x in accruals_market_cap.items()])
97weighted_accruals_data:List[float] = []
98for symbol, accruals_and_cap in accruals_market_cap.items():
99    weight:float = accruals_and_cap[1] / total_market_cap
100    weighted_accruals_data.append(accruals_and_cap[0] * weight)
101    
102aggregate_accruals:float = sum([x for x in weighted_accruals_data])
103self.aggregate_accruals.append(aggregate_accruals)
104return list(self.accrual_data.keys())
105def OnData(self, data:Slice) -> None:
106if not self.selection_flag:
107    return
108self.selection_flag = False
109# 10 years of accruals history is ready
110if len(self.market_yearly_data) == self.market_yearly_data.maxlen and \
111    len(self.aggregate_accruals) == self.aggregate_accruals.maxlen:
112    # Regression calc.
113    market_prices:np.ndarray = np.array([x[0] for x in self.market_yearly_data])
114    market_returns:np.ndarray = (market_prices[1:] - market_prices[:-1]) / market_prices[:-1]
115    
116    # shift values for regression to predict future return
117    market_returns:np.ndarray = market_returns[-(self.year_period - 1):]
118    accruals:List[float] = [x for x in self.aggregate_accruals]
119    regr_accruals = accruals[:(self.year_period - 1)]
120    
121    # Simple Linear Regression
122    # Y = α + (β ∗ X)
123    slope, intercept, r_value, p_value, std_err = stats.linregress(regr_accruals, market_returns)
124    expected_return:float = intercept + slope * accruals[-1]
125    # predict variance
126    yearly_variances:List[float] = [x[1] for x in self.market_yearly_data]
127    yearly_variances = yearly_variances[-(self.year_period - 1):]
128    
129    slope, intercept, r_value, p_value, std_err = stats.linregress(regr_accruals, yearly_variances)
130    expected_variance:float = intercept + slope * accruals[-1]
131    
132    market_weight:float = expected_return / (expected_variance * self.gamma)
133    market_weight = max(min(market_weight, self.leverage_cap), -self.leverage_cap) # leverage cap
134    tbills_weight:float = 1. - abs(market_weight) if abs(market_weight) < 1. else 0.
135            
136    # trade execution
137    if self.market in data and data[self.market] and self.tbills in data and data[self.tbills]:
138        self.SetHoldings(self.market, market_weight)
139        self.SetHoldings(self.tbills, tbills_weight)
140
141def Volatility(self, values) -> float:
142values:np.ndarray = np.array(values)
143returns:np.ndarray = (values[1:] - values[:-1]) / values[:-1]
144return np.std(returns)
145
146def CalculateAccruals(self, current_accrual_data, prev_accrual_data) -> float:
147delta_assets:float = current_accrual_data.CurrentAssets - prev_accrual_data.CurrentAssets
148delta_cash:float = current_accrual_data.CashAndCashEquivalents - prev_accrual_data.CashAndCashEquivalents
149delta_liabilities:float = current_accrual_data.CurrentLiabilities - prev_accrual_data.CurrentLiabilities
150delta_debt:float = current_accrual_data.CurrentDebt - prev_accrual_data.CurrentDebt
151delta_tax:float = current_accrual_data.IncomeTaxPayable - prev_accrual_data.IncomeTaxPayable
152dep:float = current_accrual_data.DepreciationAndAmortization
153avg_total:float = (current_accrual_data.TotalAssets + prev_accrual_data.TotalAssets) / 2
154
155bs_acc:float = ((delta_assets - delta_cash) - (delta_liabilities - delta_debt - delta_tax) - dep)
156return bs_acc
157# custom fee model
158class CustomFeeModel(FeeModel):
159def GetOrderFee(self, parameters):
160fee:float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
161return OrderFee(CashAmount(fee, "USD"))
162class AccrualsData():
163def __init__(self, current_assets:float, cash_and_cash_equivalents:float, current_liabilities:float, current_debt:float, income_tax_payable:float, depreciation_and_amortization:float, total_assets:float):
164self.CurrentAssets:float = current_assets
165self.CashAndCashEquivalents:float = cash_and_cash_equivalents
166self.CurrentLiabilities:float = current_liabilities
167self.CurrentDebt:float = current_debt
168self.IncomeTaxPayable:float = income_tax_payable
169self.DepreciationAndAmortization:float = depreciation_and_amortization
170self.TotalAssets:float = total_assets