Post-Split Drift Combined with PEAD Anomaly

1from AlgorithmImports import *
2import data_tools
3import numpy as np
4from collections import deque
5import dateutil.relativedelta
6from dateutil.relativedelta import relativedelta
7from typing import List, Dict, Set, Deque
8from numpy import isnan
9#endregion
10class PostSplitDriftCombinedWithPEADAnomaly(QCAlgorithm):
11def Initialize(self) -> None:
12self.SetStartDate(2011, 1, 1)
13self.SetCash(100_000)
14
15self.symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
16
17# Set of unique symbols to trade.
18self.long: Set = set()
19self.short: Set = set()
20
21self.leverage: int = 5
22self.period: int = 13
23self.long_symbols: int = 20
24self.short_symbols: int = 20
25self.holding_period: int = 90
26self.percentiles: List[int] = [20, 80]
27self.eps_data: Dict[Symbol, Deque[datetime, float]] = {}
28# Import earnigns data.
29self.earnings_data: Dict[datetime, str] = {}
30# Import splits data
31self.splits_data: Dict[str, datetime.date] = {}
32
33# Surprise data count needed to count standard deviation.
34self.surprise_period: int = 4
35self.earnings_surprise: Dict[Deque[float]] = {}
36
37# SUE history for previous quarter used for statistics.
38self.sue_history_previous: Deque[float] = deque()
39self.sue_history_actual: Deque[float] = deque()
40
41self.first_date: Union[None, datetime.date] = None
42earnings_data: str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
43earnings_data_json: List[dict] = json.loads(earnings_data)
44
45for obj in earnings_data_json:
46    date: datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
47    self.earnings_data[date] = []
48    if not self.first_date: self.first_date = date
49    for stock_data in obj['stocks']:
50        ticker: str = stock_data['ticker']
51        self.earnings_data[date].append(ticker)
52        
53csv_string_file: str = self.Download('data.quantpedia.com/backtesting_data/economic/splits.csv')
54lines: str = csv_string_file.split('\r\n')
55for line in lines:
56    if line == '':
57        continue
58    line_split: str = line.split(';')
59    symbol: str = line_split[0]
60    
61    self.splits_data[symbol] = []
62    
63    for i in range(1, len(line_split)):
64        if line_split[i] is not '':
65            date: datetime = datetime.strptime(line_split[i], '%m/%d/%Y').date()
66            self.splits_data[symbol].append(date)
67
68# 70 equally weighted brackets for traded symbols. - 20 symbols long, 20 symbols short, 90 days of holding.
69self.trade_manager: data_tools.TradeManager = data_tools.TradeManager(self, self.long_symbols, self.short_symbols, self.holding_period)
70
71self.month: int = 12
72self.selection_flag: bool = True
73self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
74self.settings.daily_precise_end_time = False
75self.UniverseSettings.Resolution = Resolution.Daily
76self.AddUniverse(self.FundamentalSelectionFunction)
77self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.AfterMarketOpen(self.symbol), self.Selection)
78def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
79for security in changes.AddedSecurities:
80    security.SetFeeModel(data_tools.CustomFeeModel())
81    security.SetLeverage(self.leverage)
82
83def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
84if not self.selection_flag:
85    return Universe.Unchanged
86self.selection_flag = False
87
88# S&P 500 universe from self.splits_data tickers.
89selected: List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' \
90                        and not isnan(x.EarningReports.BasicEPS.ThreeMonths) and x.EarningReports.BasicEPS.ThreeMonths != 0 \
91                        and not isnan(x.EarningReports.FileDate.Value.year) and x.EarningReports.FileDate.Value.year != 1 \
92                        and x.Symbol.Value in self.splits_data]
93# SUE data.
94sue_data: Dict[Symbol, float] = {}
95for stock in selected:
96    symbol: Symbol = stock.Symbol
97    # Store eps data.
98    if symbol not in self.eps_data:
99        self.eps_data[symbol] = deque(maxlen = self.period)
100    data: Tuple[datetime.date, float] = (stock.EarningReports.FileDate.Value.date(), stock.EarningReports.BasicEPS.ThreeMonths)
101    # NOTE: Handles duplicate values. QC fine contains duplicated stocks in some cases.
102    if data not in self.eps_data[symbol]:
103        self.eps_data[symbol].append(data)
104        
105    if len(self.eps_data[symbol]) == self.eps_data[symbol].maxlen:
106        recent_eps_data: datetime = self.eps_data[symbol][-1]
107        
108        year_range: List[int] = range(self.Time.year - 3, self.Time.year)
109        
110        last_month_date: datetime.date = recent_eps_data[0] + relativedelta(months = -1)
111        next_month_date: datetime.date = recent_eps_data[0] + relativedelta(months = 1)
112        month_range: List[int] = [last_month_date.month, recent_eps_data[0].month, next_month_date.month]
113        # Earnings with todays month number 4 years back.
114        seasonal_eps_data: List[Tuple[datetime.date, float]] = [x for x in self.eps_data[symbol] if x[0].month in month_range and x[0].year in year_range]
115        if len(seasonal_eps_data) != 3:
116            continue
117        
118        # Make sure we have a consecutive seasonal data. Same months with one year difference.
119        year_diff: np.ndarray = np.diff([x[0].year for x in seasonal_eps_data])
120        if all(x == 1 for x in year_diff):
121            seasonal_eps: List[float] = [x[1] for x in seasonal_eps_data]
122            diff_values: List[float] = np.diff(seasonal_eps)
123            drift: float = np.average(diff_values)
124            
125            # SUE calculation.
126            last_earnings: float = seasonal_eps[-1]
127            expected_earnings: float = last_earnings + drift
128            actual_earnings: float = recent_eps_data[1]
129            # Store sue value with earnigns date.
130            earnings_surprise: float = actual_earnings - expected_earnings
131            if symbol not in self.earnings_surprise:
132                self.earnings_surprise[symbol] = deque()
133            else:
134                # Surprise data is ready.
135                if len(self.earnings_surprise[symbol]) >= self.surprise_period:
136                    earnings_surprise_std:float = np.std(self.earnings_surprise[symbol])
137                    sue: float = earnings_surprise / earnings_surprise_std
138
139                    sue_data[symbol] = sue
140                        
141                    # Store pair in this month's history.
142                    self.sue_history_actual.append(sue)
143            self.earnings_surprise[symbol].append(earnings_surprise)
144        
145if len(sue_data) != 0: 
146    # Wait until we have history data for previous three months.
147    if len(self.sue_history_previous) != 0:
148        # Sort by SUE
149        top_sue_quintile: float = np.percentile(self.sue_history_previous, self.percentiles[1])
150        bottom_sue_quintile: float = np.percentile(self.sue_history_previous, self.percentiles[0])
151        self.long = [x[0] for x in sue_data.items() if x[1] >= top_sue_quintile]
152        self.short = [x[0] for x in sue_data.items() if x[1] <= bottom_sue_quintile]
153 
154return list(self.long) + list(self.short)
155
156def OnData(self, data: Slice) -> None:
157# Liquidate opened symbols after 90 days.
158self.trade_manager.TryLiquidate()
159current_date: datetime.date = (self.Time).date()
160three_months_ago: datetime.date = (current_date - dateutil.relativedelta.relativedelta(months=3))
161date_to_lookup: datetime.date = (self.Time - timedelta(days=3)).date()
162
163# If there is no earnings data yet.
164if date_to_lookup < self.first_date:
165    # Clear long and short set.
166    self.long.clear()
167    self.short.clear()
168# Open new trades.
169symbols_to_delete: List[Symbol] = []
170if date_to_lookup in self.earnings_data:
171    for symbol in self.long:
172        if symbol.Value in self.earnings_data[date_to_lookup]: # Check if stock had earnings date three days before this one
173            if symbol.Value in self.splits_data:
174                if self.CheckSplitDate(symbol.Value, three_months_ago, current_date):
175                    self.trade_manager.Add(symbol, True)
176                    symbols_to_delete.append(symbol)
177                
178    for symbol in self.short:
179        if symbol.Value in self.earnings_data[date_to_lookup]: # Check if stock had earnings date three days before this one
180            if symbol.Value in self.splits_data:
181                if self.CheckSplitDate(symbol.Value, three_months_ago, current_date):
182                    self.trade_manager.Add(symbol, False)
183                    symbols_to_delete.append(symbol)
184                    
185# Delete already traded symbols from long or short set
186for symbol in symbols_to_delete:
187    if symbol in self.long:
188        self.long.remove(symbol)
189    if symbol in self.short:
190        self.short.remove(symbol)
191
192def CheckSplitDate(self, symbol: Symbol, date_from: datetime.date, date_to: datetime.date) -> bool:
193for split_date in self.splits_data[symbol]:
194    if date_from <= split_date <= date_to:
195        return True
196    
197return False
198
199def Selection(self) -> None:
200self.selection_flag = True
201
202# Every three months.    
203if self.month % 3 == 0:
204    # Save previous month history.
205    self.sue_history_previous = [x for x in self.sue_history_actual]
206    self.sue_history_actual.clear()
207self.month += 1
208if self.month > 12:
209    self.month = 1
210    
211# Custom fee model
212class CustomFeeModel(FeeModel):
213def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
214fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
215return OrderFee(CashAmount(fee, "USD"))