Custom Time Rule Framework for Constraining Schedules to Market Hours for Intraday Algos

The Python code you've shared seems to be well-structured and doesn't appear to have any obvious bugs. It creates a custom time rule for scheduling events during market hours. The custom_time_rule function generates a FuncTimeRule object that creates event times for each date at intervals specified by the interval parameter. These event times are generated by the get_custom_times function, which creates a list of times at the specified interval between the market open and close times.

The scheduled function my_scheduled_function is set to log the execution time whenever it's called.

Please note that this analysis is based on the provided code and assumes that the QuantConnect API functions such as self.schedule.on, self.date_rules.every_day, FuncTimeRule, etc., are being used correctly according to the QuantConnect API documentation.

Here is the code you provided for reference:

class CustomTimeRuleAlgorithm(QCAlgorithm):
    def initialize(self):
        self.set_start_date(2021, 1, 4)
        self.set_end_date(2021, 1, 5)
        self.set_cash(100000)
        self.add_equity("SPY", Resolution.MINUTE)
        self.use_extended_hrs = False
        self.schedule.on(
            self.date_rules.every_day("SPY"),
            self.custom_time_rule("SPY", timedelta(minutes=5)),
            self.my_scheduled_function
        )
    def custom_time_rule(self, symbol, interval):
        return FuncTimeRule(
            name="EveryIntervalDuringMarketHours",
            createUtcEventTimesFunction=lambda dates: [
                dt for date in dates
                for dt in self.get_custom_times(symbol, date, interval)
            ]
        )
    def get_custom_times(self, symbol, date, interval):
        exchange = self.securities[symbol].exchange
        market_open = exchange.hours.get_next_market_open(date, self.use_extended_hrs)
        market_close = exchange.hours.get_next_market_close(market_open, self.use_extended_hrs)
        market_open = market_open.replace(tzinfo=timezone(timedelta(hours=-5))).astimezone(timezone.utc)
        market_close = market_close.replace(tzinfo=timezone(timedelta(hours=-5))).astimezone(timezone.utc)
        times = []
        current_time = market_open
        while current_time < market_close:
            times.append(current_time)
            current_time += interval
        return times
    def my_scheduled_function(self):
        self.log(f"Executed at: {self.time}")
        self.debug(f"Executed at: {self.time}")

Please test this code in your environment to ensure it works as expected. If you encounter any issues, feel free to ask for further assistance.