Quarterly Investment Spikes Predict Stock Returns

1from AlgorithmImports import *
2
3class QuarterlyInvestmentSpikesPredictStockReturns(QCAlgorithm):
4
5def Initialize(self):
6self.SetStartDate(2000, 1, 1)
7self.SetCash(100000)
8
9self.period:int = 4             # need n quarterly capex growths
10self.leverage:int = 10
11self.quantile:int = 3
12self.selection_months = [4, 7, 10, 1]   # MonthStart
13self.total_asset_threshold:int = 1e7
14self.capex_threshold:int = 1e5
15
16self.weight:Dict[Symbol, float] = {}
17self.capex:Dict[Symbol, RollingWindow] = {}
18self.last_fine:List[Symbol] = []
19
20market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
21
22self.rebalance_flag:bool = False
23self.selection_flag:bool = False
24self.UniverseSettings.Resolution = Resolution.Daily
25self.AddUniverse(self.FundamentalSelectionFunction)
26self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
27self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.BeforeMarketClose(market, 0), self.Selection)
28
29def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
30for security in changes.AddedSecurities:
31    security.SetFeeModel(CustomFeeModel())
32    security.SetLeverage(self.leverage)
33
34def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
35if not self.selection_flag:
36    return Universe.Unchanged
37self.selection_flag = False
38
39# filter stock symbols
40selected:List[Fundamental] = [x for x in fundamental if x.MarketCap != 0 and \
41    x.FinancialStatements.CashFlowStatement.CapExReported.ThreeMonths >= self.capex_threshold and \
42    x.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths >= self.total_asset_threshold
43    ]
44
45qspikes:Dict[Fundamental, float] = {} # storing stocks qspike keyed by stocks objects
46    
47for stock in selected:
48    symbol:Symbol = stock.Symbol
49    capex:float = stock.FinancialStatements.CashFlowStatement.CapExReported.ThreeMonths
50    
51    # make sure data are consecutive
52    if symbol not in self.last_fine:
53        self.capex[symbol] = RollingWindow[float](self.period)
54    self.capex[symbol].Add(capex)
55    
56    # calculate qspike only on rebalance and when data are ready
57    if self.rebalance_flag and self.capex[symbol].IsReady:
58        capexes:List[float] = list(self.capex[symbol])
59        
60        # get last quarter of capex growth
61        last_capex:float = capexes[0]
62        # calculate average from first three quarters of capex growth
63        avg_from_first_three:float = np.average(capexes[1:4])
64        
65        # calculate and store stock's qspike
66        qspikes[stock] = last_capex / avg_from_first_three
67
68# change last fine universe based on current fine
69self.last_fine = list(map(lambda x: x.Symbol, selected))
70        
71# not enough stocks for selection
72if len(qspikes) < self.quantile:
73    return Universe.Unchanged
74
75# perform selection
76# select stocks based on 30th percentile and 70th percentile 
77quantile:int = int(len(qspikes) / self.quantile)
78sorted_by_qspike = [x[0] for x in sorted(qspikes.items(), key=lambda item: item[1])]
79
80# long low portfolio
81long = sorted_by_qspike[:quantile]
82# short high portfolio
83short = sorted_by_qspike[-quantile:]
84
85# calculate weights
86for i, portfolio in enumerate([long, short]):
87    mc_sum:float = sum(map(lambda x: x.MarketCap, portfolio))
88    for stock in portfolio:
89        self.weight[stock.Symbol] = ((-1) ** i) * (stock.MarketCap / mc_sum)
90
91return list(self.weight.keys())
92
93def OnData(self, data: Slice) -> None:
94# rebalance yearly
95if not self.rebalance_flag:
96    return
97self.rebalance_flag = False
98
99# trade execution
100portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
101self.SetHoldings(portfolio, True)
102
103self.weight.clear()
104
105def Selection(self) -> None:
106if self.Time.month in self.selection_months:
107    self.selection_flag = True
108
109# rebalance yearly on last selection month
110if self.Time.month == self.selection_months[-1]:
111    self.rebalance_flag = True
112
113# Custom fee model
114class CustomFeeModel(FeeModel):
115def GetOrderFee(self, parameters):
116fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
117return OrderFee(CashAmount(fee, "USD"))