| Overall Statistics |
|
Total Orders 174 Average Win 7.97% Average Loss -19.81% Compounding Annual Return 1152.726% Drawdown 74.600% Expectancy 0.227 Start Equity 100000 End Equity 2299016.4 Net Profit 2199.016% Sharpe Ratio 9.195 Sortino Ratio 12.711 Probabilistic Sharpe Ratio 91.286% Loss Rate 12% Win Rate 88% Profit-Loss Ratio 0.40 Alpha 10.338 Beta 9.91 Annual Standard Deviation 1.294 Annual Variance 1.673 Information Ratio 9.699 Tracking Error 1.21 Treynor Ratio 1.2 Total Fees $3283.05 Estimated Strategy Capacity $97000000.00 Lowest Capacity Asset NQ YGT6HGVF2SQP Portfolio Turnover 131.32% |
# region imports
from datetime import timedelta
from AlgorithmImports import *
import numpy as np
import json
# endregion
class PurpleReign(QCAlgorithm):
def initialize(self):
self.set_start_date(2023, 1, 1)
self.set_end_date(2024, 6, 1)
self.set_cash(100000)
self.future_chains = None
# self.symbol = self.add_equity("SPY").symbol
# Set the symbol of the asset we want to trade
future = self.add_future(Futures.Indices.NASDAQ_100_E_MINI, Resolution.Minute)
future.SetFilter(timedelta(0), timedelta(182))
self.symbol = future.Symbol
# Set up indicators
self._macd = self.MACD(self.symbol, 8, 17, 9, MovingAverageType.Simple)
self._bb = self.BB(self.symbol, 18, 2, MovingAverageType.Simple)
self._kc = self.KCH(self.symbol, 18, 1.5, MovingAverageType.Simple)
# Create a RollingWindow to store the past 3 values of the MACD Histogram
self._macd_hist_window = RollingWindow[IndicatorDataPoint](3)
# Consolidate the data into 5-minute bars
self.Consolidate(self.symbol, timedelta(minutes=5), self.on_data_consolidated)
self.register_indicator(self.symbol, self._macd, timedelta(minutes=5))
self.register_indicator(self.symbol, self._bb, timedelta(minutes=5))
self.register_indicator(self.symbol, self._kc, timedelta(minutes=5))
# Setting stoploss and take profit
self.stop_loss_len = 20*5
self.stop_loss_indicator = self.MIN(self.symbol, self.stop_loss_len, Resolution.MINUTE)
self.stop_loss = 0
self.start_stop_loss = False
self.take_profit_1 = 0
self.take_profit_2 = 0
# Warming up engine
self.set_warm_up(5*20, Resolution.MINUTE)
self.Settings.FreePortfolioValuePercentage = 0.3
def on_margin_call_warning(self):
self.Error("Margin call warning")
def on_data_consolidated(self, data: slice):
# Check if the data strategy is warming up
if self.is_warming_up:
return
# Check if the Bollinger Bands are within the Keltner Channels
self.squeeze = self._bb.UpperBand.Current.Value < self._kc.UpperBand.Current.Value and self._bb.LowerBand.Current.Value > self._kc.LowerBand.Current.Value
# self.Log(f"Squeeze indicator: {self.squeeze}")
# Check for MACD entry signal
self._macd_hist_window.Add(self._macd.Histogram.Current)
# Ensure we have 3 data points in the window
if self._macd_hist_window.IsReady:
macd_hist = self._macd_hist_window[0].Value # Current MACD Histogram value
macd_hist_1 = self._macd_hist_window[1].Value # MACD Histogram value 1 bar ago
macd_hist_2 = self._macd_hist_window[2].Value # MACD Histogram value 2 bars ago
self.macd_long_in = (macd_hist > macd_hist_1 or macd_hist > macd_hist_2) and macd_hist > 0
# self.Log(f"MACD entry: {self.macd_long_in}")
# Find the future contract
for chain in self.future_chains:
self.popular_contracts = [contract for contract in chain.value if contract.open_interest > 1000]
if len(self.popular_contracts) == 0:
continue
sorted_bt_o_i_contracts = sorted(self.popular_contracts, key=lambda k: k.open_interest, reverse=True)
self.future_contract = sorted_bt_o_i_contracts[0]
# Entry
if not self.portfolio.invested:
if self.squeeze and self.macd_long_in:
self.log(f"Buy at {data.Close}")
self.set_holdings(self.future_contract.symbol, 1)
self.take_profit_1 = data.Close * 1.025
self.take_profit_2 = data.Close * 1.04
self.start_stop_loss = True
self.stop_loss = data.Close * 0.97
# self.log(f"Stop loss level {self.stop_loss}")
# Send notification
# self.send_webhook_notification("Buy")
# # Stop loss
# if self.start_stop_loss:
# self.stop_loss = self.stop_loss_indicator.Current.Value
# Exit
if self.portfolio.invested:
# Stop loss
# if data.Close < self.stop_loss or not self.Time.hour in range(9, 16):
if data.Close < self.stop_loss:
self.log(f"Stop loss at {data.Close}")
self.liquidate(self.future_contract.symbol)
self.start_stop_loss = False
# Send notification
# self.send_webhook_notification("Sell")
# Take profit 1
if data.Close > self.take_profit_1 and self.start_stop_loss:
self.log(f"Take profit at {data.Close}")
self.set_holdings(self.future_contract.symbol, 0.4)
# Take profit 2
if data.Close > self.take_profit_2 and self.start_stop_loss:
self.log(f"Take profit at {data.Close}")
self.liquidate(self.future_contract.symbol)
self.start_stop_loss = False
# Send notification
# self.send_webhook_notification("Sell")
def on_data(self, data: Slice):
self.future_chains = data.FutureChains
def send_webhook_notification(self, position):
"""
Send a webhook notification with the given position (Buy/Sell).
"""
url = 'https://newagewallstreet.io/version-test/api/1.1/wf/catch-trades?api_token=cc7d071598606d101e84f252c9654956'
data = {
"strat": "1716875896866x801605936682184200",
"ticker": "NQM4",
"position": position
}
headers = {'Content-Type': 'application/json'}
try:
self.notify.web(url, json.dumps(data), headers)
self.log(f"Webhook notification sent. Position: {position}")
except Exception as e:
self.log(f"Failed to send webhook notification. Error: {str(e)}")# region imports
from datetime import timedelta
import datetime
from AlgorithmImports import *
from universe import *
import numpy as np
import json
# endregion
class PurpleReign(QCAlgorithm):
def initialize(self):
self.set_start_date(2023, 3, 10)
self.set_end_date(2023, 3, 20)
self.set_cash(100000)
self.indicators = {}
self.future_chains = None
# self.symbol = self.add_equity("SPY").symbol
# Set the symbol of the asset we want to trade
# Add future universe
self.universe_settings.asynchronous = True
universe = atomik_future()
self.add_universe_selection(universe)
# Set up indicators
# Create a manual indicators
self._macd = MovingAverageConvergenceDivergence(8, 17, 9, MovingAverageType.Simple)
self._bb = BollingerBands(18, 2, MovingAverageType.Simple)
self._kc = KeltnerChannels(18, 1.5, MovingAverageType.Simple)
self._macd_hist_window = RollingWindow[IndicatorDataPoint](3)
# # Setting stoploss
# self.stop_loss_len = 20*5
# self.stop_loss_indicator = self.MIN(self.symbol, self.stop_loss_len, Resolution.MINUTE)
# self.stop_loss = None
# self.start_stop_loss = False
# Warming up engine
self.set_warm_up(5*20, Resolution.MINUTE)
self.Settings.FreePortfolioValuePercentage = 0.3
def on_margin_call_warning(self):
self.Error("Margin call warning")
def on_data_consolidated(self, sender, consolidated):
# Check if the data strategy is warming up
if self.is_warming_up:
return
# Log the indicator value for each symbol BB, KC, MACD with the symbol and expiry date
symbol = consolidated.Symbol
self.Log(f"{symbol} Open: {consolidated.Open} High: {consolidated.High} Low: {consolidated.Low} Close: {consolidated.Close}")
# # Check for MACD entry signal
# self._macd_hist_window.Add(self._macd.Histogram.Current)
# # Ensure we have 3 data points in the window
# if self._macd_hist_window.IsReady:
# macd_hist = self._macd_hist_window[0].Value # Current MACD Histogram value
# macd_hist_1 = self._macd_hist_window[1].Value # MACD Histogram value 1 bar ago
# macd_hist_2 = self._macd_hist_window[2].Value # MACD Histogram value 2 bars ago
# self.macd_long_in = (macd_hist > macd_hist_1 or macd_hist > macd_hist_2) and macd_hist > 0
# self.Log(f"MACD entry: {self.macd_long_in}")
# # Find the future contract
# for chain in self.future_chains:
# self.popular_contracts = [contract for contract in chain.value if contract.open_interest > 1000]
# if len(self.popular_contracts) == 0:
# continue
# sorted_bt_o_i_contracts = sorted(self.popular_contracts, key=lambda k: k.open_interest, reverse=True)
# self.future_contract = sorted_bt_o_i_contracts[0]
# # Entry
# if not self.portfolio.invested:
# if self.squeeze and self.macd_long_in:
# self.log(f"Price is {data.Close}")
# self.set_holdings(self.future_contract.symbol, 1)
# self.stop_loss = self.stop_loss_indicator.Current.Value
# self.log(f"Stop loss level {self.stop_loss}")
# # Send notification
# self.send_webhook_notification("Buy")
# # Exit
# if self.portfolio.invested:
# if data.Close < self.stop_loss or not self.Time.hour in range(9, 16):
# self.log(f"Stop loss at {data.Close}")
# self.liquidate(self.future_contract.symbol)
# # Send notification
# self.send_webhook_notification("Sell")
# def on_securities_changed(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
# for security in changes.added_securities:
# if security.Symbol.IsCanonical():
# self.future = security
def on_data(self, data: Slice):
# self.future_chains = data.FutureChains
# log the current value of the indicators and respective symbol
pass
# def send_webhook_notification(self, position):
# """
# Send a webhook notification with the given position (Buy/Sell).
# """
# url = 'https://newagewallstreet.io/version-test/api/1.1/wf/catch-trades?api_token=cc7d071598606d101e84f252c9654956'
# data = {
# "strat": "1716875896866x801605936682184200",
# "ticker": "MNQM4",
# "position": position
# }
# headers = {'Content-Type': 'application/json'}
# try:
# self.notify.web(url, json.dumps(data), headers)
# self.log(f"Webhook notification sent. Position: {position}")
# except Exception as e:
# self.log(f"Failed to send webhook notification. Error: {str(e)}")
def on_securities_changed(self, changes) -> None:
for security in changes.added_securities:
symbol = security.Symbol
consolidator = TradeBarConsolidator(timedelta(minutes=5))
self.subscription_manager.add_consolidator(symbol, consolidator)
consolidator.data_consolidated += self.on_data_consolidated
# self.asign_indicator(symbol, self._macd, consolidator, "MACD")
# self.asign_indicator(symbol, self._bb, consolidator, "BB")
# self.asign_indicator(symbol, self._kc, consolidator, "KC")
def asign_indicator(self, symbol, indicator, consolidator, indicator_name):
self.register_indicator(symbol, indicator, consolidator)
if symbol not in self.indicators:
self.indicators[symbol] = {}
if indicator_name not in self.indicators[symbol]:
self.indicators[symbol][indicator_name] = indicator
#region imports
from AlgorithmImports import *
from datetime import timedelta
import datetime
from Selection.FutureUniverseSelectionModel import FutureUniverseSelectionModel # type: ignore
#endregion
class atomik_future(FutureUniverseSelectionModel):
def __init__(self) -> None:
super().__init__(timedelta(1), self.select_future_chain_symbols)
def select_future_chain_symbols(self, utc_time: datetime) -> List[Symbol]:
return [
Symbol.create(Futures.Indices.NASDAQ_100_E_MINI, SecurityType.FUTURE, Market.CME),
Symbol.create(Futures.Indices.MICRO_NASDAQ_100_E_MINI, SecurityType.FUTURE, Market.CME),
Symbol.create(Futures.Indices.SP_500_E_MINI, SecurityType.FUTURE, Market.CME),
Symbol.create(Futures.Indices.MICRO_SP_500_E_MINI, SecurityType.FUTURE, Market.CME),
]