| Overall Statistics |
|
Total Orders 9464 Average Win 5.38% Average Loss -3.93% Compounding Annual Return 116.924% Drawdown 37.400% Expectancy 0.349 Start Equity 100000 End Equity 393714.42 Net Profit 293.714% Sharpe Ratio 1.634 Sortino Ratio 2.145 Probabilistic Sharpe Ratio 65.012% Loss Rate 43% Win Rate 57% Profit-Loss Ratio 1.37 Alpha 0.96 Beta -0.27 Annual Standard Deviation 0.566 Annual Variance 0.32 Information Ratio 1.364 Tracking Error 0.581 Treynor Ratio -3.429 Total Fees $5310.49 Estimated Strategy Capacity $240000.00 Lowest Capacity Asset UVXY V0H08FY38ZFP Portfolio Turnover 25.67% |
# region imports
from AlgorithmImports import *
from QuantConnect.DataSource import *
from QuantConnect.Python import PythonQuandl
from pandas.tseries.offsets import BDay
from arch import arch_model
import pickle
from datetime import datetime
from config import *
from models import *
import numpy as np
import random
# endregion
class LogicalYellowGreenRat(QCAlgorithm):
def initialize(self):
self.set_start_date(2023, 1, 1)
#self.set_end_date(2023, 12, 31)
#self.set_end_date(2024, 3, 19)
self.set_warm_up(60*8*model_setting['Random_Forest']['n_lookback'], Resolution.MINUTE)
self.set_cash(100000)
self.universe_settings.resolution = Resolution.MINUTE
self.random_seed = 60
np.random.seed(self.random_seed)
random.seed(self.random_seed)
self.ticker_symbol_map = {}
self.symbol_ticker_map = {}
for ticker in general_setting['data_']:
if general_setting['data_'][ticker]['type'] == 'index':
sym = self.add_index(ticker, Resolution.MINUTE).symbol
self.ticker_symbol_map[ticker] = sym
self.symbol_ticker_map[sym] = ticker
elif general_setting['data_'][ticker]['type'] == 'equity':
sym = self.add_equity(ticker, Resolution.MINUTE).symbol
self.ticker_symbol_map[ticker] = sym
self.symbol_ticker_map['sym'] = ticker
elif general_setting['data_'][ticker]['type'] == 'data':
sym = self.add_data(general_setting['data_'][ticker]['source'], ticker, Resolution.DAILY).symbol
self.ticker_symbol_map[ticker] = sym
self.symbol_ticker_map[sym] = ticker
elif general_setting['data_'][ticker]['type'] == 'future':
if ticker == 'VX':
self.VX = self.AddFuture(general_setting['data_'][ticker]['source'], Resolution.MINUTE, dataNormalizationMode=DataNormalizationMode.BackwardsRatio, dataMappingMode=DataMappingMode.LAST_TRADING_DAY, extendedMarketHours=True)
self.VX_sym = self.VX.symbol
self.ticker_symbol_map[ticker] = self.VX_sym
self.symbol_ticker_map[self.VX_sym] = ticker
else:
future = self.AddFuture(general_setting['data_'][ticker]['source'], Resolution.MINUTE, dataNormalizationMode=DataNormalizationMode.BackwardsRatio, dataMappingMode=DataMappingMode.OpenInterest, extendedMarketHours=True)
future_sym = future.symbol
self.ticker_symbol_map[ticker] = future_sym
self.symbol_ticker_map[future_sym] = ticker
elif general_setting['data_'][ticker]['type'] == 'external':
self.ticker_symbol_map[ticker] = ticker
self.symbol_ticker_map[ticker] = ticker
# Bollinger Band
self.SPX_bb = self.bb(self.ticker_symbol_map['SPX'], 30, 2, resolution=Resolution.MINUTE)
self.GLD_bb = self.bb(self.ticker_symbol_map['GLD'], 30, 2, resolution=Resolution.MINUTE)
# Trading Instruments
self.uvxy = self.AddEquity("UVXY", Resolution.MINUTE, leverage=2).Symbol
self.svxy = self.AddEquity("SVXY", Resolution.MINUTE, leverage=2).Symbol
self.models = {}
self.initialize_models()
self.prediction = 0
self.morning_value = 0
self.position_status = 0
self.reinvested = False
self.morning_attempt = True
self.data_dict_list = []
self.Schedule.On(self.DateRules.Every([1,2,3,4,5]), self.TimeRules.At(2, 0), self.fetch_other_data)
self.Schedule.On(self.DateRules.Every([1,2,3,4,5]), self.TimeRules.At(3, 0), self.train_ML_model)
self.entry_price = 0
self.quantity = 0
# 1: long volatility
# 0.5: long volatility after a loss from short
# 2: long valatility after taking profit from long
# -1: short volatility
# -0.5: short volatility after a loss from long
# -2: short volatility after taking profit from short
#
self.message = ""
def initialize_models(self):
# Garch Model
# SPX_history = self.history(self.SPX_sym, 100, Resolution.HOUR)[:-1]
# SPX_history = SPX_history.droplevel(['symbol'])
# returns = np.log(SPX_history['close'] / SPX_history['close'].shift(1))
# returns = returns.dropna()
# self.models['GARCH'] = GARCH_model(returns, self)
self.models['Random_Forest'] = Random_Forest_model(self.random_seed)
def train_ML_model(self):
self.prediction = self.models['Random_Forest'].train_model(self.data_dict_list, self)
self.monring_attempt = True
# Get the data that are available in the slice.
def fetch_data(self, data):
data_dict = {}
data_dict['date'] = self.Time.date()
for ticker, sym in self.ticker_symbol_map.items():
if general_setting['data_'][ticker]['type'] == 'data' or general_setting['data_'][ticker]['type'] == "external":
continue
if data.contains_key(sym):
value = data[sym].close
elif not self.history(sym, 20, Resolution.MINUTE).empty:
value = list(self.history(sym, 20, Resolution.MINUTE)['close'])[-1]
else:
self.debug(f'{sym} data is not retrieved! ')
value = float('nan')
# Y variable
if ticker == 'VX':
data_dict['Y_'+ticker] = value
else:
data_dict['Close_'+ticker] = value
self.data_dict_list.append(data_dict)
# Get data that are unavailable in the slice.
def fetch_other_data(self):
date = (self.Time.date() - BDay(1)).date()
if len(self.data_dict_list) == 0:
return
data_dict = self.data_dict_list[-1]
# Other data
for ticker, sym in self.ticker_symbol_map.items():
if general_setting['data_'][ticker]['type'] == 'data':
if ticker == "VVIX":
vvix_value = list(self.history(sym, 10, Resolution.DAILY)['close'])[-1]
if date == data_dict['date']:
data_dict['Close_'+ticker] = vvix_value
if ticker == 'USTYCR':
yield_curve_symbol = self.ticker_symbol_map[ticker]
df = self.history(USTreasuryYieldCurveRate, yield_curve_symbol, 10, Resolution.DAILY)
if 'tenyear' in df.columns:
data_dict['Close_r10'] = df[-1:]['tenyear'][0]
else:
self.debug(f'ten-year treasury yield data not available')
data_dict['Close_r10'] = float('nan')
# elif general_setting['data_'][ticker]['type'] == 'external':
# if ticker == "10_year_yield":
# data_str = self.download(general_setting['data_']["10_year_yield"]['source'])
# data_str_list = data_str.split('\n')
# columns = []
# rows = []
# top_row = True
# for row in data_str_list:
# content = row.split(',')
# if top_row:
# for i in range(len(content)):
# if content[i] != "Date":
# content[i] = content[i][1:-1]
# columns = content
# top_row = False
# else:
# rows.append(content)
# data_df = pd.DataFrame(rows, columns=columns)
# data_df = data_df.set_index(['Date'])
# data_df.index = pd.to_datetime(data_df.index)
# data_df = data_df.sort_index()
# tenyear_yield = float(data_df[data_df.index.date < self.Time.date()].iloc[-1]['10 Yr'])
# data_dict['Close_r10'] = tenyear_yield
# Indicators
bb_width_SPX = self.SPX_bb.upper_band.current.value - self.SPX_bb.lower_band.current.value
bb_width_GLD = self.GLD_bb.upper_band.current.value - self.GLD_bb.lower_band.current.value
if date == data_dict['date']:
data_dict['I_SPXbb'] = bb_width_SPX
data_dict['I_GLDbb'] = bb_width_GLD
def on_data(self, data: Slice):
# Rollover
for symbol, changed_event in data.symbol_changed_events.items():
old_symbol = changed_event.old_symbol
new_symbol = changed_event.new_symbol
tag = f"Rollover - Symbol changed at {self.time}: {old_symbol} -> {new_symbol}"
if old_symbol not in self.portfolio:
continue
quantity = self.portfolio[old_symbol].quantity
self.liquidate(old_symbol, tag=tag)
if quantity != 0: self.market_order(new_symbol, quantity, tag=tag)
uvxy_price = self.Securities[self.uvxy].Price
svxy_price = self.Securities[self.svxy].Price
#self.debug(f'uvxy_price = {uvxy_price}')
# Fetch data every day
time_for_data = (self.Time.hour == 15 and self.Time.minute == 59)
#time_for_data = (self.Time.hour == 9 and self.Time.minute == 31)
if time_for_data:
self.fetch_data(data)
if self.is_warming_up:
return
self.VX_contract = self.Securities[self.VX.mapped]
time_for_entry = (self.Time.hour == 9 and self.Time.minute >= 35) or (self.Time.hour >= 10)
if time_for_entry and self.is_market_open(self.uvxy):
obj = {"text": f"Volatility Morning Briefing ({self.Time.date()}): prediction = {self.prediction}"}
obj = json.dumps(obj)
self.Notify.web("https://hooks.slack.com/services/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", obj)
self.reinvested = False
self.morning_value = self.portfolio.total_portfolio_value
if self.prediction == 1:
if not self.portfolio.invested:
self.entry_price = uvxy_price
self.quantity = self.calculate_order_quantity(self.uvxy, 0.8)
self.message = 'Open Long Volatility position (signal)'
self.set_holdings(self.uvxy, 0.8)
self.position_status = 1
elif self.position_status == -1:
self.message = 'Closed Short Volatility position (change of signal)'
self.liquidate()
self.entry_price = uvxy_price
self.quantity = self.calculate_order_quantity(self.uvxy, 0.8)
self.message = 'Open Long Volatility position (signal)'
self.set_holdings(self.uvxy, 0.8)
self.position_status = 1
elif self.prediction == -1:
if not self.portfolio.invested:
self.entry_price = uvxy_price
self.quantity = self.calculate_order_quantity(self.uvxy, -0.5)
# self.entry_price = data[self.svxy].close
# self.quantity = self.calculate_order_quantity(self.svxy, 1.5)
self.message = 'Open Short Volatility position (signal)'
self.set_holdings(self.uvxy, -0.5)
# self.set_holdings(self.svxy, 1.5)
# self.position_status = -1
elif self.position_status == 1:
self.message = 'Close Long Volatility position (change of signal)'
self.liquidate()
self.entry_price = uvxy_price
self.quantity = self.calculate_order_quantity(self.uvxy, -0.5)
# self.entry_price = data[self.svxy].close
# self.quantity = self.calculate_order_quantity(self.svxy, 1.5)
self.message = 'Open Short Volatility position (signal)'
self.set_holdings(self.uvxy, -0.5)
# self.set_holdings(self.svxy, 1.5)
self.position_status = -1
if self.portfolio.invested and not time_for_entry:
if self.entry_price * abs(self.portfolio[self.uvxy].quantity) == 0:
self.debug(f'{self.Time}, entry_price * quantity == 0!')
return
#if (self.portfolio.total_portfolio_value - self.morning_value) / self.morning_value <= -0.12:
if self.portfolio.total_unrealized_profit / (self.entry_price * abs(self.portfolio[self.uvxy].quantity)) <= -0.1:
self.morning_value = self.portfolio.total_portfolio_value
#self.debug(f'{self.Time}, Stop Loss')
self.message = 'Close Long Volatility position (Stop Loss)' if self.position_status == 1 else 'Close Short Volatility position (Stop Loss)'
self.liquidate(asynchronous=True)
if self.position_status == 1 and not self.reinvested:
self.entry_price = uvxy_price
self.quantity = self.calculate_order_quantity(self.uvxy, -0.2)
# self.entry_price = data[self.svxy].close
# self.quantity = self.calculate_order_quantity(self.svxy, 0.8)
self.message = 'Reinvest in Short Volatility position (after SL)'
self.set_holdings(self.uvxy, -0.2)
# self.set_holdings(self.svxy, 0.8)
self.position_status = -1
self.reinvested = True
elif self.position_status == -1 and not self.reinvested:
self.entry_price = uvxy_price
self.quantity = self.calculate_order_quantity(self.uvxy, 0.3)
self.message = 'Reinvest in Long Volatility position (after SL)'
self.set_holdings(self.uvxy, 0.3)
self.position_status = 1
self.reinvested = True
#if (self.portfolio.total_portfolio_value - self.morning_value) / self.morning_value >= 0.12:
elif self.portfolio.total_unrealized_profit / (self.entry_price * abs(self.portfolio[self.uvxy].quantity)) >= 0.2:
#self.debug(F'{self.Time}, Take Profit')
self.message = 'Close Long Volatility position (Take Profit)' if self.position_status == 1 else 'Close Short Volatility position (Take Profit)'
self.liquidate(asynchronous=True)
self.morning_value = self.portfolio.total_portfolio_value
if self.position_status == 1:
self.entry_price = uvxy_price
self.quantity = self.calculate_order_quantity(self.uvxy, 0.8)
self.message = 'Reinvest in Long Volatility position (after TP)'
self.set_holdings(self.uvxy, 0.8)
elif self.position_status == -1:
self.entry_price = uvxy_price
self.quantity = self.calculate_order_quantity(self.uvxy, -0.5)
# self.entry_price = data[self.svxy].close
# self.quantity = self.calculate_order_quantity(self.svxy, 1.5)
self.message = 'Reinvest in Short Volatility position (after TP)'
self.set_holdings(self.uvxy, -0.5)
# self.set_holdings(self.svxy, 1.5)
def on_order_event(self, order_event: OrderEvent) -> None:
order = self.transactions.get_order_by_id(order_event.order_id)
symbol = order_event.symbol
fill_price = order_event.fill_price
fill_quantity = order_event.fill_quantity
direction = order_event.direction
date = self.Time.date()
hour = self.Time.hour
minute = self.Time.minute
second = str(self.Time.second)
#self.entry_price = fill_price
if order_event.status == OrderStatus.FILLED or order_event.status == OrderStatus.PARTIALLY_FILLED:
obj = {"text": f"{self.message}\n<PAPER MONEY> Time: {date} {hour}:{minute}:{second.zfill(2)}, Symbol: {symbol}, Quantity: {fill_quantity}, Price: {fill_price}"}
obj = json.dumps(obj)
self.Notify.web("https://hooks.slack.com/services/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", obj)
def on_end_of_algorithm(self) -> None:
self.debug("Algorithm done")
self.debug(f'Now: {self.Time}')
# region imports
from AlgorithmImports import *
from arch import arch_model
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.model_selection import KFold
from sklearn.model_selection import cross_validate
from sklearn import tree
from config import *
# endregion
# Your New Python File
class GARCH_model:
def __init__(self, data, algorithm):
self.model = arch_model(data, mean = 'AR', vol = 'Garch', p = 2, q = 2, dist = 'Normal')
self.model = self.model.fit(update_freq = 5, disp='off')
a = self.model.forecast(horizon=5)
def train(self, data):
self.model = arch_model(data, vol = 'Garch', p = 5, q = 5, dist = 'Normal')
self.model = self.model.fit(update_freq = 5, disp='off')
def predict_variance(self, h):
var_curr = self.model.conditional_volatility.iloc[-1,0 ]
result = self.model.forecast(horizon=h)
var_hat_1 = result.variance.iloc[0, 0]
var_hat_2 = result.variance.iloc[0, 1]
var_hat_3 = result.variance.iloc[0, 2]
#if var_hat_1 > 0 var_curr and v
return result.variance.iloc[0, :]
class Random_Forest_model:
def __init__(self, random_seed):
self.model = RandomForestClassifier(n_estimators=100,
criterion='gini',
max_depth=None,
min_samples_split=2,
min_samples_leaf=1,
min_weight_fraction_leaf=0.0,
max_features='sqrt',
max_leaf_nodes=None,
random_state=random_seed)
#self.model = tree.DecisionTreeClassifier()
self.X_train = pd.DataFrame()
self.Y_train = pd.DataFrame()
self.X_test = pd.DataFrame()
def process_data(self, data_dict_list, algorithm):
if len(data_dict_list) < model_setting['Random_Forest']['n_lookback']:
return
#df = pd.DataFrame(data_dict_list).dropna(axis=0)
df = pd.DataFrame(data_dict_list).ffill()
df = df[-model_setting['Random_Forest']['n_lookback']:].copy()
df = df.set_index(['date'], drop=True)
algorithm.debug(f'data: {df[-15:]}')
close_cols = []
indicator_cols = []
Y_col = ''
for x in df.columns:
if x.startswith('Close'):
close_cols.append(x)
elif x.startswith('I'):
indicator_cols.append(x)
elif x.startswith('Y'):
Y_col = x
for col in close_cols:
var = col[6:]
df['%_'+var] = (df[col] - df[col].shift(1)) / df[col].shift(1)
df = df.drop(columns = close_cols)
Y_var = Y_col[2:]
df['%_'+Y_var] = (df[Y_col].shift(-1) - df[Y_col]) / df[Y_col]
df.loc[df['%_'+Y_var] > general_setting['vix_up_threshold'], 'signal'] = 1
df.loc[df['%_'+Y_var] < general_setting['vix_down_threshold'], 'signal'] = -1
df.loc[(df['%_'+Y_var] <= general_setting['vix_up_threshold']) & (df['%_'+Y_var] >= general_setting['vix_down_threshold']), 'signal'] = 0
df = df.drop(columns=[Y_col, '%_'+Y_var])
#algorithm.debug(F'columns: {list(df.columns)}')
return df
def train_model(self, data_dict_list, algorithm):
df = self.process_data(data_dict_list, algorithm)
if df is None:
return
Y_col = 'signal'
X_cols = list(df.columns)
X_cols.remove(Y_col)
df = df.iloc[1:].copy()
df_train = df[:-1].dropna(axis=0)
df_test = df[-1:]
self.X_train = df_train[X_cols]
self.Y_train = df_train[[Y_col]]
self.X_test = df_test[X_cols]
self.model.fit(self.X_train, self.Y_train)
signal_hat = self.model.predict(self.X_test)[0]
return signal_hat
# region imports
from AlgorithmImports import *
# endregion
# Your New Python File
general_setting = {
'vix_up_threshold': 0.003,
'vix_down_threshold': -0.003,
'data_': {
# index
"VIX":
{"type": "index", "source": "QC"},
"SPX":
{"type": "index", "source": "QC"},
# equity
"UUP":
{"type": "equity", "source": "QC"},
# future
"VX":
{"type": "future", "source": Futures.Indices.VIX},
"CL":
{"type": "future", "source": Futures.Energies.CrudeOilWTI},
"GLD":
{"type": "future", "source": Futures.Metals.GOLD},
# other data
# "VVIX":
# {"type": "data", "source": CBOE},
"USTYCR":
{"type": "data", "source": USTreasuryYieldCurveRate},
"10_year_yield":
{"type": "external",
"source": "https://raw.githubusercontent.com/deerfieldgreen/us-department-treasury/refs/heads/main/data/daily-treasury-rates.csv",
"col_date": "Date",
"col_val": "10 Yr"}
},
}
model_setting = {
'Random_Forest': {
'n_lookback': 60,
},
'Garch': {
}
}