Using Straddles to Trade on Earnings Announcements

1from AlgorithmImports import *
2from pandas.tseries.offsets import BDay
3from typing import Dict, List, Tuple
4#endregion
5class UsingStraddlesToTradeOnEarningsAnnouncements(QCAlgorithm):
6def Initialize(self) -> None:
7self.SetStartDate(2018, 1, 1)
8self.SetCash(100_000) 
9
10self.earnings: Dict[datetime.date, str] = {}
11self.selected_for_trade: List[Tuple[Symbol, float]] = []
12
13# dj30 universe
14self.tickers: List[str] = [
15    'UNH','GS','HD','AMGN','MSFT','CAT','MCD','V','BA','HON','CRM','MMM','DIS','JNJ','JPM','TRV','AXP','IBM','WMT','PG','NKE','AAPL','CVX','MRK','DOW','VZ','INTC','KO','WBA','CSCO'
16]
17
18min_strike: int = -5
19max_strike: int = 5
20min_expiry: int = 4
21max_expiry: int = 30
22self.leverage: int = 10
23self.day_threshold: int = 5
24
25self.percentage_traded: float = 0.1  # only 10% of portfolio assets are used to perform this strategy
26
27for ticker in self.tickers:
28    # add equity
29    data: Security = self.AddEquity(ticker, Resolution.Minute)
30    data.SetFeeModel(CustomFeeModel())
31    data.SetLeverage(self.leverage)
32    
33    # add options
34    option: Option = self.AddOption(ticker, Resolution.Minute)
35    option.SetFilter(min_strike, max_strike, min_expiry, max_expiry)
36
37# source: 'https://www.nasdaq.com/market-activity/earnings
38earnings_data: str = self.Download('data.quantpedia.com/backtesting_data/economic/earnings_dates_eps.json')
39earnings_data_json: List[dict] = json.loads(earnings_data)
40
41for obj in earnings_data_json:
42    date: datetime.date = datetime.strptime(obj['date'], "%Y-%m-%d").date()
43    self.earnings[date] = []
44    for stock_data in obj['stocks']:
45        ticker: str = stock_data['ticker']
46        self.earnings[date].append(ticker)
47# sort earnings by date    
48self.earnings: Dict[datetime, str] = {date: tickers for date, tickers in sorted(self.earnings.items(), key=lambda x: x[0])}
49            
50self.earnings_date: Union[None, datetime.date] = None
51            
52self.last_day: int = -1
53self.opened_equity_count: int = 0
54def OnData(self, slice: Slice) -> None:
55# check once a day
56if self.Time.day == self.last_day:
57    return
58self.last_day = self.Time.day
59
60invested: List[str] = [x.Key.Value for x in self.Portfolio if x.Value.Invested]
61# only equities are opened, options expired
62if self.opened_equity_count != 0 and len(invested) == self.opened_equity_count:
63    self.opened_equity_count = 0
64    # liquidate hedge
65    self.Liquidate()
66
67if not self.Portfolio.Invested and len(self.selected_for_trade) == 0:
68    current_date: datetime.date = self.Time.date()
69    for date in self.earnings:
70        # get closest earnings annoucement day, which is older than 3 days from current
71        # if date > current_date and abs((date - current_date).days) > 3 and self.earnings_date != date:
72        if date > (current_date + BDay(3)).date() and self.earnings_date != date:
73            self.earnings_date = date
74            break
75    
76    # return from function, if earnings_date wasn't selected
77    if self.earnings_date is None:
78        return
79    
80    for i in slice.OptionChains:
81        chains: OptionChains = i.Value
82        # get ticker for current chain
83        ticker:str = chains.get_Symbol().get_Value()[1:]
84        
85        # check if current ticker is stored in self.earnings dictionary under self.earnings_date
86        if ticker not in self.earnings[self.earnings_date]:
87            continue
88        
89        calls: List[OptionChains] = list(filter(lambda x: x.Right == OptionRight.Call, chains))
90        puts: List[OptionChains] = list(filter(lambda x: x.Right == OptionRight.Put, chains))
91    
92        if not calls or not puts:
93            continue
94        
95        underlying_price: float = chains.Underlying.Price
96        expiries: List[datetime.date] = [i.Expiry for i in puts]
97        
98        expiry_date: datetime.date = None
99        for expiry in expiries:
100            date: datetime.date = expiry.date()
101            
102            # pick expiry_date, if difference between expiration date and earnings annoucement date is less than 10 days
103            if abs((date - self.earnings_date).days) < 10:
104                expiry_date = date
105                break
106                
107        # don't continue in current iteration, if expiry_date wasn't picked
108        if expiry_date is None:
109            continue
110        
111        strikes: List[float] = [i.Strike for i in puts]
112    
113        # determine at-the-money strike
114        strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
115        atm_call: List[OptionChains] = [i for i in calls if i.Expiry.date() == expiry_date and i.Strike == strike]
116        atm_put: List[OptionChains] = [i for i in puts if i.Expiry.date() == expiry_date and i.Strike == strike]
117        if atm_call and atm_put:
118            self.selected_for_trade.append((atm_call[0], atm_put[0], underlying_price))
119
120if (self.Time + BDay(3)).date() > self.earnings_date and len(self.selected_for_trade) != 0:
121    self.selected_for_trade.clear()
122
123# need to trade 3 days before earnings annoucement            
124if (self.Time + BDay(3)).date() == self.earnings_date and len(self.selected_for_trade) != 0:
125    length: int = len(self.selected_for_trade)
126    
127    for atm_call, atm_put, underlying_price in self.selected_for_trade:
128        if (atm_call.Symbol in slice and slice[atm_call.Symbol] and atm_put.Symbol in slice and slice[atm_put.Symbol] and atm_put.UnderlyingSymbol in slice and slice[atm_put.UnderlyingSymbol]):
129            options_q: float = int(((self.Portfolio.TotalPortfolioValue*self.percentage_traded) / length) / (underlying_price * 100))
130            
131            # buy at-the-money straddle
132            self.Buy(atm_call.Symbol, options_q)
133            self.Buy(atm_put.Symbol, options_q)
134            
135            self.SetHoldings(atm_put.UnderlyingSymbol, -self.percentage_traded / length)
136        
137            # increase number of opened equities
138            self.opened_equity_count += 1
139    
140    # clear list after trade
141    self.selected_for_trade.clear()
142   
143# Custom fee model
144class CustomFeeModel(FeeModel):
145def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
146fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
147return OrderFee(CashAmount(fee, "USD"))