| Overall Statistics |
|
Total Orders 8 Average Win 25.80% Average Loss -9.69% Compounding Annual Return 895.317% Drawdown 5.100% Expectancy 0.832 Start Equity 500000 End Equity 639148.1 Net Profit 27.830% Sharpe Ratio 12.97 Sortino Ratio 27.31 Probabilistic Sharpe Ratio 99.459% Loss Rate 50% Win Rate 50% Profit-Loss Ratio 2.66 Alpha 4.488 Beta 1.083 Annual Standard Deviation 0.332 Annual Variance 0.11 Information Ratio 14.535 Tracking Error 0.308 Treynor Ratio 3.979 Total Fees $249.40 Estimated Strategy Capacity $190000000.00 Lowest Capacity Asset NQ YOGVNNAOI1OH Portfolio Turnover 171.02% |
#region imports
from AlgorithmImports import *
#endregion
general_setting = {
"lookback": 100,
"lookback_RESOLUTION": "HOUR",
"ratio_method": "Regression",
"Take_Profit_pct": 0.3,
#"Stop_Loss_pct": 0.08,
"Stop_Loss_pct": 0.05,
"Cointegration_price": "Raw", # or could be "Log" or "Return"
"p_value_threshold": 0.1,
#"p_value_threshold": 0.025,
"enter_level": 1.5,
"exit_level": 1
}
# region imports from AlgorithmImports import * # endregion # Your New Python File
# 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(2024, 12, 1)
#self.set_end_date(2024, 7, 14)
self.set_cash(500000)
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.RTY = self.add_future(ticker=Futures.Indices.RUSSELL_2000_E_MINI, 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.RTY_sym = self.RTY.symbol
self.minute_counter = 0
self.Schedule.On(self.date_rules.every_day(), self.TimeRules.At(18,0), self.reset_minute_counter)
self.freeze = False
def reset_minute_counter(self):
self.minute_counter = 0
def freeze_over(self):
self.freeze = False
def stats(self, symbols, method="Regression", price_value='Log'):
# 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
############################################################
## 2025.01.09 - Howard Hou - adding data stats
df_ES.index = df_ES.index.droplevel(['symbol', 'expiry'])
df_NQ.index = df_NQ.index.droplevel(['symbol', 'expiry'])
start_time = df_ES.index[0]
end_time = df_ES.index[-1]
n_data = len(df_ES)
############################################################
df_ES = df_ES["close"]
df_NQ = df_NQ["close"]
if price_value == "Log":
ES = np.array(df_ES.apply(lambda x: math.log(x))) # X
NQ = np.array(df_NQ.apply(lambda x: math.log(x))) # Y
elif price_value == "Raw":
ES = np.array(df_ES)
NQ = np.array(df_NQ)
# Regression and ADF test
X = sm.add_constant(ES)
Y = NQ
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, 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)
# slope = state_means[:, 0][-1]
# intercept = state_means[:, 1][-1]
self.printed = True
return [test_passed, zscore, slope, p_value, start_time, end_time, n_data]
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:
if self.IsInvested:
# Take Profit
#if self.portfolio.total_unrealized_profit >= self.portfolio.total_portfolio_value * general_setting['Take_Profit_pct']:
if self.portfolio.total_unrealized_profit >= self.start_portfolio * general_setting['Take_Profit_pct']:
self.pnl = self.portfolio.total_unrealized_profit
self.pnl_pct = round(self.pnl / self.start_portfolio * 100, 1)
delta_logES = np.log(self.px_ES) - self.start_logES
delta_logNQ = np.log(self.px_NQ) - self.start_logNQ
roc = round(delta_logNQ / delta_logES, 2)
self.liquidate(tag=f'Take Profit, PnL={self.pnl_pct}% (d_ES={round(delta_logES,2)} d_NQ={round(delta_logNQ,2)} r={roc})')
# Stop Loss
#elif self.portfolio.total_unrealized_profit <= -self.portfolio.total_portfolio_value * general_setting['Stop_Loss_pct']:
elif self.portfolio.total_unrealized_profit <= -self.start_portfolio * general_setting['Stop_Loss_pct']:
self.pnl = self.portfolio.total_unrealized_profit
self.pnl_pct = round(self.pnl / self.start_portfolio * 100, 1)
delta_logES = np.log(self.px_ES) - self.start_logES
delta_logNQ = np.log(self.px_NQ) - self.start_logNQ
roc = round(delta_logNQ / delta_logES, 2)
self.liquidate(tag = f'Stop Loss, PnL={self.pnl_pct}% (d_ES={round(delta_logES,2)} d_NQ={round(delta_logNQ,2)} r={roc})')
self.freeze = True
hour = self.Time.hour
self.schedule.on(self.date_rules.tomorrow, self.TimeRules.At(9, 30), self.freeze_over)
# 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'], price_value = "Log")
if stats == 0:
self.minute_counter = (self.minute_counter + 1) % 60
return
self.debug(f'stats: {stats}')
adf_test_passed = stats[0]
zscore= stats[1][-1]
self.beta = stats[2]
self.p_value = stats[3]
self.start_coint_time = stats[4]
self.end_coint_time = stats[5]
self.n_coint_data = stats[6]
beta_weighted = self.beta
self.wt_ES = 1/(1+beta_weighted)
self.wt_NQ = beta_weighted/(1+beta_weighted)
# n_ES_ratio = 1 / (beta_weighted + 1)
# wt_ES_ratio = (1 * self.px_ES*50) / (1*self.px_ES*50 + beta_weighted*self.px_NQ*20)
# self.wt_ES = wt_ES_ratio
# self.wt_NQ = 1 - wt_ES_ratio
# n_ES_NQ_ratio = 1 / (1+beta_weighted)
# wt_ES_NQ_ratio = n_ES_NQ_ratio * 275000 / 400000
# self.wt_ES = wt_ES_NQ_ratio / (1 + wt_ES_NQ_ratio)
# self.wt_NQ = 1 / (1 + wt_ES_NQ_ratio)
if self.IsInvested:
if (self.ShortSpread and zscore <= self.exit) or (self.LongSpread and zscore >= -self.exit):
self.pnl = self.portfolio.total_unrealized_profit
self.pnl_pct = round(self.pnl / self.start_portfolio * 100, 1)
delta_logES = np.log(self.px_ES) - self.start_logES
delta_logNQ = np.log(self.px_NQ) - self.start_logNQ
roc = round(delta_logNQ / delta_logES, 2)
self.Liquidate(tag=f'Mean Reversion, PnL={self.pnl_pct}% (d_ES={round(delta_logES,2)} d_NQ={round(delta_logNQ,2)} r={roc})')
self.debug(f'liquidated at {self.Time}')
else:
if adf_test_passed:
pass
elif zscore > self.enter and not self.freeze:
#short spread
# self.SetHoldings(self.ES_contract.symbol, -self.wt_ES, tag=f'z={round(zscore,2)} beta={round(beta_weighted,2)}')
# self.SetHoldings(self.NQ_contract.symbol, self.wt_NQ, tag=f'z={round(zscore,2)} beta={round(beta_weighted,2)}')
self.start_portfolio = self.portfolio.total_portfolio_value
self.start_logES = np.log(self.px_ES)
self.start_logNQ = np.log(self.px_NQ)
self.SetHoldings(self.ES_contract.symbol, -0.51, tag=f'z={round(zscore,2)} beta={round(beta_weighted,2)}')
self.SetHoldings(self.NQ_contract.symbol, 0.51, tag=f'z={round(zscore,2)} beta={round(beta_weighted,2)}')
elif zscore < -self.enter and not self.freeze:
#long the spread
# self.SetHoldings(self.ES_contract.symbol, self.wt_ES, tag=f'z={round(zscore,2)} beta={round(beta_weighted,2)}')
# self.SetHoldings(self.NQ_contract.symbol, -self.wt_NQ, tag=f'z={round(zscore,2)} beta={round(beta_weighted,2)}')
self.start_portfolio = self.portfolio.total_portfolio_value
self.start_logES = np.log(self.px_ES)
self.start_logNQ = np.log(self.px_NQ)
self.SetHoldings(self.ES_contract.symbol, -0.51, tag=f'z={round(zscore,2)} beta={round(beta_weighted,2)}')
self.SetHoldings(self.NQ_contract.symbol, 0.51, tag=f'z={round(zscore,2)} beta={round(beta_weighted,2)}')
date = self.Time.date()
hour = self.Time.hour
minute = self.Time.minute
message = f"ES_NQ Heartbeat {date} {hour}:{minute}: zscore: {zscore}, beta: {self.beta}, p-value: {self.p_value}, adf_test_passed: {adf_test_passed}, \nstart_coint_time: {self.start_coint_time}, end_coint_time: {self.end_coint_time}, n_coint_data: {self.n_coint_data}"
self.debug(f'{message}')
self.plot("Plot", "Z score", zscore)
self.plot("Plot", "beta", self.beta)
#self.plot("Plot", "p value", self.p_value)
# self.plot("z score", "Raw", zscore)
# self.plot("z score", "Log", zscore_1)
# self.plot("BETA", "Raw", self.beta)
# self.plot("BETA", "Log", beta_1)
# self.plot("P_value", "Raw", self.p_value)
# self.plot("P_value", "Log", p_value_1)
self.minute_counter = (self.minute_counter + 1) % 60
# message = f'<ES_NQ HOURLY UPDATE> {self.Time}, zscore={zscore}, beta={self.beta}, p_value={self.p_value}'
# obj = {"text": message}
# self.notify.web(webhook_url, obj)
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 = {"text": f"<LIVE REALMONEY> 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)