“通过构建delta跨式期权交易美国股票期权，做多跨式期权回报最低的股票，做空最高的股票，每月持有零delta跨式期权至到期。”
资产类别: 期权 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 反转、跨式期权
I. 策略概要
投资范围包括来自OptionMetrics数据库的在美国月度周期内到期的美国股票期权。股票数据来自CRSP。在到期日，每只股票选择两对看涨和看跌期权：一对基于两种期权的未平仓合约为正，另一对看涨期权delta接近0.5。看涨期权delta在0.25-0.75之外的期权被排除，重点关注平价期权。Delta跨式期权使用看跌和看涨期权的加权delta，基于买卖中点形成。股票根据一个月形成期内的跨式期权回报分为五分位。该策略做多最低的五分位，做空最高的五分位，持有零delta跨式期权至到期。
II. 策略合理性
Jones、Khorram和Mo研究了delta对冲期权回报和波动率互换收益，发现动量是由对过去波动率冲击反应不足以及与未定价的已实现股票波动率（而非隐含波动率）季节性变化相关的季节性驱动的。短期内，横截面反转是显而易见的，期权在一个月表现良好往往在下个月表现不佳。这种反转对于滞后一期的策略而言意义重大。由于所提出的策略持有跨式期权至到期，回报取决于前期期权价格，明确地将收益与标的股票价格挂钩。短期反转表明，到期日期权收益影响未到期期权的定价，从而影响回报并突出了期权市场的反转动态。
III. 来源论文
Momentum, Reversal, and Seasonality in Option Returns [点击查看论文]
- Christopher S. Jones、Mehdi Khorram、Haitao Mo。南加州大学 – 马歇尔商学院 – 金融与商业经济系。罗切斯特理工学院（RIT）。堪萨斯大学
<摘要>
期权回报在6到36个月的形成期内表现出显著的动量，多头/空头投资组合的年化夏普比率超过1.5。短期内，期权回报呈现反转。期权在3个月和12个月滞后倍数上也显示出显著的季节性。所有这些结果在横截面和时间上都非常显著且稳定。在控制其他特征后，它们仍然强劲，并且动量和季节性在因子风险调整后仍然存在。动量主要由对过去波动率和其他冲击的反应不足来解释，而季节性则反映了股票回报波动率中未定价的季节性变化。
IV. 回测表现
| 年化回报 | 49.71% |
| 波动率 | 36.8% |
| β值 | -0.051 |
| 夏普比率 | 1.35 |
| 索提诺比率 | -1.202 |
| 最大回撤 | N/A |
| 胜率 | 37% |
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.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
# initial data feed
self.AddEquity('SPY', Resolution.Daily)
self.fundamental_count: int = 200
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.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
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
# 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]]
# 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 = data[atm_call].Value
atm_put_price = data[atm_put].Value
# store straddle sum price
straddle_price_sum = 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: 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
# 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:
straddle_performance: 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[symbol] = self.straddle_price_sum[symbol][-1] / self.straddle_price_sum[symbol][0] - 1
# reset straddle prices for next month
self.straddle_price_sum[symbol] = []
# make sure there are enough stock's for quintile selection
if len(straddle_performance) < self.quantile:
self.rebalance_flag = False
return
# perform quintile selection
quantile: int = int(len(straddle_performance) / self.quantile)
sorted_by_straddle_perf: List[Symbol] = [x[0] for x in sorted(straddle_performance.items(), key=lambda item: item[1]) if x[0] in data and data[x[0]]]
# long straddles with lowest performance
long: List[Symbol] = sorted_by_straddle_perf[:quantile]
# short straddles with highest performance
short: List[Symbol] = sorted_by_straddle_perf[-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"))