Refine handling of timers in backtest engine

Author: faysou

examples/backtest/notebooks/databento_test_order_book_deltas.py

@@ -21,6 +21,8 @@
 
 # %%
 
+import pandas as pd
+
 from nautilus_trader.adapters.databento.data_utils import databento_data
 from nautilus_trader.adapters.databento.data_utils import load_catalog
 from nautilus_trader.backtest.node import BacktestNode
@@ -91,6 +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
+        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(
+            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):
@@ -100,12 +115,12 @@ def on_order_book_deltas(self, deltas):
 
         self._deltas_count += 1
 
-    def user_log(self, msg, color=LogColor.GREEN):
-        self.log.warning(f"{msg}", color=color)
-
     def on_stop(self):
         order_book = self.cache.order_book(self.config.symbol_id)
-        self.user_log(f"{order_book}")
+        self.user_log(f"Final OrderBook: {order_book}")
+
+    def user_log(self, msg, color=LogColor.GREEN):
+        self.log.warning(f"{msg}", color=color)
 
 
 # %% [markdown]

nautilus_trader/backtest/engine.pxd

@@ -124,6 +124,7 @@ cdef class BacktestEngine:
     cdef list _response_data
 
     cdef CVec _advance_time(self, uint64_t ts_now)
+    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,12 +1531,22 @@ 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 data is not None:
+            while True:
+                if data is None:
+                    if self._process_next_timer():
+                        break
+
+                    data = self._data_iterator.next()
+                    continue
+
                 if data.ts_init > end_ns:
                     # End of backtest
                     break
@@ -1593,6 +1602,7 @@ cdef class BacktestEngine:
                     )
                     if raw_handlers.ptr != NULL:
                         vec_time_event_handlers_drop(raw_handlers)
+
                     raw_handlers_count = 0
 
                 self._iteration += 1
@@ -1612,9 +1622,15 @@ cdef class BacktestEngine:
         for exchange in self._venues.values():
             exchange.process(self._kernel.clock.timestamp_ns())
 
-        # Flush remaining events at the last data timestamp
-        if self._last_ns > 0:
-            self._flush_accumulator_events(self._last_ns)
+        # Flush remaining events up to end_ns to ensure all timers are processed
+        # This handles the case where the loop ended because data.ts_init > end_ns
+        # (when data is None, _process_remaining_timers already flushed to end_ns, but
+        # flushing again is safe since _flush_accumulator_events only processes events <= ts_now)
+        if end_ns > 0:
+            if self._last_ns < end_ns:
+                self._last_ns = end_ns
+
+            self._flush_accumulator_events(end_ns)
 
     cdef CVec _advance_time(self, uint64_t ts_now):
         # Advance clocks and process all events before ts_now in timestamp order.
@@ -1697,6 +1713,47 @@ cdef class BacktestEngine:
         empty_vec.cap = 0
         return empty_vec
 
+    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 uint64_t min_next_time = 0
+        cdef uint64_t next_timer_time
+        cdef str name
+
+        # 1. Process all timers up to current time
+        self._flush_accumulator_events(self._last_ns)
+
+        # 2. Find the absolute next timer time across all clocks
+        for clock in clocks:
+            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
+
+        # 3. Process the next closest possible timer(s)
+        self._last_ns = min_next_time
+        self._flush_accumulator_events(min_next_time)
+
+        # 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)
         cdef TestClock clock