Pre-Earnings Announcement Drift

1from AlgorithmImports import *
2import numpy as np
3from collections import deque
4from pandas.tseries.offsets import BDay
5from typing import Dict, List, Deque, Set
6class PreEarningsAnnouncementDrift(QCAlgorithm):
7
8def Initialize(self):
9self.SetStartDate(2010, 1, 1)
10self.SetCash(100000)
11self.leverage:int = 5
12self.quarter_period:int = 8
13self.ear_period:int = 30
14self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
15
16# Daily price data.
17self.data:Dict[Symbol, Deque[DateTime, float]] = {}
18
19# Quarterly ear data.
20self.ear_data:Dict[Symbol, Deque[float]] = {}
21
22# Import earnigns data.
23self.earnings_data:Dict[DateTime, List[str]] = {}
24
25# Available symbols from earning_dates.csv.
26self.symbols:Set = set()
27
28self.first_date:datetime.date|None = None
29earnings_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
30earnings_data_json:list[dict] = json.loads(earnings_data)
31
32for obj in earnings_data_json:
33    date:datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
34    self.earnings_data[date] = []
35    
36    if not self.first_date: self.first_date = date
37    for stock_data in obj['stocks']:
38        ticker:str = stock_data['ticker']
39        self.earnings_data[date].append(ticker)
40        self.symbols.add(ticker)
41        
42# EAR history for previous quarter used for statistics. 
43self.ear_previous_quarter:List[float] = []
44self.ear_actual_quarter:List[float] = []
45
46# Equally weighted brackets for traded symbols. - 10 symbols long , 10 for short, 5 days of holding.
47self.trade_manager:TradeManager = trade_manager.TradeManager(self, 10, 10, 5)
48
49self.month:int = 12
50self.selection_flag:bool = False
51self.settings.daily_precise_end_time = False
52self.settings.minimum_order_margin_portfolio_percentage = 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):
57for security in changes.AddedSecurities:
58    symbol:Symbol = security.Symbol
59    security.SetFeeModel(CustomFeeModel())
60    security.SetLeverage(self.leverage)
61    
62for security in changes.RemovedSecurities:
63    symbol:Symbol = security.Symbol
64    if symbol in self.ear_data:
65        del self.ear_data[symbol]
66def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
67# Update the rolling window every day.
68for stock in fundamental:
69    symbol:Symbol = stock.Symbol
70    # Store monthly price.
71    if symbol in self.data:
72        self.data[symbol].append((self.Time.date(), stock.AdjustedPrice))
73if not self.selection_flag:
74    return Universe.Unchanged
75self.selection_flag = False
76selection:List[Fundamental] = [x.Symbol
77    for x in sorted([x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > 5 and x.Symbol.Value in self.symbols \
78        and x.EarningReports.FileDate.HasValue and ((x.SecurityReference.ExchangeId == "NYS") or (x.SecurityReference.ExchangeId == "NAS") or (x.SecurityReference.ExchangeId == "ASE"))], key = lambda x: x.DollarVolume, reverse = True)]
79# Warmup price rolling windows.
80for symbol in selection:
81    if symbol in self.data:
82        continue
83    
84    self.data[symbol] = deque(maxlen = self.ear_period)
85    history:DataFrame = self.History(symbol, self.ear_period, Resolution.Daily)
86    if history.empty:
87        self.Log(f"Not enough data for {symbol} yet.")
88        continue
89    closes:Series = history.loc[symbol].close
90    for time, close in closes.items():
91        self.data[symbol].append((time.date(),close))
92
93# Stocks with last month's earnings.
94last_month_date:DateTime = self.Time - timedelta(self.Time.day)
95filtered_selection = [x for x in fundamental if (x.EarningReports.FileDate.Value.year == last_month_date.year and x.EarningReports.FileDate.Value.month == last_month_date.month)]    
96
97for stock in filtered_selection:
98    symbol:Symbol = stock.Symbol
99    # Add symbol to ear data dict.
100    if symbol not in self.ear_data:
101        self.ear_data[symbol] = deque(maxlen = self.quarter_period)
102    # Month of data is ready.
103    if symbol in self.data and len(self.data[symbol]) == self.data[symbol].maxlen:
104        
105        earnings_day:DateTime = stock.EarningReports.FileDate.Value.date()
106        day_before_earnings:DateTime = earnings_day - BDay(2)
107        two_days_after_earnings:DateTime = earnings_day + BDay(2)
108        day_range:List[DateTime, DateTime] = [day_before_earnings.date(), two_days_after_earnings.date()]
109        
110        # Store performance around earnings.
111        ear_prices:List[float] = [x[1] for x in self.data[symbol] if x[0] >= day_range[0] and x[0] <= day_range[-1]]
112        if len(ear_prices) == 5:
113            ear:float = ear_prices[-1] / ear_prices[0] - 1
114            self.ear_data[symbol].append(ear)
115
116return selection
117                
118def OnData(self, data: Slice) -> None:
119date_to_lookup:DateTime = (self.Time + BDay(5)).date()
120
121# Liquidate opened symbols after five days.
122self.trade_manager.TryLiquidate()
123
124ear_avg:Dict[Symbol, float] = {}
125for symbol in self.data:
126    # EAR data is ready.
127    if symbol in self.ear_data and len(self.ear_data[symbol]) == self.ear_data[symbol].maxlen:
128        if date_to_lookup in self.earnings_data:
129            # Earnings is in next two day for the symbol.
130            if symbol.Value in self.earnings_data[date_to_lookup]:
131                # Avg ear calc.
132                ear_values:List[float] = [x for x in self.ear_data[symbol]]
133                avg:float = np.mean(ear_values)
134                ear_avg[symbol] = avg
135                
136                # Store average return in this month's history.
137                self.ear_actual_quarter.append(avg)
138    
139# Wait until we have history data for previous three months.
140if len(self.ear_previous_quarter) != 0:
141    # Sort by EAR.
142    ear_values:List[float] = self.ear_previous_quarter
143    top_ear_quintile:float = np.percentile(ear_values, 90)
144    bottom_ear_quintile:float = np.percentile(ear_values, 10)
145    
146    # Store symbol to set.
147    short:List[Symbol] = [x[0] for x in ear_avg.items() if x[1] <= bottom_ear_quintile]
148    long:List[Symbol] = [x[0] for x in ear_avg.items() if x[1] >= top_ear_quintile]
149    
150    # Open new trades.
151    for symbol in long:
152        if symbol in data and data[symbol]:
153            self.trade_manager.Add(symbol, True)
154    
155    for symbol in short:
156        if symbol in data and data[symbol]:
157            self.trade_manager.Add(symbol, False)
158def Selection(self):
159self.selection_flag = True
160
161# Every three months.    
162if self.month % 3 == 0:
163    # Save quarter history.
164    self.ear_previous_quarter = [x for x in self.ear_actual_quarter]
165    self.ear_actual_quarter.clear()
166self.month += 1
167if self.month > 12:
168    self.month = 1
169# Custom fee model
170class CustomFeeModel(FeeModel):
171def GetOrderFee(self, parameters):
172fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
173return OrderFee(CashAmount(fee, "USD"))