“该策略交易国际ETF期权，买入低隐含波动率回报三分位数，卖出高隐含波动率回报三分位数，每月重新平衡，利用隐含波动率的错误定价获取系统性回报。”
资产类别: 期权 | 地区: 全球 | 周期: 每月 | 市场: 股票 | 关键词: 波动率、套利
I. 策略概要
该策略的目标是国际ETF期权，重点关注隐含波动率回报。隐含波动率回报计算为1-（上一年已实现波动率/当前平值隐含波动率），使用过去12个月的每日数据。平值跨式期权按前一日波动率回报排名，并分组为三个等权重三分位投资组合。多空投资组合卖出高（昂贵）三分位数的期权，买入低（便宜）三分位数的期权，并在每个月的第四个星期五重新平衡。这种方法利用隐含波动率的错误定价，从基于波动率的低效率中产生回报。
II. 策略合理性
国际期权市场表现出显著的波动率偏差，导致衍生品定价不一致。这些偏差提供了重要的经济和统计机会，与国内市场相比，国际市场的年化风险调整后回报更高。对冲基金主要利用国内期权，对国际期权策略的敞口有限，这表明在国外波动率套利方面存在尚未开发的机遇。国际定价行为的关键决定因素包括较大的波动率偏差、合约规格的异质性以及近期发行的国际ETP产品。尽管异常回报可观，但国际多空期权策略表现出正偏度、低波动率和中性股票市场敞口，使其对寻求多元化和稳定回报的从业者具有吸引力。
III. 来源论文
International Volatility Arbitrage [点击查看论文]
- 阿德里亚诺·托西，惠灵顿管理公司
<摘要>
国际上，交易所交易产品（ETP）和指数的期权定价是否一致？国际期权回报的横截面通过按事前波动率回报排序，表现出错误定价。此外，卖出国际ETP期权并买入其相应的指数期权可获得正的风险溢价。这两个实证发现都在国际上具有很大的经济意义和普遍性，而它们在国内则相对较小。虽然波动率对冲基金对国内期权产品有敞口，但它们忽略了从事国外波动率套利的可能性。这些发现意味着，寻求阿尔法的投资者可能会将其视野扩大到乍一看相似但机构上不同的国际衍生品。
IV. 回测表现
| 年化回报 | 16.38% |
| 波动率 | 8.93% |
| β值 | 0.004 |
| 夏普比率 | 1.83 |
| 索提诺比率 | N/A |
| 最大回撤 | N/A |
| 胜率 | 44% |
V. 完整的 Python 代码
from AlgorithmImports import *
import calendar
import datetime
#endregion
class InternationalVolatilityArbitrage(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(1000000)
self.min_expiry = 20
self.max_expiry = 90
self.percentage_traded = 0.2 # traded percentage of the portfolio
self.period = 12 * 21 # need 12 months of daily prices
self.prices = {} # storing daily prices
self.contracts = {} # storing option contracts
self.tickers_symbols = {} # storing symbols under their tickers
self.tickers = [
"EWA", # iShares MSCI Australia Index ETF
"EWO", # iShares MSCI Austria Investable Mkt Index ETF
"EWK", # iShares MSCI Belgium Investable Market Index ETF
"EWZ", # iShares MSCI Brazil Index ETF
"EWC", # iShares MSCI Canada Index ETF
"FXI", # iShares China Large-Cap ETF
"EWQ", # iShares MSCI France Index ETF
"EWG", # iShares MSCI Germany ETF
"EWH", # iShares MSCI Hong Kong Index ETF
"EWI", # iShares MSCI Italy Index ETF
"EWJ", # iShares MSCI Japan Index ETF
"EWM", # iShares MSCI Malaysia Index ETF
"EWW", # iShares MSCI Mexico Inv. Mt. Idx
"EWN", # iShares MSCI Netherlands Index ETF
"EWS", # iShares MSCI Singapore Index ETF
"EZA", # iShares MSCI South Africe Index ETF
"EWY", # iShares MSCI South Korea ETF
"EWP", # iShares MSCI Spain Index ETF
"EWD", # iShares MSCI Sweden Index ETF
"EWL", # iShares MSCI Switzerland Index ETF
"EWT", # iShares MSCI Taiwan Index ETF
"THD", # iShares MSCI Thailand Index ETF
"EWU", # iShares MSCI United Kingdom Index ETF
"SPY", # SPDR S&P 500 ETF
]
for ticker in self.tickers:
# subscribe to etf
security = self.AddEquity(ticker, Resolution.Minute)
# change normalization to raw to allow adding etf contracts
security.SetDataNormalizationMode(DataNormalizationMode.Raw)
# set fee model and leverage
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(5)
# get etf symbol
symbol = security.Symbol
# store etf symbol under etf ticker
self.tickers_symbols[ticker] = symbol
# create RollingWindow for daily prices
self.prices[symbol] = RollingWindow[float](self.period)
self.fourth_friday = self.FindFourthFriday(self.Time.year, self.Time.month)
self.day = -1
self.selection_flag = False
def OnData(self, data):
# execute once a day
if self.day == self.Time.day:
return
self.day = self.Time.day
# update RollingWindow with daily prices
for _, symbol in self.tickers_symbols.items():
# update RollingWindow with daily prices
if symbol in data and data[symbol]:
self.prices[symbol].Add(data[symbol].Value)
if data.OptionChains.Count >= 3 and self.selection_flag:
# stop rebalance
self.selection_flag = False
self.Liquidate()
vol_metric = {} # storing volatility differences for each etf
for kvp in data.OptionChains:
chain = kvp.Value
# get etf symbol
symbol = self.tickers_symbols[chain.Underlying.Symbol.Value]
# get contracts
contracts = [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.prices[symbol].IsReady or symbol not in self.contracts:
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 = (call_iv + put_iv) / 2
# get historical volatility
hv = self.GetHistoricalVolatility(self.prices[symbol])
# store metrics 1 - ratio between historical and implied volatility
vol_metric[symbol] = 1 - (hv / iv)
# can't perform selection when there aren't enough contracts
if len(vol_metric) > 3:
# perform selection
tercile = int(len(vol_metric) / 3)
sorted_by_vol_metric = [x[0] for x in sorted(vol_metric.items(), key=lambda item: item[1])]
# short expensive (high) tercile
short = sorted_by_vol_metric[-tercile:]
# long cheap (low) tercile
long = sorted_by_vol_metric[:tercile]
# trade execution
self.Liquidate()
# trade long
self.TradeOptions(long, True)
# trade short
self.TradeOptions(short, False)
# rebalance on fourth friday
if self.fourth_friday <= self.Time.date():
next_month = 1 if self.Time.month == 12 else self.Time.month + 1
year = self.Time.year + 1 if next_month == 1 else self.Time.year
# find fourth friday of next month
self.fourth_friday = self.FindFourthFriday(year, next_month)
# remove old contracts on rebalance
for _, symbol in self.tickers_symbols.items():
if symbol in self.contracts:
# remove Contracts object for current symbol
del self.contracts[symbol]
# perform new selection
self.selection_flag = True
self.Liquidate()
# subscribe to new contracts
for _, symbol in self.tickers_symbols.items():
# don't subscribe contracts of already subscribed symbols
if symbol in self.contracts:
continue
# get all contracts for current etf
contracts = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
# get current price for etf
underlying_price = self.Securities[symbol].Price
# get strikes from commodity future contracts
strikes = [i.ID.StrikePrice for i in contracts]
# can't filter contracts, if there isn't any strike price
if len(strikes) <= 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) and len(puts):
# sort by expiry and select contracts with latest expiry
call = sorted(calls, key=lambda x: x.ID.Date)[0]
put = sorted(puts, key=lambda x: x.ID.Date)[0]
subscriptions = self.SubscriptionManager.SubscriptionDataConfigService.GetSubscriptionDataConfigs(call.Underlying)
if subscriptions:
# add call contract
self.AddContract(call)
# add put contract
self.AddContract(put)
# store contracts with expiry date under etf symbol
self.contracts[symbol] = Contracts(underlying_price, [call, put])
def FindFourthFriday(self, year, month):
date = datetime.datetime(year, month, 1).date()
week_day = date.weekday()
# Taken from https://stackoverflow.com/questions/28680896/how-can-i-get-the-3rd-friday-of-a-month-in-python
calendar_obj = calendar.Calendar(firstweekday=week_day)
monthcal = calendar_obj.monthdatescalendar(year, month)
fridays = [day for week in monthcal for day in week if \
day.weekday() == calendar.FRIDAY and \
day.month == month]
fourth_friday = fridays[3] if len(fridays) > 3 else fridays[-1]
return fourth_friday
def FilterContracts(self, strikes, contracts, underlying_price):
''' 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 = call_strike
calls = [] # storing call contracts
puts = [] # 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 AddContract(self, contract):
''' subscribe option contract, set price mondel and normalization mode '''
option = self.AddOptionContract(contract, Resolution.Minute)
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
def GetImpliedVolatilities(self, contracts):
''' retrieve implied volatility of contracts from contracts parameteres '''
''' returns call and put implied volatility '''
call_iv = None
put_iv = 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):
''' calculate historical volatility based on daily prices in rolling_window_prices parameter '''
prices = np.array([x for x in rolling_window_prices])
returns = (prices[:-1] - prices[1:]) / prices[1:]
return np.std(returns)
def TradeOptions(self, symbols, long_flag):
''' on long signal buy call and put option contract '''
''' on short signal sell call and put option contract '''
length = len(symbols)
# trade etf's call and put contracts
for symbol in symbols:
# get call and put contract
contracts = self.contracts[symbol].contracts
call = contracts[0]
put = contracts[1]
# get underlying price
underlying_price = self.contracts[symbol].underlying_price
options_q = int(((self.Portfolio.TotalPortfolioValue*self.percentage_traded) / length) / (underlying_price * 100))
if self.Securities[call].IsTradable and self.Securities[put].IsTradable:
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, underlying_price, contracts):
self.underlying_price = underlying_price
self.contracts = contracts
# custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))