Overall Statistics |
Total Orders 3139 Average Win 3.56% Average Loss -2.07% Compounding Annual Return 2141.200% Drawdown 66.800% Expectancy 0.165 Start Equity 100000 End Equity 4729797.35 Net Profit 4629.797% Sharpe Ratio 14.275 Sortino Ratio 23.41 Probabilistic Sharpe Ratio 96.129% Loss Rate 57% Win Rate 43% Profit-Loss Ratio 1.72 Alpha 17.634 Beta 9.045 Annual Standard Deviation 1.335 Annual Variance 1.782 Information Ratio 14.978 Tracking Error 1.262 Treynor Ratio 2.107 Total Fees $561476.80 Estimated Strategy Capacity $60000000.00 Lowest Capacity Asset NQ YGT6HGVF2SQP Portfolio Turnover 7640.63% |
# 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 self.stop_loss_len = 20*5 self.stop_loss_indicator = self.MIN(self.symbol, self.stop_loss_len, Resolution.MINUTE) self.lowest_low = 0 self.stop_loss = 0 self.start_stop_loss = False self.closed_window = RollingWindow[TradeBar](2) # Warming up engine self.set_warm_up(5*20, Resolution.MINUTE) self.settings.free_portfolio_value_percentage = 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 # Track the last 2 bars self.closed_window.Add(data) # 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.start_stop_loss = True self.stop_loss = self.lowest_low # self.log(f"Stop loss level {self.stop_loss}") # Send notification # self.send_webhook_notification("Buy") # Exit if self.portfolio.invested and self.start_stop_loss: # Register stop loss and take profit levels if self.closed_window.IsReady: if self.closed_window[0].Close > self.closed_window[1].Close: self.stop_loss += (self.closed_window[0].Close - self.closed_window[1].Close) # 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 def on_data(self, data: Slice): self.future_chains = data.FutureChains self.lowest_low = self.stop_loss_indicator.Current.Value 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), ]