Post-Earnings Announcement Drift for Friday Evening Announcers

1import numpy as np
2from AlgorithmImports import *
3from collections import deque
4from dateutil.relativedelta import relativedelta
5from typing import List, Dict, Tuple
6from numpy import isnan
7class PostEarningsAnnouncementDriftFridayEveningAnnouncers(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2000, 1, 1)
10self.SetCash(100000)
11self.period:int = 13
12self.leverage:int = 5
13self.min_share_price:int = 2
14
15# EPS quarterly data.
16self.eps_data:Dict[Symbol, List[Tuple[float]]] = {}
17
18market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
19# Surprise data count needed to count standard deviation.
20self.surprise_period:int = 4
21self.earnings_surprise:Dict[Symbol, List[float]] = {}
22
23# SUE history for previous year used for statistics.
24self.sue_previous_year:List[float] = []
25self.sue_actual_year:List[float] = []
26
27# Trenching.
28self.holding_period:int = 12
29self.managed_queue:List[RebalanceQueueItem] = []
30self.market_cap_threshold: float = 100_000_000
31
32# Last fundamental stock price
33self.last_price:Dict[Symbol, float] = {}
34
35self.month:int = 12
36self.fundamental_sorting_key = lambda x: x.DollarVolume
37self.fundamental_count:int = 1000
38self.selection_flag:bool = False
39self.UniverseSettings.Resolution = Resolution.Daily
40self.AddUniverse(self.FundamentalSelectionFunction)
41self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
42self.Schedule.On(self.DateRules.MonthEnd(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
43self.settings.daily_precise_end_time = False
44def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
45for security in changes.AddedSecurities:
46    security.SetFeeModel(CustomFeeModel())
47    security.SetLeverage(self.leverage)
48
49def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
50if not self.selection_flag:
51    return Universe.Unchanged
52
53selected:List[Fundamental] = [
54    x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' and not\
55    isnan(x.EarningReports.BasicEPS.ThreeMonths) and (x.EarningReports.BasicEPS.ThreeMonths != 0) and x.MarketCap > self.market_cap_threshold
56]
57if len(selected) > self.fundamental_count:
58    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
59
60for stock in selected:
61    self.last_price[stock.Symbol] = stock.AdjustedPrice
62
63# Stocks with last month's earnings.
64last_month_date:datetime = self.Time - timedelta(self.Time.day)
65filered_fundamental:List[Symbol] = [x for x in selected if (x.EarningReports.FileDate.Value.year == last_month_date.year and x.EarningReports.FileDate.Value.month == last_month_date.month)]    
66
67sue_data:Dict[Symbol, float] = {}
68
69for stock in filered_fundamental:
70    symbol:Symbol = stock.Symbol
71    
72    # Store eps data.
73    if symbol not in self.eps_data:
74        self.eps_data[symbol] = deque(maxlen = self.period)
75    data:Tuple[float] = (stock.EarningReports.FileDate.Value.date(), stock.EarningReports.BasicEPS.ThreeMonths)
76    # NOTE: Handles duplicate values. QC fundamental contains duplicated stocks in some cases.
77    if data not in self.eps_data[symbol]:
78        self.eps_data[symbol].append(data)
79        
80    if len(self.eps_data[symbol]) == self.eps_data[symbol].maxlen:
81        recent_eps_data:Tuple[float] = self.eps_data[symbol][-1]
82        
83        year_range:range = range(self.Time.year - 3, self.Time.year)
84        
85        last_month_date:datetime = recent_eps_data[0] + relativedelta(months = -1)
86        next_month_date:datetime = recent_eps_data[0] + relativedelta(months = 1)
87        month_range:List[int] = [last_month_date.month, recent_eps_data[0].month, next_month_date.month]
88        # Earnings with todays month number 4 years back.
89        seasonal_eps_data:List[Tuple[float]] = [x for x in self.eps_data[symbol] if x[0].month in month_range and x[0].year in year_range]
90        if len(seasonal_eps_data) != 3: continue
91        
92        # Make sure we have a consecutive seasonal data. Same months with one year difference.
93        year_diff:float = np.diff([x[0].year for x in seasonal_eps_data])
94        if all(x == 1 for x in year_diff):
95            seasonal_eps:List[float] = [x[1] for x in seasonal_eps_data]
96            diff_values:List[float] = np.diff(seasonal_eps)
97            drift:float = np.average(diff_values)
98            
99            # SUE calculation.
100            last_earnings:Tuple[float] = seasonal_eps[-1]
101            expected_earnings:Tuple[float] = last_earnings + drift
102            actual_earnings:Tuple[float] = recent_eps_data[1]
103            # Store sue value with earnigns date.
104            earnings_surprise:Tuple[float] = actual_earnings - expected_earnings
105            if symbol not in self.earnings_surprise:
106                self.earnings_surprise[symbol] = deque()
107            else:
108                # Surprise data is ready.
109                if len(self.earnings_surprise[symbol]) >= self.surprise_period:
110                    earnings_surprise_std:float = np.std(self.earnings_surprise[symbol])
111                    sue:float = earnings_surprise / earnings_surprise_std
112                    
113                    # Store sue in this years's history of friday's earnings.
114                    self.sue_actual_year.append(sue)
115            
116                    # Only stocks with last month's earnings on friday.
117                    if stock.EarningReports.FileDate.Value.weekday() == 4:
118                        sue_data[symbol] = sue
119            
120            self.earnings_surprise[symbol].append(earnings_surprise)
121
122long:List[Symbol] = []
123short:List[Symbol] = []
124
125# Wait until we have history data for previous year.
126if len(sue_data) != 0 and len(self.sue_previous_year) != 0: 
127    positive_sue_values:List[float] = [x for x in self.sue_previous_year if x > 0]
128    positive_sue_median:float = np.median(positive_sue_values)
129    
130    negative_sue_values:List[float] = [x for x in self.sue_previous_year if x <= 0]
131    negative_sue_median:float = np.median(negative_sue_values)
132    
133    long = [x[0] for x in sue_data.items() if x[1] > 0 and x[1] >= positive_sue_median]
134    short = [x[0] for x in sue_data.items() if x[1] <= 0 and x[1] <= negative_sue_median]
135    
136    long_symbol_q:List[Tuple[Symbol, float]] = []
137    short_symbol_q:List[Tuple[Symbol, float]] = []
138    if len(long) != 0:
139        long_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(long)
140        long_symbol_q = [(x, np.floor(long_w / self.last_price[x])) for x in long if self.last_price[x] != 0]
141    
142    if len(short) != 0:
143        short_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period / len(short)
144        short_symbol_q = [(x, -np.floor(short_w / self.last_price[x])) for x in short if self.last_price[x] != 0]
145    
146    self.managed_queue.append(RebalanceQueueItem(long_symbol_q, short_symbol_q))                
147    
148return long + short
149
150def OnData(self, data: Slice) -> None:
151if not self.selection_flag:
152    return
153self.selection_flag = False
154# Trade execution
155remove_item:Union[None, RebalanceQueueItem] = None
156
157# Rebalance portfolio
158for item in self.managed_queue:
159    if item.holding_period == self.holding_period:
160        for symbol, quantity in item.long_symbol_q + item.short_symbol_q:
161            if symbol in data and data[symbol]:
162                if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
163                    self.MarketOrder(symbol, -quantity)
164                    
165        remove_item = item
166        
167    elif item.holding_period == 0:
168        open_long_symbol_q:List[Tuple[Symbol, float]] = []
169        open_short_symbol_q:List[Tuple[Symbol, float]] = []
170        
171        for symbol, quantity in item.long_symbol_q + item.short_symbol_q:
172            if symbol in data and data[symbol]:
173                if self.Securities[symbol].Price != 0 and self.Securities[symbol].IsTradable:
174                    self.MarketOrder(symbol, quantity)
175                    open_long_symbol_q.append((symbol, quantity))
176                    
177        # Only opened orders will be closed        
178        item.long_symbol_q = open_long_symbol_q
179        item.short_symbol_q = open_short_symbol_q
180        
181    item.holding_period += 1
182    
183# We need to remove closed part of portfolio after loop. Otherwise it will miss one item in self.managed_queue.
184if remove_item:
185    self.managed_queue.remove(remove_item)
186
187def Selection(self) -> None:
188self.selection_flag = True
189
190# Save yearly history.
191if self.month == 12:
192    self.sue_previous_year = list(self.sue_actual_year)
193    self.sue_actual_year.clear()
194self.month += 1
195if self.month > 12:
196    self.month = 1
197class RebalanceQueueItem():
198def __init__(self, long_symbol_q:List[Tuple[Symbol, float]], short_symbol_q:List[Tuple[Symbol, float]]):
199# symbol/quantity collections
200self.long_symbol_q:List[Tuple[Symbol, float]] = long_symbol_q  
201self.short_symbol_q:List[Tuple[Symbol, float]] = short_symbol_q
202self.holding_period:int = 0
203# Custom fee model.
204class CustomFeeModel(FeeModel):
205def GetOrderFee(self, parameters):
206fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
207return OrderFee(CashAmount(fee, "USD"))