The strategy trades US stocks with liquid options, shorting high O/S ratio changes and going long on low changes, with equal weighting and monthly rebalancing based on six-month averages.
ASSET CLASS: stocks | REGION: United States | FREQUENCY:
Monthly | MARKET: equities | KEYWORD: Change, Option/Stock, Volume, Ratio, Stock, Returns
I. STRATEGY IN A NUTSHELL
This strategy focuses on U.S. stocks with active options, excluding CEFs, REITs, ADRs, and stocks under $1. Each month, it measures the change in the option-to-stock (O/S) volume ratio over six months. Stocks with high O/S changes are shorted, and those with low changes are bought. The portfolio is equally weighted and rebalanced monthly to capture predictive signals from options activity.
II. ECONOMIC RATIONALE
High O/S ratio changes predict lower future stock returns because shorting equities is costly. Informed traders often use options to act on negative information, so large option activity signals bearish sentiment, which drives the predictive relationship.
III. SOURCE PAPER
The Option to Stock Volume Ratio and Future Returns [Click to Open PDF]
Johnson, So, Stanford University – Graduate School of Business
<Abstract>
We examine the information content of option and equity volumes when trade direction is unobserved. In a multimarket symmetric information model, we show that equity short-sale costs result in a negative relation between relative option volume and future firm value. In our empirical tests, firms in the lowest decile of the option to stock volume ratio (O/S) outperform the highest decile by 1.47% per month on a risk-adjusted basis. Our model and empirics both indicate that O/S is a stronger signal when short-sale costs are high or option leverage is low. O/S also predicts future firm-specific earnings news, consistent with O/S reflecting private information.
IV. BACKTEST PERFORMANCE
| Annualised Return | 14.65% |
| Volatility | 15.5% |
| Beta | -0.03 |
| Sharpe Ratio | 0.69 |
| Sortino Ratio | -0.119 |
| Maximum Drawdown | N/A |
| Win Rate | 50% |
V. FULL PYTHON CODE
from AlgorithmImports import *
from typing import List, Dict
from dataclasses import dataclass
#endregion
class ChangeInOptionStockVolumeRatioPredictsStockReturns(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2014, 1, 1)
self.SetCash(100000)
self.tickers_to_ignore: List[str] = ['AMD', 'TSLA', 'MSFT']
self.min_expiry: int = 20
self.max_expiry: int = 30
self.period: int = 6 # need n monthly volumes
self.min_period_len: int = 14 # need at least n daily volumes and at least n minute volumes
self.quantile: int = 5
self.leverage: int = 15
self.min_share_price: int = 5
self.current_fundamental: List[Symbol] = []
self.previous_fundamental: List[Symbol] = []
self.data: Dict[Symbol, SymbolData] = {}
self.subscribed_contracts: Dict[Symbol, Contracts] = {} # subscribed option universe
# initial data feed
self.AddEquity('SPY', Resolution.Minute)
self.months_counter: int = 1
self.fundamental_count: int = 100
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag: bool = True
self.subscribing_flag: bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Minute
self.settings.daily_precise_end_time = False
self.AddUniverse(self.FundamentalSelectionFunction)
self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
for security in changes.AddedSecurities:
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(self.leverage)
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# rebalance yearly
if not self.selection_flag:
return Universe.Unchanged
# change flags values
self.selection_flag = False
self.subscribing_flag = True
# filter top n U.S. 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
and x.Symbol.Value not in self.tickers_to_ignore
]
if len(selected) > self.fundamental_count:
selected = [
x for x in sorted(
selected,
key=self.fundamental_sorting_key,
reverse=True)[:self.fundamental_count]]
for stock in selected:
symbol: Symbol = stock.Symbol
self.current_fundamental.append(symbol)
# make sure data are consecutive
if symbol not in self.data or symbol not in self.previous_fundamental:
self.data[symbol] = SymbolData(self.period)
# return newly selected symbols
return self.current_fundamental
def OnData(self, data: Slice) -> None:
for stock_symbol in self.current_fundamental:
# stock has to have subscribed option contracts
if stock_symbol not in self.subscribed_contracts:
continue
if self.Securities[stock_symbol].IsDelisted:
continue
# check if any of the subscribed contracts expired
if self.subscribed_contracts[stock_symbol].expiry_date - timedelta(days=1) <= self.Time.date():
for c in self.subscribed_contracts[stock_symbol].contracts:
self.RemoveOptionContract(c)
self.subscribed_contracts[stock_symbol].contracts.clear()
# remove Contracts object for current symbol
del self.subscribed_contracts[stock_symbol]
else:
# collect volumes
stock_volume: Union[None, float] = data[stock_symbol].Value if stock_symbol in data and data[stock_symbol] else None
option_volumes: List[float] = []
option_contracts: List[Symbol] = self.subscribed_contracts[stock_symbol].contracts
for option_contract in option_contracts:
if option_contract in data and data[option_contract]:
# option volume isn't in data object
option_volumes.append(self.Securities[option_contract].Volume)
# make sure all volumes were collected
if stock_volume is not None and len(option_volumes) == len(option_contracts):
# store volumes
if stock_symbol not in self.data:
self.data[stock_symbol] = SymbolData(self.period)
# store minute stock volume
self.data[stock_symbol].stock_minute_volumes.append(stock_volume)
# store total minute option volume
self.data[stock_symbol].options_minute_volumes.append(sum(option_volumes))
if stock_symbol in self.data:
if self.Time.hour == 16 and self.Time.minute == 0 and self.data[stock_symbol].minute_volumes_ready(self.min_period_len):
self.data[stock_symbol].update_daily_volumes()
# perform trade, when there are no active contracts for current selection
if len(self.subscribed_contracts) == 0 and not self.subscribing_flag and self.Time.hour != 0:
OS_ratio_change: Dict[Symbol, float] = {}
for stock_symbol in self.current_fundamental:
if stock_symbol not in self.data:
continue
if self.Securities[stock_symbol].IsDelisted:
continue
symbol_obj: SymbolData = self.data[stock_symbol]
# each stock has to have at least minimum daily volumes
if symbol_obj.daily_volumes_ready(self.min_period_len):
symbol_obj.update_os_ratios()
else:
self.data[stock_symbol].clear_data()
del self.data[stock_symbol]
# OS ratios data has to be ready
if not symbol_obj.is_ready():
continue
OS_ratios_values: List[float] = [x for x in symbol_obj.os_ratios]
mean_os_ratios_value: float = np.mean(OS_ratios_values)
OS_ratio_change_value: float = (OS_ratios_values[0] - mean_os_ratios_value) / mean_os_ratios_value
# store OS ratio change keyed by stock symbol
OS_ratio_change[stock_symbol] = OS_ratio_change_value
# make sure there are enough stocks with data
if len(OS_ratio_change) >= self.quantile:
# perform selection
quantile: int = int(len(OS_ratio_change) / self.quantile)
sorted_by_ratio: List[Symbol] = [x[0] for x in sorted(OS_ratio_change.items(), key=lambda item: item[1])]
# long low and short high
long: List[Symbol] = sorted_by_ratio[:quantile]
short: List[Symbol] = sorted_by_ratio[-quantile:]
# trade execution
targets: List[PortfolioTarget] = []
for i, portfolio in enumerate([long, short]):
for symbol in portfolio:
if symbol in data and data[symbol]:
targets.append(PortfolioTarget(symbol, ((-1) ** i) / len(portfolio)))
self.SetHoldings(targets, True)
elif not self.selection_flag:
# liquidate all positions from previous selection
self.Liquidate()
if len(self.current_fundamental) != 0 and self.months_counter % 12 == 0:
# reinitialize previous fundamental
self.previous_fundamental = list(map(lambda symbol: symbol, self.current_fundamental))
# make space for new stocks from fundamental
self.current_fundamental.clear()
# increase month counter
self.months_counter += 1
# perform next selection
self.selection_flag = True
elif not self.subscribing_flag and len(self.current_fundamental) != 0:
# perform new subscribtion without selection
self.subscribing_flag = True
# increase months counter
self.months_counter += 1
return # skip to firstly perform fundamental selection and then contracts subscribing
# subscribe to new contracts after selection
if len(self.subscribed_contracts) == 0 and self.subscribing_flag:
for symbol in self.current_fundamental:
if self.Securities[symbol].IsDelisted:
continue
# 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.Securities[symbol].Price
# 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
atm_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
itm_strike: float = min(strikes, key=lambda x: abs(x-(underlying_price*0.95)))
otm_strike: float = min(strikes, key=lambda x: abs(x-(underlying_price*1.05)))
# filter calls and puts contracts with one month expiry
atm_calls, atm_puts = self.FilterContracts(atm_strike, contracts, underlying_price)
itm_calls, itm_puts = self.FilterContracts(itm_strike, contracts, underlying_price)
otm_calls, otm_puts = self.FilterContracts(otm_strike, contracts, underlying_price)
# make sure, there is at least one call and put contract
if len(atm_calls) > 0 and len(atm_puts) > 0 and len(itm_calls) > 0 and len(itm_puts) > 0 and len(otm_calls) > 0 and len(otm_puts) > 0:
# sort by expiry
atm_call, atm_put = self.SortByExpiry(atm_calls, atm_puts)
itm_call, itm_put = self.SortByExpiry(itm_calls, itm_puts)
otm_call, otm_put = self.SortByExpiry(otm_calls, otm_puts)
atm_call_subscriptions: List[SubscriptionDataConfig] = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(atm_call.Underlying)
# check if stock's call and put contract was successfully subscribed
if atm_call_subscriptions:
selected_contracts: List[Symbol] = [atm_call, atm_put, itm_call, itm_put, otm_call, otm_put]
for contract in selected_contracts:
# add contract
self.AddOptionContract(contract, Resolution.Minute)
# 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()
# store contracts with expiry date under stock's symbol
self.subscribed_contracts[symbol] = Contracts(expiry_date, underlying_price, selected_contracts)
# at least one stock has to have successfully subscribed all option contracts, to stop subscribing
if len(self.subscribed_contracts) > 0:
self.subscribing_flag = False
def FilterContracts(self,
strike: float,
contracts: List[Symbol],
underlying_price: float) -> List[Symbol]:
''' filter call and put contracts from contracts parameter '''
''' return call and put contracts '''
calls: List[Symbol] = [] # storing call contracts
puts: List[Symbol] = [] # storing put contracts
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 SortByExpiry(self,
calls: List[Symbol],
puts: List[Symbol]) -> List[Symbol]:
''' return option call and option put with farest expiry '''
call: List[Symbol] = sorted(calls, key = lambda x: x.ID.Date, reverse=True)[0]
put: List[Symbol] = sorted(puts, key = lambda x: x.ID.Date, reverse=True)[0]
return call, put
class SymbolData:
def __init__(self, period: int) -> None:
self.os_ratios: RollingWindow = RollingWindow[float](period)
self.stock_minute_volumes: List[float] = []
self.options_minute_volumes: List[float] = []
self.stock_daily_volumes: List[float] = []
self.options_daily_volumes: List[float] = []
def update_daily_volumes(self) -> None:
self.stock_daily_volumes.append(sum(self.stock_minute_volumes))
self.options_daily_volumes.append(sum(self.options_minute_volumes))
self.stock_minute_volumes.clear()
self.options_minute_volumes.clear()
def update_os_ratios(self) -> None:
os_ratio_value = sum(self.options_daily_volumes) / sum(self.stock_daily_volumes)
self.os_ratios.Add(os_ratio_value)
self.stock_daily_volumes.clear()
self.options_daily_volumes.clear()
def clear_data(self) -> None:
self.stock_minute_volumes.clear()
self.options_minute_volumes.clear()
self.stock_daily_volumes.clear()
self.options_daily_volumes.clear()
def daily_volumes_ready(self, period: int) -> bool:
return len(self.stock_daily_volumes) >= period and len(self.options_daily_volumes) >= period
def minute_volumes_ready(self, period: int) -> bool:
return len(self.stock_minute_volumes) >= period and len(self.options_minute_volumes) >= period
def is_ready(self) -> bool:
return self.os_ratios.IsReady
@dataclass
class Contracts():
expiry_date: datetime.date
underlying_price: float
contracts: List[Symbol]
# Custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters: OrderFeeParameters) -> OrderFee:
fee: float = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))