Reversal in Post-Earnings Announcement Drift

1from data_tools import SymbolData, CustomFeeModel, TradeManager
2from AlgoLib import *
3import numpy as np
4from collections import deque
5from typing import Dict, List
6from pandas.core.frame import DataFrame
7from pandas.tseries.offsets import BDay
8from dateutil.relativedelta import relativedelta
9
10class ReversalPostEarningsAnnouncementDrift(XXX):
11
12def Initialize(self) -> None:
13self.SetStartDate(2009, 1, 1) # earnings dates starts in 2010
14self.SetCash(100_000)
15
16self.long_symbols: int = 10
17self.short_symbols: int = 10
18self.holding_period: int = 2
19self.lookback_period: int = 3
20
21self.leverage: int = 5
22self.ear_period: int = 30
23self.prev_month_year: int = -1
24self.prev_month: int = -1
25self.percentiles: List[int] = [10, 90]
26        
27self.data: Dict[Symbol, SymbolData] = {}
28
29# EAR last quarter data
30self.ear_data: Dict[Symbol, List[datetime.date, float]] = {}
31
32self.earnings_data: Dict[datetime.date, List[str]] = {}
33self.eps_data: Dict[int, Dict[int, Dict[str, Dict[datetime.date, float]]]] = {}
34
35self.first_date: Union[datetime.date, None] = None
36
37earnings_data: str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
38earnings_data_json: List[dict] = json.loads(earnings_data)
39
40for obj in earnings_data_json:
41    date: datetime.date = datetime.strptime(obj['date'], '%Y-%m-%d').date()
42    year: int = date.year
43    month: int = date.month
44
45    self.earnings_data[date] = []
46
47    if not self.first_date: self.first_date = date
48    
49    for stock_data in obj['stocks']:
50        ticker: str = stock_data['ticker']
51
52        self.earnings_data[date].append(ticker)
53
54        if stock_data['eps'] == '':
55            continue
56
57        if year not in self.eps_data:
58            self.eps_data[year] = {}
59
60        if month not in self.eps_data[year]:
61            self.eps_data[year][month] = {}
62
63        if ticker not in self.eps_data[year][month]:
64            self.eps_data[year][month][ticker] = {}
65
66        self.eps_data[year][month][ticker][date] = float(stock_data['eps'])
67
68# EAR quarters history
69self.current_quarter_ears: List[float] = []
70self.previous_quarter_ears: List[float] = []
71
72# equally weighted brackets for traded symbols - 10 symbols long and short, 2 days of holding
73self.trade_manager: TradeManager = TradeManager(self, self.long_symbols, self.short_symbols, self.holding_period)
74
75self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
76
77self.selection_flag: bool = False
78self.store_sales_data_flag: bool = True
79self.sales_growth_sort_flag: bool = False
80self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
81self.UniverseSettings.Resolution = Resolution.Daily
82self.AddUniverse(self.FundamentalSelectionFunction)
83self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
84
85def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
86for security in changes.AddedSecurities:
87    security.SetFeeModel(CustomFeeModel())
88    security.SetLeverage(self.leverage)
89
90def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
91# update daily prices
92for stock in fundamental:
93    symbol: Symbol = stock.Symbol
94    
95    if symbol in self.data:
96        self.data[symbol].update(self.Time, stock.AdjustedPrice)
97
98# monthly selection
99if not self.selection_flag:
100    return Universe.Unchanged
101self.selection_flag = False
102
103prev_month_date: datetime.date = (self.Time - relativedelta(months=1)).date()
104self.prev_month_year: int = prev_month_date.year
105self.prev_month: int = prev_month_date.month
106
107if self.prev_month_year not in self.eps_data or self.prev_month not in self.eps_data[self.prev_month_year]:
108    return Universe.Unchanged
109
110# select every stock, which had earnings in previous month
111stocks_with_prev_month_eps: Dict[str, Dict[datetime.date, float]] = self.eps_data[self.prev_month_year][self.prev_month]
112selected_symbols: List[Symbol] = [x.Symbol for x in fundamental if x.Symbol.Value in stocks_with_prev_month_eps]
113
114for symbol in selected_symbols + [self.symbol]:
115    if symbol not in self.data:   
116        # warm up stock prices
117        self.data[symbol] = SymbolData(self.ear_period)
118        history: DataFrame = self.History(symbol, self.ear_period, Resolution.Daily)
119        if history.empty:
120            continue
121        
122        closes: Series = history.loc[symbol].close
123        for time, close in closes.iteritems():
124            self.data[symbol].update(self.Time, close)
125
126for symbol in selected_symbols:
127    if not self.data[symbol].is_ready():
128        continue
129    ticker: str = symbol.Value
130
131    # get all stock's eps from previous month
132    stock_prev_month_eps: Dict[datetime.date, float] = self.eps_data[self.prev_month_year][self.prev_month][ticker]
133    # get the date of the latest eps in previous month
134    stock_latest_eps_date: datetime.date = list(stock_prev_month_eps.keys())[-1]
135
136    # get 4 days around earnings and calculate EAR
137    date_from: datetime = stock_latest_eps_date - BDay(2)
138    date_to: datetime = stock_latest_eps_date + BDay(1)
139    
140    market_return: float = self.data[self.symbol].get_prices([date_from, date_to])
141    stock_return: float = self.data[symbol].get_prices([date_from, date_to])
142    
143    # check if returns are ready 
144    if market_return and stock_return:
145        ear: float = stock_return - market_return
146
147        ear_data: List[datetime.date] = (stock_latest_eps_date, ear)
148        self.ear_data[symbol] = ear_data
149            
150        # store ear in this month's history
151        self.current_quarter_ears.append(ear)
152
153# check if there are any symbols, which can be traded                
154if len(self.ear_data) == 0:
155    return Universe.Unchanged
156
157# return symbols from self.ear_data, because they will be traded
158return list(self.ear_data.keys())
159
160def OnData(self, data: Slice) -> None:
161# open trades on earnings day
162date_to_lookup: datetime.date = self.Time.date()
163
164# if there is no earnings data yet
165if date_to_lookup < self.first_date:
166    return
167
168# liquidate opened symbols after holding period
169self.trade_manager.TryLiquidate()
170
171# wait until we have history data for previous three months
172if len(self.previous_quarter_ears) == 0:
173    return
174
175ear_values: List[float] = [x for x in self.previous_quarter_ears]
176top_ear_decile: float = np.percentile(ear_values, self.percentiles[1])
177bottom_ear_decile: float = np.percentile(ear_values, self.percentiles[0])
178
179# Open new trades.
180if date_to_lookup in self.earnings_data:
181    symbols_to_trade: List[Symbol] = [symbol for symbol in self.ear_data if symbol.Value in self.earnings_data[date_to_lookup]]
182
183    symbols_to_delete: List[Symbol] = []
184    for symbol in symbols_to_trade:
185        # last earnings was less than three months ago
186        last_earnings_date: datetime.date = self.ear_data[symbol][0]
187
188        if last_earnings_date >= (self.Time - relativedelta(months=self.lookback_period)).date():
189            if symbol in data and data[symbol]:
190                if self.ear_data[symbol][1] >= top_ear_decile:
191                    self.trade_manager.Add(symbol, True)
192                    symbols_to_delete.append(symbol)
193                elif self.ear_data[symbol][1] <= bottom_ear_decile:
194                    self.trade_manager.Add(symbol, False)
195                    symbols_to_delete.append(symbol)
196    
197    # delete already traded symbols from symbol to trade
198    for symbol in symbols_to_delete:
199        del self.ear_data[symbol]
200
201def Selection(self) -> None:
202self.selection_flag = True
203
204if self.Time.month % 3 == 0:
205    # store previous quarter's history
206    self.previous_quarter_ears = [x for x in self.current_quarter_ears]
207    self.current_quarter_ears.clear()