Earnings Response Elasticity

1from AlgorithmImports import *
2import numpy as np
3from collections import deque
4from pandas.tseries.offsets import BDay
5from trade_manager import TradeManager
6from typing import List, Dict, Deque
7from pandas.core.frame import DataFrame
8from pandas.core.series import Series
9#endregion
10class EarningsResponseElasticity(QCAlgorithm):
11def Initialize(self) -> None:
12self.SetStartDate(2010, 1, 1)
13self.SetCash(100000)
14self.period: int = 13
15self.ear_period: int = 4
16self.surprise_period: int = 4
17self.holding_period: int = 60
18self.long_size: int = 50
19self.short_size: int = 50
20self.leverage: int = 5
21self.threshold: int = 3
22self.percentile: int = 20
23
24# market daily price data
25self.market_prices: Deque = deque(maxlen = self.ear_period)
26
27self.earnings_surprise: Dict[Symbol, float] = {}
28self.long: List[Symbol] = []
29self.short: List[Symbol] = []
30self.ere_history_previous: List[float] = []
31self.ere_history_actual: List[float] = []
32self.eps: Dict[Symbol, deque] = {} 
33self.earnings_data: Dict[datetime.date, Dict[str, float]] = {}
34
35# 50 equally weighted brackets for traded symbols
36self.trade_manager: trade_manager.TradeManager = TradeManager(self, self.long_size, self.short_size, self.holding_period)
37self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
38earnings_data: str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
39earnings_data_json: List[dict] = json.loads(earnings_data)
40
41for obj in earnings_data_json:
42    date: datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
43    self.earnings_data[date] = {}
44    
45    for stock_data in obj['stocks']:
46        ticker: str = stock_data['ticker']
47        if stock_data['eps'] != '':
48            self.earnings_data[date][ticker] = float(stock_data['eps'])
49
50self.month: int = 12
51self.settings.daily_precise_end_time = False
52self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
53self.UniverseSettings.Resolution = Resolution.Daily
54self.AddUniverse(self.FundamentalSelectionFunction)
55self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
56def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
57for security in changes.AddedSecurities:
58    security.SetFeeModel(CustomFeeModel())
59    security.SetLeverage(self.leverage)
60
61def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
62# stocks with yesterday's earnings
63yesterday: datetime.date = (self.Time - BDay(1)).date()
64if yesterday not in self.earnings_data:
65    return Universe.Unchanged
66tickers_with_yesterday_earnings: List[str] = list(self.earnings_data[yesterday].keys())
67# stocks with yesterday's earnings
68selected: List[Fundamental] = [x for x in fundamental if x.Symbol.Value in tickers_with_yesterday_earnings]
69# ERE data
70ere_data: Dict[Symbol, List[float, float, float]] = {}
71
72for stock in selected:
73    symbol: Symbol = stock.Symbol
74    ticker: str = symbol.Value
75    
76    # Store eps data.
77    if symbol not in self.eps:
78        self.eps[symbol] = deque(maxlen = self.period)
79    self.eps[symbol].append([yesterday, self.earnings_data[yesterday][ticker]])
80    if len(self.eps[symbol]) == self.eps[symbol].maxlen:
81        year_range: range = range(self.Time.year - 3, self.Time.year)
82        month_range: List[int] = [self.Time.month-1, self.Time.month, self.Time.month+1]
83        
84        # earnings 4 years back
85        seasonal_eps_data: List[List[datetime.date, float]] = [x for x in self.eps[symbol] if x[0].month in month_range and x[0].year in year_range]
86        if len(seasonal_eps_data) != self.threshold: continue
87        recent_eps_data: List[datetime.date, float] = self.eps[symbol][-1]
88        
89        # Make sure we have a consecutive seasonal data. Same months with one year difference.
90        year_diff: np.ndarray = np.diff([x[0].year for x in seasonal_eps_data])
91        if all(x == 1 for x in year_diff):
92            seasonal_eps: List[float] = [x[1] for x in seasonal_eps_data]
93            diff_values: np.ndarray = np.diff(seasonal_eps)
94            drift: float = np.average(diff_values)
95            
96            # SUE calculation
97            last_earnings: float = seasonal_eps[-1]
98            expected_earnings: float = last_earnings + drift
99            actual_earnings: float = recent_eps_data[1]
100            # store sue value with earnigns date
101            earnings_surprise: float = actual_earnings - expected_earnings
102            if symbol not in self.earnings_surprise:
103                self.earnings_surprise[symbol] = deque(maxlen = self.surprise_period)
104            elif len(self.earnings_surprise[symbol]) >= self.surprise_period:
105                earnings_surprise_std: float = np.std(self.earnings_surprise[symbol])
106                sue: float = earnings_surprise / earnings_surprise_std
107        
108                # EAR calc
109                if len(self.market_prices) == self.market_prices.maxlen:
110                    # abnormal return calc
111                    history: DataFrame = self.History(symbol, self.ear_period, Resolution.Daily)
112                    if len(history) == self.ear_period and 'close' in history:
113                        stock_closes: Series = history['close']
114                        ear: float = Return(stock_closes) - Return(self.market_prices)
115                        ere: float = abs(ear) / sue
116                        ere_data[symbol] = [ere, ear, sue]
117                        
118                        # store ere data in this month's history
119                        self.ere_history_actual.append(ere)
120            self.earnings_surprise[symbol].append(earnings_surprise)
121        
122if len(ere_data) != 0 and len(self.ere_history_previous) != 0:
123    # sort by ERE
124    bottom_ere_quintile:float = np.percentile(self.ere_history_previous, self.percentile)
125    self.long = [x[0] for x in ere_data.items() if x[1][0] <= bottom_ere_quintile and x[1][1] > 0 and x[1][2] > 0]
126    self.short = [x[0] for x in ere_data.items() if x[1][0] <= bottom_ere_quintile and x[1][1] < 0 and x[1][2] < 0]
127    
128return self.long + self.short
129def OnData(self, data: Slice) -> None:
130if self.symbol in data and data[self.symbol]:
131    self.market_prices.append(data[self.symbol].Value)
132# open new trades
133for symbol in self.long:
134    self.trade_manager.Add(symbol, True)
135for symbol in self.short:
136    self.trade_manager.Add(symbol, False)
137
138self.trade_manager.TryLiquidate()
139
140self.long.clear()
141self.short.clear()
142
143def Selection(self) -> None:
144# every three months  
145if self.month % 3 == 0:
146    # save previous history
147    self.ere_history_previous = [x for x in self.ere_history_actual]
148    self.ere_history_actual.clear()
149self.month += 1
150if self.month > 12:
151    self.month = 1
152# Custom fee model.
153class CustomFeeModel(FeeModel):
154def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
155fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
156return OrderFee(CashAmount(fee, "USD"))
157def Return(values: np.ndarray) -> float:
158return (values[-1] - values[0]) / values[0]