Post-Earnings Announcement Drift Based on Price Signal Alone

1from data_tools import CustomFeeModel, TradeManager, SymbolData
2from AlgorithmImports import *
3import numpy as np
4from collections import deque
5from typing import Dict, List, Set
6from pandas.tseries.offsets import BDay
7class PostEarningsAnnouncementDrift(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2010, 1, 1) # earnings data start in 2010
10self.SetCash(100000)
11self.period:int = 61
12self.leverage:int = 5
13self.long:List[Symbol] = []
14self.short:List[Symbol] = []
15self.long_count:int = 5
16self.short_count:int = 5
17self.holding_period:int = 40
18
19# monthly selected universe
20self.last_selection:List[Symbol] = []
21
22self.data:Dict[Symbol, SymbolData] = {}
23self.tickers:Set(str) = set()
24# EPS quarterly data
25self.eps:Dict[Symbol, deque] = {}
26self.earnings_data:Dict[datetime.date, List[str]] = {} 
27earnings_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
28earnings_data_json:list[dict] = json.loads(earnings_data)
29
30for obj in earnings_data_json:
31    date:datetime.date = datetime.strptime(obj['date'], '%Y-%m-%d').date()
32    year:int = date.year
33    month:int = date.month
34    self.earnings_data[date] = []
35    for stock_data in obj['stocks']:
36        ticker:str = stock_data['ticker']
37        self.earnings_data[date].append(ticker)
38        self.tickers.add(ticker)
39
40# equally weighted brackets for traded symbols
41self.trade_manager:TradeManager = TradeManager(self, self.long_count, self.short_count, self.holding_period)
42
43self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
44self.fundamental_sorting_key = lambda x: x.DollarVolume
45self.fundamental_count:int = 500
46self.last_month:int = -1
47self.UniverseSettings.Resolution = Resolution.Daily
48self.AddUniverse(self.FundamentalSelectionFunction)
49self.settings.daily_precise_end_time = False
50self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
51
52def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
53for security in changes.AddedSecurities:
54    security.SetFeeModel(CustomFeeModel())
55    security.SetLeverage(self.leverage)
56
57for security in changes.RemovedSecurities:
58    if security.Symbol in self.data:
59        if security.Symbol != self.market:
60            del self.data[security.Symbol]
61def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
62if self.Time.month != self.last_month:
63    self.last_month = self.Time.month
64
65    # in fundamental always select whole universe (stocks, which are in QP earnings data),
66    # because prices of each stock in this universe are updated in OnData (due to open price)
67    selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol.Value in self.tickers]
68    if len(selected) > self.fundamental_count:
69        selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
70    self.last_selection = [x.Symbol for x in selected]
71# warm up prices
72for symbol in self.last_selection + [self.market]:
73    if symbol in self.data:
74        continue
75    
76    self.data[symbol] = SymbolData(self.period)
77    history:DataFrame = self.History(symbol, self.period+1, Resolution.Daily)
78    if history.empty:
79        continue
80    
81    closes:Series = history.close
82    opens:Series = history.open
83    
84    for (_, open_price), (_, close) in zip(opens.items(), closes.items()):
85        self.data[symbol].update(close, open_price)
86
87# market prices has to be ready
88if self.market not in self.data or not self.data[self.market].is_ready(): 
89    return self.last_selection
90prev_business_day:datetime.date = (self.Time - BDay(1)).date()
91if prev_business_day not in self.earnings_data:
92    return self.last_selection
93# filter stocks, which had earnings on prev business day
94prev_bussiness_day_earnings:List[str] = self.earnings_data[prev_business_day]
95selected_fundamental:List[Symbol] = [x for x in self.last_selection if x.Value in prev_bussiness_day_earnings]
96
97market_intraday_returns:List[float] = [x if x != 0.0 else 1.0 for x in self.data[self.market].get_intraday_returns()]
98
99for symbol in selected_fundamental:
100    # symbol:Symbol = stock.Symbol
101    if not self.data[symbol].is_ready():
102        continue
103    stock_intraday_returns:List[float] = self.data[symbol].get_intraday_returns()
104    daily_moves:List[float] = [(stock_intrady_ret / market_intraday_ret) for stock_intrady_ret, market_intraday_ret \
105        in zip(stock_intraday_returns, market_intraday_returns)]
106        
107    std:float = np.std(daily_moves)
108    mean:float = np.mean(daily_moves)
109
110    if daily_moves[0] > mean + 5 * std:
111        self.long.append(symbol)
112    elif daily_moves[0] < mean - 5 * std:
113        self.short.append(symbol)
114
115return self.last_selection
116def OnData(self, data:Slice) -> None:
117# update stock data for current universe
118for symbol in self.last_selection + [self.market]:
119    if symbol in self.data and symbol in data and data[symbol]:
120        close:float = data[symbol].Close
121        open:float = data[symbol].Open
122        self.data[symbol].update(close, open)
123
124self.trade_manager.TryLiquidate()
125# open new trades
126for symbol in self.long:
127    if symbol in data and data[symbol]:
128        self.trade_manager.Add(symbol, True)
129for symbol in self.short:
130    if symbol in data and data[symbol]:
131        self.trade_manager.Add(symbol, False)
132self.long.clear()
133self.short.clear()