Cross-Sectional Six-Month Equity ATM Straddle Trading Strategy

1from AlgorithmImports import *
2from typing import List, Dict
3from pandas.core.frame import DataFrame
4from pandas.core.series import Series
5#endregion
6class CrossSectionalSixMonthEquityATMStraddleTradingStrategy(QCAlgorithm):
7def Initialize(self) -> None:
8self.SetStartDate(2015, 1, 1)
9self.SetCash(1000000)
10
11self.min_expiry: int = 180
12self.max_expiry: int = 240
13
14self.period: int = 6 * 21 # need n of stock daily prices
15self.percentage_traded: float = 0.1
16self.selection_threshold: int = 10
17self.min_share_price: int = 10
18self.quantile: int = 10
19self.leverage: int = 10
20
21self.data: Dict[Symbol, RollingWindow[float]] = {}
22self.symbols_by_ticker: Dict[str, Symbol] = {}
23self.subscribed_contracts: Dict[Symbol, Contracts] = {}
24
25symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
26
27self.day: int = -1
28self.fundamental_count: int = 100
29self.fundamental_sorting_key = lambda x: x.DollarVolume
30self.selection_flag: bool = False
31self.settings.daily_precise_end_time = False
32self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
33self.UniverseSettings.Resolution = Resolution.Daily
34self.AddUniverse(self.FundamentalSelectionFunction)
35self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
36self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.BeforeMarketClose(symbol), self.Selection)
37def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
38for security in changes.AddedSecurities:
39    security.SetFeeModel(CustomFeeModel())
40    security.SetLeverage(self.leverage)
41def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
42# update daily prices of stocks in self.data dictionary
43for stock in fundamental:
44    symbol:Symbol = stock.Symbol
45    
46    if symbol in self.data:
47        self.data[symbol].Add(stock.AdjustedPrice)
48
49# rebalance monthly
50if not self.selection_flag:
51    return Universe.Unchanged
52
53# select top n stocks by dollar volume
54selected: List[Fundamental] = [
55    x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price
56]
57
58if len(selected) > self.fundamental_count:
59    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
60        
61for stock in selected:
62    symbol: Symbol = stock.Symbol
63    ticker: str = symbol.Value 
64    self.symbols_by_ticker[ticker] = symbol
65    
66    if symbol in self.data:
67        continue
68    
69    self.data[symbol] = RollingWindow[float](self.period)
70    history: DataFrame = self.History(symbol, self.period, Resolution.Daily)
71    if history.empty:
72        continue
73    closes: Series = history.loc[symbol].close
74    for time, close in closes.items():
75        self.data[symbol].Add(close)
76# return newly selected symbols
77return list(map(lambda x: x.Symbol, selected))
78def OnData(self, data: Slice) -> None:
79# execute once a day
80if self.day == self.Time.day:
81    return
82self.day = self.Time.day
83
84# subscribe to new contracts after selection
85if len(self.subscribed_contracts) == 0 and self.selection_flag:
86    for _, symbol in self.symbols_by_ticker.items():
87        if self.Securities[symbol].IsDelisted:
88            continue
89        if symbol in data and data[symbol]:
90            if symbol in self.data and self.data[symbol].IsReady:
91                # get all contracts for current stock symbol
92                contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
93                # get current price for etf
94                underlying_price: float = self.data[symbol][0]
95                
96                # get strikes from commodity future contracts
97                strikes: List[float] = [i.ID.StrikePrice for i in contracts]
98                
99                # can't filter contracts, if there isn't any strike price
100                if len(strikes) <= 0 or underlying_price == 0:
101                    continue
102                
103                # filter calls and puts contracts with one month expiry
104                calls, puts = self.FilterContracts(strikes, contracts, underlying_price)
105                
106                # make sure, there is at least one call and put contract
107                if len(calls) > 0 and len(puts) > 0:
108                    # sort by expiry
109                    call: Symbol = sorted(calls, key = lambda x: x.ID.Date, reverse=True)[0]
110                    put: Symbol = sorted(puts, key = lambda x: x.ID.Date, reverse=True)[0]
111                    
112                    subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(call.Underlying)
113                    if subscriptions:
114                        # add call contract
115                        self.AddContract(call)
116                        # add put contract
117                        self.AddContract(put)
118                        
119                        # retrieve expiry date for contracts
120                        expiry_date: dateTime.date = call.ID.Date.date()
121                        # store contracts with expiry date under stock's symbol
122                        self.subscribed_contracts[symbol] = Contracts(expiry_date, underlying_price, [call, put])
123        
124# calculate term structure and trade options
125elif len(self.subscribed_contracts) != 0 and data.OptionChains.Count != 0 and self.selection_flag:
126    self.selection_flag = False # this makes sure, there will be no other trades until next selection
127    
128    term_structure: Dict[Symbol, float] = {} # storing term structures keyed by stock's symbol
129    
130    for kvp in data.OptionChains:
131        chain: OptionChain = kvp.Value
132        
133        ticker: str = chain.Underlying.Symbol.Value
134        if ticker in self.symbols_by_ticker:
135            # get stock's symbol
136            symbol: Symbol = self.symbols_by_ticker[ticker]
137            if symbol in data and data[symbol]:
138                # get contracts
139                contracts: List[Symbol] = [x for x in chain]
140                
141                # check if there are enough contracts for option and daily prices are ready
142                if len(contracts) < 2 or not self.data[symbol].IsReady:
143                    continue
144                
145                # get call and put implied volatility
146                call_iv, put_iv = self.GetImpliedVolatilities(contracts)
147                
148                if call_iv and put_iv:
149                    # make mean from call implied volatility and put implied volatility
150                    iv: float = (call_iv + put_iv) / 2 
151                    # get historical volatility
152                    hv: float = self.GetHistoricalVolatility(self.data[symbol])
153                    
154                    # store stock's term structure
155                    term_structure[symbol] = (iv - hv) / hv
156        
157    # can't perform selection
158    if len(term_structure) < self.selection_threshold:
159        return
160    
161    # perform quantile selection
162    quantile: int = int(len(term_structure) / self.quantile)
163    sorted_by_term_structure: List[Symbol] = [x[0] for x in sorted(term_structure.items(), key=lambda item: item[1])]
164    
165    # long bottom
166    long: List[Symbol] = sorted_by_term_structure[:quantile]
167    # short top
168    short: List[Symbol] = sorted_by_term_structure[-quantile:]
169    
170    # trade long
171    self.TradeOptions(data, long, True)
172    # trade short
173    self.TradeOptions(data, short, False)
174    
175def Selection(self) -> None:
176self.selection_flag = True  # perform new selection
177self.Liquidate()            # rebalance monthly, so liquidate all holdings
178
179# clear dictionary for subscribed contracts, because there will be new selection
180self.subscribed_contracts.clear()
181# clear dictionary of tickers and their symbols, because new stocks will be selected
182self.symbols_by_ticker.clear()
183
184def FilterContracts(self, strikes: List[float], contracts: List[Symbol], underlying_price: float) -> List[Symbol]:
185''' filter call and put contracts from contracts parameter '''
186''' return call and put contracts '''
187
188# straddle
189call_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
190put_strike: float = call_strike
191
192calls: List[Symbol] = []  # storing call contracts
193puts: List[Symbol] = []   # storing put contracts
194
195for contract in contracts:
196    # check if contract has six months expiry
197    if self.min_expiry < (contract.ID.Date - self.Time).days < self.max_expiry:
198        # check if contract is call
199        if contract.ID.OptionRight == OptionRight.Call and contract.ID.StrikePrice == call_strike:
200            calls.append(contract)
201        # check if contract is put
202        elif contract.ID.OptionRight == OptionRight.Put and contract.ID.StrikePrice == put_strike:
203            puts.append(contract)
204
205# return filtered calls and puts with one month expiry
206return calls, puts
207
208def AddContract(self, contract: Symbol) -> None:
209''' subscribe option contract, set price mondel and normalization mode '''
210option: Option = self.AddOptionContract(contract, Resolution.Daily)
211option.PriceModel = OptionPriceModels.CrankNicolsonFD()
212
213def GetImpliedVolatilities(self, contracts: List[Symbol]) -> List[float]:
214''' retrieve implied volatility of contracts from contracts parameteres '''
215''' returns call and put implied volatility '''
216call_iv: Union[None, float] = None
217put_iv: Union[None, float] = None
218        
219# go through option contracts
220for c in contracts:
221    if c.Right == OptionRight.Call:
222        # found call option
223        call_iv = c.ImpliedVolatility
224    else:
225        # found put option
226        put_iv = c.ImpliedVolatility
227    
228return call_iv, put_iv
229
230def GetHistoricalVolatility(self, rolling_window_prices: RollingWindow) -> float:
231''' calculate historical volatility based on daily prices in rolling_window_prices parameter '''
232prices: np.ndarray = np.array([x for x in rolling_window_prices])
233returns: np.ndarray = (prices[:-1] - prices[1:]) / prices[1:]
234return np.std(returns)
235
236def TradeOptions(self, data: Slice, symbols: List[Symbol], long_flag: bool):
237''' on long signal buy call and put option contract '''
238''' on short signal sell call and put option contract '''
239count: int = len(symbols)
240
241# trade etf's call and put contracts
242for symbol in symbols:
243    if symbol in self.subscribed_contracts:
244        # check if contracts are tradebale and don't have 0 price
245        # for contract in self.subscribed_contracts[symbol].contracts:
246        #     if not self.Securities[contract].IsTradable or self.Securities[contract].Price == 0:
247        #         return
248
249        # get call and put contract
250        call, put = self.subscribed_contracts[symbol].contracts
251        # get underlying price
252        underlying_price: float = self.subscribed_contracts[symbol].underlying_price
253        
254        options_q: int = int(((self.Portfolio.MarginRemaining * self.percentage_traded) / count) / (underlying_price * 100))
255        
256        if call in data and data[call] and put in data and data[put]:
257            if long_flag:
258                self.Buy(call, options_q)
259                self.Buy(put, options_q)
260            else:
261                self.Sell(call, options_q)
262                self.Sell(put, options_q)
263
264class Contracts():
265def __init__(self, expiry_date: datetime.date, underlying_price: float, contracts: List[Symbol]):
266self.expiry_date: datetime.date = expiry_date
267self.underlying_price: float = underlying_price
268self.contracts: List[Symbol] = contracts
269# Custom fee model
270class CustomFeeModel(FeeModel):
271def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
272fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
273return OrderFee(CashAmount(fee, "USD"))