| Overall Statistics |
|
Total Orders 488 Average Win 6.27% Average Loss -5.02% Compounding Annual Return 57.202% Drawdown 24.300% Expectancy 0.105 Start Equity 1000000 End Equity 1619479 Net Profit 61.948% Sharpe Ratio 1.065 Sortino Ratio 1.314 Probabilistic Sharpe Ratio 49.298% Loss Rate 51% Win Rate 49% Profit-Loss Ratio 1.25 Alpha 0.486 Beta -0.481 Annual Standard Deviation 0.394 Annual Variance 0.155 Information Ratio 0.673 Tracking Error 0.421 Treynor Ratio -0.873 Total Fees $39861.00 Estimated Strategy Capacity $380000000.00 Lowest Capacity Asset NQ YQYHC5L1GPA9 Portfolio Turnover 1142.48% |
#region imports
from AlgorithmImports import *
#endregion
general_setting = {
"lookback": 100,
"lookback_RESOLUTION": "HOUR",
"ratio_method": "Regression",
"Take_Profit_pct": 0.08,
"Stop_Loss_pct": 0.08,
"p_value_threshold": 0.1,
"enter_level": 1.5,
"exit_level": 1.2
}# region imports
from AlgorithmImports import *
import numpy as np
import pandas as pd
import math
import statsmodels.api as sm
from pandas.tseries.offsets import BDay
from pykalman import KalmanFilter
from statsmodels.tsa.stattools import coint, adfuller
# endregion
from config import general_setting
class HipsterYellowGreenHamster(QCAlgorithm):
def initialize(self):
self.set_warm_up(timedelta(days=100))
self.debug(f'Starting new algo.')
self.set_start_date(2024, 1, 1)
self.set_cash(1000000)
self.universe_settings.asynchronous = True
self.universe_settings.resolution = Resolution.MINUTE
# lookback frequency settings
self.lookback = general_setting['lookback']
self.lookback_RESOLUTION = general_setting['lookback_RESOLUTION']
self.enter = general_setting["enter_level"]
self.exit = general_setting["exit_level"]
self.ES = self.add_future(ticker=Futures.Indices.SP500EMini, resolution=Resolution.MINUTE, market="CME", fill_forward=True, leverage=3,
extended_market_hours=True, data_normalization_mode = DataNormalizationMode.BackwardsRatio,
data_mapping_mode = DataMappingMode.LastTradingDay, contract_depth_offset = 0)
self.ES_sym = self.ES.symbol
self.NQ = self.add_future(Futures.Indices.NASDAQ100EMini, resolution=Resolution.MINUTE, market="CME", fill_forward=True, leverage=3,
extended_market_hours=True, data_normalization_mode = DataNormalizationMode.BackwardsRatio,
data_mapping_mode = DataMappingMode.LastTradingDay, contract_depth_offset = 0)
self.NQ_sym = self.NQ.symbol
self.minute_counter = 0
self.Schedule.On(self.date_rules.every_day(), self.TimeRules.At(18,0), self.reset_minute_counter)
self.strategy = 'Mean_Diversion'
self.spread_position = None
self.Zscore_container = []
self.Zscore_low = None
self.Zscore_high = None
self.Beta_container = []
self.Beta_low = None
self.Beta_high = None
self.flag_directional = False
def reset_minute_counter(self):
self.minute_counter = 0
def stats(self, symbols, method="Regression"):
# lookback here refers to market hour, whereas additional extended-market-hour data are also included.
if self.lookback_RESOLUTION == "MINUTE":
df_ES = self.History(symbols[0], self.lookback, Resolution.MINUTE)
df_NQ = self.History(symbols[1], self.lookback, Resolution.MINUTE)
elif self.lookback_RESOLUTION == "HOUR":
df_ES = self.History(symbols[0], self.lookback, Resolution.HOUR)
df_NQ = self.History(symbols[1], self.lookback, Resolution.HOUR)
else:
df_ES = self.History(symbols[0], self.lookback, Resolution.Daily)
df_NQ = self.History(symbols[1], self.lookback, Resolution.Daily)
if df_ES.empty or df_NQ.empty:
return 0
#df_ES.index = df_ES.index.droplevel(['symbol', 'expiry'])
#df_NQ.index = df_NQ.index.droplevel(['symbol', 'expiry'])
df_ES = df_ES["close"]
df_NQ = df_NQ["close"]
ES_log = np.array(df_ES.apply(lambda x: math.log(x))) # X
NQ_log = np.array(df_NQ.apply(lambda x: math.log(x))) # Y
# Regression and ADF test
X = sm.add_constant(ES_log)
Y = NQ_log
model = sm.OLS(Y, X)
results = model.fit()
sigma = math.sqrt(results.mse_resid)
slope = results.params[1]
intercept = results.params[0]
res = results.resid
zscore = res/sigma
adf = adfuller(res)
p_value = adf[1]
test_passed = p_value <= general_setting['p_value_threshold']
# Kalman Filtering to get parameters
if method == "Kalman_Filter":
obs_mat = sm.add_constant(ES_log, prepend=False)[:, np.newaxis]
trans_cov = 1e-5 / (1 - 1e-5) * np.eye(2)
kf = KalmanFilter(n_dim_obs=1, n_dim_state=2,
initial_state_mean=np.ones(2),
initial_state_covariance=np.ones((2, 2)),
transition_matrices=np.eye(2),
observation_matrices=obs_mat,
observation_covariance=0.5,
transition_covariance=0.000001 * np.eye(2))
state_means, state_covs = kf.filter(NQ_log)
slope = state_means[:, 0][-1]
intercept = state_means[:, 1][-1]
self.printed = True
return [test_passed, zscore, slope]
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}"
quantity = self.portfolio[old_symbol].quantity
self.liquidate(old_symbol, tag=tag)
if quantity != 0: self.market_order(new_symbol, quantity, tag=tag)
self.log(tag)
# get contracts
self.ES_contract = self.Securities[self.ES.mapped]
self.NQ_contract = self.Securities[self.NQ.mapped]
# Portfolio status
self.IsInvested = (self.Portfolio[self.ES_contract.symbol].Invested) or (self.Portfolio[self.NQ_contract.symbol].Invested)
self.ShortSpread = self.Portfolio[self.ES_contract.symbol].IsShort
self.LongSpread = self.Portfolio[self.ES_contract.symbol].IsLong
self.pos_ES = self.Portfolio[self.ES_contract.symbol].Quantity
self.px_ES = self.Portfolio[self.ES_contract.symbol].Price
self.pos_NQ = self.Portfolio[self.NQ_contract.symbol].Quantity
self.px_NQ = self.Portfolio[self.NQ_contract.symbol].Price
self.equity =self.Portfolio.TotalPortfolioValue
# Take profit or stop loss every 1 minute
if self.minute_counter != 0:
# Take Profit
if self.Portfolio[self.ES_contract.symbol].unrealized_profit + self.Portfolio[self.NQ_contract.symbol].unrealized_profit >= self.portfolio.total_portfolio_value * general_setting['Take_Profit_pct']:
self.liquidate()
self.debug(f'liquidated (Take Profit) at {self.Time}')
# Stop Loss
if self.Portfolio[self.ES_contract.symbol].unrealized_profit + self.Portfolio[self.NQ_contract.symbol].unrealized_profit <= -self.portfolio.total_portfolio_value * general_setting['Stop_Loss_pct']:
self.liquidate()
self.debug(f'liquidated (Stop Loss) at {self.Time}')
# Do cointegration and pairs open/close every 1 hour
else:
stats = self.stats(symbols=[self.ES.mapped, self.NQ.mapped], method=general_setting['ratio_method'])
if stats == 0:
self.minute_counter = (self.minute_counter + 1) % 60
return
adf_test_passed = stats[0]
zscore= stats[1][-1]
self.beta = stats[2]
self.Beta_container.append(self.beta)
if len(self.Beta_container) > 100:
self.Beta_low = np.quantile(self.Beta_container, 0.2)
self.Beta_high = np.quantile(self.Beta_container, 0.8)
self.flag_directional = True if self.beta > self.Beta_high else False
self.wt_ES = self.beta/(1+self.beta)
self.wt_NQ = 1/(1+self.beta)
self.wt_ES_directional = 1/(1+self.beta)
self.wt_NQ_directional = self.beta/(1+self.beta)
if self.IsInvested:
# Close and open mean-reverting portfolio
if self.strategy == 'Mean_Diversion':
if self.spread_position == 'long':
if zscore <= 1.3:
self.liquidate()
# if self.flag_directional:
# self.SetHoldings(self.ES_contract.symbol, self.wt_ES_directional)
# self.SetHoldings(self.NQ_contract.symbol, -self.wt_NQ_directional)
# else:
self.SetHoldings(self.ES_contract.symbol, self.wt_ES)
self.SetHoldings(self.NQ_contract.symbol, -self.wt_NQ)
self.strategy = 'Mean_Reversion'
self.spread_position = 'short'
elif self.spread_position == 'short':
if zscore >= -1.3:
self.liquidate()
# if self.flag_directional:
# self.SetHoldings(self.ES_contract.symbol, -self.wt_ES_directional)
# self.SetHoldings(self.NQ_contract.symbol, self.wt_NQ_directional)
# else:
self.SetHoldings(self.ES_contract.symbol, -self.wt_ES)
self.SetHoldings(self.NQ_contract.symbol, self.wt_NQ)
self.strategy = 'Mean_Reversion'
self.spread_position = 'long'
# Close the Mean reversion portfolio
elif self.strategy == 'Mean_Reversion':
if self.spread_position == 'short':
if zscore <= 1:
self.liquidate()
self.strategy = 'Mean_Diversion'
self.spread_position = None
elif self.spread_position == 'long':
if zscore >= -1:
self.liquidate()
self.strategy = 'Mean_Diversion'
self.spread_position = None
# Consider opening position only every 1 hour
else:
if adf_test_passed:
pass
if zscore > self.enter:
# long the spread
if self.flag_directional:
self.SetHoldings(self.ES_contract.symbol, -self.wt_ES_directional)
self.SetHoldings(self.NQ_contract.symbol, self.wt_NQ_directional)
else:
self.SetHoldings(self.ES_contract.symbol, -self.wt_ES)
self.SetHoldings(self.NQ_contract.symbol, self.wt_NQ)
self.strategy = 'Mean_Diversion'
self.spread_position = 'long'
elif zscore < -self.enter:
if self.flag_directional:
self.SetHoldings(self.ES_contract.symbol, self.wt_ES_directional)
self.SetHoldings(self.NQ_contract.symbol, -self.wt_NQ_directional)
# short the spread
else:
self.SetHoldings(self.ES_contract.symbol, self.wt_ES)
self.SetHoldings(self.NQ_contract.symbol, -self.wt_NQ)
self.strategy = 'Mean_Diversion'
self.spread_position = 'short'
self.plot("z score", "z value", zscore)
self.plot("BETA", "beta", self.beta)
self.minute_counter = (self.minute_counter + 1) % 60
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)
if order_event.status == OrderStatus.FILLED or order_event.status == OrderStatus.PARTIALLY_FILLED:
obj = {"content": f"<LIVE REALMONEY> Time: {date} {hour}:{minute}:{second.zfill(2)}, Symbol: {symbol}, Quantity: {fill_quantity}, Price: {fill_price}"}
obj = json.dumps(obj)
#################
# 2024.09.25 - Kevin Stoll - old webhook. Slack invalidated this for some reason. New URL is below
# self.Notify.web("https://hooks.slack.com/services/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", obj)
#################
######### 2024.09.25 - Kevin Stoll - new webhook, from updated Slack bot.
self.Notify.web("https://hooks.slack.com/services/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", obj)
###### 2024.12.06 - Kevin Stoll - adding discord webhook #channel - heartbeats
self.Notify.web("https://discord.com/api/webhooks/1312611480046403614/5NFPHqBJI3kvYS9-DiGh3SeUh_jxGD603zFZIx-c9NFuwNeN-S7GuA0cY9JYgHH2A00C", obj)
# region imports from AlgorithmImports import * # endregion # Your New Python File