Earnings Announcement Size-Sorted Short-Term Reversal Strategy

1import data_tools
2from AlgoLib import *
3import numpy as np
4from typing import Dict, List
5from collections import deque
6
7class ReversalDuringEarningsAnnouncements(XXX):
8
9def Initialize(self):
10self.SetStartDate(2010, 1, 1) # earnings dates start in 2010
11self.SetCash(100000)
12
13self.long_count:int = 20
14self.short_count:int = 20
15self.holding_period:int = 3
16
17self.leverage:int = 5
18self.ear_period:int = 4
19self.symbol:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
20
21# Daily price data.
22self.data:Dict[Symbol, RollingWindow] = {}
23
24# Import earnigns data.
25self.earnings_data:Dict[datetime, List[str]] = {}
26
27# Available symbols from earning_dates dataset.
28self.tickers:Set(str) = set()
29
30self.first_date:Union[None, datetime.date] = None
31
32earnings_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
33earnings_data_json:list[dict] = json.loads(earnings_data)
34
35for obj in earnings_data_json:
36    date:datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
37    self.earnings_data[date] = []
38    
39    if not self.first_date: self.first_date = date
40
41    for stock_data in obj['stocks']:
42        ticker:str = stock_data['ticker']
43
44        self.earnings_data[date].append(ticker)
45        self.tickers.add(ticker)
46
47# EAR history for previous quarter used for statistics. 
48self.ear_previous_quarter:List[float] = []
49self.ear_actual_quarter:List[float] = []
50
51# 5 equally weighted brackets for traded symbols. - 20 symbols long , 20 for short, 3 days of holding.
52self.trade_manager = data_tools.TradeManager(self, self.long_count, self.short_count, self.holding_period)
53
54self.month:int = 0
55self.fundamental_sorting_key = lambda x: x.DollarVolume
56self.fundamental_count:int = 1000
57self.selection_flag = False
58self.rebalance_flag = False
59self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
60self.UniverseSettings.Resolution = Resolution.Daily
61self.AddUniverse(self.FundamentalSelectionFunction)
62self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
63
64def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
65for security in changes.AddedSecurities:
66    security.SetFeeModel(data_tools.CustomFeeModel())
67    security.SetLeverage(self.leverage)
68
69def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
70# update daily prices
71for stock in fundamental:
72    symbol:Symbol = stock.Symbol
73    
74    if symbol in self.data:
75        self.data[symbol].Add(stock.AdjustedPrice)
76
77if not self.selection_flag:
78    return Universe.Unchanged
79self.selection_flag = False
80
81selected:List[Symbol] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Symbol.Value in self.tickers]
82
83if len(selected) > self.fundamental_count:
84    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
85
86for stock in selected:
87    symbol:Symbol = stock.Symbol
88    if symbol in self.data:
89        continue
90    
91    self.data[symbol] = RollingWindow[float](self.ear_period)
92    history:DataFrame = self.History(symbol, self.ear_period, Resolution.Daily)
93    if history.empty:
94        self.Log(f"Not enough data for {symbol} yet")
95        continue
96    
97    closes:Series = history.loc[symbol].close
98    for time, close in closes.items():
99        self.data[symbol].Add(close)
100
101return list(map(lambda x: x.Symbol, selected))
102
103def OnData(self, data: Slice) -> None:
104date_to_lookup = (self.Time + timedelta(days=1)).date()
105
106# Liquidate opened symbols after three days.
107self.trade_manager.TryLiquidate()
108
109ret_t4_t2 = {}
110
111for symbol in self.data:
112    # Data is ready.
113    if self.data[symbol].IsReady:
114        # Earnings is in next two day for the symbol.
115        if date_to_lookup in self.earnings_data and symbol.Value in self.earnings_data[date_to_lookup]:
116            closes = [x for x in self.data[symbol]]
117            if closes[-1] != 0:
118                # Calculate t-4 to t-2 return.
119                ret = (closes[0] - closes[-1]) / closes[-1]
120                ret_t4_t2[symbol] = ret
121                
122                # Store return in this month's history.
123                self.ear_actual_quarter.append(ret)
124    
125# Wait until we have history data for previous three months.
126if len(self.ear_previous_quarter) != 0:
127    # Sort by EAR.
128    ear_values = self.ear_previous_quarter
129    top_ear_quintile = np.percentile(ear_values, 80)
130    bottom_ear_quintile = np.percentile(ear_values, 20)
131    
132    # Store symbol to set.
133    long = [x[0] for x in ret_t4_t2.items() if x[1] <= bottom_ear_quintile and x[0] in data and data[x[0]]]
134    short = [x[0] for x in ret_t4_t2.items() if x[1] >= top_ear_quintile and x[0] in data and data[x[0]]]
135    
136    # Open new trades.
137    for symbol in long:
138        self.trade_manager.Add(symbol, True)
139    for symbol in short:
140        self.trade_manager.Add(symbol, False)
141
142def Selection(self) -> None:
143# There is no earnings data yet.
144if self.Time.date() < self.first_date:
145    return
146
147self.selection_flag = True
148
149# Every three months.    
150if self.month % 3 == 0:
151    # Save quarter history.
152    self.ear_previous_quarter = [x for x in self.ear_actual_quarter]
153    self.ear_actual_quarter.clear()
154
155self.month += 1