Implied Put-Call Volatility Spread in US Equities

1from AlgorithmImports import *
2from typing import List, Dict
3# endregion
4
5class ImpliedPutCallVolatilitySpreadinUSEquities(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2015, 1, 1)
9self.SetCash(100000)
10
11self.leverage:int = 5
12self.quantile:int = 5
13
14self.min_contracts:int = 3 # number of nearest options
15self.one_stock_contracts:int = self.min_contracts * 6 # 2 ATM, 2 ITM and 2 OTM options time self.min_contracts expiries
16
17self.min_expiry:int = 35
18self.max_expiry:int = 360
19
20self.exchanges:List[str] = ['NYS', 'NAS', 'ASE']
21
22self.subscribed_options:Dict[str, List[OptionContract]] = {}
23self.symbols_by_tickers:Dict[str, Symbol] = {}
24self.selected_universe:List[Symbol] = []
25self.daily_VS:Dict[str, List[float]] = {}
26
27self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
28
29self.coarse_count:int = 100
30self.selection_flag:bool = False
31self.UniverseSettings.Resolution = Resolution.Daily
32self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction)
33self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
34self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.BeforeMarketClose(self.market, 0), self.Selection)
35
36def OnSecuritiesChanged(self, changes:SecurityChanges) -> None:
37for security in changes.AddedSecurities:
38    security.SetFeeModel(CustomFeeModel())
39    security.SetLeverage(self.leverage)
40
41def CoarseSelectionFunction(self, coarse:List[CoarseFundamental]) -> List[Symbol]:
42if not self.selection_flag:
43    return Universe.Unchanged
44
45selected:List[CoarseFundamental] = sorted([x for x in coarse if x.HasFundamentalData and x.AdjustedPrice >= 5],
46        key=lambda x: x.DollarVolume, reverse=True)[:self.coarse_count]
47
48return list(map(lambda stock: stock.Symbol, selected))
49
50def FineSelectionFunction(self, fine:List[FineFundamental]) -> List[Symbol]:
51fine:List[FineFundamental] = [x.Symbol for x in fine if x.MarketCap != 0 and x.SecurityReference.ExchangeId in self.exchanges]
52self.selected_universe = fine
53return self.selected_universe
54
55def OnData(self, data:Slice) -> None:
56# store VS every day
57if len(self.subscribed_options) != 0 and data.OptionChains.Count != 0:
58    for kvp in data.OptionChains:
59        chain:OptionChain = kvp.Value
60        ticker:str = chain.Underlying.Symbol.Value
61        if ticker not in self.subscribed_options:
62            continue
63        
64        # calculate VS
65        contracts:List[OptionContract] = [x for x in chain]
66        sorted_contracts:List[OptionContract] = sorted(contracts, key=lambda item: (item.Right, item.Expiry, item.Strike))
67        calls:List[OptionContract] = sorted_contracts[:self.one_stock_contracts // 2]
68        puts:List[OptionContract] = sorted_contracts[-self.one_stock_contracts // 2:]
69
70        VS:float = np.sum(np.array(list(map(lambda c_p: (c_p[0].OpenInterest + c_p[1].OpenInterest) / 2, zip(calls, puts)))) * (np.array(list(map(lambda c: c.ImpliedVolatility, calls))) - np.array(list(map(lambda p: p.ImpliedVolatility, puts)))))
71
72        # store VS
73        if ticker not in self.daily_VS:
74            self.daily_VS[ticker] = []
75        self.daily_VS[ticker].append(VS)
76        
77if self.selection_flag:
78    self.selection_flag = False
79    self.Liquidate()
80
81    # VS sort
82    monthly_VS:Dict[Symbol, float] = {}
83    for ticker, VS_list in self.daily_VS.items():
84        if len(VS_list) != 0:
85            if ticker in self.symbols_by_tickers:
86                symbol:Symbol = self.symbols_by_tickers[ticker]
87                monthly_VS[symbol] = np.mean(VS_list)
88    
89    # order execution
90    if len(monthly_VS) >= self.quantile:
91        quantile:int = len(monthly_VS) // self.quantile
92        sorted_by_monthly_VS:List[Symbol] = sorted(monthly_VS, key=monthly_VS.get, reverse=True)
93        long_leg:List[Symbol] = sorted_by_monthly_VS[:quantile]
94        short_leg:List[Symbol] = sorted_by_monthly_VS[-quantile:]
95
96        for symbol in long_leg:
97            self.SetHoldings(symbol, 1 / len(long_leg))
98
99        for symbol in short_leg:
100            self.SetHoldings(symbol, -1 / len(short_leg))
101
102    # remove old option contracts
103    for ticker in self.subscribed_options:
104        for option in self.subscribed_options[ticker]:
105            self.RemoveOptionContract(option.Symbol)
106    
107    # reset last month's data
108    self.subscribed_options.clear()
109    self.daily_VS.clear()
110    self.symbols_by_tickers.clear()
111
112    for symbol in self.selected_universe:
113        # subscribe to contract
114        contracts:List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
115        underlying_price:float = self.Securities[symbol].Price
116        
117        strikes:List[float] = [i.ID.StrikePrice for i in contracts]
118        
119        if len(strikes) <= 0:
120            continue
121        
122        atm_strike:float = min(strikes, key=lambda x: abs(x-underlying_price))
123        itm_strike:float = min(strikes, key=lambda x: abs(x-(underlying_price*0.95)))
124        otm_strike:float = min(strikes, key=lambda x: abs(x-(underlying_price*1.05)))
125
126        atm_calls:List[Symbol] = self.select_contracts(contracts, OptionRight.Call, atm_strike)
127        atm_puts:List[Symbol] = self.select_contracts(contracts, OptionRight.Put, atm_strike)
128
129        itm_calls:List[Symbol] = self.select_contracts(contracts, OptionRight.Call, itm_strike)
130        itm_puts:List[Symbol] = self.select_contracts(contracts, OptionRight.Put, itm_strike)
131
132        otm_calls:List[Symbol] = self.select_contracts(contracts, OptionRight.Call, otm_strike)
133        otm_puts:List[Symbol] = self.select_contracts(contracts, OptionRight.Put, otm_strike)
134            
135        # make sure there are enough contracts
136        if not all(len(x) >= self.min_contracts for x in [atm_calls, atm_puts, itm_calls, itm_puts, otm_calls, otm_puts]):
137            continue
138        
139        # store stock's symbol under it's ticker, because it is the only possibility how to access it from option contract
140        self.symbols_by_tickers[symbol.Value] = symbol
141
142        # sort by expiry and subscribe n nearest contracts
143        subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(atm_calls[0].Underlying)
144        if subscriptions:
145            if self.Securities[symbol].DataNormalizationMode == DataNormalizationMode.Raw:
146                for selected_contracts in [atm_calls, atm_puts, itm_calls, itm_puts, otm_calls, otm_puts]:
147                    nearest_contracts:List[Symbol] = sorted(selected_contracts, key=lambda item: item.ID.Date)[:self.min_contracts]
148
149                    for contract in nearest_contracts:
150                        option:OptionContract = self.AddOptionContract(contract, Resolution.Daily)
151                        option.PriceModel = OptionPriceModels.CrankNicolsonFD()
152
153                        # after trade execution subscribed options will be unsubscribed,
154                        # to stop receiving unncessary data, which slows down strategy
155                        if symbol.Value not in self.subscribed_options:
156                            self.subscribed_options[symbol.Value] = []
157
158                        self.subscribed_options[symbol.Value].append(option)
159
160def select_contracts(self, contracts:List[Symbol], option_right:int, strike:float) -> List:
161return [i for i in contracts if i.ID.OptionRight == option_right \
162    and i.ID.StrikePrice == strike and self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
163
164def Selection(self) -> None:
165self.selection_flag = True
166
167# Custom fee model
168class CustomFeeModel(FeeModel):
169def GetOrderFee(self, parameters):
170fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
171return OrderFee(CashAmount(fee, "USD"))