“该策略通过形成平值跨式期权投资美国股票期权，并按隐含波动率斜率排序。它买入波动率斜率向上的期权，卖出波动率斜率向下的期权，每月重新平衡。”
资产类别: 期权 | 地区: 美国 | 周期: 每月 | 市场: 股票 | 关键词: 平值跨式期权
I. 策略概要
投资范围包括所有美国股票期权，重点是每月期权到期后形成的平值跨式期权。排除违反套利条件或标的股票价格低于10美元的期权。仅包括delta在±0.35至±0.65之间的平值期权。投资组合基于隐含波动率期限结构的斜率。跨式期权根据其斜率分为十分位数，第一十分位数包含波动率斜率最大的期权。该策略从第一十分位数买入，从第十分位数卖出，持有至到期。投资组合等权重，并由于策略的偏度风险，将投资限制在20%。
II. 策略合理性
该策略的功能基于一个原则，即风险的价格在较长的时间范围内会降低。这种关系适用于各种资产类别，表明了投资者风险偏好的一个基本方面。隐含波动率期限结构与短期到期期权价格的过度反应相关，不同时间范围内的已实现波动率有助于解释短期和长期到期隐含波动率。波动率期限结构的斜率与短期到期期权的波动率风险溢价之间存在很强的联系。随着期限结构反转，短期风险溢价增加，而长期风险溢价减少，显示出与回报的负相关。
III. 来源论文
Jump Risk and Option Returns [点击查看论文]
- 吉姆·坎帕萨诺 (Jim Campasano) 和马修·林恩 (Matthew Linn)。 马萨诸塞大学阿默斯特分校 – 艾斯本商学院 (Isenberg School of Management)；堪萨斯州立大学 – 金融系。马萨诸塞大学艾斯本商学院 (Isenberg School of Management, University of Massachusetts)
<摘要>
我们表明，股票波动率的期限结构可以有力地预测标的股票的跳跃。我们的分析为文献中一些最大的基于期权的异常现象提供了基于风险的解释。我们表明，基于期限结构斜率不同度量的期权策略的回报反映了每个度量预测标的股票跳跃的时间范围。这进一步支持了与期限结构相关的溢价是由于跳跃风险的理论。此外，我们表明，期限结构优于文献中现有的跳跃预测指标。
IV. 回测表现
| 年化回报 | 31.32% |
| 波动率 | 18.46% |
| β值 | -0.025 |
| 夏普比率 | 1.7 |
| 索提诺比率 | -0.96 |
| 最大回撤 | N/A |
| 胜率 | 24% |
V. 完整的 Python 代码
from AlgorithmImports import *
from typing import List, Dict
from pandas.core.frame import dataframe
from pandas.core.series import Series
#endregion
class CrossSectionalSixMonthEquityATMStraddleTradingStrategy(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2015, 1, 1)
self.SetCash(1000000)
self.min_expiry: int = 180
self.max_expiry: int = 240
self.period: int = 6 * 21 # need n of stock daily prices
self.percentage_traded: float = 0.1
self.selection_threshold: int = 10
self.min_share_price: int = 10
self.quantile: int = 10
self.leverage: int = 10
self.data: Dict[Symbol, RollingWindow[float]] = {}
self.symbols_by_ticker: Dict[str, Symbol] = {}
self.subscribed_contracts: Dict[Symbol, Contracts] = {}
symbol: Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
self.day: int = -1
self.fundamental_count: int = 100
self.fundamental_sorting_key = lambda x: x.DollarVolume
self.selection_flag: bool = False
self.settings.daily_precise_end_time = False
self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Daily
self.AddUniverse(self.FundamentalSelectionFunction)
self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
self.Schedule.On(self.DateRules.MonthStart(symbol), self.TimeRules.BeforeMarketClose(symbol), self.Selection)
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]:
# update daily prices of stocks in self.data dictionary
for stock in fundamental:
symbol:Symbol = stock.Symbol
if symbol in self.data:
self.data[symbol].Add(stock.AdjustedPrice)
# rebalance monthly
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]]
for stock in selected:
symbol: Symbol = stock.Symbol
ticker: str = symbol.Value
self.symbols_by_ticker[ticker] = symbol
if symbol in self.data:
continue
self.data[symbol] = RollingWindow[float](self.period)
history: dataframe = self.History(symbol, self.period, Resolution.Daily)
if history.empty:
continue
closes: Series = history.loc[symbol].close
for time, close in closes.items():
self.data[symbol].Add(close)
# return newly selected symbols
return list(map(lambda x: x.Symbol, selected))
def OnData(self, data: Slice) -> None:
# execute once a day
if self.day == self.Time.day:
return
self.day = self.Time.day
# subscribe to new contracts after selection
if len(self.subscribed_contracts) == 0 and self.selection_flag:
for _, symbol in self.symbols_by_ticker.items():
if self.Securities[symbol].IsDelisted:
continue
if symbol in data and data[symbol]:
if symbol in self.data and self.data[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.data[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
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)
if subscriptions:
# add call contract
self.AddContract(call)
# add put contract
self.AddContract(put)
# retrieve expiry date for contracts
expiry_date: dateTime.date = call.ID.Date.date()
# store contracts with expiry date under stock's symbol
self.subscribed_contracts[symbol] = Contracts(expiry_date, underlying_price, [call, put])
# calculate term structure and trade options
elif len(self.subscribed_contracts) != 0 and data.OptionChains.Count != 0 and self.selection_flag:
self.selection_flag = False # this makes sure, there will be no other trades until next selection
term_structure: Dict[Symbol, float] = {} # storing term structures keyed by stock's symbol
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]
if symbol in data and data[symbol]:
# get contracts
contracts: List[Symbol] = [x for x in chain]
# check if there are enough contracts for option and daily prices are ready
if len(contracts) < 2 or not self.data[symbol].IsReady:
continue
# get call and put implied volatility
call_iv, put_iv = self.GetImpliedVolatilities(contracts)
if call_iv and put_iv:
# make mean from call implied volatility and put implied volatility
iv: float = (call_iv + put_iv) / 2
# get historical volatility
hv: float = self.GetHistoricalVolatility(self.data[symbol])
# store stock's term structure
term_structure[symbol] = (iv - hv) / hv
# can't perform selection
if len(term_structure) < self.selection_threshold:
return
# perform quantile selection
quantile: int = int(len(term_structure) / self.quantile)
sorted_by_term_structure: List[Symbol] = [x[0] for x in sorted(term_structure.items(), key=lambda item: item[1])]
# long bottom
long: List[Symbol] = sorted_by_term_structure[:quantile]
# short top
short: List[Symbol] = sorted_by_term_structure[-quantile:]
# trade long
self.TradeOptions(data, long, True)
# trade short
self.TradeOptions(data, short, False)
def Selection(self) -> None:
self.selection_flag = True # perform new selection
self.Liquidate() # rebalance monthly, so liquidate all holdings
# clear dictionary for subscribed contracts, because there will be new selection
self.subscribed_contracts.clear()
# clear dictionary of tickers and their symbols, because new stocks will be selected
self.symbols_by_ticker.clear()
def FilterContracts(self, strikes: List[float], contracts: List[Symbol], underlying_price: float) -> List[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 six months 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 AddContract(self, contract: Symbol) -> None:
''' subscribe option contract, set price mondel and normalization mode '''
option: Option = self.AddOptionContract(contract, Resolution.Daily)
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
def GetImpliedVolatilities(self, contracts: List[Symbol]) -> List[float]:
''' retrieve implied volatility of contracts from contracts parameteres '''
''' returns call and put implied volatility '''
call_iv: Union[None, float] = None
put_iv: Union[None, float] = None
# go through option contracts
for c in contracts:
if c.Right == OptionRight.Call:
# found call option
call_iv = c.ImpliedVolatility
else:
# found put option
put_iv = c.ImpliedVolatility
return call_iv, put_iv
def GetHistoricalVolatility(self, rolling_window_prices: RollingWindow) -> 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)
def TradeOptions(self, data: Slice, symbols: List[Symbol], long_flag: bool):
''' on long signal buy call and put option contract '''
''' on short signal sell call and put option contract '''
count: int = len(symbols)
# trade etf's call and put contracts
for symbol in symbols:
if symbol in self.subscribed_contracts:
# check if contracts are tradebale and don't have 0 price
# for contract in self.subscribed_contracts[symbol].contracts:
# if not self.Securities[contract].IsTradable or self.Securities[contract].Price == 0:
# return
# get call and put contract
call, put = self.subscribed_contracts[symbol].contracts
# get underlying price
underlying_price: float = self.subscribed_contracts[symbol].underlying_price
options_q: int = int(((self.Portfolio.MarginRemaining * self.percentage_traded) / count) / (underlying_price * 100))
if call in data and data[call] and put in data and data[put]:
if long_flag:
self.Buy(call, options_q)
self.Buy(put, options_q)
else:
self.Sell(call, options_q)
self.Sell(put, options_q)
class Contracts():
def __init__(self, expiry_date: datetime.date, underlying_price: float, contracts: List[Symbol]):
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"))