diff --git a/docs/user-guide/feeds.md b/docs/user-guide/feeds.md index 0311e7d..a296611 100644 --- a/docs/user-guide/feeds.md +++ b/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. diff --git a/src/ml4t/live/feeds/aggregator.py b/src/ml4t/live/feeds/aggregator.py index 1516d1a..e2f36c3 100644 --- a/src/ml4t/live/feeds/aggregator.py +++ b/src/ml4t/live/feeds/aggregator.py @@ -323,7 +323,7 @@ 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 @@ -331,9 +331,8 @@ def _truncate_to_bar(self, dt: datetime) -> datetime: 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, diff --git a/tests/unit/test_bar_aggregator.py b/tests/unit/test_bar_aggregator.py index 2c6ade4..9529cff 100644 --- a/tests/unit/test_bar_aggregator.py +++ b/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(