“该策略交易具有流动性期权的美国股票，卖空高期权/股票成交量比率的股票，买入低期权/股票成交量比率的股票，等权重配置，并每月再平衡以利用交易差异。”
资产类别: 股票 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 成交量
I. 策略概要
该策略目标是具有流动性期权的美国公司，排除CEFs、REITs、ADRs和价格低于1美元的股票。每月，投资者计算O/S比率，即总期权成交量（所有行权价，针对月底后五天开始的30个交易日内到期的期权）除以总股票成交量，进行标准化以考虑代表100股的期权合约。股票按O/S比率排名，投资者卖空高O/S比率的股票，买入低O/S比率的股票。投资组合等权重，每月再平衡，并寻求利用期权和股票交易活动之间的差异。
II. 策略合理性
研究表明，期权/股票成交量比率（O/S）与未来回报之间的负相关关系源于股票市场的卖空成本。这些成本使得期权市场成为交易者对负面消息采取行动的首选场所。资本约束和卖空股票的困难导致知情交易者更多地依赖期权来表达负面信号。因此，较高的相对期权成交量（O/S）表示看跌情绪并预测较低的未来股票回报。这种动态凸显了期权市场在反映股票市场不易交易的负面信息方面的作用。
III. 来源论文
期权与股票成交量比率与未来回报 [点击查看论文]
- 约翰逊，苏，斯坦福大学商学院，斯坦福大学商学院
<摘要>
我们研究在交易方向未知时期权和股票成交量的信息内容。在一个多市场对称信息模型中，我们展示了股票卖空成本导致相对期权成交量与未来公司价值之间存在负相关关系。在我们的实证检验中，期权/股票成交量比率（O/S）最低十分位数的公司在风险调整基础上每月比最高十分位数的公司高出1.47%。我们的模型和实证研究都表明，当卖空成本高或期权杠杆低时，O/S是一个更强的信号。O/S还预测未来的公司特定收益消息，这与O/S反映私人信息相一致。
IV. 回测表现
| 年化回报 | 14.54% |
| 波动率 | 19.2% |
| β值 | 0.06 |
| 夏普比率 | 0.55 |
| 索提诺比率 | -0.049 |
| 最大回撤 | N/A |
| 胜率 | 49% |
V. 完整的 Python 代码
from AlgorithmImports import *
from typing import List, Dict
#endregion
class OptionStockVolumeRatioPredictsStockReturns(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2020, 1, 1)
self.SetCash(100_000)
self.min_expiry: int = 20
self.max_expiry: int = 30
self.min_period_len: int = 14 # need at least n daily volumes
self.quantile: int = 5
self.leverage: int = 5
self.min_share_price: int = 5
self.last_fundamental: List[Symbol] = []
self.data: Dict[Symbol, SymbolData] = {} # list of stocks volumes and list of total option volumes in selection
self.subscribed_contracts: Dict[Symbol, Contracts] = {} # subscribed option universe
# initial data feed
self.AddEquity('SPY', Resolution.Minute)
self.fundamental_count: int = 50
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag: bool = True
self.subscribing_flag: bool = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.settings.daily_precise_end_time = False
self.UniverseSettings.Resolution = Resolution.Minute
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 monthly
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 != 'GOOG']
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
# filter top n U.S. stocks by dollar volume
# initialize new fundamental
self.last_fundamental = [x.Symbol for x in selected]
# return newly selected symbols
return self.last_fundamental
def OnData(self, data: Slice) -> None:
for stock_symbol in self.last_fundamental:
# stock has to have subscribed option contracts
if stock_symbol not in self.subscribed_contracts:
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()
# store stock volume in all stocks volumes in this day
self.data[stock_symbol].stock_minute_volumes.append(stock_volume)
# store total option volume in all total option volumes in this day
self.data[stock_symbol].options_minute_volumes.append(sum(option_volumes))
# execute once a day for storing stock and options volumes
if self.Time.hour == 16 and self.Time.minute == 00 and stock_symbol in self.data:
self.data[stock_symbol].update_daily_volumes()
# perform trade, then perform next selection, 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:
# calculate OS ratio
OS_ratio: Dict[Symbol, float] = {}
for stock_symbol, symbol_obj in self.data.items():
# make sure volumes data are ready
if symbol_obj.is_ready(self.min_period_len):
month_stock_volume: float = sum(symbol_obj.stock_daily_volumes)
if month_stock_volume != 0:
month_total_option_volume: float = sum(symbol_obj.total_option_daily_volumes)
OS_ratio_value: float = month_total_option_volume / month_stock_volume
# store OS ratio keyed by stock symbol
OS_ratio[stock_symbol] = OS_ratio_value
# clear last selection data
symbol_obj.clear_data()
if len(OS_ratio) >= self.quantile:
# perform selection
quantile: int = int(len(OS_ratio) / self.quantile)
sorted_by_ratio: List[Symbol] = [x[0] for x in sorted(OS_ratio.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:]
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()
# clear for next selection
self.last_fundamental.clear()
self.selection_flag = True
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.last_fundamental:
# 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 = min([c.ID.Date.date() for c in selected_contracts])
# 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) -> None:
self.stock_minute_volumes: List[float] = []
self.options_minute_volumes: List[float] = []
self.stock_daily_volumes: List[float] = []
self.total_option_daily_volumes: List[float] = []
def update_daily_volumes(self) -> None:
self.stock_daily_volumes.append(sum(self.stock_minute_volumes))
self.total_option_daily_volumes.append(sum(self.options_minute_volumes))
self.stock_minute_volumes.clear()
self.options_minute_volumes.clear()
def clear_data(self) -> None:
self.stock_minute_volumes.clear()
self.options_minute_volumes.clear()
self.stock_daily_volumes.clear()
self.total_option_daily_volumes.clear()
def is_ready(self, period: int) -> bool:
return len(self.stock_daily_volumes) >= period and len(self.total_option_daily_volumes) >= period
class Contracts():
def __init__(self, expiry_date: datetime.date, underlying_price: float, contracts: List[Symbol]) -> None:
self.expiry_date: datetime.date = expiry_date
self.underlying_price: float = underlying_price
self.contracts: List[Symbol] = contracts
# 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"))