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,10 +1622,6 @@ 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)
-
     cdef CVec _advance_time(self, uint64_t ts_now):
         # Advance clocks and process all events before ts_now in timestamp order.
         #
@@ -1697,6 +1703,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

tests/unit_tests/backtest/test_engine_timers.py

@@ -0,0 +1,241 @@
+# -------------------------------------------------------------------------------------------------
+#  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
+#  https://nautechsystems.io
+#
+#  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
+#  You may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+# -------------------------------------------------------------------------------------------------
+
+import pandas as pd
+
+from nautilus_trader.backtest.engine import BacktestEngine
+from nautilus_trader.backtest.engine import BacktestEngineConfig
+from nautilus_trader.common.actor import Actor
+from nautilus_trader.config import LoggingConfig
+from nautilus_trader.model.currencies import USD
+from nautilus_trader.model.enums import AccountType
+from nautilus_trader.model.enums import OmsType
+from nautilus_trader.model.identifiers import Venue
+from nautilus_trader.model.objects import Money
+from nautilus_trader.test_kit.providers import TestInstrumentProvider
+from nautilus_trader.test_kit.stubs.data import TestDataStubs
+
+
+class TimerActor(Actor):
+    def __init__(self, instrument_id):
+        super().__init__()
+        self.instrument_id = instrument_id
+        self.timer_fired_count = 0
+        self.last_timer_time = 0
+        self.received_data = []
+
+    def on_start(self):
+        self.subscribe_quote_ticks(self.instrument_id)
+
+    def timer_callback(self, event):
+        self.timer_fired_count += 1
+        self.last_timer_time = self.clock.timestamp_ns()
+
+    def on_quote_tick(self, tick):
+        self.received_data.append(tick)
+
+
+class TestBacktestEngineTimers:
+    def setup_method(self):
+        self.engine = BacktestEngine(
+            BacktestEngineConfig(logging=LoggingConfig(bypass_logging=True)),
+        )
+        self.engine.add_venue(
+            venue=Venue("SIM"),
+            oms_type=OmsType.HEDGING,
+            account_type=AccountType.MARGIN,
+            base_currency=USD,
+            starting_balances=[Money(1000000, USD)],
+        )
+        self.instrument = TestInstrumentProvider.default_fx_ccy("EUR/USD")
+        self.engine.add_instrument(self.instrument)
+
+    def test_timer_execution_no_data(self):
+        # Test that timers fire correctly even when no data is provided
+        actor = TimerActor(self.instrument.id)
+        self.engine.add_actor(actor)
+
+        start_time = pd.Timestamp("2024-01-01", tz="UTC")
+        end_time = start_time + pd.Timedelta(seconds=10)
+        timer_time = start_time + pd.Timedelta(seconds=5)
+
+        actor.clock.set_time_alert("test_timer", timer_time, actor.timer_callback)
+
+        self.engine.run(start=start_time, end=end_time)
+
+        assert actor.timer_fired_count == 1
+        assert actor.last_timer_time == timer_time.value
+
+    def test_on_the_fly_data_loading_from_timer(self):
+        # Test that data added via add_data_iterator in a timer callback is processed correctly.
+        # This mirrors the real-world pattern where subscriptions use generators for on-the-fly loading.
+        actor = TimerActor(self.instrument.id)
+        self.engine.add_actor(actor)
+
+        start_time = pd.Timestamp("2024-01-01", tz="UTC")
+        end_time = start_time + pd.Timedelta(seconds=10)
+        timer_time = start_time + pd.Timedelta(seconds=5)
+        data_time = start_time + pd.Timedelta(seconds=6)
+
+        def data_generator():
+            # Generator that yields data when called from timer callback
+            # This simulates how _handle_subscribe uses generators for on-the-fly loading
+            yield [TestDataStubs.quote_tick(self.instrument, ts_init=data_time.value)]
+
+        def timer_callback_with_data(event):
+            actor.timer_callback(event)
+            # Load data on the fly using add_data_iterator (like _handle_subscribe does)
+            self.engine.add_data_iterator("on_the_fly_data", data_generator())
+
+        actor.clock.set_time_alert("test_timer", timer_time, timer_callback_with_data)
+
+        self.engine.run(start=start_time, end=end_time)
+
+        assert actor.timer_fired_count == 1
+        assert len(actor.received_data) == 1
+        assert actor.received_data[0].ts_init == data_time.value
+
+    def test_multiple_timers_same_timestamp(self):
+        # Test that multiple timers at the exact same timestamp all fire
+        actor = TimerActor(self.instrument.id)
+        self.engine.add_actor(actor)
+
+        start_time = pd.Timestamp("2024-01-01", tz="UTC")
+        end_time = start_time + pd.Timedelta(seconds=10)
+        timer_time = start_time + pd.Timedelta(seconds=5)
+
+        actor.clock.set_time_alert("timer1", timer_time, actor.timer_callback)
+        actor.clock.set_time_alert("timer2", timer_time, actor.timer_callback)
+        actor.clock.set_time_alert("timer3", timer_time, actor.timer_callback)
+
+        self.engine.run(start=start_time, end=end_time)
+
+        assert actor.timer_fired_count == 3
+        assert actor.last_timer_time == timer_time.value
+
+    def test_chained_timers(self):
+        # Test that a timer scheduling another timer works
+        actor = TimerActor(self.instrument.id)
+        self.engine.add_actor(actor)
+
+        start_time = pd.Timestamp("2024-01-01", tz="UTC")
+        end_time = start_time + pd.Timedelta(seconds=10)
+        timer1_time = start_time + pd.Timedelta(seconds=5)
+        timer2_time = start_time + pd.Timedelta(seconds=6)
+
+        def timer1_callback(event):
+            actor.timer_callback(event)
+            actor.clock.set_time_alert("timer2", timer2_time, actor.timer_callback)
+
+        actor.clock.set_time_alert("timer1", timer1_time, timer1_callback)
+
+        self.engine.run(start=start_time, end=end_time)
+
+        assert actor.timer_fired_count == 2
+        assert actor.last_timer_time == timer2_time.value
+
+    def test_chained_timers_same_timestamp(self):
+        # Test that a timer scheduling another timer for the SAME timestamp works
+        actor = TimerActor(self.instrument.id)
+        self.engine.add_actor(actor)
+
+        start_time = pd.Timestamp("2024-01-01", tz="UTC")
+        end_time = start_time + pd.Timedelta(seconds=10)
+        timer_time = start_time + pd.Timedelta(seconds=5)
+
+        def timer1_callback(event):
+            actor.timer_callback(event)
+            # Schedule another one for the same time
+            actor.clock.set_time_alert("timer2", timer_time, actor.timer_callback)
+
+        actor.clock.set_time_alert("timer1", timer_time, timer1_callback)
+
+        self.engine.run(start=start_time, end=end_time)
+
+        assert actor.timer_fired_count == 2
+        assert actor.last_timer_time == timer_time.value
+
+    def test_timers_alphabetical_order_same_timestamp(self):
+        # Test that multiple timers at the same timestamp fire regardless of name order
+        actor = TimerActor(self.instrument.id)
+        self.engine.add_actor(actor)
+
+        start_time = pd.Timestamp("2024-01-01", tz="UTC")
+        end_time = start_time + pd.Timedelta(seconds=10)
+        timer_time = start_time + pd.Timedelta(seconds=5)
+
+        fired_order = []
+
+        def callback_z(event):
+            fired_order.append("z")
+
+        def callback_a(event):
+            fired_order.append("a")
+
+        def callback_m(event):
+            fired_order.append("m")
+
+        actor.clock.set_time_alert("z_timer", timer_time, callback_z)
+        actor.clock.set_time_alert("a_timer", timer_time, callback_a)
+        actor.clock.set_time_alert("m_timer", timer_time, callback_m)
+
+        self.engine.run(start=start_time, end=end_time)
+
+        # They should all fire. In current implementation, order might be alphabetical
+        # due to Rust's BTreeMap/ordered keys, but the key thing is they ALL fire.
+        assert len(fired_order) == 3
+        assert fired_order == ["a", "m", "z"]
+
+    def test_timers_and_data_interwoven(self):
+        # Test that timers and data are processed in the correct interleaved order
+        start_time = pd.Timestamp("2024-01-01", tz="UTC")
+        end_time = start_time + pd.Timedelta(seconds=10)
+
+        # Timeline:
+        # T+2: Data1
+        # T+4: Timer1
+        # T+6: Data2
+        # T+8: Timer2
+
+        events = []
+
+        def timer_callback(event):
+            events.append(f"timer_{event.ts_event}")
+
+        class InterwovenActor(TimerActor):
+            def on_quote_tick(self, tick):
+                events.append(f"data_{tick.ts_init}")
+
+        actor = InterwovenActor(self.instrument.id)
+        self.engine.add_actor(actor)
+
+        t2 = (start_time + pd.Timedelta(seconds=2)).value
+        t4 = (start_time + pd.Timedelta(seconds=4)).value
+        t6 = (start_time + pd.Timedelta(seconds=6)).value
+        t8 = (start_time + pd.Timedelta(seconds=8)).value
+
+        self.engine.add_data(
+            [
+                TestDataStubs.quote_tick(self.instrument, ts_init=t2),
+                TestDataStubs.quote_tick(self.instrument, ts_init=t6),
+            ],
+        )
+        actor.clock.set_time_alert("timer1", pd.Timestamp(t4, unit="ns", tz="UTC"), timer_callback)
+        actor.clock.set_time_alert("timer2", pd.Timestamp(t8, unit="ns", tz="UTC"), timer_callback)
+
+        self.engine.run(start=start_time, end=end_time)
+
+        expected = [f"data_{t2}", f"timer_{t4}", f"data_{t6}", f"timer_{t8}"]
+        assert events == expected