Volatility Of Volatility Effect in Stocks
Log in to collectAcademic paper
Strategy in a nutshell
A long–short NYSE equity strategy that exploits volatility-of-volatility by going long low-vol-of-vol stocks and short high-vol-of-vol stocks, rebalanced monthly using large, liquid stocks.
Economic rationale
Options markets embed predictive information that diffuses slowly into equities; the negative vol-of-vol effect reflects delayed price discovery between options and stock markets.
Backtest performance
Annualised return10.56%
Volatility12.2%
Beta0.057
Sharpe ratio0.54
Sortino ratio0.002
Win rate51%
Full Python code
from AlgorithmImports import *
import numpy as np
from typing import List, Dict
#endregion
class VolatilityOfVolatilityEffectinStocks(QCAlgorithm):
def Initialize(self) -> None:
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
# self.min_expiry = 30
# self.max_expiry = 60
self.tickers_to_ignore: List[str] = ['XOM', 'AAPL', 'AMZN']
self.period: int = 20 # need n of daily implied volatility values
self.leverage: int = 5
self.min_share_price: int = 5
self.quantile: int = 5
self.data: Dict[Symbol, SymbolData] = {} # storing daily IV
self.contracts: Dict[Symbol, Contract] = {} # storing option contracts
self.tickers_symbols: Dict[str, Symbol] = {} # storing symbols under their tickers
symbol: Symbol = self.AddEquity('SPY', Resolution.Minute).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.MinimumOrderMarginPortfolioPercentage = 0.
self.UniverseSettings.Resolution = Resolution.Minute
self.settings.daily_precise_end_time = False
self.AddUniverse(self.FundamentalSelectionFunction)
self.SetSecurityInitializer(lambda x: x.SetDataNormalizationMode(DataNormalizationMode.Raw))
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]:
# rebalance weekly
if not self.selection_flag:
return Universe.Unchanged
# select top n stocks by dollar volume with price higher than 5
selected: List[Fundamental] = sorted([
x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and x.MarketCap != 0 and x.Price > self.min_share_price and x.Symbol.Value not in self.tickers_to_ignore
], key=lambda x: x.DollarVolume, reverse=True)[:self.fundamental_count]
if len(selected) > self.fundamental_count:
selected = [x for x in sorted(selected, key=self.fundamental_sorting_key, reverse=True)[:self.fundamental_count]]
selected_symbols: List[Symbol] = [] # storing symbols of selected stocks
selected_tickers: List[str] = [] # storing tickers of selected stocks
# add new stocks to dictionaries
for stock in selected:
symbol: Symbol = stock.Symbol
ticker: str = symbol.Value
market_cap: float = stock.MarketCap
# remove duplicate stocks from fine
if ticker in selected_tickers:
# check if symbol of duplicated ticker was stored in self.data
if symbol in self.data and symbol in self.contracts:
# remove stock's contracts
for contract in self.contracts[symbol].contracts:
self.RemoveSecurity(contract)
del self.data[symbol]
del self.contracts[symbol]
continue
# add stock symbol to list of selected stocks
selected_symbols.append(symbol)
# add stock ticker to list of selected tickers
selected_tickers.append(ticker)
# don't override data, if they are consecutive
if ticker in self.tickers_symbols and symbol in self.data and symbol in self.contracts:
# update market cap
self.data[symbol].update_market_cap(market_cap)
continue
# store stock market cap and create RollingWindwow for IV values
self.data[symbol] = SymbolData(self.period, market_cap)
# store symbol under stock ticker
self.tickers_symbols[ticker] = symbol
# create object from Contracts class for stock symbol
self.contracts[symbol] = Contract(self.Time.date(), [])
# make sure, data are consecutive
remove_tickers_symbols: List[Tuple[str, Symbol]] = [] # storing tuple (ticker, symbol)
for ticker, symbol in self.tickers_symbols.items():
# add stocks, which weren't selected to remove list
if symbol not in selected_symbols:
remove_tickers_symbols.append((ticker, symbol))
# remove not selected stocks from dictionaries
for ticker, symbol in remove_tickers_symbols:
if symbol in self.contracts:
# remove stock's contracts
for contract in self.contracts[symbol].contracts:
self.RemoveSecurity(contract)
del self.contracts[symbol]
if symbol in self.data:
# delete stock from dictionaries
del self.data[symbol]
del self.tickers_symbols[ticker]
# return symbols of selected stocks
return selected_symbols
def OnData(self, data: Slice) -> None:
# each day store implied volatility for selected stocks
if self.Time.hour == 9:
if data.OptionChains.Count != 0:
for kvp in data.OptionChains:
chain: OptionChain = kvp.Value
symbol: Symbol = chain.Underlying.Symbol
# get option ticker from option symbol
ticker: str = symbol.Value
if ticker not in self.tickers_symbols:
continue
# based on option ticker get stock symbol
stock_symbol: Symbol = self.tickers_symbols[ticker]
# make sure, IV is updated once in a day
if stock_symbol not in self.data or self.data[stock_symbol].updated_date == self.Time.date():
continue
contracts: List[OptionContract] = [x for x in chain]
# make sure, there are enough contracts for stock
if len(contracts) < 2:
continue
call_iv: Union[None, float] = None
put_iv: Union[None, float] = None
# get atm call and atm put contract
for c in contracts:
if c.Right == OptionRight.Call:
# found atm call
call_iv = c.ImpliedVolatility
elif c.Right == OptionRight.Put:
# found atm put
put_iv = c.ImpliedVolatility
# check if there are both contracts
if call_iv and put_iv:
# calculate IV
iv: float = (put_iv + call_iv) / 2
# update RollingWindow for stock's IV values
self.data[stock_symbol].update_implied_vol(iv, self.Time.date())
# execute once a day
if self.day == self.Time.day:
return
self.day = self.Time.day
# check expiry of contracts
for symbol in self.data:
# remove contract, when it has 1 day to expiry
if symbol in self.contracts and ((self.contracts[symbol].expiry_date - timedelta(days=1)) <= self.Time.date()):
# remove expired contracts
for contract in self.contracts[symbol].contracts:
self.RemoveSecurity(contract)
# remove Contracts object for current symbol
del self.contracts[symbol]
# subscribe to contracts, if stock symbol doesn't have any
if symbol not in self.contracts:
# get new contracts after expiration
self.SubscribeOptionContracts(symbol)
# rebalance monthly
if not self.selection_flag:
return
self.selection_flag = False
vov: Dict[Symbol, float] = { x : self.data[x].volatility_of_volatility() for x in self.data if self.data[x].is_ready() }
# make sure there are enough stocks for quantile selection
if len(vov) < self.quantile:
self.Liquidate()
return
# sort stocks by VOV and perform quantile selection
quantile: int = int(len(vov) / self.quantile)
sorted_by_vov: List[Symbol] = [x[0] for x in sorted(vov.items(), key=lambda item: item[1])]
# long stocks with lowest VOV
long: List[Symbol] = sorted_by_vov[:quantile]
# short stocks with highest VOV
short: List[Symbol] = sorted_by_vov[-quantile:]
weight: Dict[Symbol, float] = {}
# trade execution
for i, portfolio in enumerate([long, short]):
mc_sum: float = sum(list(map(lambda x: self.data[x].market_cap, portfolio)))
for symbol in portfolio:
weight[symbol] = (((-1)**i) * self.data[symbol].market_cap / mc_sum)
portfolio: List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in weight.items() if symbol in data and data[symbol]]
self.SetHoldings(portfolio, True)
def SubscribeOptionContracts(self, symbol: Symbol) -> None:
''' get atm and atm strike for specific symbol then it filters atm call and atm put '''
''' if there are enough atm calls and atm puts this function subscribes one of their contracts based on expiry and store expiry date '''
# get all contracts for current commodity future
contracts: List[Symbol] = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
# get current price for commodity future
underlying_price: float = self.Securities[symbol].Price
# get strikes from commodity future contracts
strikes: List[float] = [i.ID.StrikePrice for i in contracts]
# check if there is at least one strike
if len(strikes) <= 0:
return
# at the money
atm_strike: Union[None, float] = None
atm_strike = min(strikes, key=lambda x: abs(x-underlying_price))
# filtred contracts based on option rights and strikes
atm_calls: List[Symbol] = self.FilterContracts(contracts, OptionRight.Call, atm_strike)
atm_puts: List[Symbol] = self.FilterContracts(contracts, OptionRight.Put, atm_strike)
# make sure there are enough contracts
if len(atm_calls) > 0 and len(atm_puts) > 0:
# sort by expiry
atm_call: Symbol = sorted(atm_calls, key = lambda item: item.ID.Date, reverse=True)[0]
atm_put: Symbol = sorted(atm_puts, key = lambda x: x.ID.Date, reverse=True)[0]
# add contracts
for contract in [atm_call, atm_put]:
self.AddContract(contract)
# get expiry date of contracts
expiry_date: datetime.date = atm_call.ID.Date.date()
# create new Contracts object for stock
self.contracts[symbol] = Contract(expiry_date, [atm_call, atm_put])
def FilterContracts(self, contracts: List[Symbol], option_right: float, strike: float) -> List[Symbol]:
''' filter contracts based on option_right and select only contracts with expiry in next month are selected '''
# filter contracts based on option right and select only contracts with next month expiry
filtered_contracts: List[Symbol] = [i for i in contracts if i.ID.OptionRight == option_right and
i.ID.StrikePrice == strike and
(self.Time.month + 1) == i.ID.Date.month]
# self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
# return filtered contracts
return filtered_contracts
def AddContract(self, contract: Symbol) -> None:
''' subcribe to contract, set price model and normalization mode '''
option = self.AddOptionContract(contract, Resolution.Minute)
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
option.SetDataNormalizationMode(DataNormalizationMode.Raw)
def Selection(self) -> None:
self.selection_flag = True
class Contract():
def __init__(self, expiry_date: datetime.date, contracts: List[Symbol]) -> None:
self.expiry_date: datetime.date = expiry_date
self.contracts: List[Symbol] = contracts
class SymbolData():
def __init__(self, period: int, market_cap: float):
self.implied_vol: RollingWindow = RollingWindow[float](period)
self.market_cap: float = market_cap
self.updated_date: Union[None, datetime.date] = None
def update_implied_vol(self, implied_vol: float, updated_date: datetime.date) -> None:
self.implied_vol.Add(implied_vol)
self.updated_date = updated_date
def update_market_cap(self, market_cap: float) -> None:
self.market_cap = market_cap
def volatility_of_volatility(self) -> float:
iv_values: np.ndarray = np.array([x for x in self.implied_vol])
mean_iv: float = np.mean(iv_values)
vov: float = np.sqrt(np.mean((iv_values - mean_iv)**2)) / mean_iv
return vov
def is_ready(self) -> bool:
return self.implied_vol.IsReady and self.market_cap
# 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"))