Refine handling of timers in backtest engine

Author: faysou

examples/backtest/notebooks/databento_test_order_book_deltas.py

@@ -93,18 +93,19 @@ def __init__(self, config: TestOrderBookDeltasConfig):
 
     def on_start(self):
         self.request_instrument(self.config.symbol_id)
-        # Set time alert for 1 second after start
 
-        self.on_subscribe_timer(1)
-        # alert_time = self.clock.utc_now() + pd.Timedelta(seconds=1)
-        # self.clock.set_time_alert(
-        #     "subscribe_alert",
-        #     alert_time,
-        #     self.on_subscribe_timer
-        # )
+        # Set time alert for 1 second after start
+        alert_time = self.clock.utc_now() + pd.Timedelta(seconds=1)
+        self.clock.set_time_alert(
+            "subscribe_alert",
+            alert_time,
+            self.on_subscribe_timer,
+        )
 
     def on_subscribe_timer(self, event):
-        self.user_log("Subscribing to order book deltas after 1 second delay")
+        self.user_log(
+            f"Subscribing to order book deltas after 1 second delay, clock={self.clock.utc_now()}",
+        )
         self.subscribe_order_book_deltas(self.config.symbol_id)
 
     def on_order_book_deltas(self, deltas):
@@ -194,7 +195,7 @@ def user_log(self, msg, color=LogColor.GREEN):
 configs = [
     BacktestRunConfig(
         engine=engine_config,
-        data=[],
+        data=[],  # data,
         venues=venues,
         chunk_size=None,  # use None when loading custom data
         start=start_time,

nautilus_trader/backtest/engine.pxd

@@ -124,7 +124,7 @@ cdef class BacktestEngine:
     cdef list _response_data
 
     cdef CVec _advance_time(self, uint64_t ts_now)
-    cdef bint _process_remaining_timers(self, uint64_t end_ns)
+    cdef bint _process_next_timer(self)
     cdef void _flush_accumulator_events(self, uint64_t ts_now)
     cdef void _process_raw_time_event_handlers(
         self,

nautilus_trader/backtest/engine.pyx

@@ -823,7 +823,6 @@ cdef class BacktestEngine:
 
         if validate:
             first = data[0]
-
             if hasattr(first, "instrument_id"):
                 Condition.is_true(
                     first.instrument_id in self._kernel.cache.instrument_ids(),
@@ -1532,14 +1531,17 @@ cdef class BacktestEngine:
                     break
 
         # -- MAIN BACKTEST LOOP -----------------------------------------------#
-        self._last_ns = 0
+        self._last_ns = start_ns
         cdef uint64_t raw_handlers_count = 0
         cdef Data data = self._data_iterator.next()
         cdef CVec raw_handlers
+        raw_handlers.ptr = NULL
+        raw_handlers.len = 0
+        raw_handlers.cap = 0
         try:
             while True:
                 if data is None:
-                    if self._process_remaining_timers(end_ns):
+                    if self._process_next_timer():
                         break
 
                     data = self._data_iterator.next()
@@ -1711,55 +1713,46 @@ cdef class BacktestEngine:
         empty_vec.cap = 0
         return empty_vec
 
-    cdef bint _process_remaining_timers(self, uint64_t end_ns):
-        # Process remaining timers when data is exhausted.
-        # Returns True if there are no more timers (should break), False otherwise (should continue).
+    cdef bint _process_next_timer(self):
+        # Process the next chronological timer when data is exhausted.
+        #
+        # This method is used when the data stream is empty, but timers (alerts) might
+        # still be active. Instead of jumping to the end of the backtest, it finds the
+        # absolute next timer time across all component clocks and advances to it.
+        #
+        # This allows for scenarios where a timer callback might load new data on-the-fly
+        # (e.g. via a subscription), which should then be processed in proper sequence.
+        #
+        # Returns True if there are no more timers within the backtest range (should break),
+        # False otherwise (should continue the backtest loop to check for new data).
 
         cdef list[TestClock] clocks = get_component_clocks(self._instance_id)
         cdef TestClock clock
-        cdef bint has_timers
+        cdef uint64_t min_next_time = 0
         cdef uint64_t next_timer_time
+        cdef str name
 
-        # If we've already advanced to end_ns or beyond, and there are still timers,
-        # they must be beyond end_ns, so we should break to avoid infinite loop
-        if self._last_ns >= end_ns:
-            return True
+        # 1. Process all timers up to current time
+        self._flush_accumulator_events(self._last_ns)
 
-        # Advance accumulator to capture timers from clocks
+        # 2. Find the absolute next timer time across all clocks
         for clock in clocks:
-            time_event_accumulator_advance_clock(
-                &self._accumulator,
-                &clock._mem,
-                end_ns,
-                False,
-            )
-
-        # Check if there are any timers that can be processed (next time <= end_ns)
-        next_timer_time = time_event_accumulator_peek_next_time(&self._accumulator)
-        if next_timer_time == 0:
-            # No timers in accumulator, break
+            for name in clock.timer_names:
+                next_timer_time = clock.next_time_ns(name)
+                if next_timer_time > self._last_ns:
+                    if min_next_time == 0 or next_timer_time < min_next_time:
+                        min_next_time = next_timer_time
+
+        if min_next_time == 0 or min_next_time > self._end_ns:
+            # No more timers in the backtest range
             return True
 
-        if next_timer_time > end_ns:
-            # All timers are beyond end_ns, break to avoid infinite loop
-            return True
-
-        # Advance time and flush events up to end_ns to process all remaining timers
-        self._last_ns = end_ns
-        self._flush_accumulator_events(end_ns)
-
-        # Set clock time after flushing
-        for clock in clocks:
-            clock.set_time(end_ns)
-
-        set_logging_clock_static_time(end_ns)
-
-        if LOGGING_PYO3:
-            nautilus_pyo3.logging_clock_set_static_time(end_ns)
+        # 3. Process the next closest possible timer(s)
+        self._last_ns = min_next_time
+        self._flush_accumulator_events(min_next_time)
 
-        # After flushing up to end_ns, any remaining timers must be beyond end_ns
-        # Always break to avoid infinite loop (we've processed all timers up to end_ns)
-        return True
+        # Return False to indicate we should continue checking for data
+        return False
 
     cdef void _flush_accumulator_events(self, uint64_t ts_now):
         cdef list[TestClock] clocks = get_component_clocks(self._instance_id)