Put-Call Spread Predicts Earnings Announcement Returns

1from AlgorithmImports import *
2import numpy as np
3from pandas.tseries.offsets import BDay
4import data_tools
5from typing import List, Dict, Set, Tuple
6#endregion
7class PutCallSpreadPredictsEarningsAnnouncementReturns(QCAlgorithm):
8def Initialize(self) -> None:
9self.SetStartDate(2010, 1, 1)
10self.SetCash(100_000)
11
12self.leverage: int = 5
13self.min_share_price: int = 5
14self.spread_threshold: int = 5
15self.top_percentile: int = 80
16self.bottom_percentile: int = 20
17# self.min_expiry = 30
18# self.max_expiry = 60
19self.fundamental_count: int = 100
20self.fundamental_sorting_key = lambda x: x.DollarVolume
21
22self.data: Dict[Symbol, SymbolData] = {}               # storing daily IV
23self.contracts: Dict[Symbol, data_tools.Contract] = {} # storing option contracts
24self.tickers_symbols: Dict[str, Symbol] = {}           # storing symbols under their tickers
25
26# quarterly stored volatility spread values for stocks
27self.actual_quarter_spread_values: List[float] = []
28self.prev_quarter_spread_values: List[float] = []
29
30# parse earnings data
31self.earnings_universe: List[str] = [] # stored earnings tickers
32self.earnings_by_date: Dict[datetime, str] = {}
33earnings_set: Set = set()
34# self.first_date:datetime.date|None = None
35earnings_data: str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
36earnings_data_json: List[dict] = json.loads(earnings_data)
37
38for obj in earnings_data_json:
39    date: datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
40    self.earnings_by_date[date] = []
41    
42    # if not self.first_date: self.first_date = date
43    for stock_data in obj['stocks']:
44        ticker: str = stock_data['ticker']
45        self.earnings_by_date[date].append(ticker)
46        earnings_set.add(ticker)
47for ticker in earnings_set:
48    self.earnings_universe.append(ticker)
49self.symbol: Symbol = self.AddEquity('SPY', Resolution.Minute).Symbol
50# equally weighted brackets for traded symbols
51self.trade_manager: data_tools.TradeManager = data_tools.TradeManager(self, 10, 10, 2)
52
53self.long: List[Symbol] = []   # long stocks with the highest implied volatility spreads on pre-announcement day
54self.short: List[Symbol] = []  # short stocks with the lowest implied volatility spreads on pre-announcement day
55
56self.selection_flag: bool = False
57self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
58self.settings.daily_precise_end_time = False
59self.UniverseSettings.Resolution = Resolution.Minute
60self.AddUniverse(self.FundamentalSelectionFunction)
61self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
62self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
63self.Schedule.On(self.DateRules.MonthStart(self.symbol), self.TimeRules.BeforeMarketClose(self.symbol), self.Selection)
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# rebalance weekly
71if not self.selection_flag:
72    return Universe.Unchanged
73
74# select top n stocks by dollar volume with price higher than 5
75selected: List[Fundamental] = [
76    x for x in fundamental 
77    if x.HasFundamentalData 
78    and x.MarketCap != 0 
79    and x.Market == 'usa' 
80    and x.Price > self.min_share_price 
81    and x.Symbol.Value in self.earnings_universe
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
86self.selection_flag = False
87
88selected_symbols: List[Symbol] = [] # storing symbols of selected stocks
89selected_tickers: List[str] = [] # storing tickers of selected stocks
90
91# add new stocks to dictionaries
92for stock in selected:
93    symbol: Symbol = stock.Symbol
94    ticker: str = symbol.Value
95    market_cap: float = stock.MarketCap
96    
97    # remove duplicate stocks from selected
98    if ticker in selected_tickers:
99        # check if symbol of duplicated ticker was stored in self.data
100        if symbol in self.data and symbol in self.contracts:
101            # remove stock's contracts
102            for contract in self.contracts[symbol].contracts:
103                self.RemoveSecurity(contract)
104                
105            del self.data[symbol]
106            del self.contracts[symbol]
107        
108        continue
109    
110    # add stock symbol to list of selected stocks
111    selected_symbols.append(symbol)
112    # add stock ticker to list of selected tickers
113    selected_tickers.append(ticker)
114    
115    # don't override data, if they are consecutive
116    if ticker in self.tickers_symbols and symbol in self.data and symbol in self.contracts:
117        # update market cap
118        self.data[symbol].market_cap = market_cap
119        continue
120    
121    self.data[symbol] = data_tools.SymbolData(market_cap, None)
122    # store symbol under stock ticker            
123    self.tickers_symbols[ticker] = symbol
124    # create object from Contract class for stock symbol
125    self.contracts[symbol] = data_tools.Contract(self.Time.date(), [])
126
127# make sure, data are consecutive
128remove_tickers_symbols:List[Tuple[str, Symbol]] = [] # storing tuple (ticker, symbol)
129
130for ticker, symbol in self.tickers_symbols.items():
131    # add stocks, which weren't selected to remove list
132    if symbol not in selected_symbols:
133        remove_tickers_symbols.append((ticker, symbol))
134        
135# remove not selected stocks from dictionaries
136for ticker, symbol in remove_tickers_symbols:
137    if symbol in self.contracts:
138        # remove stock's contracts
139        for contract in self.contracts[symbol].contracts:
140            self.RemoveSecurity(contract)
141            
142        del self.contracts[symbol]
143        
144    if symbol in self.data:   
145        # delete stock from dictionaries
146        del self.data[symbol]
147        del self.tickers_symbols[ticker]
148
149# return symbols of selected stocks    
150return selected_symbols
151def OnData(self, data: Slice) -> None:
152# each day store implied volatility for selected stocks
153if self.Time.hour == 9:
154    if data.OptionChains.Count != 0:
155        # there is no earnings next day
156        date_to_check: datetime.date = (self.Time.date() + BDay(1)).date()
157        if date_to_check not in self.earnings_by_date:
158            return
159        # top and bottom percentile of volatility spread values
160        top_percentile: Union[None, float] = None
161        bottom_percentile: Union[None, float] = None
162        
163        # spread values are stored for previous quarter
164        if len(self.prev_quarter_spread_values) > self.spread_threshold:
165            top_percentile = np.percentile(self.prev_quarter_spread_values, self.top_percentile)
166            bottom_percentile = np.percentile(self.prev_quarter_spread_values, self.bottom_percentile)
167        for kvp in data.OptionChains:
168            chain: OptionChains = kvp.Value
169            symbol: Symbol = chain.Underlying.Symbol
170            
171            # get option ticker from option symbol
172            ticker: str = symbol.Value
173                
174            # stock has earnings in one day
175            if ticker not in self.earnings_by_date[date_to_check]:
176                continue
177            if ticker not in self.tickers_symbols:
178                continue
179            
180            # based on option ticker get stock symbol
181            stock_symbol: Symbol = self.tickers_symbols[ticker]
182            
183            # make sure, spread is updated once in a day
184            if stock_symbol not in self.data or self.data[stock_symbol].updated_date == self.Time.date():
185                continue
186            
187            contracts: List[OptionContracts] = [x for x in chain]
188            
189            # make sure, there are enough contracts for stock
190            if len(contracts) < 2:
191                continue
192            
193            call_iv: Union[None, float] = None
194            put_iv: Union[None, float] = None
195            
196            # get atm call and atm put contract
197            for c in contracts:
198                if c.Right == OptionRight.Call:
199                    # found atm call
200                    call_iv = c.ImpliedVolatility
201                elif c.Right == OptionRight.Put:
202                    # found atm put
203                    put_iv = c.ImpliedVolatility
204            
205            # check if there are both contracts
206            if call_iv and put_iv:
207                # calculate volatility spread
208                oi: float = c.OpenInterest
209                vol_spread: float = oi*(put_iv - call_iv)
210                # bid_ask = abs(c.BidPrice - c.AskPrice)
211                
212                self.actual_quarter_spread_values.append(vol_spread)
213                self.data[stock_symbol].updated_date = self.Time.date()
214                
215                # top and bottom percentile of voltility spread was found
216                if top_percentile and bottom_percentile:
217                    if vol_spread >= top_percentile:
218                        self.long.append(stock_symbol)
219                    elif vol_spread <= bottom_percentile:
220                        self.short.append(stock_symbol)
221
222if self.Time.hour == 15 and self.Time.minute == 59:
223    # check expiry of contracts
224    for symbol in self.data:
225        # remove contract, when it has 1 day to expiry
226        if symbol in self.contracts and ((self.contracts[symbol].expiry_date - timedelta(days=1)) <= self.Time.date()):
227            # remove expired contracts
228            for contract in self.contracts[symbol].contracts:
229                self.RemoveSecurity(contract)
230            # remove Contracts object for current symbol
231            del self.contracts[symbol]
232            
233        # subscribe to contracts, if stock symbol doesn't have any
234        if symbol not in self.contracts:
235            # get new contracts after expiration
236            self.SubscribeOptionContracts(symbol)
237    
238    # try liquidate
239    self.trade_manager.TryLiquidate()
240    
241    # open new trades
242    for symbol in self.long:
243        if symbol in data and data[symbol]:
244            self.trade_manager.Add(symbol, True)
245    for symbol in self.short:
246        if symbol in data and data[symbol]:
247            self.trade_manager.Add(symbol, False)
248    
249    self.long = []
250    self.short = []
251
252def SubscribeOptionContracts(self, symbol: Symbol) -> None:
253''' get atm and atm strike for specific symbol then it filters atm call and atm put ''' 
254''' if there are enough atm calls and atm puts this function subscribes one of their contracts based on expiry and store expiry date ''' 
255
256# get all contracts for current commodity future
257contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
258# get current price for commodity future
259underlying_price: float = self.Securities[symbol].Price
260
261# get strikes from commodity future contracts
262strikes: List[float] = [i.ID.StrikePrice for i in contracts]
263
264# check if there is at least one strike    
265if len(strikes) <= 0:
266    return
267
268# at the money
269atm_strike: Union[None, float] = None
270atm_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
271
272# filtred contracts based on option rights and strikes
273atm_calls: List[Symbol] = self.FilterContracts(contracts, OptionRight.Call, atm_strike)
274atm_puts: List[Symbol] = self.FilterContracts(contracts, OptionRight.Put, atm_strike)
275
276# make sure there are enough contracts
277if len(atm_calls) > 0 and len(atm_puts) > 0:
278    # sort by expiry
279    atm_call: List[Symbol] = sorted(atm_calls, key = lambda item: item.ID.Date, reverse=True)[0]
280    atm_put: List[Symbol] = sorted(atm_puts, key = lambda x: x.ID.Date, reverse=True)[0]
281    
282    # add contracts
283    for contract in [atm_call, atm_put]:
284        self.AddContract(contract)
285    
286    # get expiry date of contracts
287    expiry_date: datetime.date = atm_call.ID.Date.date()
288    # create new Contract object for stock            
289    self.contracts[symbol] = data_tools.Contract(expiry_date, [atm_call, atm_put])   
290    
291def FilterContracts(self, 
292                contracts: List[Symbol], 
293                option_right: float, 
294                strike: float) -> List[Symbol]:
295''' filter contracts based on option_right and select only contracts with expiry in next month are selected '''
296
297# filter contracts based on option right and select only contracts with next month expiry
298filtered_contracts: List[Symbol] = [i for i in contracts if i.ID.OptionRight == option_right and 
299                                         i.ID.StrikePrice == strike and 
300                                         (self.Time.month + 1) == i.ID.Date.month]
301                                        #  self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
302# return filtered contracts
303return filtered_contracts
304
305def AddContract(self, contract: Symbol) -> None:
306''' subcribe to contract, set price model and normalization mode '''
307option: Option = self.AddOptionContract(contract, Resolution.Minute)
308option.PriceModel = OptionPriceModels.CrankNicolsonFD()
309option.SetDataNormalizationMode(DataNormalizationMode.Raw)
310
311def Selection(self) -> None:
312self.selection_flag = True
313
314# store spread values quarterly 
315if self.Time.month % 3 == 0:
316    self.prev_quarter_spread_values = self.actual_quarter_spread_values
317    self.actual_quarter_spread_values = []