Option Gamma Predicts Stock Returns

1from AlgorithmImports import *
2from typing import List, Dict
3# endregion
4
5class OptionGammaPredictsStockReturns(QCAlgorithm):
6
7def Initialize(self):
8self.SetStartDate(2015, 1, 1)
9self.SetCash(100_000)
10
11self.leverage: int = 5
12self.quantile: int = 10
13
14self.min_contracts: int = 3
15self.one_stock_contracts: int = 6 * self.min_contracts
16
17self.min_expiry: int = 35
18self.max_expiry: int = 360
19
20self.exchanges: List[str] = ['NYS', 'NAS', 'ASE']
21
22self.market_cap: Dict[Symbol, float] = {}
23self.subscribed_options: List[OptionContract] = []
24self.symbols_by_tickers: Dict[str, Symbol] = {}
25
26self.market_symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
27
28self.fundamental_count: int = 300
29self.fundamental_sorting_key = lambda x: x.MarketCap
30
31self.selection_flag: bool = False
32self.UniverseSettings.Resolution = Resolution.Daily
33self.AddUniverse(self.FundamentalSelectionFunction)
34self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
35self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
36
37self.Schedule.On(self.DateRules.MonthStart(self.market_symbol), self.TimeRules.BeforeMarketClose(self.market_symbol, 0), self.Selection)
38
39def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
40for security in changes.AddedSecurities:
41    security.SetFeeModel(CustomFeeModel())
42    security.SetLeverage(self.leverage)
43
44def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
45if not self.selection_flag:
46    return Universe.Unchanged
47
48selected: List[Fundamental] = [
49    f for f in fundamental if f.HasFundamentalData and \
50    f.MarketCap != 0 and \
51    f.SecurityReference.ExchangeId in self.exchanges
52]
53
54if len(selected) > self.fundamental_count:
55    selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
56
57self.subscribed_options.clear()
58self.market_cap = { stock.Symbol: stock.MarketCap for stock in selected }
59
60return list(self.market_cap.keys()) 
61
62def OnData(self, data: Slice) -> None:
63if self.selection_flag:
64    self.selection_flag = False
65    self.Liquidate()
66
67    for symbol in self.market_cap:
68        # subscribe to contract
69        contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
70        underlying_price: float = self.Securities[symbol].Price
71        
72        strikes: List[float] = [i.ID.StrikePrice for i in contracts]
73        
74        if len(strikes) <= 0:
75            continue
76        
77        atm_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
78        itm_strike: float = min(strikes, key=lambda x: abs(x-(underlying_price*0.95)))
79        otm_strike: float = min(strikes, key=lambda x: abs(x-(underlying_price*1.05)))
80
81        atm_calls: List[Symbol] = self.SelectContracts(contracts, OptionRight.Call, atm_strike)
82        atm_puts: List[Symbol] = self.SelectContracts(contracts, OptionRight.Put, atm_strike)
83
84        itm_calls: List[Symbol] = self.SelectContracts(contracts, OptionRight.Call, itm_strike)
85        itm_puts: List[Symbol] = self.SelectContracts(contracts, OptionRight.Put, itm_strike)
86
87        otm_calls: List[Symbol] = self.SelectContracts(contracts, OptionRight.Call, otm_strike)
88        otm_puts: List[Symbol] = self.SelectContracts(contracts, OptionRight.Put, otm_strike)
89            
90        # make sure there are enough contracts
91        if not all(len(x) >= self.min_contracts for x in [atm_calls, atm_puts, itm_calls, itm_puts, otm_calls, otm_puts]):
92            continue
93        
94        # store stock's symbol under it's ticker, because it is the only possibility how to access it from option contract
95        self.symbols_by_tickers[symbol.Value] = symbol
96
97        # sort by expiry and subscribe n nearest contracts
98        for selected_contracts in [atm_calls, atm_puts, itm_calls, itm_puts, otm_calls, otm_puts]:
99            nearest_contracts: List[Symbol] = sorted(selected_contracts, key=lambda item: item.ID.Date)[:self.min_contracts]
100
101            for contract in nearest_contracts:
102                option: OptionContract = self.AddOptionContract(contract, Resolution.Daily)
103                option.PriceModel = OptionPriceModels.CrankNicolsonFD()
104
105                # after trade execution subscribed options will be unsubscribed,
106                # to stop receiving unncessary data, which slows down strategy
107                self.subscribed_options.append(option)
108
109if len(self.symbols_by_tickers) != 0 and data.OptionChains.Count != 0:
110    gamma_weighted_sum: Dict[Symbol, float] = {}
111    
112    for kvp in data.OptionChains:
113        chain: OptionChain = kvp.Value
114        ticker: str = chain.Underlying.Symbol.Value
115        if ticker not in self.symbols_by_tickers:
116            continue
117
118        symbol: Symbol = self.symbols_by_tickers[ticker]
119        contracts: List[OptionContract] = [x for x in chain]
120        
121        # check if there are enough contracts for option
122        if len(contracts) < self.one_stock_contracts:
123            continue
124        
125        gamma_weighted_sum_value: float = sum([(c.Greeks.Gamma * c.OpenInterest) for c in contracts])
126        gamma_weighted_sum[symbol] = gamma_weighted_sum_value
127
128    self.symbols_by_tickers.clear() # clear list to prevent next gamma weight sum calculation
129    # unsubscribe option contracts, to stop receiving data
130    for option in self.subscribed_options:
131        self.RemoveOptionContract(option.Symbol)
132
133    # make sure there are enough stocks with gamma value for selection
134    if len(gamma_weighted_sum) < self.quantile:
135        return
136
137    quantile: int = int(len(gamma_weighted_sum) / self.quantile)
138    sorted_by_gamma_weighted_sum: List[Symbol] = [x[0] for x in sorted(gamma_weighted_sum.items(), key=lambda item: item[1])]
139    long_leg: List[Symbol] = sorted_by_gamma_weighted_sum[-quantile:]
140    short_leg: List[Symbol] = sorted_by_gamma_weighted_sum[:quantile]
141
142    # trade execution
143    total_long_cap: float = sum(list(map(lambda symbol: self.market_cap[symbol], long_leg)))
144    for symbol in long_leg:
145        self.SetHoldings(symbol, self.market_cap[symbol] / total_long_cap)
146
147    total_short_cap: float = sum(list(map(lambda symbol: self.market_cap[symbol], short_leg)))
148    for symbol in short_leg:
149        self.SetHoldings(symbol, -self.market_cap[symbol] / total_short_cap)
150
151def SelectContracts(self, contracts: List[Symbol], option_right: int, strike: float) -> List[Symbol]:
152return [i for i in contracts if i.ID.OptionRight == option_right \
153    and i.ID.StrikePrice == strike and self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
154
155def Selection(self) -> None:
156self.selection_flag = True
157
158# Custom fee model
159class CustomFeeModel(FeeModel):
160def GetOrderFee(self, parameters):
161fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
162return OrderFee(CashAmount(fee, "USD"))