Does quantconnect force you to trade on the next day when using external signals in backtesting

from AlgorithmImports import *
import pandas as pd
import numpy as np
from io import StringIO
from datetime import timedelta, datetime, date

class PredictionBasedStrategy(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2024, 12, 2)   # Backtest start date
        self.SetCash(100000)            # Starting capital

        self.UniverseSettings.Resolution = Resolution.Daily
        
        # Add SPY to ensure we can schedule properly on market days
        self.spySymbol = self.AddEquity("SPY", Resolution.Daily).Symbol

        # Attempt to load predictions from Object Store
        predictions_file_key = "test_predictions.csv"
        if not self.ObjectStore.ContainsKey(predictions_file_key):
            self.Error(f"{predictions_file_key} not found in Object Store.")
            return

        file_bytes_net = self.ObjectStore.ReadBytes(predictions_file_key)
        if file_bytes_net is None:
            self.Error(f"Could not read {predictions_file_key} from Object Store.")
            return

        file_bytes = bytes(bytearray(file_bytes_net))
        csv_data = StringIO(file_bytes.decode('utf-8'))
        self.predictions_df = pd.read_csv(csv_data, parse_dates=True, index_col=0)

        # Build a dictionary: { date => {ticker => prediction} }
        self.predictions_by_date = {}
        for dt, row in self.predictions_df.iterrows():
            # Convert the DataFrame index (a Timestamp) to a raw date() for easy dictionary lookup
            date_key = dt.to_pydatetime().date()
            self.predictions_by_date[date_key] = row.to_dict()

        # Add each ticker as an Equity asset
        self.tickers = list(self.predictions_df.columns)
        for ticker in self.tickers:
            self.AddEquity(ticker, Resolution.Daily)

        # How many stocks to trade on each side
        self.max_stocks = 1

        # True = trade at next day’s CLOSE; False = next day’s OPEN
        # Both use only a 1-day delay in this example
        self.tradeAtNextDayClose = False

        # If True, strategy goes only Long; if False, it does Long & Short
        self.long_only_mode = False

        # Because we want "higher is better," we do NOT invert predictions
        self.is_prediction_inverted = False

        # Schedule daily rebalancing time
        if self.tradeAtNextDayClose:
            # Rebalance near the close
            self.Schedule.On(
                self.DateRules.EveryDay(),
                self.TimeRules.BeforeMarketClose(self.spySymbol, 10),  # e.g., 10 min before close
                self.Rebalance
            )
            self.Log("Trading Mode: Next Day CLOSE (T+1 close), 1-day lookback.")
        else:
            # Rebalance shortly after the open
            self.Schedule.On(
                self.DateRules.EveryDay(),
                self.TimeRules.AfterMarketOpen(self.spySymbol, 5),     # e.g., 5 min after open
                self.Rebalance
            )
            self.Log("Trading Mode: Next Day OPEN (T+1 open), 1-day lookback.")

        # End-of-day portfolio log
        self.Schedule.On(
            self.DateRules.EveryDay(),
            self.TimeRules.AfterMarketClose(self.spySymbol, 1),
            self.LogPortfolio
        )

        self.lastRebalanceDate = None
        self.current_targets = None  # to track changes in positions

    def Rebalance(self):
        """
        Rebalance once per day, reading predictions from 1 day ago.
        """
        today = self.Time.date()
        if self.lastRebalanceDate == today:
            return
        self.lastRebalanceDate = today

        # The key difference: always 1-day offset
        # If you want day (T) predictions for day (T+1) trades,
        # we look at "yesterday" from the perspective of 'today'.
        days_back = 1

        # This method tries to find a valid trading day 'days_back' days before 'today',
        # skipping weekends/holidays
        prediction_day = self.GetDelayedTradingDay(today, days_back, self.spySymbol)

        if prediction_day not in self.predictions_by_date:
            self.Log(f"{today}: No predictions available for {prediction_day}; skipping rebalance.")
            return

        # Pull out the predictions for 'prediction_day'
        preds = self.predictions_by_date[prediction_day]

        # If we are not inverting, we do nothing special; "higher is better" means top are longs
        effective_preds = {}
        for ticker, value in preds.items():
            effective_preds[ticker] = -value if self.is_prediction_inverted else value

        # Sort descending so highest predictions go to LONG
        sorted_items = sorted(effective_preds.items(), key=lambda x: x[1], reverse=True)
        top_n = sorted_items[:self.max_stocks]      # highest = long
        bottom_n = sorted_items[-self.max_stocks:]  # lowest = short

        # Convert to just ticker symbols
        long_targets = [x[0] for x in top_n]
        short_targets = [x[0] for x in bottom_n]

        if self.long_only_mode:
            self.Log(f"{today}: Long‐only mode. Ignoring short targets.")
            short_targets = []

        new_targets = (sorted(long_targets), sorted(short_targets))
        if self.current_targets == new_targets:
            self.Log(f"{today}: No change in targets. Holding positions.")
            return

        # Update stored targets
        self.current_targets = new_targets
        time_str = "CLOSE" if self.tradeAtNextDayClose else "OPEN"
        self.Log(f"Rebalance on {today} at {time_str}, using predictions from {prediction_day}. "
                 f"New targets: LONG {long_targets}, SHORT {short_targets}")

        # First, liquidate anything not in the new target set
        targetSymbols = set(long_targets + short_targets)
        for s in list(self.Portfolio.Keys):
            if self.Portfolio[s].Invested and s.Value not in targetSymbols:
                self.Liquidate(s)

        # Then open new positions. We'll do an equal weighting among all chosen instruments.
        long_weight = 1.0 / len(long_targets) if long_targets else 0
        short_weight = 0 if self.long_only_mode else (-1.0 / len(short_targets) if short_targets else 0)

        for ticker in long_targets:
            symbol = self.Symbol(ticker)
            self.SetHoldings(symbol, long_weight)

        for ticker in short_targets:
            symbol = self.Symbol(ticker)
            self.SetHoldings(symbol, short_weight)

    def OnOrderEvent(self, orderEvent):
        if orderEvent.Status != OrderStatus.Filled:
            return
        symbol = orderEvent.Symbol
        qty = orderEvent.FillQuantity
        direction = "LONG" if qty > 0 else "SHORT" if qty < 0 else "FLAT"
        self.Log(f"Order filled: {symbol.Value} {direction}, Qty: {qty}, Price: {orderEvent.FillPrice}")

    def LogPortfolio(self):
        positions = []
        for s in self.Portfolio.Keys:
            if self.Portfolio[s].Invested:
                pos = self.Portfolio[s]
                pos_type = "LONG" if pos.Quantity > 0 else "SHORT" if pos.Quantity < 0 else "FLAT"
                positions.append(f"{s.Value} ({pos_type}): {pos.Quantity}")
        if positions:
            self.Log("Portfolio at EOD: " + ", ".join(positions))
        else:
            self.Log("No positions held at end of day.")

    def OnData(self, data):
        pass

    def GetDelayedTradingDay(self, baseDate: date, delay: int, symbol: Symbol) -> date:
        """
        Looks back `delay` valid trading days from `baseDate`.
        """
        candidate = baseDate
        for _ in range(delay):
            candidate = self.GetPreviousTradingDay(candidate, symbol)
        return candidate

    def GetPreviousTradingDay(self, baseDate: date, symbol: Symbol) -> date:
        """
        Steps backward day by day until we find a date the market was open.
        """
        candidate = baseDate - timedelta(days=1)
        while True:
            dt = datetime.combine(candidate, datetime.min.time())
            if self.Securities[symbol].Exchange.Hours.IsDateOpen(dt):
                return candidate
            candidate -= timedelta(days=1)

