Contrast Effect During the Earnings Announcements

1from AlgorithmImports import *
2import numpy as np
3from collections import deque
4from pandas.tseries.offsets import BDay
5from dateutil.relativedelta import relativedelta
6from typing import Dict, List
7#endregion
8class ContrastEffectDuringtheEarningsAnnouncements(QCAlgorithm):
9def Initialize(self):
10self.SetStartDate(2010, 1, 1)   # scheduled earnings data starts in 2010
11self.SetCash(100000)
12self.leverage:int = 5
13self.seasonal_eps_count:int = 3
14self.holding_period:int = 2
15self.surprise_period:int = 4
16self.period:int = 13
17# trenching
18self.managed_queue:List[RebalanceQueueItem] = []
19
20# surprise data count needed to count standard deviation
21self.earnings_surprise:Dict[Symbol, deque] = {}
22   
23self.last_price:Dict[Symbol, float] = {}
24
25# SUE and EAR history for previous quarter used for statistics
26self.surprise_history_previous:deque = deque()
27self.surprise_history_actual:deque = deque()
28self.eps:Dict[Symbol, deque] = {}
29data = self.AddEquity('SPY', Resolution.Daily)
30data.SetFeeModel(CustomFeeModel())
31data.SetLeverage(self.leverage)
32self.symbol:Symbol = data.Symbol
33# Earning data parsing.
34self.earnings_data:Dict[datetime.date, Dict[str, float]] = {}
35self.tickers:Set(str) = set() 
36earnings_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
37earnings_data_json:list[dict] = json.loads(earnings_data)
38
39for obj in earnings_data_json:
40    date:datetime.date = datetime.strptime(obj['date'], '%Y-%m-%d').date()
41    self.earnings_data[date] = {}
42    
43    for stock_data in obj['stocks']:
44        ticker:str = stock_data['ticker']
45        if stock_data['eps'] != '':
46            self.earnings_data[date][ticker] = float(stock_data['eps'])
47            
48        self.tickers.add(ticker)
49
50self.month:int = 0
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.MonthStart(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]:
62self.last_price.clear()
63for equity in fundamental:
64    symbol:Symbol = equity.Symbol
65    ticker:str = symbol.Value
66    if ticker in self.tickers or symbol == self.symbol:
67        self.last_price[symbol] = equity.AdjustedPrice
68selected:List[FineFundamental] = [x for x in fundamental if x.MarketCap != 0]
69# make sure there are some stocks with yesterday's earnings
70yesterday:datetime.date = (self.Time - BDay(1)).date()
71if yesterday not in self.earnings_data:
72    return Universe.Unchanged
73tickers_with_yesterday_earnings:List[str] = list(self.earnings_data[yesterday].keys())
74# stocks with yesterday's earnings
75filtered_fine:List[Fundamental] = [x for x in selected if x.Symbol.Value in tickers_with_yesterday_earnings] 
76# SUE data
77sue_data:Dict[Symbol, float] = {}
78
79for stock in filtered_fine:
80    symbol:Symbol = stock.Symbol
81    ticker:str = symbol.Value
82    # store eps data
83    if symbol not in self.eps:
84        self.eps[symbol] = deque(maxlen = self.period)
85    data:List[datetime.date, float] = [yesterday, self.earnings_data[yesterday][ticker]]
86    self.eps[symbol].append(data)
87    
88    # consecutive EPS data
89    if len(self.eps[symbol]) == self.eps[symbol].maxlen:
90        recent_eps_data:float = self.eps[symbol][-1]
91        
92        year_range:range = range(self.Time.year - 3, self.Time.year)
93        
94        last_month_date:datetime.date = recent_eps_data[0] - relativedelta(months=1)
95        next_month_date:datetime.date = recent_eps_data[0] + relativedelta(months=1)
96        month_range:List[int] = [last_month_date.month, recent_eps_data[0].month, next_month_date.month]
97        # earnings with todays month number 4 years back
98        seasonal_eps_data:List[List[datetime.date, float]] = [x for x in self.eps[symbol] \
99            if x[0].month in month_range and x[0].year in year_range]
100        if len(seasonal_eps_data) != self.seasonal_eps_count: continue
101        
102        # Make sure we have a consecutive seasonal data. Same months with one year difference.
103        year_diff:np.array = np.diff([x[0].year for x in seasonal_eps_data])
104        if all(x == 1 for x in year_diff):
105            seasonal_eps:List[float] = [x[1] for x in seasonal_eps_data]
106            diff_values:np.array = np.diff(seasonal_eps)
107            drift:float = np.average(diff_values)
108            
109            # SUE calculation
110            last_earnings:float = seasonal_eps[-1]
111            expected_earnings:float = last_earnings + drift
112            actual_earnings:float = recent_eps_data[1]
113            # store sue value with earnigns date
114            earnings_surprise:float = actual_earnings - expected_earnings
115
116            if symbol not in self.earnings_surprise:
117                self.earnings_surprise[symbol] = deque(maxlen=self.surprise_period)
118            elif len(self.earnings_surprise[symbol]) >= self.surprise_period:
119                earnings_surprise_std:float = np.std(self.earnings_surprise[symbol])
120                sue:float = earnings_surprise / earnings_surprise_std
121                
122                sue_data[symbol] = sue
123            self.earnings_surprise[symbol].append(earnings_surprise)
124
125if len(sue_data) == 0:
126    return Universe.Unchanged
127
128long_symbol_q:List[Symbol, float] = []
129short_symbol_q:List[Symbol, float] = []
130
131# store total yesterday's surprise in this month's history
132yesterdays_surprises:float = sum([x[1] for x in sue_data.items()])
133# wait until there is surprise history data for previous three months
134if len(self.surprise_history_previous) != 0:
135    # find symbols with next day scheduled earnings
136    earnings_date = (self.Time + BDay(1)).date()
137    if earnings_date in self.earnings_data:
138        surprise_values:List = [x for x in self.surprise_history_previous]
139    
140        top_surprise_percentile:float  = np.percentile(surprise_values, 75)
141        bottom_surprise_percentile:float = np.percentile(surprise_values, 25)
142        
143        traded_symbols:List[List[Symbol, float]] = []
144        for stock in selected:
145            symbol:Symbol = stock.Symbol
146            ticker:str = symbol.Value
147            # stock has earnings in 1 day
148            if ticker in self.earnings_data[earnings_date]:
149                traded_symbols.append([symbol, stock.MarketCap])
150        
151        if len(traded_symbols) != 0:
152            if self.symbol in self.last_price:
153                total_market_cap:float = sum([x[1] for x in traded_symbols])
154                
155                stocks_w:float = self.Portfolio.TotalPortfolioValue / self.holding_period
156                spy_quantity:float = self.Portfolio.TotalPortfolioValue / self.holding_period / self.last_price[self.symbol]
157                
158                if yesterdays_surprises > top_surprise_percentile:
159                    long_symbol_q = [(x[0], np.floor(stocks_w * (x[1] / total_market_cap) / self.last_price[x[0]])) for x in traded_symbols]
160                    
161                    # Quantity instead of weight is used in case of SPY.
162                    short_symbol_q = [(self.symbol, -spy_quantity)]
163                    
164                elif yesterdays_surprises < bottom_surprise_percentile:
165                    # Quantity instead of weight is used in case of SPY.
166                    long_symbol_q = [(self.symbol, spy_quantity)]
167                    
168                    short_symbol_q = [(x[0], -np.floor(stocks_w * (x[1] / total_market_cap) / self.last_price[x[0]])) for x in traded_symbols]  
169    
170self.surprise_history_actual.append(yesterdays_surprises)
171    
172self.managed_queue.append(RebalanceQueueItem(long_symbol_q + short_symbol_q))
173    
174return [x[0] for x in long_symbol_q + short_symbol_q] if len(long_symbol_q + short_symbol_q) != 0 else Universe.Unchanged
175def OnData(self, data: Slice) -> None:
176# trade execution
177remove_item:Union[RebalanceQueueItem, None] = None
178
179# rebalance portfolio
180for item in self.managed_queue:
181    if item.holding_period == self.holding_period:
182        for symbol, quantity in item.symbol_q:
183            self.MarketOrder(symbol, -quantity)
184                    
185        remove_item = item
186        
187    elif item.holding_period == 0:
188        open_symbol_q:List[List[Symbol, float]] = []
189        
190        for symbol, quantity in item.symbol_q:
191            if symbol in data and data[symbol]:
192                self.MarketOrder(symbol, quantity)
193                open_symbol_q.append((symbol, quantity))
194                    
195        # Only opened orders will be closed        
196        item.symbol_q = open_symbol_q
197        
198    item.holding_period += 1
199    
200# remove closed part of portfolio after loop
201# otherwise it will miss one item in self.managed_queue
202if remove_item:
203    self.managed_queue.remove(remove_item)
204    
205def Selection(self) -> None:        
206# every three months
207if self.month % 3 == 0:
208    # save history
209    self.surprise_history_previous = [x for x in self.surprise_history_actual]
210    self.surprise_history_actual.clear()
211
212self.month += 1
213    
214class RebalanceQueueItem:
215def __init__(self, symbol_q):
216# symbol/quantity collections
217self.symbol_q:List[List[Symbol, float]] = symbol_q  
218self.holding_period:int = 0
219
220# Custom fee model
221class CustomFeeModel(FeeModel):
222def GetOrderFee(self, parameters):
223fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
224return OrderFee(CashAmount(fee, "USD"))