Options Convexity Predicts Consecutive Stock Returns

1import numpy as np
2from AlgorithmImports import *
3from typing import Dict, List, Tuple
4class OptionsConvexityPredictsConsecutiveStockReturns(QCAlgorithm):
5def Initialize(self) -> None:
6self.SetStartDate(2015, 1, 1)
7self.SetCash(100000)
8
9self.min_expiry: int = 25
10self.max_expiry: int = 35
11self.quantile: int = 5
12self.leverage: int = 5
13self.min_share_price: int = 5
14self.contracts_count: int = 3
15self.thresholds: List[int] = [0.95, 1.05]
16self.next_expiry: Union[None, datetime.date] = None
17
18self.fundamental_count: int = 100
19self.fundamental_sorting_key = lambda x: x.DollarVolume
20self.selection_flag: bool = False
21self.stock_universe: List[Symbol] = []
22self.option_universe: Dict[Symbol, List[Symbol]] = {}
23self.contracts_expiry: Dict[Symbol, datetime.date] = {} # storing contracts expiry date under symbols
24self.settings.daily_precise_end_time = False
25self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
26self.UniverseSettings.Resolution = Resolution.Minute
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
29self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
30self.current_day: int = -1
31symbol: Symbol = self.AddEquity("SPY", Resolution.Daily).Symbol
32self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.AfterMarketOpen(symbol), self.Selection)
33
34def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
35for security in changes.AddedSecurities:
36    security.SetFeeModel(CustomFeeModel())
37    security.SetLeverage(self.leverage)
38# remove old option contracts
39for security in changes.RemovedSecurities:
40    symbol = security.Symbol
41    if symbol in self.option_universe:
42        for option in self.option_universe[symbol]:
43            self.RemoveSecurity(option)
44        del self.option_universe[symbol]
45        self.Liquidate(symbol)
46def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
47# new symbol selection once a quarter
48if not self.selection_flag:
49    return Universe.Unchanged
50selected: List[Fundamental] = [
51    x for x in fundamental 
52    if x.HasFundamentalData 
53    and x.Market == 'usa' 
54    and x.Price > self.min_share_price
55]
56if len(selected) > self.fundamental_count:
57    selected = [
58    x for x in sorted(
59        selected, 
60        key=self.fundamental_sorting_key, 
61        reverse=True)[:self.fundamental_count]
62        ]     
63
64self.stock_universe = [x.Symbol for x in selected]
65return self.stock_universe
66
67def Selection(self) -> None:
68if self.Time.month % 3 == 0:
69    self.selection_flag = True
70    self.Liquidate()
71
72def OnData(self, data: Slice) -> None:
73# rebalance daily
74if self.current_day == self.Time.day:
75    return
76self.current_day = self.Time.day
77
78if self.next_expiry and self.Time.date() >= self.next_expiry.date():
79    for symbol in self.option_universe:
80        for option in self.option_universe[symbol]:
81            self.RemoveSecurity(option)
82    self.Liquidate()
83# for symbol in self.option_universe:
84#     # subscribe to new contracts, because current ones has expiried
85#     if symbol not in self.contracts_expiry or self.contracts_expiry[symbol] <= self.Time.date():
86    
87if not self.Portfolio.Invested:
88    for symbol in self.stock_universe:
89        contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
90        underlying_price: float = self.Securities[symbol].Price
91        
92        if self.Securities[symbol].IsDelisted:
93            continue
94        strikes: List[float] = [i.ID.StrikePrice for i in contracts]
95        if len(strikes) > 0:
96            atm_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
97            itm_strike: float = min(strikes, key=lambda x: abs(x-(underlying_price*min(self.thresholds))))
98            otm_strike: float = min(strikes, key=lambda x: abs(x-(underlying_price*max(self.thresholds))))
99            
100            atm_calls: List[Symbol] = [i for i in contracts if i.ID.OptionRight == OptionRight.Call and 
101                                                 i.ID.StrikePrice == atm_strike and 
102                                                 self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
103
104            itm_puts: List[Symbol] = [i for i in contracts if i.ID.OptionRight == OptionRight.Put and 
105                                                 i.ID.StrikePrice == itm_strike and 
106                                                 self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
107
108            otm_puts: List[Symbol] = [i for i in contracts if i.ID.OptionRight == OptionRight.Put and 
109                                                 i.ID.StrikePrice == otm_strike and 
110                                                 self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
111            
112            if len(atm_calls) > 0 and len(itm_puts) > 0 and len(otm_puts) > 0:
113                # sort by expiry
114                atm_call: List[Symbol] = sorted(atm_calls, key = lambda x: x.ID.Date)[0]
115                itm_put: List[Symbol] = sorted(itm_puts, key = lambda x: x.ID.Date)[0]
116                otm_put: List[Symbol] = sorted(otm_puts, key = lambda x: x.ID.Date)[0]
117                
118                # store expiry date
119                # self.contracts_expiry[symbol] = itm_put.ID.Date.date()
120                self.next_expiry = atm_call.ID.Date
121                # add contracts
122                option: Option = self.AddOptionContract(atm_call, Resolution.Minute)
123                option.PriceModel = OptionPriceModels.CrankNicolsonFD()
124                
125                option: Option = self.AddOptionContract(itm_put, Resolution.Minute)
126                option.PriceModel = OptionPriceModels.CrankNicolsonFD()
127                
128                option: Option = self.AddOptionContract(otm_put, Resolution.Minute)
129                option.PriceModel = OptionPriceModels.CrankNicolsonFD()
130                
131                options: List[Symbol] = [atm_call, itm_put, otm_put]
132                self.option_universe[symbol] = options           
133    
134    iv_convexity: Dict[Symbol, float] = {} 
135    if data.OptionChains.Count != 0:
136        for kvp in data.OptionChains:
137            chain: OptionChain = kvp.Value
138            contracts: List[Symbol] = [x for x in chain]
139            if len(contracts) == self.contracts_count:
140                atm_call_iv: Union[None, float] = None
141                itm_put_iv: Union[None, float] = None
142                otm_put_iv: Union[None, float] = None
143                symbol: Symbol = chain.Underlying.Symbol
144                for c in contracts:
145                    if c.Right == OptionRight.Call:
146                        # found atm call
147                        atm_call_iv = c.ImpliedVolatility
148                    else:
149                        # found put option
150                        underlying_price:float = self.Securities[c.UnderlyingSymbol].Price
151                        if c.Strike < underlying_price:
152                            # found itm put
153                            itm_put_iv = c.ImpliedVolatility
154                        else:
155                            # found otm put
156                            otm_put_iv = c.ImpliedVolatility
157                
158                if atm_call_iv and itm_put_iv and otm_put_iv:
159                    iv_convexity[symbol] = itm_put_iv + otm_put_iv - (2*atm_call_iv)
160    long: List[Symbol] = []
161    short: List[Symbol] = []
162
163    # convexity sorting
164    if len(iv_convexity) >= self.quantile:
165        sorted_by_convexity: List[Tuple[Symbol, float]] = sorted(iv_convexity.items(), key = lambda x: x[1], reverse = True)
166        quantile: int = int(len(sorted_by_convexity) / self.quantile)
167        long = [x[0] for x in sorted_by_convexity[-quantile:]]
168        short = [x[0] for x in sorted_by_convexity[:quantile]]
169    
170    # trade execution
171    targets: List[PortfolioTarget] = []
172    for i, portfolio in enumerate([long, short]):
173        for symbol in portfolio:
174            if symbol in data and data[symbol]:
175                targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
176    self.SetHoldings(targets, True)
177# custom fee model
178class CustomFeeModel(FeeModel):
179def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
180fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
181return OrderFee(CashAmount(fee, "USD"))