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),
        ]