| Overall Statistics |
|
Total Orders 16 Average Win 12.32% Average Loss -10.26% Compounding Annual Return 7.568% Drawdown 4.900% Expectancy 0.100 Start Equity 1000000 End Equity 1004607.4 Net Profit 0.461% Sharpe Ratio 0.091 Sortino Ratio 0.105 Probabilistic Sharpe Ratio 42.714% Loss Rate 50% Win Rate 50% Profit-Loss Ratio 1.20 Alpha 0.055 Beta -0.087 Annual Standard Deviation 0.19 Annual Variance 0.036 Information Ratio -1.835 Tracking Error 0.226 Treynor Ratio -0.2 Total Fees $890.10 Estimated Strategy Capacity $350000000.00 Lowest Capacity Asset NQ YQYHC5L1GPA9 Portfolio Turnover 611.79% |
#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.debug(f'Starting new algo.')
self.set_start_date(2025, 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'
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.wt_ES = 1/(1+self.beta)
self.wt_NQ = self.beta/(1+self.beta)
self.wt_ES = self.beta/(1+self.beta)
self.wt_NQ = 1/(1+self.beta)
# Close Pairs for mean-reversion
if self.IsInvested:
if (self.ShortSpread and zscore <= self.exit) or (self.LongSpread and zscore >= -self.exit):
self.Liquidate()
self.debug(f'liquidated at {self.Time}')
# Consider opening position only every 1 hour
else:
if adf_test_passed:
pass
if zscore > self.enter:
#short spread
self.SetHoldings(self.ES_contract.symbol, -self.wt_ES)
self.SetHoldings(self.NQ_contract.symbol, self.wt_NQ)
if self.Portfolio[self.ES_contract.symbol].Quantity != 0:
self.debug(f'short spread at {self.Time}')
elif zscore < -self.enter:
#long the spread
self.SetHoldings(self.ES_contract.symbol, self.wt_ES)
self.SetHoldings(self.NQ_contract.symbol, -self.wt_NQ)
if self.Portfolio[self.ES_contract.symbol].Quantity != 0:
self.debug(f'long spread at {self.Time}')
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