“该策略通过将商品按波动率分组进行投资。它从“低”波动率组买入，从“高”波动率组卖出，每月重新平衡并等权重。”
资产类别: 差价合约、期货 | 地区: 全球 | 周期: 每月 | 市场: 大宗商品 | 关键词: 隐含波动率
I. 策略概要
投资范围包括25种商品，根据30天隐含波动率（通过过去12个月的平均波动率去除趋势）分为四组。“低”组包含波动率最低的25%的商品，而“高”组包含波动率最高的25%的商品。该策略为多空策略，买入“低”组中的商品，卖出“高”组中的商品。投资组合等权重，每月重新平衡。
II. 策略合理性
VOL策略的回报主要由现货回报的可预测性驱动。波动率保险昂贵的商品价格往往下跌，而保险便宜的商品价格往往上涨。这种可预测性与套利限制有关。当对冲成本上升时，资本受限的对冲者会减少库存，从而造成卖压。相反，当波动率较低且市场条件稳定时，对冲成本较低，允许对冲者对冲更多产量。这种机制有助于商品市场的价格动态，从而形成VOL策略中观察到的模式。
III. 来源论文
Commodity Option Implied Volatilities and the Expected Futures Returns [点击查看论文]
- 高，卢森堡金融学院；卢森堡大学
<摘要>
商品期权的去趋势隐含波动率（VOL）显著预测商品期货回报的横截面。做多低VOL商品和做空高VOL商品的零成本策略产生的年化回报为12.66%，夏普比率为0.69。值得注意的是，基于波动率策略的超额回报主要来自其对未来现货成分的预测能力，这与迄今为止文献中研究的其他所有受展期回报驱动的商品策略不同。该策略与其他策略（如动量或基差）的相关性较低（低于10%），并且在经济衰退期间表现尤为出色。在控制了流动性不足、其他商品定价因素以及对整体商品市场波动率的敞口后，我们的结果仍然稳健。VOL指标与期货，尤其是期权市场的对冲压力有关。新闻媒体也有助于放大不确定性影响。与投资者彩票偏好和市场摩擦相关的变量能够解释部分预测关系。
IV. 回测表现
| 年化回报 | 12.66% |
| 波动率 | 18.48% |
| β值 | -0.079 |
| 夏普比率 | 0.69 |
| 索提诺比率 | 0.342 |
| 最大回撤 | N/A |
| 胜率 | 50% |
V. 完整的 Python 代码
from AlgorithmImports import *
#endregion
# https://quantpedia.com/strategies/commodity-option-implied-volatility-strategy/
#
# The investment universe consists of 25 commodities.
# Commodities are sorted into four groups based on the 30-days implied volatility de-trended by the previous 12 months mean of implied volatility (see page 8 for exact formula).
# The “Low” (“High”) group contains the top 25% of all commodities with the lowest (highest) volatilities.
# The portfolio is long-short and buys commodities from the group “Low” and sells commodities from the group “High”.
# The portfolio is equally-weighted and is rebalanced on a monthly basis.
#
# QC Implementation:
import numpy as np
class CommodityOptionImpliedVolatilityStrategy(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(100000)
self.min_expiry = 25
self.max_expiry = 35
self.period = 12 # need n of implied volatilities
self.iv = {} # storing implied volatilies in RollingWindow
self.contracts = {} # storing option contracts
self.tickers_symbols = {} # storing commodities symbols under their tickers
self.tickers = ['GLD', 'USO', 'UNG', 'SLV', 'DBA', 'DBB', 'PPLT', 'PALL']
self.next_expiry = None
for ticker in self.tickers:
# subscribe to commodity
security = self.AddEquity(ticker, Resolution.Minute)
# change normalization to raw to allow adding contracts
security.SetDataNormalizationMode(DataNormalizationMode.Raw)
# set fee model and leverage
security.SetFeeModel(CustomFeeModel())
security.SetLeverage(5)
# get commodity symbol
symbol = security.Symbol
# store symbol under ticker
self.tickers_symbols[ticker] = symbol
# create RollingWindow for implied volatilities
self.iv[symbol] = RollingWindow[float](self.period)
self.day = -1
def OnData(self, data):
# rebalance daily
if self.day == self.Time.day:
return
self.day = self.Time.day
if self.next_expiry and self.Time.date() >= self.next_expiry.date():
self.Liquidate()
for symbol in self.tickers_symbols:
if symbol in self.contracts:
# remove expired contracts
for contract in self.contracts[symbol]:
self.RemoveSecurity(contract)
# remove contracts from dictionary
del self.contracts[symbol]
if not self.Portfolio.Invested:
for symbol in self.tickers_symbols:
if symbol not in self.contracts:
# get all contracts for current commodity
contracts = self.OptionChainProvider.GetOptionContractList(symbol, self.Time)
# get current price for commodity
underlying_price = self.Securities[symbol].Price
# get strikes from commodity contracts
strikes = [i.ID.StrikePrice for i in contracts]
if len(strikes) > 0:
# get at the money strike
atm_strike:float = min(strikes, key=lambda x: abs(x-underlying_price))
atm_calls:list = [i for i in contracts if i.ID.OptionRight == OptionRight.Call and
i.ID.StrikePrice == atm_strike and
self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
atm_puts:list = [i for i in contracts if i.ID.OptionRight == OptionRight.Put and
i.ID.StrikePrice == atm_strike and
self.min_expiry < (i.ID.Date - self.Time).days < self.max_expiry]
if len(atm_calls) and len(atm_puts):
# sort by expiry
atm_call = sorted(atm_calls, key = lambda x: x.ID.Date)[0]
atm_put = sorted(atm_puts, key = lambda x: x.ID.Date)[0]
self.next_expiry = min(atm_call.ID.Date, atm_put.ID.Date)
# add contracts
option = self.AddOptionContract(atm_call, Resolution.Minute)
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
option.SetDataNormalizationMode(DataNormalizationMode.Raw)
option = self.AddOptionContract(atm_put, Resolution.Minute)
option.PriceModel = OptionPriceModels.CrankNicolsonFD()
option.SetDataNormalizationMode(DataNormalizationMode.Raw)
# store atm contracts by symbol
self.contracts[symbol] = [atm_call, atm_put]
iv_detrend = {} # storing detrend implied volatility for options
if data.OptionChains.Count != 0:
for kvp in data.OptionChains:
chain = kvp.Value
contracts = [x for x in chain]
# check if there are enough contracts for option
if len(contracts) < 2:
continue
atm_call_iv = None
atm_put_iv = None
# get ticker
ticker = chain.Underlying.Symbol.Value
# go through option contracts
for c in contracts:
if c.Right == OptionRight.Call:
# found atm call
atm_call_iv = c.ImpliedVolatility
else:
# found put option
atm_put_iv = c.ImpliedVolatility
if atm_call_iv and atm_put_iv:
# make mean from atm call implied volatility and atm put implied volatility
iv = (atm_call_iv + atm_put_iv) / 2
# get symbol based on ticker from option contract
commodity_symbol = self.tickers_symbols[ticker]
# check if there are enough data of mean implied volatilities
if self.iv[commodity_symbol].IsReady:
# calculate mean of previous mean implied volatilities
vol_mean = np.mean([x for x in self.iv[commodity_symbol]])
# calculate detrend implied volatility and store it by symbol
iv_detrend[commodity_symbol] = iv - vol_mean
# add current mean of implied volatility
self.iv[commodity_symbol].Add(iv)
# can't perform quintile selection
if len(iv_detrend) < 4:
self.Liquidate()
return
quintile = int(len(iv_detrend) / 4)
sorted_by_iv_detrend = [x[0] for x in sorted(iv_detrend.items(), key=lambda item: item[1])]
# go long smallest quintile
long = sorted_by_iv_detrend[:quintile]
# go short largest quintile
short = sorted_by_iv_detrend[-quintile:]
# trade execution
long_length = len(long)
short_length = len(short)
for symbol in long:
self.SetHoldings(symbol, 1 / long_length)
for symbol in short:
self.SetHoldings(symbol, -1 / short_length)
# custom fee model
class CustomFeeModel(FeeModel):
def GetOrderFee(self, parameters):
fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
return OrderFee(CashAmount(fee, "USD"))