the predictions are ml predictions for the next day where highest should be long and lowest should be short

QuantConnect does not automatically force trades to occur on the next day. Instead, trading execution in backtesting depends on your algorithm's specific logic and the scheduled events you set up. Here are some key points on how to integrate trading signals and manage trades in QuantConnect:

1. Trade Execution in Backtesting:

   • The timing of trade executions relies on the resolution of the security data (e.g., minute, second). Your algorithm processes data and executes trades based on the rules you set, rather than being automatically executed on the next day.

2. Integrating External Signals:

   • To use external signals in your backtesting, your algorithm needs to incorporate these signals using on_data or scheduled events. There is no automatic trading if not explicitly programmed.

3. Scheduled Events:

   • You can set up scheduled events with the schedule.on() method provided by the QCAlgorithm class. This allows you to execute trades at specific times or intervals, which can be crucial if your strategy requires precise timing.

4. Rule-based Trading:

   • Utilize the QCAlgorithm class's on_data method and other event handlers to implement logic for executing trades based on your strategy's rules. This ensures that trades occur under the conditions you've defined.

5. Testing and Validation:

   • It is essential to validate your trade logic by running comprehensive backtests over historical data to ensure the expected behavior of trade execution.

By properly implementing these elements, you can effectively manage when and how trades occur in response to external trading signals in QuantConnect. If specific conditions or operations need trades to happen the next day, you must configure this within your algorithm's logic.

