International Volatility Arbitrage
Log in to collectAcademic paper
Strategy in a nutshell
The strategy trades international ETF options using implied volatility returns, calculated as 1 minus the ratio of prior-year realized volatility to current ATM implied volatility. ATM straddles are ranked daily and grouped into three terciles, with a long-short portfolio buying low-volatility (cheap) options and selling high-volatility (expensive) options. Portfolios are rebalanced monthly on the fourth Friday.
Economic rationale
Volatility deviations across international option markets create exploitable mispricings. The strategy captures these inefficiencies, delivering strong risk-adjusted returns, low volatility, and neutral equity exposure, with opportunities largely untapped by hedge funds focused on domestic markets.
Backtest performance
Annualised return16.38%
Volatility8.93%
Beta0.004
Sharpe ratio1.83
Win rate44%
Full Python code
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"))