“The strategy trades S&P 500 futures using a predictive model based on implied volatility spreads and macroeconomic variables, dynamically allocating between equity index and risk-free rate daily.“
ASSET CLASS: CFDs, ETFs, futures | REGION: United States | FREQUENCY:
Daily | MARKET: bonds, equities | KEYWORD: Implied Volatility
I. STRATEGY IN A NUTSHELL
The strategy trades S&P 500 index futures based on a predictive model using implied volatility spread and macroeconomic variables. Implied volatility spread is calculated daily as the difference between the volume-weighted average volatility spreads for OTM puts (moneyness 0.80-0.95) and ATM calls (moneyness 0.95-1.05). This spread, along with explanatory variables like VIX squared, default spread changes, term spread changes, detrended riskless rate, dividend-to-price ratio, and lagged index returns, is used in a regression analysis to predict excess market returns. Using data from the midpoint of the dataset to day t, coefficients forecast excess returns. If the forecast is positive, the strategy invests 100% in the equity index; if negative, it invests in the risk-free rate. Daily returns are realized, and regressions are updated with an expanded data window for continuous forecasting and adjustment. The model integrates volatility and macroeconomic dynamics for systematic equity market investments.
II. ECONOMIC RATIONALE
Investors with positive market expectations increase demand for call options and reduce demand for puts, raising call option volatility and lowering put option volatility. A lower put-call implied volatility spread indicates higher expected returns for the underlying asset. The effect is more pronounced during informationally intense periods, such as earnings announcements or significant news about cash flow and discount rates, and when consumer sentiment index values are at extremes, highlighting the spread’s sensitivity to market sentiment and information-driven conditions.
III. SOURCE PAPER
Implied Volatility Spreads and Expected Market Returns [Click to Open PDF]
Yigit Atilgan, Turan G. Bali, K. Ozgur Demirtas
<Abstract>
This paper investigates the intertemporal relation between volatility spreads and expected returns on the aggregate stock market. We provide evidence for a significantly negative link between volatility spreads and expected returns at the daily and weekly frequencies. We argue that this link is driven by the information flow from option markets to stock markets. The documented relation is significantly stronger for the periods during which (i) S&P 500 constituent firms announce their earnings; (ii) cash flow and discount rate news are large in magnitude; and (iii) consumer sentiment index takes extreme values. The intertemporal relation remains strongly negative after controlling for conditional volatility, variance risk premium and macroeconomic variables. Moreover, a trading strategy based on the intertemporal relation with volatility spreads has higher portfolio returns compared to a passive strategy of investing in the S&P 500 index, after transaction costs are taken into account.
IV. BACKTEST PERFORMANCE
| Annualised Return | 11.09% |
| Volatility | 14.43% |
| Beta | 0.193 |
| Sharpe Ratio | 0.77 |
| Sortino Ratio | -0.019 |
| Maximum Drawdown | N/A |
| Win Rate | 46% |
V. FULL PYTHON CODE
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()