Sources: Scheduled Events - QuantConnect.com, Algorithm Engine - QuantConnect.com, Dual Thrust Trading Algorithm, US Equities - QuantConnect.com, Scheduled Events Algorithm

from AlgorithmImports import *
import pandas as pd
import numpy as np
from io import StringIO
from datetime import timedelta, datetime, date


class PredictionBasedStrategy(QCAlgorithm):
    def initialize(self):
        self.set_start_date(2024, 12, 2)  # Backtest start date
        self.set_cash(100000)  # Starting capital
        self.universe_settings.resolution = Resolution.DAILY

        # Add SPY to ensure we can schedule properly on market days
        self.spy_symbol = self.add_equity("SPY", Resolution.DAILY).Symbol

        # Attempt to load predictions from Object Store
        predictions_file_key = "test_predictions.csv"
        if not self.object_store.contains_key(predictions_file_key):
            self.error(f"{predictions_file_key} not found in Object Store.")
            return

        file_bytes_net = self.object_store.read_bytes(predictions_file_key)
        if file_bytes_net is None:
            self.error(f"Could not read {predictions_file_key} from Object Store.")
            return

        file_bytes = bytes(bytearray(file_bytes_net))
        csv_data = StringIO(file_bytes.decode("utf-8"))
        self.predictions_df = pd.read_csv(csv_data, parse_dates=True, index_col=0)

        # Build a dictionary: { date => {ticker => prediction} }
        self.predictions_by_date = {}
        for dt, row in self.predictions_df.iterrows():
            # Convert the DataFrame index (a Timestamp) to a raw date() for easy dictionary lookup
            date_key = dt.date()
            self.predictions_by_date[date_key] = row.to_dict()

        # Add each ticker as an Equity asset
        self.tickers = list(self.predictions_df.columns)
        for ticker in self.tickers:
            self.add_equity(ticker, Resolution.DAILY)

        # How many stocks to trade on each side
        self.max_stocks = 1

        # True = trade at next day’s CLOSE; False = next day’s OPEN
        # Both use only a 1-day delay in this example
        self.trade_at_next_day_close = False

        # If True, strategy goes only Long; if False, it does Long & Short
        self.long_only_mode = False

        # Because we want "higher is better," we do NOT invert predictions
        self.is_prediction_inverted = False

        # Schedule daily rebalancing time
        if self.trade_at_next_day_close:
            # Rebalance near the close
            self.schedule.on(
                self.date_rules.every_day(),
                self.time_rules.before_market_close(self.spy_symbol, 10),  # e.g., 10 min before close
                self.rebalance,
            )
            self.log("Trading Mode: Next Day CLOSE (T+1 close), 1-day lookback.")
        else:
            # Rebalance shortly after the open
            self.schedule.on(
                self.date_rules.every_day(),
                self.time_rules.after_market_open(self.spy_symbol, 5),  # e.g., 5 min after open
                self.rebalance,
            )
            self.log("Trading Mode: Next Day OPEN (T+1 open), 1-day lookback.")

        # End-of-day portfolio log
        self.schedule.on(
            self.date_rules.every_day(),
            self.time_rules.after_market_close(self.spy_symbol, 1),
            self.log_portfolio,
        )

        self.last_rebalance_date = None
        self.current_targets = None  # to track changes in positions

    def rebalance(self):
        """
        Rebalance once per day, reading predictions from 1 day ago.
        """
        today = self.time.date()
        if self.last_rebalance_date == today:
            return
        self.last_rebalance_date = today

        # The key difference: always 1-day offset
        # If you want day (T) predictions for day (T+1) trades,
        # we look at "yesterday" from the perspective of 'today'.
        days_back = 1

        # This method tries to find a valid trading day 'days_back' days before 'today',
        # skipping weekends/holidays
        prediction_day = self.get_delayed_trading_day(today, days_back, self.spy_symbol)

        if prediction_day not in self.predictions_by_date:
            self.log(
                f"{today}: No predictions available for {prediction_day}; skipping rebalance."
            )
            return

        # Pull out the predictions for 'prediction_day'
        preds = self.predictions_by_date[prediction_day]

        # If we are not inverting, we do nothing special; "higher is better" means top are longs
        effective_preds = {}
        for ticker, value in preds.items():
            effective_preds[ticker] = -value if self.is_prediction_inverted else value

        # Sort descending so highest predictions go to LONG
        sorted_items = sorted(effective_preds.items(), key=lambda x: x[1], reverse=True)
        top_n = sorted_items[: self.max_stocks]  # highest = long
        bottom_n = sorted_items[-self.max_stocks :]  # lowest = short

        # Convert to just ticker symbols
        long_targets = [x[0] for x in top_n]
        short_targets = [x[0] for x in bottom_n]

        if self.long_only_mode:
            self.log(f"{today}: Long‐only mode. Ignoring short targets.")
            short_targets = []

        new_targets = (sorted(long_targets), sorted(short_targets))
        if self.current_targets == new_targets:
            self.log(f"{today}: No change in targets. Holding positions.")
            return

        # Update stored targets
        self.current_targets = new_targets
        time_str = "CLOSE" if self.trade_at_next_day_close else "OPEN"
        self.log(
            f"Rebalance on {today} at {time_str}, using predictions from {prediction_day}. "
            f"New targets: LONG {long_targets}, SHORT {short_targets}"
        )

        # First, liquidate anything not in the new target set
        target_symbols = set(long_targets + short_targets)
        for s in list(self.portfolio):
            if self.portfolio[s].invested and s.value not in target_symbols:
                self.liquidate(s)

        # Then open new positions. We'll do an equal weighting among all chosen instruments.
        long_weight = 1.0 / len(long_targets) if long_targets else 0
        short_weight = (
            0
            if self.long_only_mode
            else (-1.0 / len(short_targets) if short_targets else 0)
        )

        for ticker in long_targets:
            symbol = self.symbol(ticker)
            self.set_holdings(symbol, long_weight)

        for ticker in short_targets:
            symbol = self.symbol(ticker)
            self.set_holdings(symbol, short_weight)

    def on_order_event(self, order_event):
        if order_event.status != OrderStatus.FILLED:
            return
        symbol = order_event.Symbol
        qty = order_event.fill_quantity
        direction = "LONG" if qty > 0 else "SHORT" if qty < 0 else "FLAT"
        self.log(
            f"Order filled: {symbol.value} {direction}, Qty: {qty}, Price: {order_event.fill_price}"
        )

    def log_portfolio(self):
        positions = []
        for s in self.portfolio:
            if self.portfolio[s].invested:
                pos = self.portfolio[s]
                pos_type = (
                    "LONG"
                    if pos.quantity > 0
                    else "SHORT"
                    if pos.quantity < 0
                    else "FLAT"
                )
                positions.append(f"{s.value} ({pos_type}): {pos.quantity}")
        if positions:
            self.log("Portfolio at EOD: " + ", ".join(positions))
        else:
            self.log("No positions held at end of day.")

    def on_data(self, data):
        pass

    def get_delayed_trading_day(self, base_date: date, delay: int, symbol: Symbol) -> date:
        """
        Looks back `delay` valid trading days from `base_date`.
        """
        candidate = base_date
        for _ in range(delay):
            candidate = self.get_previous_trading_day(candidate, symbol)
        return candidate

    def get_previous_trading_day(self, base_date: date, symbol: Symbol) -> date:
        """
        Steps backward day by day until we find a date the market was open.
        """
        candidate = base_date - timedelta(days=1)
        while True:
            dt = datetime.combine(candidate, datetime.min.time())
            if self.securities[symbol].exchange.hours.is_date_open(dt):
                return candidate
            candidate -= timedelta(days=1)

Sources: Bybit Crypto Futures Regression Algorithm, Overnight Anomaly, Momentum Effect In Commodities Futures, Seasonality Effect Based On Same Calendar Month Returns, CAPM Alpha Ranking Strategy On Dow 30 Companies