Exploiting Option Information in the Equity Market
Log in to collectAcademic paper
Strategy in a nutshell
Trade the 100 largest U.S. stocks using options-implied signals. Combine four standardized volatility-based measures into a composite score to rank stocks weekly. Go long top-quintile (bullish signals), short bottom-quintile (bearish), rebalancing equally each week.
Economic rationale
Options markets often reveal informed trading earlier than equities. Due to leverage and lower frictions, option prices embed private information first—creating predictive power for subsequent stock returns.
Backtest performance
Annualised return7.61%
Volatility6.69%
Beta0.417
Sharpe ratio0.54
Sortino ratio-0.009
Win rate59%
Full Python code
from AlgorithmImports import *
from typing import Dict, List
class ExploitingOptionInformationintheEquityMarket(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2015, 1, 1)
self.SetCash(1000000)
self.min_expiry:int = 25
self.max_expiry:int = 35
self.daily_period:int = 21 # stock daily price period
self.vol_skew_change_period:int = 5 # weekly change in ATM vol skew period
self.min_share_price:int = 5
self.leverage:int = 20
self.quantile:int = 5
self.price_threshold:float = [0.8 , 0.95]
self.prices:Dict[Symbol, RollingWindow] = {}
self.symbols_by_ticker:Dict[str, Symbol] = {}
self.subscribed_contracts:Dict[Symbol, Contracts] = {}
self.ATM_volatility_skew_values:Dict[Symbol, RollingWindow] = {}
self.day:int = -1
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.fundamental_count:int = 100
self.selection_flag:bool = True
self.rebalance_flag:bool = False
self.wait_one_day = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Minute
self.AddUniverse(self.FundamentalSelectionFunction)
self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
market = self.AddEquity('SPY', Resolution.Minute).Symbol
self.Schedule.On(self.DateRules.Every(DayOfWeek.Tuesday), self.TimeRules.BeforeMarketClose(market), self.Rebalance)
def Rebalance(self) -> None:
self.rebalance_flag = True
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
for security in changes.RemovedSecurities:
symbol:Symbol = security.Symbol
if symbol in self.ATM_volatility_skew_values:
# delete ATM vol skew values
del self.ATM_volatility_skew_values[symbol]
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# update daily prices of stocks in self.data dictionary
for stock in fundamental:
symbol:Symbol = stock.Symbol
if symbol in self.prices:
self.prices[symbol].Add(stock.AdjustedPrice)
# rebalance, when contracts expiried
if not self.selection_flag:
return Universe.Unchanged
# select top n stocks by dollar volume
selected:List[Fundamental] = [
x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.Price > self.min_share_price
]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
selected_symbols:List[Symbol] = []
for stock in selected:
symbol:Symbol = stock.Symbol
ticker:str = symbol.Value
selected_symbols.append(symbol)
self.symbols_by_ticker[ticker] = symbol
if symbol in self.prices:
continue
self.prices[symbol] = RollingWindow[float](self.daily_period)
history:DataFrame = self.History(symbol, self.daily_period, Resolution.Daily)
if history.empty:
continue
closes:Series = history.loc[symbol].close
for time, close in closes.items():
self.prices[symbol].Add(close)
# return newly selected symbols
return selected_symbols
def OnData(self, data: Slice) -> None:
# execute once a day
if self.day == self.Time.day:
return
if not (self.Time.hour == 9 and self.Time.minute == 31):
return
self.day = self.Time.day
# check if any of the subscribed contracts expired
for _, symbol in self.symbols_by_ticker.items():
if symbol in self.subscribed_contracts and self.subscribed_contracts[symbol].expiry_date <= self.Time.date():
# remove expired contracts
for contract in [self.subscribed_contracts[symbol].atm_call, self.subscribed_contracts[symbol].atm_put, self.subscribed_contracts[symbol].otm_put]:
if self.Securities[contract].IsTradable:
# self.RemoveSecurity(contract)
self.Liquidate(contract)
# remove Contracts object for current symbol
del self.subscribed_contracts[symbol]
# perform next selection, when there are no active contracts
if len(self.subscribed_contracts) == 0 and not self.selection_flag:
# liquidate leftovers
if self.Portfolio.Invested:
self.Liquidate()
self.symbols_by_ticker.clear()
self.selection_flag = True
return
# subscribe to new contracts after selection
elif len(self.subscribed_contracts) == 0 and self.selection_flag:
self.selection_flag = False
for _, symbol in self.symbols_by_ticker.items():
if symbol in self.prices:
if self.prices[symbol].IsReady:
# get all contracts for current stock symbol
contracts:List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
# get current price for etf
underlying_price:float = self.prices[symbol][0]
# get strikes from commodity future contracts
strikes:List[float] = [i.ID.StrikePrice for i in contracts]
# can't filter contracts, if there isn't any strike price
if len(strikes) <= 0 or underlying_price == 0:
continue
# filter calls and puts contracts with one month expiry
atm_calls, atm_puts = self.FilterContracts(strikes, contracts, underlying_price, OptionType.ATM)
_, otm_puts = self.FilterContracts(strikes, contracts, underlying_price, OptionType.OTM)
# make sure, there is at least one call and put contract
if len(atm_calls) > 0 and len(atm_puts) > 0 and len(otm_puts) > 0:
# sort by expiry
atm_call:Symbol = sorted(atm_calls, key = lambda x: x.ID.Date, reverse=True)[0]
atm_put:Symbol = sorted(atm_puts, key = lambda x: x.ID.Date, reverse=True)[0]
otm_put:Symbol = sorted(otm_puts, key = lambda x: x.ID.Date, reverse=True)[0]
subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(atm_call.Underlying)
if subscriptions:
# add contracts
self.AddContract(atm_call)
self.AddContract(atm_put)
self.AddContract(otm_put)
# retrieve expiry date for contracts
expiry_date:datetime.date = atm_call.ID.Date.date() if atm_call.ID.Date.date() < atm_put.ID.Date.date() else atm_put.ID.Date.date()
expiry_date:datetime.date = expiry_date if expiry_date < otm_put.ID.Date.date() else otm_put.ID.Date.date()
# store contracts with expiry date under stock's symbol
self.subscribed_contracts[symbol] = Contracts(expiry_date, underlying_price, atm_call, atm_put, otm_put)
# calculate measures
elif len(self.subscribed_contracts) != 0 and data.OptionChains.Count != 0:
if self.rebalance_flag:
OTM_volatility_skew:Dict[Symbol, float] = {}
volatility_spread:Dict[Symbol, float] = {}
ATM_volatility_skew:Dict[Symbol, float] = {}
ATM_volatility_skew_change:Dict[Symbol, float] = {}
for kvp in data.OptionChains:
chain:OptionChain = kvp.Value
ticker:str = chain.Underlying.Symbol.Value
if ticker in self.symbols_by_ticker:
# get stock's symbol
symbol:Symbol = self.symbols_by_ticker[ticker]
# get contracts
contracts:List[OptionConctract] = [x for x in chain]
# check if there are enough contracts for option and daily prices are ready
if len(contracts) < 3 or not self.prices[symbol].IsReady:
continue
# get call and put implied volatility
atm_call_iv, atm_put_iv, otm_put_iv = self.GetImpliedVolatilities(contracts)
if atm_call_iv is not None and atm_put_iv is not None and otm_put_iv is not None:
ATM_vol_skew:float = atm_put_iv - atm_call_iv
# store daily vol skew value
if symbol not in self.ATM_volatility_skew_values:
self.ATM_volatility_skew_values[symbol] = RollingWindow[float](self.vol_skew_change_period)
self.ATM_volatility_skew_values[symbol].Add(ATM_vol_skew)
if self.rebalance_flag:
if self.ATM_volatility_skew_values[symbol].IsReady:
if self.ATM_volatility_skew_values[symbol][4] != 0 and self.ATM_volatility_skew_values[symbol][0] != 0:
# 1.measure
OTM_volatility_skew[symbol] = otm_put_iv - atm_call_iv
# 2.measure
hv:float = self.GetHistoricalVolatility(self.prices[symbol])
atm_iv:float = (atm_call_iv + atm_put_iv) / 2
volatility_spread[symbol] = hv - atm_iv
# 3.measure
ATM_volatility_skew[symbol] = ATM_vol_skew
# 4.measure
# weekly change in ATM vol skew
ATM_volatility_skew_change[symbol] = self.ATM_volatility_skew_values[symbol][0] / self.ATM_volatility_skew_values[symbol][4] - 1
if self.rebalance_flag:
self.rebalance_flag = False
# can't perform selection
if len(OTM_volatility_skew) < self.quantile:
self.Liquidate()
return
# z-score normalization
OTM_volatility_skew_score:Dict[Symbol, float] = self.StandardizedZscore(OTM_volatility_skew)
volatility_spread_score:Dict[Symbol, float] = self.StandardizedZscore(volatility_spread)
ATM_volatility_skew_score:Dict[Symbol, float] = self.StandardizedZscore(ATM_volatility_skew)
ATM_volatility_skew_change_score:Dict[Symbol, float] = self.StandardizedZscore(ATM_volatility_skew_change)
# score calculation
combined_zscore:Dict[Symbol, float] = { x[0] : np.mean([x[1], volatility_spread_score[x[0]], ATM_volatility_skew_score[x[0]], ATM_volatility_skew_change_score[x[0]]]) for x in OTM_volatility_skew_score.items() }
# perform selection
quantile:int = int(len(combined_zscore) / self.quantile)
sorted_by_zscore:List[Symbol] = [x[0] for x in sorted(combined_zscore.items(), key=lambda item: item[1], reverse=True)]
# long top
long:List[Symbol] = sorted_by_zscore[-quantile:]
# short bottom
short:List[Symbol] = sorted_by_zscore[:quantile]
# trade execution
invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
for symbol in invested:
if symbol not in long + short:
# self.Liquidate(symbol)
self.MarketOnCloseOrder(symbol, -self.Portfolio[symbol].Quantity)
long_n:int = len(long)
short_n:int = len(short)
for symbol in long:
if symbol in data and data[symbol]:
q = int((self.Portfolio.TotalPortfolioValue / long_n) / self.prices[symbol][0])
self.MarketOnCloseOrder(symbol, q)
for symbol in short:
if symbol in data and data[symbol]:
q = int((self.Portfolio.TotalPortfolioValue / short_n) / self.prices[symbol][0])
self.MarketOnCloseOrder(symbol, -q)
def StandardizedZscore(self, collection:dict) -> dict:
collection_values = list(collection.values())
# median
collection_values_median:float = np.median(collection_values)
# avg
collection_values_avg:float = np.mean(collection_values)
# median absolute deviation
collection_med:float = np.median(np.array([abs(x-collection_values_median) for x in collection_values]))
max_cap:int = 3
result:dict = { x[0]: min(max_cap, max((x[1] - collection_values_avg) / collection_med, -max_cap)) for x in collection.items() }
return result
def FilterContracts(self, strikes, contracts, underlying_price, option_type) -> tuple():
''' filter call and put contracts from contracts parameter '''
''' returns call and put contracts tuple '''
strike:float = None
if option_type == OptionType.ATM:
# at the money strike
strikes:List[float] = [x for x in strikes if x > self.price_threshold[1] *underlying_price]
if len(strikes) != 0:
strike:float = min(strikes, key=lambda x: abs(x-underlying_price))
elif option_type == OptionType.OTM:
# out the money
strikes:List[float] = [x for x in strikes if x <= self.price_threshold[1] *underlying_price and x > self.price_threshold[0] *underlying_price]
if len(strikes) != 0:
strike:float = min(strikes, key=lambda x: abs(x-(underlying_price* self.price_threshold[0])))
calls:List[Symbol] = [] # storing call contracts
puts:List[Symbol] = [] # storing put contracts
if strike is not None:
for contract in contracts:
# check if contract has one month expiry
if self.min_expiry < (contract.ID.Date - self.Time).days < self.max_expiry:
# check if contract is call
if contract.ID.OptionRight == OptionRight.Call and contract.ID.StrikePrice == strike:
calls.append(contract)
# check if contract is put
elif contract.ID.OptionRight == OptionRight.Put and contract.ID.StrikePrice == strike:
puts.append(contract)
# return filtered calls and puts with one month expiry
return calls, puts
def AddContract(self, contract):
''' subscribe option contract, set price model and normalization mode '''
option:OptionContract = self.AddOptionContract(contract, Resolution.Minute)
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
def GetImpliedVolatilities(self, contracts):
''' retrieve implied volatility of contracts from contracts parameteres '''
''' returns call and put implied volatility '''
atm_call_iv:Union[None, float] = None
atm_put_iv:Union[None, float] = None
otm_put_iv:Union[None, float] = None
underlying_symbol:Symbol = contracts[0].Symbol.Underlying
if underlying_symbol in self.subscribed_contracts:
# go through option contracts
for c in contracts:
# atm call IV
if c.Symbol.Value == self.subscribed_contracts[underlying_symbol].atm_call.Value:
atm_call_iv = c.ImpliedVolatility
# atm put IV
elif c.Symbol.Value == self.subscribed_contracts[underlying_symbol].atm_put.Value:
atm_put_iv = c.ImpliedVolatility
# otm put IV
elif c.Symbol.Value == self.subscribed_contracts[underlying_symbol].otm_put.Value:
otm_put_iv = c.ImpliedVolatility
return atm_call_iv, atm_put_iv, otm_put_iv
def GetHistoricalVolatility(self, rolling_window_prices) -> float:
''' calculate historical volatility based on daily prices in rolling_window_prices parameter '''
prices:np.ndarray = np.array([x for x in rolling_window_prices])
returns:np.ndarray = (prices[:-1] - prices[1:]) / prices[1:]
return np.std(returns)
from enum import Enum
class OptionType(Enum):
ATM:str = 'atm'
OTM:str = 'otm'
ITM:str = 'itm'
class Contracts():
def __init__(self, expiry_date:datetime.date, underlying_price:float, atm_call:Symbol, atm_put:Symbol, otm_put:Symbol) -> None:
self.expiry_date:datetime.date = expiry_date
self.underlying_price:float = underlying_price
# self.contracts = contracts # = [atm_call, atm_put, otm_put]
self.atm_call:Symbol = atm_call
self.atm_put:Symbol = atm_put
self.otm_put:Symbol = otm_put
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))