Conglomerates Post-Earnings Announcement Drift

1import numpy as np
2from AlgorithmImports import *
3from trade_manager import TradeManager
4from collections import deque
5from pandas.tseries.offsets import BDay
6from dateutil.relativedelta import relativedelta
7class ConglomeratesPostEarningsAnnouncementDrift(QCAlgorithm):
8def Initialize(self):
9self.SetStartDate(2010, 1, 1)
10self.SetCash(100000)
11self.period:int = 13
12self.leverage:int = 5
13self.seasonal_eps_count:int = 3
14self.eps:Dict[Symbol, List[datetime.date, float]] = {}
15self.earnings_data:Dict[datetime.date, Dict[str, float]] = {}
16
17market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol        
18
19self.long:List[Symbol] = []
20
21# equally weighted brackets for traded symbols
22self.trade_manager:TradeManager = TradeManager(self, 20, 20, 58)
23earnings_data:str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
24earnings_data_json:list[dict] = json.loads(earnings_data)
25
26for obj in earnings_data_json:
27    date:datetime.date = datetime.strptime(obj['date'], '%Y-%m-%d').date()
28    self.earnings_data[date] = {}
29    
30    for stock_data in obj['stocks']:
31        ticker:str = stock_data['ticker']
32        if stock_data['eps'] != '':
33            self.earnings_data[date][ticker] = float(stock_data['eps'])
34self.UniverseSettings.Resolution = Resolution.Daily
35self.settings.daily_precise_end_time = False
36self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
37self.AddUniverse(self.FundamentalSelectionFunction)
38def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
39for security in changes.AddedSecurities:
40    security.SetFeeModel(CustomFeeModel())
41    security.SetLeverage(self.leverage)
42def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
43# stocks with yesterday's earnings
44yesterday:datetime.date = (self.Time - BDay(1)).date()
45if yesterday not in self.earnings_data:
46    return Universe.Unchanged
47tickers_with_yesterday_earnings:List[str] = list(self.earnings_data[yesterday].keys())
48
49# stocks with yesterday's earnings
50selected:List[Fundamental] = [x for x in fundamental if x.AssetClassification.MorningstarIndustryGroupCode == MorningstarIndustryGroupCode.Conglomerates and \
51    x.Symbol.Value in tickers_with_yesterday_earnings] 
52for stock in selected:
53    symbol:Symbol = stock.Symbol
54    ticker:str = symbol.Value
55    # store eps data
56    if symbol not in self.eps:
57        self.eps[symbol] = deque(maxlen=self.period)
58    data:List[datetime.date, float] = [yesterday, self.earnings_data[yesterday][ticker]]
59    self.eps[symbol].append(data)
60    if len(self.eps[symbol]) == self.eps[symbol].maxlen:
61        recent_eps_data:List[datetime.date, float] = self.eps[symbol][-1]
62        
63        year_range:range = range(self.Time.year - 3, self.Time.year)
64        
65        last_month_date:datetime.date = recent_eps_data[0] - relativedelta(months=1)
66        next_month_date:datetime.date = recent_eps_data[0] + relativedelta(months=1)
67        month_range:List[int] = [last_month_date.month, recent_eps_data[0].month, next_month_date.month]
68        # earnings with todays month number 4 years back
69        seasonal_eps_data:List[List[datetime.date, float]] = [x for x in self.eps[symbol] if \
70            x[0].month in month_range and x[0].year in year_range]
71        if len(seasonal_eps_data) != self.seasonal_eps_count: continue
72        
73        # make sure we have a consecutive seasonal data
74        # same months with one year difference
75        year_diff:np.array = np.diff([x[0].year for x in seasonal_eps_data])
76        if all(x == 1 for x in year_diff):
77            seasonal_eps:List[float] = [x[1] for x in seasonal_eps_data]
78            diff_values:np.array = np.diff(seasonal_eps)
79            drift:float = np.average(diff_values)
80            
81            # SUE calculation
82            last_earnings:float = seasonal_eps[-1]
83            expected_earnings:float = last_earnings + drift
84            actual_earnings:float = recent_eps_data[1]
85        
86            # earning beat
87            if actual_earnings > expected_earnings:
88                self.long.append(symbol)
89return self.long
90def OnData(self, data):
91self.trade_manager.TryLiquidate()
92
93# Open new trades.
94for symbol in self.long:
95    self.trade_manager.Add(symbol, True)
96
97self.long.clear()
98# Custom fee model
99class CustomFeeModel(FeeModel):
100def GetOrderFee(self, parameters):
101fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
102return OrderFee(CashAmount(fee, "USD"))