How to Perform Heavy Calculations Before Market Open Using Scheduled Events in QuantConnect

To run a heavy calculation once a day before trading starts in QuantConnect, you should use the Scheduled Events API. This allows you to trigger a specific method at a precise time (e.g., 30 minutes before market open) without blocking the data feed during trading hours.

If the calculation is extremely intensive (taking more than 10 minutes, such as training a Machine Learning model), you should utilize the self.train() method within your scheduled event to prevent the algorithm from timing out.

Solution: Scheduled Event Pre-Market

The following strategy demonstrates how to:

1. Schedule a function (perform_heavy_computation) to run every day at 9:00 AM (30 minutes before US Equity market open).
2. Perform a "heavy" history request and statistical calculation.
3. Store the results to be used later in on_data.

# region imports
from AlgorithmImports import *
import numpy as np
# endregion

class PreMarketCalculationAlgorithm(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2023, 1, 1)
        self.set_end_date(2023, 6, 1)
        self.set_cash(100000)

        # Add a reference symbol (SPY) to determine market days
        self.spy = self.add_equity("SPY", Resolution.MINUTE).symbol
        
        # Add a list of symbols to analyze
        self.symbols = [
            self.add_equity(ticker, Resolution.MINUTE).symbol 
            for ticker in ["AAPL", "MSFT", "GOOG", "AMZN", "TSLA"]
        ]

        # Variable to store our pre-calculated targets
        self.target_symbol = None

        # ------------------------------------------------------------
        # SCHEDULED EVENT
        # Run 'perform_heavy_computation' every trading day at 9:00 AM
        # (30 minutes before the standard US market open at 9:30 AM)
        # ------------------------------------------------------------
        self.schedule.on(
            self.date_rules.every_day(self.spy),
            self.time_rules.at(9, 0),
            self.perform_heavy_computation
        )

    def perform_heavy_computation(self):
        """
        This runs before the market opens.
        We perform heavy history calls and math here.
        """
        self.debug(f"Starting heavy calculation at {self.time}")

        # 1. Request heavy amount of historical data
        # (e.g., last 30 days of minute data for all symbols)
        history = self.history(self.symbols, 30 * 390, Resolution.MINUTE)
        
        if history.empty:
            return

        highest_volatility = -1
        selected_symbol = None

        # 2. Perform heavy math (e.g., calculating volatility on large data)
        # We iterate through symbols to find the one with highest std dev
        for symbol in self.symbols:
            if symbol not in history.index.levels[0]:
                continue
                
            # Extract close prices for this symbol
            closes = history.loc[symbol]['close']
            
            # Calculate volatility (Standard Deviation)
            volatility = np.std(closes)
            
            if volatility > highest_volatility:
                highest_volatility = volatility
                selected_symbol = symbol

        # 3. Store the result for the trading day
        self.target_symbol = selected_symbol
        
        if self.target_symbol:
            self.log(f"Pre-market calculation complete. Target: {self.target_symbol} (Vol: {highest_volatility})")

    def on_data(self, data: Slice):
        """
        Standard trading logic. This runs fast because the heavy lifting
        was done in perform_heavy_computation.
        """
        # Ensure we have a target and data for it
        if self.target_symbol is None or not data.contains_key(self.target_symbol):
            return

        # Simple logic: Buy the selected volatile stock if we aren't invested
        if not self.portfolio.invested:
            self.set_holdings(self.target_symbol, 1.0)

    def on_end_of_day(self, symbol):
        # Liquidate at end of day to reset for the next day's calculation
        self.liquidate()
        self.target_symbol = None

Key Concepts

1. self.schedule.on(...)

This is the primary mechanism for timing.

• DateRules: self.date_rules.every_day(symbol) ensures it only runs on days the market is open.
• TimeRules: self.time_rules.at(9, 0) sets the specific wall-clock time. Since US Equities open at 9:30 AM, setting this to 9:00 AM ensures it finishes before trading begins.

2. Handling Timeouts (self.train)

If your calculation involves training a Neural Network or processing years of tick data and takes longer than 10 minutes, the standard method above might time out. In that case, wrap the logic in self.train:

def perform_heavy_computation(self):
    # self.train runs in a separate thread (in live trading) 
    # and prevents timeout warnings.
    self.train(self.my_training_method)

def my_training_method(self):
    # Your very heavy ML training logic here
    pass

Q&A

Q: Can I use TimeRules.AfterMarketOpen with negative minutes?
A: While technically possible in some contexts, it is often safer and clearer to use TimeRules.At(hour, minute) for pre-market activities to avoid confusion regarding Extended Market Hours or specific brokerage opening times.

Q: Does history() request data from the current day if run at 9:00 AM?
A: No. If you run a request at 9:00 AM, it will return data up until the previous market close (or the most recent available tick before 9:00 AM). It will not include future data.

Q: What happens if the calculation takes longer than the time until market open?
A: In backtesting, the engine pauses time until your calculation finishes, so you won't "miss" the market open. In Live Trading, the engine does not pause. If your calculation starts at 9:00 AM and takes 40 minutes, on_data will start firing at 9:30 AM, but your target_symbol variable won't be updated until 9:40 AM.