| Overall Statistics |
|
Total Trades 171 Average Win 2.81% Average Loss -1.03% Compounding Annual Return 9.745% Drawdown 17.400% Expectancy 0.861 Net Profit 264.925% Sharpe Ratio 0.608 Probabilistic Sharpe Ratio 11.337% Loss Rate 50% Win Rate 50% Profit-Loss Ratio 2.72 Alpha 0.049 Beta 0.095 Annual Standard Deviation 0.094 Annual Variance 0.009 Information Ratio -0.166 Tracking Error 0.16 Treynor Ratio 0.599 Total Fees $1937.72 Estimated Strategy Capacity $5900000.00 Lowest Capacity Asset VIXY UT076X30D0MD Portfolio Turnover 1.05% |
from AlgorithmImports import *
import numpy as np
import pandas as pd
from arch import arch_model
class CombiningVIXFuturesTermStructureStrategySP500Index(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2010, 1, 1)
self.SetCash(500000)
self.tickers = ['VIXY', 'SPY']
self.period = 120
self.data = {}
self.garch_fit_counter = 0
self.days_to_fit = 120
# Initialize default GARCH parameters
self.garch_params_spy = {'alpha_0': 0.0002, 'alpha_1': 0.1, 'beta_1': 0.8}
self.garch_params_vix = {'alpha_0': 0.0003, 'alpha_1': 0.2, 'beta_1': 0.7}
#futures
# self.vix_future = self.AddFuture(Futures.Indices.VIX, Resolution.Daily)
for ticker in self.tickers:
self.AddEquity(ticker, Resolution.Daily).Symbol
self.data[ticker] = RollingWindow[float](self.period)
self.vix = self.AddData(CBOE, 'VIX', Resolution.Daily).Symbol
self.vix3M = self.AddData(CBOE, 'VIX3M', Resolution.Daily).Symbol
self.Schedule.On(self.DateRules.EveryDay(), self.TimeRules.AfterMarketOpen('SPY', 10), self.FitGarchModel)
def OnData(self, data: Slice) -> None:
for ticker in self.tickers:
if ticker in data and data[ticker]:
self.data[ticker].Add(data[ticker].Value)
if all(x in data and data[x] for x in [self.vix, self.vix3M]):
if all(self.data[x].IsReady for x in self.tickers):
vix = data[self.vix].Value
vix3m = data[self.vix3M].Value
vix_volatility = self.calculate_volatility(list(self.data['VIXY'])[::-1], is_spx=False)
market_volatility = self.calculate_volatility(list(self.data['SPY'])[::-1], is_spx=True)
total_volatility = 1 / vix_volatility + 1 / market_volatility
w = (1.0 / vix_volatility) / total_volatility
if vix3m >= vix:
if not self.Portfolio['VIXY'].IsShort:
self.SetHoldings('VIXY', -w)
else:
if not self.Portfolio['VIXY'].IsLong:
self.SetHoldings('VIXY', w)
else:
self.Liquidate()
def FitGarchModel(self):
self.garch_fit_counter += 1
if self.garch_fit_counter >= self.days_to_fit:
for ticker in self.tickers:
if self.data[ticker].IsReady:
returns = self.CalculateReturns(list(self.data[ticker]))
model = arch_model(returns, vol='Garch', p=1, q=1)
res = model.fit(update_freq=5, disp='off')
if ticker == 'SPY':
self.garch_params_spy = {'alpha_0': res.params['omega'], 'alpha_1': res.params['alpha[1]'], 'beta_1': res.params['beta[1]']}
else:
self.garch_params_vix = {'alpha_0': res.params['omega'], 'alpha_1': res.params['alpha[1]'], 'beta_1': res.params['beta[1]']}
self.garch_fit_counter = 0
# def FitGarchModel(self):
# # Increment the counter
# self.garch_fit_counter += 1
# # Perform fitting every 20 days
# if self.garch_fit_counter >= self.days_to_fit:
# if self.data['SPY'].IsReady:
# returns_spy = self.CalculateReturns(list(self.data['SPY']))
# # Fit the GARCH model for SPY and update self.garch_params_spy
# if self.data['VIXY'].IsReady:
# returns_vix = self.CalculateReturns(list(self.data['VIXY']))
# # Fit the GARCH model for VIXY and update self.garch_params_vix
# self.garch_fit_counter = 0
def CalculateReturns(self, values):
return pd.Series((np.array(values)[1:] - np.array(values)[:-1]) / np.array(values)[:-1])
def calculate_volatility(self, values, min_observations=2, is_spx=True):
returns = pd.Series((np.array(values)[1:] - np.array(values)[:-1]) / np.array(values)[:-1])
volatilities = pd.Series(index=returns.index, dtype=float)
if is_spx:
params = self.garch_params_spy
else:
params = self.garch_params_vix
for i in range(len(returns)):
if i < min_observations:
volatilities.iloc[i] = returns.iloc[:i+1].std()
else:
epsilon_t_minus_1 = returns.iloc[i-1]
sigma_previous = volatilities.iloc[i-1]
sigma_squared = params['alpha_0'] + params['alpha_1'] * epsilon_t_minus_1**2 + params['beta_1'] * sigma_previous**2
volatilities.iloc[i] = np.sqrt(sigma_squared)
return volatilities.iloc[-1]