“该策略使用基于隐含波动率价差和宏观经济变量的预测模型交易标准普尔500指数期货，每日在股指和无风险利率之间动态分配。”
资产类别: 差价合约、ETF、期货 | 地区: 美国 | 周期: 每日 | 市场: 债券、股票 | 关键词: 隐含波动率价差
I. 策略概要
该策略基于一个使用隐含波动率价差和宏观经济变量的预测模型交易标准普尔500指数期货。隐含波动率价差每天计算，其值为虚值看跌期权（行权价0.80-0.95）和平值看涨期权（行权价0.95-1.05）的成交量加权平均波动率价差之差。该价差以及VIX平方、违约利差变化、期限利差变化、去趋势无风险利率、股息收益率和滞后指数回报等解释变量，用于回归分析以预测超额市场回报。使用从数据集中间点到t日的数据，系数预测超额回报。如果预测为正，策略将100%投资于股指；如果为负，则投资于无风险利率。每日回报实现，回归通过扩展数据窗口进行更新，以实现持续预测和调整。该模型整合了波动率和宏观经济动态，用于系统性股票市场投资。
II. 策略合理性
投资者对市场持积极预期时，会增加对看涨期权的需求并减少对看跌期权的需求，从而提高看涨期权波动率并降低看跌期权波动率。较低的看跌-看涨隐含波动率价差表明标的资产的预期回报更高。这种效应在信息密集时期（例如收益公告或有关现金流和折现率的重大新闻）以及消费者信心指数值处于极端时更为明显，这突显了价差对市场情绪和信息驱动条件的敏感性。
III. 来源论文
Implied Volatility Spreads and Expected Market Returns [点击查看论文]
- Yigit Atilgan, Turan G. Bali, and K. Ozgur Demirtas.
<摘要>
本文研究了波动率价差与股票市场总预期回报之间的跨期关系。我们提供了证据表明，在每日和每周频率上，波动率价差与预期回报之间存在显著的负相关关系。我们认为这种联系是由期权市场到股票市场的信息流驱动的。记录的关系在以下时期显著更强：（i）标准普尔500成分股公司宣布其收益；（ii）现金流和折现率新闻幅度较大；以及（iii）消费者信心指数取极端值。在控制了条件波动率、方差风险溢价和宏观经济变量后，跨期关系仍然保持强烈的负相关。此外，考虑到交易成本后，基于与波动率价差的跨期关系的交易策略比投资标准普尔500指数的被动策略具有更高的投资组合回报。
IV. 回测表现
| 年化回报 | 11.09% |
| 波动率 | 14.43% |
| β值 | 0.193 |
| 夏普比率 | 0.77 |
| 索提诺比率 | -0.019 |
| 最大回撤 | N/A |
| 胜率 | 46% |
V. 完整的 Python 代码
from AlgorithmImports import *
import data_tools
import statsmodels.api as sm
import pandas as pd
from QuantConnect.DataSource import *
#endregion
class ImpliedVolatilitySpreadsandExpectedMarketReturnsinSP500(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2005, 1, 1)
self.SetCash(100_000)
self.leverage: int = 10
self.vix: Symbol = self.AddData(CBOE, 'VIX', Resolution.Daily).Symbol
self.put_call: Symbol = self.AddData(data_tools.PutCallRatio, "PutCallRatio", Resolution.Daily).Symbol
# bond yield data
self.us_yield_10y: Symbol = self.AddData(data_tools.BondYield, 'US10YT', Resolution.Daily).Symbol
self.us_yield_3m: Symbol = self.AddData(data_tools.InterestRate3M, 'IR3TIB01USM156N', Resolution.Daily).Symbol
self.us_yield_3m_sma: SimpleMovingAverage = SimpleMovingAverage(12)
# warmup 3m yield history
us_yield_3m_history: dataframe = self.History(self.us_yield_3m, 12*30, Resolution.Daily)
yields: Series = us_yield_3m_history.loc[self.us_yield_3m].value
for time, _yield in yields.items():
self.us_yield_3m_sma.Update(time, _yield)
self.daaa_yield: Symbol = self.AddData(data_tools.BondYield, 'DAAA', Resolution.Daily).Symbol
self.dbaa_yield: Symbol = self.AddData(data_tools.BondYield, 'DBAA', Resolution.Daily).Symbol
self.last_default_spread: float|None = None
# risk free etf
symbol_data: Equity = self.AddEquity('BIL', Resolution.Daily)
self.risk_free_asset: Symbol= symbol_data.Symbol
symbol_data.SetLeverage(self.leverage)
# SPY and SPTR data
symbol_data: Equity = self.AddEquity('SPY', Resolution.Daily)
symbol_data.SetLeverage(self.leverage)
self.spy: Symbol = symbol_data.Symbol
self.sptr: Symbol = self.AddData(data_tools.SPTRIndex, 'SP500TR', Resolution.Daily).Symbol
self.period: int = 21
self.SetWarmUp(self.period, Resolution.Daily)
self.spy_history: RollingWindow = RollingWindow[float](self.period)
self.sptr_history: RollingWindow = RollingWindow[float](self.period)
self.regression_data: List = []
self.min_regression_period: int = 12 * 21
def OnData(self, slice: Slice) -> None:
# check if data is still comming in
put_call_last_update_date: Dict[str, datetime.date] = data_tools.PutCallRatio.get_last_update_date()
bond_yield_last_update_date: Dict[str, datetime.date] = data_tools.BondYield.get_last_update_date()
ir_last_update_date: Dict[str, datetime.date] = data_tools.InterestRate3M.get_last_update_date()
index_last_update_date: Dict[str, datetime.date] = data_tools.SPTRIndex.get_last_update_date()
if not ((self.put_call.Value in put_call_last_update_date and self.Time.date() < put_call_last_update_date[self.put_call.Value]) and \
(all(symbol.Value in bond_yield_last_update_date and self.Time.date() < bond_yield_last_update_date[symbol.Value] for symbol in [self.us_yield_10y, self.daaa_yield, self.dbaa_yield])) and \
(self.us_yield_3m.Value in ir_last_update_date and self.Time.date() < ir_last_update_date[self.us_yield_3m.Value]) and \
(self.sptr.Value in index_last_update_date and self.Time.date() < index_last_update_date[self.sptr.Value])):
self.Liquidate()
return
# update 3M yield moving average
if self.us_yield_3m in slice:
yield_3m: float = slice[self.us_yield_3m].Value
self.us_yield_3m_sma.Update(self.Time, yield_3m)
# update SPY and SPTR history
if self.spy in slice and self.sptr in slice:
spy_price: float = slice[self.spy].Value
sptr_price: float = slice[self.sptr].Value
if spy_price != 0 and sptr_price != 0:
self.spy_history.Add(spy_price)
self.sptr_history.Add(sptr_price)
# calculate independent variables
reg_data: Dict[str, float] = {}
if (self.vix in slice and self.put_call in slice) and \
(self.daaa_yield in slice and self.dbaa_yield in slice) and \
(self.us_yield_3m_sma.IsReady and self.Securities.ContainsKey(self.us_yield_3m) and self.us_yield_10y in slice) and \
(self.spy_history.IsReady and self.sptr_history.IsReady):
aaa_yield: float = slice[self.daaa_yield].Value
baa_yield: float = slice[self.dbaa_yield].Value
default_spread: float = baa_yield - aaa_yield
if self.last_default_spread:
change_in_default_spread: float = default_spread - self.last_default_spread
reg_data['change_in_default_spread'] = change_in_default_spread #independend variable
vixsq: float = slice[self.vix].Value ** 2
reg_data['vixsq'] = vixsq #independend variable
put_call_spread: float = slice[self.put_call].Value
reg_data['put_call_spread'] = put_call_spread #independend variable
yield_3m: float = self.Securities[self.us_yield_3m].Price
yield_sma_3m: float = self.us_yield_3m_sma.Current.Value
detrended_riskless_rate: float = yield_3m - yield_sma_3m
reg_data['detrended_riskless_rate'] = detrended_riskless_rate #independend variable
yield_10y: float = slice[self.us_yield_10y].Value
term_spread: float = yield_10y - yield_3m
reg_data['term_spread'] = term_spread #independend variable
spy_prices: List[float] = list(self.spy_history)
sptr_prices: List[float] = list(self.sptr_history)
spy_return: float = spy_prices[0] / spy_prices[-1] - 1
reg_data['spy_return'] = spy_return #dependend variable
sptr_return: float = sptr_prices[0] / sptr_prices[-1] - 1
if spy_return != 0:
dividend_price_ratio: float = sptr_return / spy_return
reg_data['dividend_price_ratio'] = dividend_price_ratio #independend variable
# one day lagged returns.
spy_prices = list(self.spy_history)[1:]
lagged_return: float = spy_prices[0] / spy_prices[-1] - 1
reg_data['lagged_return'] = lagged_return #independend variable
# update regression data
# NOTE We might want to ommit concat or append wor functions.
# "It is worth noting that concat() (and therefore append()) makes a full copy of the data, and that constantly reusing this function can create a significant performance hit.
# If you need to use the operation over several datasets, use a list comprehension."
# - Source: https://pandas.pydata.org/pandas-docs/stable/user_guide/merging.html
self.regression_data.append(reg_data)
# regression data is ready
if len(self.regression_data) >= self.min_regression_period:
regression_df: dataframe = pd.dataframe(self.regression_data)
x: dataframe = regression_df.loc[:, regression_df.columns != 'spy_return'][:-1] # offset x. Last x entry is used to get Y prediction.
x = sm.add_constant(x)
y: dataframe = regression_df['spy_return'][1:]
y = y.reset_index(drop=True)
lm = sm.OLS(y, x).fit()
last_x = x.tail(1).reset_index(drop=True)
y_predicted: float = lm.predict(last_x)[0]
if y_predicted > 0:
if self.Portfolio[self.risk_free_asset].Invested:
self.Liquidate(self.risk_free_asset)
self.SetHoldings(self.spy, 1)
else:
if self.Portfolio[self.spy].Invested:
self.Liquidate(self.spy)
self.SetHoldings(self.risk_free_asset, 1)
self.last_default_spread = default_spread
else:
self.Liquidate()