Post-Loss/Profit Announcement Drift in Stocks

1import numpy as np
2from AlgorithmImports import *
3import data_tools
4from collections import deque
5from pandas.tseries.offsets import BDay
6from dateutil.relativedelta import relativedelta
7from typing import Dict, List, Deque
8from numpy import isnan
9class SwitchingBetweenValueMomentum(QCAlgorithm):
10def Initialize(self) -> None:
11self.SetStartDate(2000, 1, 1)
12self.SetCash(100_000)
13self.period: int = 13
14self.eps_data: Dict[Symbol, Deque[datetime.date]] = {} # EPS quarterly data
15
16self.coarse_count: int = 500
17self.leverage: int = 5
18self.min_share_price: int = 5
19self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
20
21self.long: List[Symbol] = []
22self.short: List[Symbol] = []
23
24self.last_selected: List[str] = []     # This year's selected stocks.
25
26# Surprise data count needed to count standard deviation.
27self.surprise_period: int = 4
28self.earnings_surprise: Dict[Symbol, float] = {}
29# SUE history for previous quarter used for statistics.
30self.sue_history_previous: Deque[float] = deque()
31self.sue_history_actual: Deque[float] = deque()
32
33# 50 equally weighted brackets for traded symbols.
34self.trade_manager: data_tools.TradeManager = data_tools.TradeManager(self, 15, 15, 60)
35
36self.month: int = 12
37self.selection_flag: bool = False
38self.rebalance_flag: bool = False
39self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
40self.settings.daily_precise_end_time = False
41self.UniverseSettings.Resolution = Resolution.Daily
42self.AddUniverse(self.FundamentalSelectionFunction)
43
44self.Schedule.On(self.DateRules.MonthEnd(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
45def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
46for security in changes.AddedSecurities:
47    security.SetFeeModel(data_tools.CustomFeeModel())
48    security.SetLeverage(self.leverage)
49def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
50# At the begining of the year pick whole new set of stocks.
51if self.selection_flag: 
52    self.selection_flag = False
53    selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Price > self.min_share_price and x.Market == 'usa' and not isnan(x.EarningReports.BasicEPS.ThreeMonths) and (x.EarningReports.BasicEPS.ThreeMonths != 0)]
54    if len(selected) > self.coarse_count:
55        selected = sorted(selected, key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
56    
57    self.last_selected = [x.Symbol.Value for x in selected]
58# During the year, filter just already picked stocks.
59# Stock with yesterdays earnings.
60filtered_fundamental: List[Fundamental] = [x for x in fundamental if x.Symbol.Value in self.last_selected and x.EarningReports.FileDate.Value.year != 1  and (self.Time.date() == (x.EarningReports.FileDate.Value + BDay(1)).date())]      
61symbol_sue: Dict[Symbol, float] = {}     # SUE for this day.
62
63for stock in filtered_fundamental:
64    symbol: Symbol = stock.Symbol
65    
66    # Store eps data.
67    if symbol not in self.eps_data:
68        self.eps_data[symbol] = deque(maxlen = self.period)
69    data: Tuple[float] = (stock.EarningReports.FileDate.Value.date(), stock.EarningReports.BasicEPS.ThreeMonths)
70    # NOTE: Handles duplicate values. QC fine contains duplicated stocks in some cases.
71    if data not in self.eps_data[symbol]:
72        self.eps_data[symbol].append(data)
73        
74    if len(self.eps_data[symbol]) == self.eps_data[symbol].maxlen:
75        recent_eps_data = self.eps_data[symbol][-1]
76        
77        year_range: List[int] = range(self.Time.year - 3, self.Time.year)
78        
79        last_month_date: datetime.date = recent_eps_data[0] + relativedelta(months = -1)
80        next_month_date: datetime.date = recent_eps_data[0] + relativedelta(months = 1)
81        month_range: List[datetime.date] = [last_month_date.month, recent_eps_data[0].month, next_month_date.month]
82        # Earnings with todays month number 4 years back.
83        seasonal_eps_data: List[float] = [x for x in self.eps_data[symbol] if x[0].month in month_range and x[0].year in year_range]
84        if len(seasonal_eps_data) != 3: continue
85        
86        # Make sure we have a consecutive seasonal data. Same months with one year difference.
87        year_diff:np.ndarray = np.diff([x[0].year for x in seasonal_eps_data])
88        if all(x == 1 for x in year_diff):
89            seasonal_eps: List[float] = [x[1] for x in seasonal_eps_data]
90            diff_values: np.ndarray = np.diff(seasonal_eps)
91            drift: float = np.average(diff_values)
92            
93            # SUE calculation.
94            last_earnings: float = seasonal_eps[-1]
95            expected_earnings: float = last_earnings + drift
96            actual_earnings: float = recent_eps_data[1]
97            # Store sue value with earnigns date.
98            earnings_surprise: float = actual_earnings - expected_earnings
99            if symbol not in self.earnings_surprise:
100                self.earnings_surprise[symbol] = deque(maxlen = self.surprise_period)
101            else:
102                # Surprise data is ready.
103                if len(self.earnings_surprise[symbol]) >= self.surprise_period:
104                    earnings_surprise_std: float = np.std(self.earnings_surprise[symbol])
105                    sue: float = earnings_surprise / earnings_surprise_std
106                    symbol_sue[symbol] = sue
107                    
108                    self.sue_history_actual.append(sue)
109            
110            self.earnings_surprise[symbol].append(earnings_surprise)
111            
112if len(symbol_sue) != 0:
113    # Wait until we have history data for previous three months.
114    if len(self.sue_history_previous) != 0:
115        # Sort by SUE.
116        sue_values: List[float] = [x for x in self.sue_history_previous]
117        top_sue_decile: float  = np.percentile(sue_values, 90)
118        bottom_sue_decile: float = np.percentile(sue_values, 10)
119    
120        self.long = [x[0] for x in symbol_sue.items() if x[1] >= top_sue_decile]
121        self.short = [x[0] for x in symbol_sue.items() if x[1] <= bottom_sue_decile]
122self.rebalance_flag = True
123
124# Return symbols
125return self.long + self.short
126
127def Selection(self) -> None:
128self.selection_flag = True
129# Every three months.    
130if self.month % 3 == 0:
131    # Save previous month history.
132    self.sue_history_previous = [x for x in self.sue_history_actual]
133    self.sue_history_actual.clear()
134self.month += 1
135if self.month > 12:
136    self.month = 1
137    
138def OnData(self, data: Slice) -> None:
139if not self.rebalance_flag:
140    return
141self.rebalance_flag = False
142
143# Open new trades.
144for i, portfolio in enumerate([self.short, self.long]):
145    for symbol in portfolio:
146        if symbol in data and data[symbol]:
147            self.trade_manager.Add(symbol, bool(i))
148        
149self.trade_manager.TryLiquidate()
150
151self.long.clear()
152self.short.clear()