“通过构建delta跨式期权交易美国股票期权，做多回报最高的五分位，做空回报最低的五分位，基于12个月平均回报，持有头寸至到期。”
资产类别: 期权 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 动量、跨式期权
I. 策略概要
投资范围包括来自OptionMetrics数据库的在美国月度周期内到期的美国股票期权，股票数据来自CRSP。在到期日，选择两对次月到期的看涨和看跌期权：一对基于未平仓合约为正，另一对看涨期权delta接近0.5。delta在0.25-0.75之外的期权被排除，重点关注平价期权。Delta跨式期权使用看跌和看涨期权的加权delta，基于买卖中点创建。股票根据12个月形成期（跳过最近一个月）的平均滞后回报分为五分位。该策略做多回报最高的五分位，做空回报最低的五分位，持有零delta跨式期权至到期。
II. 策略合理性
Jones、Khorram和Mo研究了delta对冲期权回报与波动率互换收益之间的关系，发现动量是由对过去波动率冲击反应不足以及由未定价的已实现波动率（而非隐含波动率）变化引起的季节性驱动的。他们方法的一个关键创新是基于形成期内的平均回报计算动量信号，这减轻了异常值的影响并创建了更对称的信号。他们的分析表明，最强的动量策略使用2到12个月的滞后，具有显著的回报差和减少的反转效应。反转仅在一个月短期的回溯期内显而易见，而长期研究结果证实，在6-36个月内表现良好的期权将继续表现良好。
III. 来源论文
Momentum, Reversal, and Seasonality in Option Returns [点击查看论文]
- Christopher S. Jones、Mehdi Khorram、Haitao Mo。南加州大学 – 马歇尔商学院 – 金融与商业经济系。罗切斯特理工学院（RIT）。堪萨斯大学
<摘要>
期权回报在6到36个月的形成期内显示出显著的动量，多头/空头投资组合的年化夏普比率超过1.5。短期内，期权回报呈现反转。期权在3个月和12个月滞后倍数上也显示出显著的季节性。所有这些结果在横截面和时间上都非常显著且稳定。在控制其他特征后，它们仍然强劲，并且动量和季节性在因子风险调整后仍然存在。动量主要由对过去波动率和其他冲击的反应不足来解释，而季节性则反映了股票回报波动率中未定价的季节性变化。
IV. 回测表现
| 年化回报 | 114.83% |
| 波动率 | 41.56% |
| β值 | 0.112 |
| 夏普比率 | 2.76 |
| 索提诺比率 | -0.78 |
| 最大回撤 | N/A |
| 胜率 | 41% |
V. 完整的 Python 代码
from AlgorithmImports import *
from typing import List, Dict
from dataclasses import dataclass
#endregion
class ReversalOnStraddles(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2015, 1, 1)
self.SetCash(100_000)
self.leverage: int = 20
self.quantile: int = 5
self.min_share_price: int = 5
self.min_expiry: int = 20
self.max_expiry: int = 30
self.min_daily_period: int = 14 # need n straddle prices
self.monthly_period: int = 12 # monthly straddle performance values
self.last_fundamental: List[Symbol] = []
self.straddle_price_sum: Dict[Symbol, float] = {} # call and put price sum
self.subscribed_contracts: Dict[Symbol, Contracts] = {} # subscribed option universe
self.monthly_straddle_returns: Dict[Symbol, RollingWindow] = {} # monthly straddle return values
# initial data feed
self.AddEquity('SPY', Resolution.Daily)
self.recent_year: int = -1
self.fundamental_count: int = 100
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag: bool = False
self.rebalance_flag: bool = False
self.UniverseSettings.Leverage = self.leverage
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.settings.daily_precise_end_time = False
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())
def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
# rebalance monthly
if not self.selection_flag:
return Universe.Unchanged
self.rebalance_flag = True
if self.Time.month % 12 != 0 or self.recent_year == self.Time.year:
return self.last_fundamental
self.recent_year = self.Time.year
# 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]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
# make sure monthly returns are consecutive
for stock in selected:
symbol: Symbol = stock.Symbol
if symbol not in self.last_fundamental:
if symbol in self.monthly_straddle_returns:
del self.monthly_straddle_returns[symbol]
# 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:
# execute once a day
# if not (self.Time.hour == 9 and self.Time.minute == 31):
# return
for symbol in self.last_fundamental:
# check if any of the subscribed contracts expired
if symbol in self.subscribed_contracts and self.subscribed_contracts[symbol].expiry_date - timedelta(days=1) <= self.Time.date():
# remove expired contracts
for contract in self.subscribed_contracts[symbol].contracts:
self.Liquidate(contract)
# liquidate hedge
if self.Portfolio[symbol].Quantity != 0:
self.MarketOrder(symbol, -self.Portfolio[symbol].Quantity)
# remove Contracts object for current symbol
del self.subscribed_contracts[symbol]
# check if stock has subscribed contracts
elif symbol in self.subscribed_contracts:
atm_call, atm_put = self.subscribed_contracts[symbol].contracts
if atm_call in data and atm_put in data and data[atm_call] and data[atm_put]:
# store straddle price
atm_call_price: float = data[atm_call].Value
atm_put_price: float = data[atm_put].Value
# store straddle sum price
straddle_price_sum: float = atm_call_price + atm_put_price
if symbol not in self.straddle_price_sum:
self.straddle_price_sum[symbol] = []
self.straddle_price_sum[symbol].append(straddle_price_sum)
# 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.selection_flag = True
return
# subscribe to new contracts after selection
if len(self.subscribed_contracts) == 0 and self.selection_flag:
self.selection_flag = False
for symbol in self.last_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 = 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
# filter calls and puts contracts with one month expiry
calls, puts = self.FilterContracts(strikes, contracts, underlying_price)
# make sure, there is at least one call and put contract
if len(calls) > 0 and len(puts) > 0:
# sort by expiry
call: Symbol = sorted(calls, key = lambda x: x.ID.Date, reverse=True)[0]
put: Symbol = sorted(puts, key = lambda x: x.ID.Date, reverse=True)[0]
subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(call.Underlying)
# check if stock's call and put contract was successfully subscribed
if subscriptions:
# add call contract
self.AddOptionContract(call, Resolution.Daily)
# add put contract
self.AddOptionContract(put, Resolution.Daily)
# retrieve expiry date for contracts
expiry_date: datetime.date = call.ID.Date.date() if call.ID.Date.date() < put.ID.Date.date() else put.ID.Date.date()
# store contracts with expiry date under stock's symbol
self.subscribed_contracts[symbol] = Contracts(expiry_date, underlying_price, [call, put])
return # one day skip for rebalance
# trade subscribed options
if len(self.subscribed_contracts) != 0 and self.rebalance_flag:
momentum: Dict[Symbol, float] = {}
for symbol in self.subscribed_contracts:
# make sure stock's symbol was lastly selected in fundamental and have straddle prices ready
if symbol not in self.last_fundamental or symbol not in self.straddle_price_sum or not len(self.straddle_price_sum[symbol]) > self.min_daily_period:
continue
# calculate straddle performance
straddle_performance: float = self.straddle_price_sum[symbol][-1] / self.straddle_price_sum[symbol][0] - 1
# calculate average straddle performance
if symbol not in self.monthly_straddle_returns:
self.monthly_straddle_returns[symbol] = RollingWindow[float](self.monthly_period)
self.monthly_straddle_returns[symbol].Add(straddle_performance)
# calculate straddle momentum
if self.monthly_straddle_returns[symbol].IsReady:
momentum[symbol] = np.mean([x for x in self.monthly_straddle_returns[symbol]][1:]) # skip last month
# reset straddle prices for next month
self.straddle_price_sum[symbol] = []
# make sure there are enough stock's for quintile selection
if len(momentum) < self.quantile:
self.rebalance_flag = False
return
# perform quintile selection
quantile: int = int(len(momentum) / self.quantile)
sorted_by_momentum: List[Symbol] = [x[0] for x in sorted(momentum.items(), key=lambda item: item[1]) if x[0] in data and data[x[0]]]
# long the quintile with the highest return
long: List[Symbol] = sorted_by_momentum[-quantile:]
# short the quintile with the lowest return
short: List[Symbol] = sorted_by_momentum[:quantile]
# trade long
self.TradeOptions(long, True)
# trade short
self.TradeOptions(short, False)
self.rebalance_flag = False
def FilterContracts(self, strikes: List[float], contracts: List[Symbol], underlying_price: float) -> Symbol:
''' filter call and put contracts from contracts parameter '''
''' return call and put contracts '''
# Straddle
call_strike: float = min(strikes, key=lambda x: abs(x-underlying_price))
put_strike: float = call_strike
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 == call_strike:
calls.append(contract)
# check if contract is put
elif contract.ID.OptionRight == OptionRight.Put and contract.ID.StrikePrice == put_strike:
puts.append(contract)
# return filtered calls and puts with one month expiry
return calls, puts
def TradeOptions(self, symbols: List[Symbol], long_flag: bool) -> None:
''' on long signal buy call and put option contract '''
''' on short signal sell call and put option contract '''
length: int = len(symbols)
# trade etf's call and put contracts
for symbol in symbols:
# get call and put contract
call, put = self.subscribed_contracts[symbol].contracts
# get underlying price
underlying_price: float = self.subscribed_contracts[symbol].underlying_price
# calculate option and hedge quantity
options_q: int = int((self.Portfolio.TotalPortfolioValue / length) / (underlying_price * 100))
hedge_q: int = options_q*50
if long_flag:
self.Buy(call, options_q)
self.Buy(put, options_q)
# initial delta hedge
self.Sell(symbol, hedge_q)
else:
self.Sell(call, options_q)
self.Sell(put, options_q)
# initial delta hedge
self.Buy(symbol, hedge_q)
@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"))