Earnings Acceleration Effect in Stocks

1from AlgorithmImports import *
2from typing import Dict, List
3from numpy import isnan
4class EarningsAccelerationEffectinStocks(QCAlgorithm):
5
6def Initialize(self):
7self.SetStartDate(2004, 1, 1)
8self.SetCash(100000)
9self.exchange_codes:List[str] = ['NYS', 'NAS', 'ASE']	
10
11self.leverage:int = 5
12self.quantile:int = 10
13self.min_share_price:int = 5
14self.quarters_count:int = 6       # Number of quarters to calculate the earning acceleration indicator
15self.fundamental_count:int = 1000
16self.fundamental_sorting_key = lambda x: x.DollarVolume
17
18self.weight:Dict[Symbol, float] = {}
19self.eps_by_symbol:Dict[Symbol, RollingWindow] = {}       # Contains RollingWindow objects for every stock
20self.last_fundamental:List[Symbol] = []
21
22self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
23
24self.selection_flag:bool = False
25self.settings.daily_precise_end_time = False
26self.settings.minimum_order_margin_portfolio_percentage = 0.
27self.UniverseSettings.Resolution = Resolution.Daily
28self.AddUniverse(self.FundamentalSelectionFunction)
29self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
30
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(CustomFeeModel())
34    security.SetLeverage(self.leverage)
35def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
36'''Drop securities which have no fundamental data or have too low prices.
37Select those with the highest dollar volume'''
38
39if not self.selection_flag:
40    return Universe.Unchanged
41selected:List[Fundamental] = [
42    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.MarketCap != 0 \
43    and not isnan(x.EarningReports.BasicEPS.ThreeMonths) and x.EarningReports.BasicEPS.ThreeMonths > 0 \
44    and x.SecurityReference.ExchangeId in self.exchange_codes
45]
46
47if len(selected) > self.fundamental_count:
48    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
49
50ea_by_stock:Dict[Symbol, float] = {}
51
52for stock in selected:
53    symbol:Symbol = stock.Symbol
54    
55    # If stock wasn't in last_fundamental we need to reintialize it to make sure data are consecutive
56    if symbol not in self.eps_by_symbol or symbol not in self.last_fundamental:
57        self.eps_by_symbol[symbol] = RollingWindow[float](self.quarters_count)
58        
59    # update rolling window for every stock
60    self.eps_by_symbol[symbol].Add(stock.EarningReports.BasicEPS.ThreeMonths)
61    if self.eps_by_symbol[symbol].IsReady:
62        rw:List[float] = [x for x in self.eps_by_symbol[symbol]]
63        eps_fraction1:float = (rw[0] - rw[4]) / rw[1]
64        eps_fraction2:float = (rw[1] - rw[5]) / rw[2]
65        ea_by_stock[stock] = eps_fraction1 - eps_fraction2 # That's the earnings acceleration we want
66
67sorted_by_ea:List[Fundamental] = [x[0] for x in sorted(ea_by_stock.items(), key = lambda item: item[1], reverse = True)]
68# Create long and short quantile
69quantile:int = int(len(sorted_by_ea) / self.quantile)
70long:List[Fundamental] = sorted_by_ea[:quantile]
71short:List[Fundamental] = sorted_by_ea[-quantile:]
72
73# Calculate weight for each stock in portfolio
74for i, portfolio in enumerate([long, short]):
75    mc_sum:float = sum(list(map(lambda stock: stock.MarketCap, portfolio)))
76    for stock in portfolio:
77        self.weight[stock.Symbol] = ((-1) ** i) * stock.MarketCap /  mc_sum
78
79# Change last fundamental to make sure data are consecutive
80self.last_fundamental = [x.Symbol for x in fundamental]
81    
82return list(self.weight.keys())
83def OnData(self, data: Slice) -> None:
84if not self.selection_flag:
85    return
86self.selection_flag = False
87
88# Trade execution.
89portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
90self.SetHoldings(portfolio, True)
91self.weight.clear()
92
93def Selection(self) -> None:
94self.selection_flag = True
95
96# Custom fee model
97class CustomFeeModel(FeeModel):
98def GetOrderFee(self, parameters):
99fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
100return OrderFee(CashAmount(fee, "USD"))