| Overall Statistics |
|
Total Orders 174 Average Win 8.50% Average Loss -5.48% Compounding Annual Return 123.973% Drawdown 28.100% Expectancy 0.216 Start Equity 500000 End Equity 972124.9 Net Profit 94.425% Sharpe Ratio 1.812 Sortino Ratio 2.322 Probabilistic Sharpe Ratio 65.967% Loss Rate 52% Win Rate 48% Profit-Loss Ratio 1.55 Alpha 1.076 Beta -1.101 Annual Standard Deviation 0.508 Annual Variance 0.258 Information Ratio 1.443 Tracking Error 0.54 Treynor Ratio -0.836 Total Fees $8737.60 Estimated Strategy Capacity $750000000.00 Lowest Capacity Asset NQ YOGVNNAOI1OH Portfolio Turnover 516.30% |
#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,
"enter_level": 1.5,
"exit_level": 1
}# 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, 1, 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.minute_counter = 0
self.Schedule.On(self.date_rules.every_day(), self.TimeRules.At(18,0), self.reset_minute_counter)
def reset_minute_counter(self):
self.minute_counter = 0
def stats(self, symbols, method="Regression", price_value='Raw'):
# 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 = 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]
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(tag='Take Profit')
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(tag = 'Stop Loss')
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'], price_value = "Log")
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.p_value = stats[3]
beta_weighted = self.beta
self.wt_ES = 1/(1+beta_weighted)
self.wt_NQ = beta_weighted/(1+beta_weighted)
# Close Pairs for mean-reversion
if self.IsInvested:
if (self.ShortSpread and zscore <= self.exit) or (self.LongSpread and zscore >= -self.exit):
self.Liquidate(tag='Mean Reversion')
self.debug(f'liquidated at {self.Time}')
# Consider opening position only every 1 hour
else:
if adf_test_passed:
pass
elif 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("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
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)