fix: preserve aggregation intervals across hours and UTC dates

docs/user-guide/feeds.md

@@ -188,6 +188,8 @@ Optional symbol filtering:
 feed = BarAggregator(raw_feed, bar_size_minutes=5, assets=["SPY", "QQQ"])
 ```
 
+Intervals align to `1970-01-01T00:00:00Z` and keep their configured duration across UTC midnight.
+
 The aggregator emits one event per asset so sparse symbols retain independent interval boundaries.
 An elapsed interval emits once as `complete`. If shutdown occurs before the current interval ends,
 its buffered value emits once as `evolving`. Late input cannot reopen an already completed interval.

src/ml4t/live/feeds/aggregator.py

@@ -323,17 +323,16 @@ def _signal_stop(self) -> None:
         self._queue.finish(discard=False)
 
     def _truncate_to_bar(self, dt: datetime) -> datetime:
-        """Truncate datetime to bar boundary.
+        """Truncate datetime to a fixed UTC epoch-aligned bar boundary.
 
         Args:
             dt: Datetime to truncate
 
         Returns:
             Datetime truncated to bar boundary (e.g., 10:35:42 -> 10:35:00 for 1-min bars)
         """
-        minutes_per_bar = int(self.bar_size.total_seconds() // 60)
-        truncated_minutes = (dt.minute // minutes_per_bar) * minutes_per_bar
-        return dt.replace(minute=truncated_minutes, second=0, microsecond=0)
+        epoch = datetime(1970, 1, 1, tzinfo=UTC)
+        return epoch + ((dt - epoch) // self.bar_size) * self.bar_size
 
     async def _emit_asset_bar(
         self,

tests/unit/test_bar_aggregator.py

@@ -102,6 +102,53 @@ async def test_truncates_to_configured_boundary(self) -> None:
 
         assert aggregator._truncate_to_bar(timestamp) == datetime(2024, 1, 1, 10, 35, tzinfo=UTC)
 
+    async def test_bar_windows_do_not_restart_at_the_hour(self) -> None:
+        """Sizes that do not divide 60 minutes must stay aligned across the hour."""
+        start = datetime(2024, 1, 1, 10, 30, tzinfo=UTC)
+        events = await collect(
+            BarAggregator(
+                MockDataFeed(
+                    [
+                        trade("AAPL", start, 100.0, 1),
+                        trade("AAPL", start + timedelta(minutes=59), 110.0, 1),
+                        trade("AAPL", start + timedelta(minutes=90), 120.0, 1),
+                    ]
+                ),
+                bar_size_minutes=90,
+            )
+        )
+
+        assert [event.event_time for event in events] == [
+            start,
+            start + timedelta(minutes=90),
+        ]
+        assert events[0].payload == BarPayload(100.0, 110.0, 100.0, 110.0, 2.0)
+        assert events[1].payload == BarPayload(120.0, 120.0, 120.0, 120.0, 1.0)
+
+    async def test_bar_windows_do_not_overlap_across_midnight(self) -> None:
+        start = datetime(2024, 1, 1, 23, tzinfo=UTC)
+        events = await collect(
+            BarAggregator(
+                MockDataFeed(
+                    [
+                        trade("AAPL", start + timedelta(minutes=50), 100.0, 1),
+                        trade("AAPL", start + timedelta(minutes=80), 110.0, 1),
+                        trade("AAPL", start + timedelta(minutes=100), 120.0, 1),
+                    ]
+                ),
+                bar_size_minutes=100,
+            )
+        )
+
+        assert datetime.fromisoformat(events[0].metadata["bar_end"]) <= events[1].event_time
+        assert [event.event_time for event in events] == [
+            start,
+            start + timedelta(minutes=100),
+        ]
+        assert events[0].metadata["bar_end"] == (start + timedelta(minutes=100)).isoformat()
+        assert events[0].payload == BarPayload(100.0, 110.0, 100.0, 110.0, 2.0)
+        assert events[1].payload == BarPayload(120.0, 120.0, 120.0, 120.0, 1.0)
+
     async def test_boundary_and_shutdown_emit_exact_completed_bars(self) -> None:
         start = datetime(2024, 1, 1, 10, tzinfo=UTC)
         events = await collect(