Refine handling of skip_first_non_full_bar in TimeBarAggregator (#3395)

Author: faysou

nautilus_trader/data/aggregation.pxd

@@ -149,6 +149,7 @@ cdef class TimeBarAggregator(BarAggregator):
     cdef readonly uint64_t interval_ns
     cdef readonly uint64_t next_close_ns
     cdef readonly uint64_t stored_open_ns
+    cdef readonly uint64_t first_close_ns
 
     cdef str _timer_name
     cdef bint _is_left_open

nautilus_trader/data/aggregation.pyx

@@ -189,6 +189,7 @@ cdef class BarBuilder:
         self._open = None
         self._high = None
         self._low = None
+        self._close = None
 
         self.volume = Quantity.zero_c(precision=self.size_precision)
         self.count = 0
@@ -1406,6 +1407,7 @@ cdef class TimeBarAggregator(BarAggregator):
         self.interval_ns = self._get_interval_ns()
         self.stored_open_ns = 0
         self.next_close_ns = 0
+        self.first_close_ns = 0
         self.historical_mode = False
         self._historical_events = []
 
@@ -1436,46 +1438,55 @@ cdef class TimeBarAggregator(BarAggregator):
         # Closing a partial bar at the transition from historical to backtest data
         cdef bint fire_immediately = (start_time == now)
 
-        self._skip_first_non_full_bar = self._skip_first_non_full_bar and now > start_time
-
-        if self.bar_type.spec.aggregation not in (BarAggregation.MONTH, BarAggregation.YEAR):
-            self._clock.set_timer(
-                name=self._timer_name,
-                interval=self.interval,
-                start_time=start_time,
-                stop_time=None,
-                callback=self._build_bar,
-                allow_past=True,
-                fire_immediately=fire_immediately,
-            )
+        # Calculate the next close time based on aggregation type
+        cdef datetime close_time
+        if fire_immediately:
+            close_time = start_time
+        elif self.bar_type.spec.aggregation == BarAggregation.MONTH:
+            close_time = start_time + pd.DateOffset(months=self.bar_type.spec.step)
+        elif self.bar_type.spec.aggregation == BarAggregation.YEAR:
+            close_time = start_time + pd.DateOffset(years=self.bar_type.spec.step)
+        else:
+            close_time = start_time + self.interval
 
-            if fire_immediately:
-                self.next_close_ns = dt_to_unix_nanos(start_time)
-            else:
-                self.next_close_ns = dt_to_unix_nanos(start_time + self.interval)
+        self.next_close_ns = dt_to_unix_nanos(close_time)
 
-            self.stored_open_ns = self.next_close_ns - self.interval_ns
+        # The stored open time needs to be defined as a subtraction with respect to the first closing time
+        if self.bar_type.spec.aggregation == BarAggregation.MONTH:
+            self.stored_open_ns = dt_to_unix_nanos(close_time - pd.DateOffset(months=self.bar_type.spec.step))
+        elif self.bar_type.spec.aggregation == BarAggregation.YEAR:
+            self.stored_open_ns = dt_to_unix_nanos(close_time - pd.DateOffset(years=self.bar_type.spec.step))
         else:
-            # The monthly/yearly alert time is defined iteratively at each alert time as there is no regular interval
-            if self.bar_type.spec.aggregation == BarAggregation.MONTH:
-                alert_time = start_time + (pd.DateOffset(months=self.bar_type.spec.step) if not fire_immediately else pd.Timedelta(0))
-            elif self.bar_type.spec.aggregation == BarAggregation.YEAR:
-                alert_time = start_time + (pd.DateOffset(years=self.bar_type.spec.step) if not fire_immediately else pd.Timedelta(0))
-            else:
-                alert_time = start_time
+            self.stored_open_ns = self.next_close_ns - self.interval_ns
+
+        if self._skip_first_non_full_bar:
+            self.first_close_ns = self.next_close_ns
 
+        if self.bar_type.spec.aggregation in (BarAggregation.MONTH, BarAggregation.YEAR):
+            # The monthly/yearly alert time is defined iteratively at each alert time as there is no regular interval
             self._clock.set_time_alert(
                 name=self._timer_name,
-                alert_time=alert_time,
+                alert_time=close_time,
                 callback=self._build_bar,
                 override=True,
                 allow_past=True,
             )
-            self.next_close_ns = alert_time.value
-            self.stored_open_ns = start_time.value
+        else:
+            self._clock.set_timer(
+                name=self._timer_name,
+                interval=self.interval,
+                start_time=start_time,
+                stop_time=None,
+                callback=self._build_bar,
+                allow_past=True,
+                fire_immediately=fire_immediately,
+            )
 
-        self._log.debug(f"Started timer {self._timer_name}, {start_time=}, {self.historical_mode=}, "
-                        f"{fire_immediately=}, {start_time=}, {now=}, {self._bar_build_delay=}")
+        self._log.debug(f"[start_timer] fire_immediately={fire_immediately}, "
+                        f"_skip_first_non_full_bar={self._skip_first_non_full_bar}, "
+                        f"now={now}, start_time={start_time}, "
+                        f"first_close_ns={unix_nanos_to_dt(self.first_close_ns)}, "
+                        f"next_close_ns={unix_nanos_to_dt(self.next_close_ns)}")
 
     cpdef void stop_timer(self):
         cdef str timer_name = str(self.bar_type)
@@ -1631,10 +1642,11 @@ cdef class TimeBarAggregator(BarAggregator):
             self.next_close_ns = self._clock.next_time_ns(self._timer_name)
 
     cdef void _build_and_send(self, uint64_t ts_event, uint64_t ts_init):
-        if self._skip_first_non_full_bar:
+        if self._skip_first_non_full_bar and ts_init <= self.first_close_ns:
             self._builder.reset()
-            self._skip_first_non_full_bar = False
         else:
+            # We set set _skip_first_non_full_bar to False for the transition from historical to live data
+            self._skip_first_non_full_bar = False
             BarAggregator._build_and_send(self, ts_event, ts_init)
 
 

nautilus_trader/data/engine.pyx

@@ -2689,14 +2689,20 @@ cdef class DataEngine(Component):
 
         if bar_type.spec.is_time_aggregated():
             time_bars_origin_offset = self._time_bars_origin_offset.get(bar_type.spec.aggregation) or params.get("time_bars_origin_offset")
+
+            if "skip_first_non_full_bar" in params:
+                time_bars_skip_first_non_full_bar = params.get("skip_first_non_full_bar", False)
+            else:
+                time_bars_skip_first_non_full_bar = self._time_bars_skip_first_non_full_bar
+
             aggregator = TimeBarAggregator(
                 instrument=instrument,
                 bar_type=aggregated_bar_type,
                 handler=self.process,
                 clock=self._clock,
                 interval_type=self._time_bars_interval_type,
                 timestamp_on_close=self._time_bars_timestamp_on_close,
-                skip_first_non_full_bar=self._time_bars_skip_first_non_full_bar,
+                skip_first_non_full_bar=time_bars_skip_first_non_full_bar,
                 build_with_no_updates=self._time_bars_build_with_no_updates,
                 time_bars_origin_offset=time_bars_origin_offset,
                 bar_build_delay=self._time_bars_build_delay,

tests/unit_tests/data/test_aggregation.py

@@ -3524,7 +3524,7 @@ def test_skip_first_non_full_bar_when_starting_on_bar_boundary(self):
         bar_spec = BarSpecification(2, BarAggregation.SECOND, PriceType.LAST)
         bar_type = BarType(instrument_id, bar_spec, AggregationSource.INTERNAL)
 
-        # Start exactly at a 2-second boundary (2024-12-01 00:00:00)
+        # Act - process ticks and advance time to trigger bars
         start_time_ns = dt_to_unix_nanos(pd.Timestamp("2024-12-01 00:00:00", tz="UTC"))
         clock.set_time(start_time_ns)
 
@@ -3538,6 +3538,10 @@ def test_skip_first_non_full_bar_when_starting_on_bar_boundary(self):
         )
         aggregator.start_timer()
 
+        events = clock.advance_time(start_time_ns)
+        for event in events:
+            event.handle()
+
         # Create trade ticks at 0.5 second intervals
         tick1 = TradeTick(
             instrument_id=instrument_id,
@@ -3549,6 +3553,9 @@ def test_skip_first_non_full_bar_when_starting_on_bar_boundary(self):
             ts_init=start_time_ns + 500_000_000,
         )
 
+        aggregator.handle_trade_tick(tick1)
+        clock.set_time(tick1.ts_event)
+
         tick2 = TradeTick(
             instrument_id=instrument_id,
             price=Price.from_str("1.00002"),
@@ -3559,6 +3566,9 @@ def test_skip_first_non_full_bar_when_starting_on_bar_boundary(self):
             ts_init=start_time_ns + 1_000_000_000,
         )
 
+        aggregator.handle_trade_tick(tick2)
+        clock.set_time(tick2.ts_event)
+
         tick3 = TradeTick(
             instrument_id=instrument_id,
             price=Price.from_str("1.00003"),
@@ -3569,25 +3579,18 @@ def test_skip_first_non_full_bar_when_starting_on_bar_boundary(self):
             ts_init=start_time_ns + 1_500_000_000,
         )
 
-        # Act - process ticks and advance time to trigger first bar
-        aggregator.handle_trade_tick(tick1)
-        clock.set_time(tick1.ts_event)
-
-        aggregator.handle_trade_tick(tick2)
-        clock.set_time(tick2.ts_event)
-
         aggregator.handle_trade_tick(tick3)
         clock.set_time(tick3.ts_event)
 
         # Advance to exactly 2 seconds to trigger the first bar
         events = clock.advance_time(start_time_ns + 2_000_000_000)
-        if events:
-            events[0].handle()
+        for event in events:
+            event.handle()
 
         # Assert - we should have received the first bar since we had data for the full period
         assert len(handler) == 1, f"Expected 1 bar but got {len(handler)} bars"
-        # The bar closes at start_time (00:00:00) because fire_immediately=True
-        assert handler[0].ts_event == start_time_ns
+
+        assert handler[0].ts_event == start_time_ns + 2_000_000_000
         assert handler[0].open == Price.from_str("1.00001")
         assert handler[0].close == Price.from_str("1.00003")
         assert handler[0].volume == Quantity.from_int(300000)