from AlgoLib import *
import numpy as np
from collections import deque
from pandas.tseries.offsets import BDay
from dateutil.relativedelta import relativedelta
from typing import Dict, List, Tuple, Deque
class PostEarningsAnnouncementEffectAlgorithm(XXX):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.earnings_surprise: Dict[Symbol, float] = {}
self.min_seasonal_eps_period: int = 4
self.min_surprise_period: int = 4
self.leverage: int = 5
self.percentile_range: List[int] = [80, 20]
self.long_positions: List[Symbol] = []
self.sue_ear_history_previous: List[Tuple[float, float]] = []
self.sue_ear_history_actual: List[Tuple[float, float]] = []
self.eps_by_ticker: Dict[str, float] = {}
self.price_data_with_date: Dict[Symbol, Deque[float]] = {}
self.price_period: int = 63
self.market: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.price_data_with_date[self.market] = deque(maxlen=self.price_period)
self.first_date: Union[datetime.date, None] = None
earnings_data: str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
earnings_data_json: list[dict] = json.loads(earnings_data)
for obj in earnings_data_json:
date: datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
if not self.first_date: self.first_date = date
for stock_data in obj['stocks']:
ticker: str = stock_data['ticker']
if stock_data['eps'] == '':
continue
if ticker not in self.eps_by_ticker:
self.eps_by_ticker[ticker] = {}
self.eps_by_ticker[ticker][date] = float(stock_data['eps'])
self.month: int = 12
self.selection_flag: bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.AfterMarketOpen(self.market), self.Selection)
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
for security in changes.RemovedSecurities:
symbol: Symbol = security.Symbol
if symbol in self.earnings_surprise:
del self.earnings_surprise[symbol]
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
for stock in fundamental:
symbol: Symbol = stock.Symbol
if symbol in self.price_data_with_date:
self.price_data_with_date[symbol].append((self.Time.date(), stock.AdjustedPrice))
if not self.selection_flag:
return Universe.Unchanged
self.selection_flag = False
selected: List[Symbol] = [x.Symbol for x in fundamental if x.Symbol.Value in self.eps_by_ticker]
sue_ear: Dict[Symbol, float] = {}
current_date: datetime.date = self.Time.date()
prev_three_months: datetime = current_date - relativedelta(months=3)
for symbol in selected:
ticker: str = symbol.Value
recent_eps_data: Union[None, datetime.date] = None
if symbol not in self.price_data_with_date:
self.price_data_with_date[symbol] = deque(maxlen=self.price_period)
history: DataFrame = self.History(symbol, self.price_period, Resolution.Daily)
if history.empty:
self.Log(f"Not enough data for {symbol} yet.")
continue
closes: Series = history.loc[symbol].close
for time, close in closes.iteritems():
self.price_data_with_date[symbol].append((time.date(), close))
if len(self.price_data_with_date[self.market]) != self.price_data_with_date[self.market].maxlen:
return Universe.Unchanged
if len(self.price_data_with_date[symbol]) != self.price_data_with_date[symbol].maxlen:
continue
for date in self.eps_by_ticker[ticker]:
if date < current_date and date >= prev_three_months:
EPS_value: float = self.eps_by_ticker[ticker][date]
recent_eps_data: Tuple[datetime.date, float] = (date, EPS_value)
break
if recent_eps_data:
last_earnings_date: datetime.date = recent_eps_data[0]
earnings_eps_history: List[Tuple[datetime.date, float]] = [(x, self.eps_by_ticker[ticker][x]) for x in self.eps_by_ticker[ticker] if x < last_earnings_date]
seasonal_eps_data: List[Tuple[datetime.date, float]] = [x for x in earnings_eps_history if x[0].month == last_earnings_date.month]
if len(seasonal_eps_data) >= self.min_seasonal_eps_period:
year_diff: np.ndarray = np.diff([x[0].year for x in seasonal_eps_data])
if all(x == 1 for x in year_diff):
seasonal_eps: List[float] = [x[1] for x in seasonal_eps_data]
diff_values: np.ndarray = np.diff(seasonal_eps)
drift: float = np.average(diff_values)
last_earnings_eps: float = seasonal_eps[-1]
expected_earnings: float = last_earnings_eps + drift
actual_earnings: float = recent_eps_data[1]
earnings_surprise: float = actual_earnings - expected_earnings
if symbol not in self.earnings_surprise:
self.earnings_surprise[symbol] = []
elif len(self.earnings_surprise[symbol]) >= self.min_surprise_period:
earnings_surprise_std: float = np.std(self.earnings_surprise[symbol])
sue: float = earnings_surprise / earnings_surprise_std
min_day: datetime.date = last_earnings_date - BDay(2)
max_day: datetime.date = last_earnings_date + BDay(1)
stock_closes_around_earnings: List[Symbol] = [x for x in self.price_data_with_date[symbol] if x[0] >= min_day and x[0] <= max_day]
market_closes_around_earnings: List[Symbol] = [x for x in self.price_data_with_date[self.market] if x[0] >= min_day and x[0] <= max_day]
if len(stock_closes_around_earnings) == 4 and len(market_closes_around_earnings) == 4:
stock_return: float = stock_closes_around_earnings[-1][1] / stock_closes_around_earnings[0][1] - 1
market_return: float = stock_closes_around_earnings[-1][1] / stock_closes_around_earnings[0][1] - 1
ear: float = stock_return - market_return
sue_ear[symbol] = (sue, ear)
self.sue_ear_history_actual.append((sue, ear))
self.earnings_surprise[symbol].append(earnings_surprise)
if len(sue_ear) != 0 and len(self.sue_ear_history_previous) != 0:
sue_values: List[float] = [x[0] for x in self.sue_ear_history_previous]
ear_values: List[float] = [x[1] for x in self.sue_ear_history_previous]
top_sue_quantile: float = np.percentile(sue_values, self.percentile_range[0])
bottom_sue_quantile: float = np.percentile(sue_values, self.percentile_range[1])
top_ear_quantile: float = np.percentile(ear_values, self.percentile_range[0])
bottom_ear_quantile: float = np.percentile(ear_values, self.percentile_range[1])
self.long_positions: List[Symbol] = [x[0] for x in sue_ear.items() if x[1][0] >= top_sue_quantile and x[1][1] >= top_ear_quantile]
return self.long_positions
def OnData(self, data: Slice) -> None:
targets: List[PortfolioTarget] = []
for symbol in self.long_positions:
if symbol in data and data[symbol]:
targets.append(PortfolioTarget(symbol, 1. / len(self.long_positions)))
self.SetHoldings(targets, True)
self.long_positions.clear()
def Selection(self):
self.selection_flag = True
if self.month % 3 == 0:
self.sue_ear_history_previous = self.sue_ear_history_actual
self.sue_ear_history_actual.clear()
self.month += 1
if self.month > 12:
self.month = 1
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))
