From 42aa185363696d5c9116957de67918c1c2ac4525 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:17:52 -0400 Subject: [PATCH 01/12] test: reproduce calendar disable and session boundary errors --- tests/test_session_calendar_contract.py | 137 ++++++++++++++++++++++++ 1 file changed, 137 insertions(+) create mode 100644 tests/test_session_calendar_contract.py diff --git a/tests/test_session_calendar_contract.py b/tests/test_session_calendar_contract.py new file mode 100644 index 0000000..31e0e78 --- /dev/null +++ b/tests/test_session_calendar_contract.py @@ -0,0 +1,137 @@ +"""Real provider/session workflows preserve declared calendar and timestamp meaning.""" + +from datetime import date + +import polars as pl +import pytest +from polars.testing import assert_frame_equal + +from ml4t.data import DataManager +from ml4t.data.core.models import Metadata +from ml4t.data.providers.synthetic import SyntheticProvider +from ml4t.data.sessions import SessionAssigner, SessionCompleter +from ml4t.data.storage.config import StorageConfig +from ml4t.data.storage.hive import HiveStorage +from ml4t.specs import FeedSpec + + +def provider_frame(frequency="daily", *, continuous=False): + with SyntheticProvider(calendar_mode="continuous" if continuous else "equity") as provider: + return provider.fetch_ohlcv("SPY", "2025-07-07", "2025-07-07", frequency) + + +@pytest.fixture(params=[False, True]) +def manager(request, tmp_path): + storage = HiveStorage(StorageConfig(base_path=tmp_path)) if request.param else None + with DataManager(storage=storage) as instance: + yield instance + + +def test_facade_default_is_nyse_not_a_missing_argument_error(manager): + frame = provider_frame() + assigned = manager.assign_sessions(frame, bar_frequency="daily") + assert assigned["session_date"].to_list() == [date(2025, 7, 7)] + + +@pytest.mark.parametrize("method", ["assign_sessions", "complete_sessions"]) +def test_explicit_none_preserves_provider_rows_without_session_or_imputation(manager, method): + frame = provider_frame() + assert_frame_equal(getattr(manager, method)(frame, calendar=None), frame) + + +def test_continuous_midnight_event_belongs_to_the_new_session(): + frame = provider_frame("hourly", continuous=True).head(2) + assigned = SessionAssigner("24/7").assign_sessions(frame, bar_frequency="intraday") + assert assigned["session_date"].to_list() == [date(2025, 7, 7), date(2025, 7, 7)] + + +def test_no_fill_flag_does_not_expand_a_provider_gap_without_storage(manager): + frame = provider_frame("minute", continuous=True).filter( + (pl.col("timestamp").dt.hour() == 13) & pl.col("timestamp").dt.minute().is_in([30, 32]) + ) + assigned = manager.complete_sessions(frame, calendar="NYSE", fill_gaps=False) + assert assigned.height == 2 + assert "is_imputed" not in assigned.columns + assert_frame_equal(assigned.select(frame.columns), frame) + + +def test_contradictory_explicit_exchange_alias_fails_without_input_mutation(manager): + frame = provider_frame() + original = frame.clone() + with pytest.raises(ValueError, match="conflict"): + manager.assign_sessions(frame, exchange="NYSE", calendar="LSE") + assert_frame_equal(frame, original) + + +@pytest.mark.parametrize( + "semantics, expected", [("event_time", date(2025, 7, 7)), ("bar_close", date(2025, 7, 6))] +) +def test_explicit_timestamp_meaning_overrides_daily_and_midnight_heuristics(semantics, expected): + frame = provider_frame("hourly", continuous=True).head(1) + feed = FeedSpec(calendar="24/7", timestamp_semantics=semantics) + assigned = SessionAssigner(feed_spec=feed).assign_sessions(frame, bar_frequency="daily") + assert assigned["session_date"].to_list() == [expected] + + +@pytest.mark.parametrize("calendar, hour", [("NYSE", 20), ("HKEX", 4)]) +def test_close_labeled_provider_bar_includes_close_or_break_start(calendar, hour): + frame = provider_frame("hourly", continuous=True).filter( + pl.col("timestamp").dt.hour() == hour - 1 + ) + frame = frame.with_columns(pl.col("timestamp") + pl.duration(hours=1)) + feed = FeedSpec(calendar=calendar, timestamp_semantics="bar_close") + assigned = SessionAssigner(feed_spec=feed).assign_sessions(frame) + assert assigned["session_date"].to_list() == [date(2025, 7, 7)] + event = SessionAssigner( + feed_spec=FeedSpec(calendar=calendar, timestamp_semantics="event_time") + ).assign_sessions(frame) + assert event["session_date"].to_list() == [None] + + +@pytest.mark.parametrize( + "declaration, expected", + [ + ({}, "NYSE"), + ({"calendar": None}, "NYSE"), + ({"time_contract_version": 2}, "NYSE"), + ({"time_contract_version": 2, "calendar": None}, None), + ({"time_contract_version": 2, "calendar": "24/7"}, "24/7"), + ], +) +def test_data_metadata_json_preserves_calendar_declaration_presence(declaration, expected): + metadata = Metadata(provider="synthetic", symbol="SPY", asset_class="equities", **declaration) + restored = Metadata.model_validate_json(metadata.model_dump_json()) + assert FeedSpec.from_any(restored).resolve_calendar() == expected + assert FeedSpec.from_mapping(metadata.model_dump()).resolve_calendar() == expected + assert FeedSpec.from_any(restored.model_copy()).resolve_calendar() == expected + if declaration == {"time_contract_version": 2}: + assert "calendar" not in restored.model_dump() + + +def test_feed_disable_survives_metadata_and_public_facade(manager): + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + time_contract_version=2, + calendar=None, + ) + restored = Metadata.model_validate_json(metadata.model_dump_json()) + frame = provider_frame() + assert_frame_equal(manager.assign_sessions(frame, feed_spec=restored), frame) + assigned = manager.assign_sessions( + frame, feed_spec=restored, calendar="NYSE", bar_frequency="daily" + ) + assert assigned["session_date"].to_list() == [date(2025, 7, 7)] + + +def test_calendar_name_alias_is_compatible_but_conflicts_are_not_silently_ignored(): + assert SessionAssigner(calendar_name="NYSE").calendar_name == "NYSE" + assert SessionCompleter(calendar_name="NYSE").calendar_name == "NYSE" + with pytest.raises(ValueError, match="conflict"): + SessionAssigner(calendar="NYSE", calendar_name="LSE") + + +def test_calendar_free_low_level_completion_is_not_a_continuous_grid(): + frame = provider_frame() + assert_frame_equal(SessionCompleter(calendar=None).complete_sessions(frame), frame) From 56f15621fa4ed41996d3f27e1c0c4ffc298a379e Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:23:57 -0400 Subject: [PATCH 02/12] fix: preserve calendar declarations and session boundary semantics --- docs/user-guide/sessions.md | 41 ++++++- src/ml4t/data/core/models.py | 62 ++++++++++- src/ml4t/data/data_manager.py | 44 +++----- src/ml4t/data/managers/metadata_manager.py | 31 ++---- src/ml4t/data/sessions/assigner.py | 118 +++++++++++++++++---- src/ml4t/data/sessions/completer.py | 61 +++++++---- tests/test_session_calendar_contract.py | 75 +++++++++++++ 7 files changed, 330 insertions(+), 102 deletions(-) diff --git a/docs/user-guide/sessions.md b/docs/user-guide/sessions.md index 84f2765..c3582c5 100644 --- a/docs/user-guide/sessions.md +++ b/docs/user-guide/sessions.md @@ -10,9 +10,9 @@ dependency is included with the package. ## Assign Session Dates -`SessionAssigner` requires a Polars `Datetime` column named `timestamp`. Naive timestamps are -interpreted as UTC; timezone-aware timestamps are converted to UTC before they are matched to a -schedule. +`SessionAssigner` accepts Polars `Datetime` instants or `Date` session labels in `timestamp`. +Naive instants use UTC unless the feed declares a timezone; timezone-aware instants are matched in +UTC. Ambiguous or nonexistent local times fail rather than being guessed. ```python from datetime import datetime @@ -55,6 +55,35 @@ fall within an open interval: Set `bar_frequency="daily"` when timestamps are daily labels. The default, `"auto"`, treats a series as daily when every non-null timestamp is midnight. +Explicit `FeedSpec.timestamp_semantics` overrides these heuristics: `session_label` preserves the +civil label date, `event_time` matches an instant, and `bar_close` matches the ending bar boundary. +Event intervals include the open and exclude the close; close-labeled bars exclude the open and +include the close. The corresponding boundaries apply to lunch breaks. An explicit `bar_label` +also selects the declared side; `event_time` alone does not infer a bar side. This assignment does +not validate intraday grid completeness. + +```python +from ml4t.specs import FeedSpec + +feed = FeedSpec(calendar="24/7", timestamp_semantics="event_time", time_contract_version=2) +assigned = SessionAssigner(feed_spec=feed).assign_sessions(bars) +unchanged = SessionAssigner(calendar=None).assign_sessions(bars) +``` + +Use the canonical `calendar` keyword on constructors and facade methods. Explicit selection wins +over feed metadata, then NYSE applies. Explicit `None` returns observations unchanged, without +adding a session column or reconstructing a continuous grid. `calendar="24/7"` remains a real +schedule, including an event exactly at midnight in the new day's session. Unknown names fail. +The positional calendar argument, `calendar_name` constructor keyword and explicit `exchange` +selector remain compatibility forms; contradictory explicit selectors raise an error. Descriptive +exchange identity in feed metadata does not select a calendar. + +Data's `Metadata` supports the same versioned temporal declaration. Version 1 retains legacy null +inheritance; version 2 preserves omitted calendar versus explicit null through JSON, copying and +shared FeedSpec projection. Use `model_dump(mode="json")` before a YAML writer, and ordinary +`model_dump_json()` for JSON; do not deliberately exclude null fields when preserving a disable +declaration. Declared timezone, timestamp semantics and fixed-grid fields survive these bridges. + ## Select an Exchange Calendar `SessionAssigner.from_exchange()` recognizes these exchange codes: @@ -103,6 +132,12 @@ Session completion rejects duplicate or non-minute-aligned timestamps and observ selected schedule. Use session assignment first when you need to inspect how individual timestamps map to the exchange calendar. +Completion remains a one-minute, open-labeled operation, not a general bar-grid validator. Explicit +close-labeled or session-label metadata is rejected. `calendar=None` returns the input unchanged. +`DataManager.complete_sessions(fill_gaps=False)` performs assignment only and never expands rows, +whether or not storage is configured. Feed declarations are accepted through the facade's +`feed_spec` keyword with the same precedence as the low-level classes. + ## Use Session Dates in Validation With scikit-learn installed, group observations by `session_date` when a split must not divide one diff --git a/src/ml4t/data/core/models.py b/src/ml4t/data/core/models.py index f2993ac..a892f69 100644 --- a/src/ml4t/data/core/models.py +++ b/src/ml4t/data/core/models.py @@ -7,9 +7,18 @@ from typing import Any import polars as pl -from pydantic import BaseModel, Field, model_validator +from pydantic import BaseModel, Field, field_validator, model_serializer, model_validator from ml4t.data.assets.asset_class import AssetClass # noqa: F401 — re-exported +from ml4t.specs import ( + UNSET_CALENDAR, + BarGridAnchor, + BarLabel, + CalendarUnset, + FeedSpec, + PartialBarPolicy, + TimestampSemantics, +) class SchemaVersion(StrEnum): @@ -73,7 +82,13 @@ class Metadata(BaseModel): # Exchange and calendar (for session-aware operations) exchange: str = "UNKNOWN" - calendar: str | None = None # pandas_market_calendars name (e.g., "CME_Globex_Crypto") + calendar: str | None | CalendarUnset = UNSET_CALENDAR + time_contract_version: int = 1 + timezone: str | None = None + timestamp_semantics: TimestampSemantics | str | None = None + bar_grid_anchor: BarGridAnchor | str | None = None + bar_label: BarLabel | str | None = None + partial_bar_policy: PartialBarPolicy | str | None = None # Tracking start_date: datetime | None = None @@ -89,6 +104,49 @@ class Metadata(BaseModel): model_config = {"use_enum_values": True} + @field_validator("calendar", mode="wrap") + @classmethod + def preserve_calendar_presence(cls, value: Any, handler: Any) -> Any: + return value if value is UNSET_CALENDAR else handler(value) + + @field_validator("time_contract_version", mode="before") + @classmethod + def validate_time_contract_version(cls, value: Any) -> int: + return FeedSpec(time_contract_version=value).time_contract_version + + def _time_declaration(self) -> FeedSpec: + return FeedSpec( + calendar=self.calendar, + time_contract_version=self.time_contract_version, + timezone=self.timezone, + timestamp_semantics=self.timestamp_semantics, + bar_grid_anchor=self.bar_grid_anchor, + bar_label=self.bar_label, + partial_bar_policy=self.partial_bar_policy, + ) + + @model_validator(mode="after") + def normalize_time_declaration(self) -> Metadata: + object.__setattr__(self, "calendar", self._time_declaration().calendar) + return self + + @model_serializer(mode="wrap") + def serialize_time_declaration(self, handler: Any) -> dict[str, Any]: + data = handler(self) + declaration = self._time_declaration().to_dict() + for name in ( + "calendar", + "time_contract_version", + "bar_grid_anchor", + "bar_label", + "partial_bar_policy", + ): + if name in declaration and name in data: + data[name] = declaration[name] + elif name not in declaration: + data.pop(name, None) + return data + @property def frequency(self) -> str: """Get frequency string for backward compatibility. diff --git a/src/ml4t/data/data_manager.py b/src/ml4t/data/data_manager.py index 0eca1e4..42cead4 100644 --- a/src/ml4t/data/data_manager.py +++ b/src/ml4t/data/data_manager.py @@ -29,6 +29,7 @@ from ml4t.data.managers.metadata_manager import MetadataManager from ml4t.data.managers.provider_manager import ProviderManager, ProviderRouter from ml4t.data.managers.storage_manager import StorageManager +from ml4t.specs import UNSET_CALENDAR, CalendarUnset logger = structlog.get_logger() @@ -453,51 +454,30 @@ def assign_sessions( self, df: pl.DataFrame, exchange: str | None = None, - calendar: str | None = None, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, bar_frequency: Literal["auto", "daily", "intraday"] = "auto", + *, + feed_spec: Any | None = None, ) -> pl.DataFrame: """Assign sessions to daily period labels or intraday timestamps.""" - if not self._metadata_manager: - # Fall back to direct implementation - from ml4t.data.sessions import SessionAssigner - - if calendar is None: - if exchange is None: - raise ValueError("Must provide either exchange or calendar parameter") - assigner = SessionAssigner.from_exchange(exchange) - else: - assigner = SessionAssigner(calendar) - return assigner.assign_sessions(df, bar_frequency=bar_frequency) - - return self._metadata_manager.assign_sessions(df, exchange, calendar, bar_frequency) + manager = self._metadata_manager or MetadataManager(None) + return manager.assign_sessions(df, exchange, calendar, bar_frequency, feed_spec=feed_spec) def complete_sessions( self, df: pl.DataFrame, exchange: str | None = None, - calendar: str | None = None, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, fill_gaps: bool = True, fill_method: str = "forward", zero_volume: bool = True, + *, + feed_spec: Any | None = None, ) -> pl.DataFrame: """Complete sessions by filling gaps.""" - if not self._metadata_manager: - # Fall back to direct implementation - from ml4t.data.sessions import SessionCompleter - from ml4t.data.sessions.assigner import EXCHANGE_CALENDARS - - if calendar is None: - if exchange is None: - raise ValueError("Must provide either exchange or calendar parameter") - calendar = EXCHANGE_CALENDARS.get(exchange.upper()) - if not calendar: - raise ValueError(f"Unknown exchange '{exchange}'") - - completer = SessionCompleter(calendar) - return completer.complete_sessions(df, fill_method=fill_method, zero_volume=zero_volume) - - return self._metadata_manager.complete_sessions( - df, exchange, calendar, fill_gaps, fill_method, zero_volume + manager = self._metadata_manager or MetadataManager(None) + return manager.complete_sessions( + df, exchange, calendar, fill_gaps, fill_method, zero_volume, feed_spec=feed_spec ) # ======================================================================== diff --git a/src/ml4t/data/managers/metadata_manager.py b/src/ml4t/data/managers/metadata_manager.py index 55e9db4..1605fd8 100644 --- a/src/ml4t/data/managers/metadata_manager.py +++ b/src/ml4t/data/managers/metadata_manager.py @@ -16,6 +16,7 @@ from ml4t.data.storage.backend import normalize_storage_metadata from ml4t.data.storage.keys import storage_key_path +from ml4t.specs import UNSET_CALENDAR, CalendarUnset logger = structlog.get_logger() @@ -156,8 +157,10 @@ def assign_sessions( self, df: pl.DataFrame, exchange: str | None = None, - calendar: str | None = None, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, bar_frequency: Literal["auto", "daily", "intraday"] = "auto", + *, + feed_spec: Any | None = None, ) -> pl.DataFrame: """Assign session_date column to DataFrame based on exchange calendar. @@ -175,12 +178,7 @@ def assign_sessions( """ from ml4t.data.sessions import SessionAssigner - if calendar is None: - if exchange is None: - raise ValueError("Must provide either exchange or calendar parameter") - assigner = SessionAssigner.from_exchange(exchange) - else: - assigner = SessionAssigner(calendar) + assigner = SessionAssigner(calendar, exchange=exchange, feed_spec=feed_spec) return assigner.assign_sessions(df, bar_frequency=bar_frequency) @@ -188,10 +186,12 @@ def complete_sessions( self, df: pl.DataFrame, exchange: str | None = None, - calendar: str | None = None, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, fill_gaps: bool = True, fill_method: str = "forward", zero_volume: bool = True, + *, + feed_spec: Any | None = None, ) -> pl.DataFrame: """Complete sessions by filling gaps. @@ -213,24 +213,13 @@ def complete_sessions( """ from ml4t.data.sessions import SessionCompleter - if calendar is None: - if exchange is None: - raise ValueError("Must provide either exchange or calendar parameter") - - from ml4t.data.sessions.assigner import EXCHANGE_CALENDARS - - calendar = EXCHANGE_CALENDARS.get(exchange.upper()) - if not calendar: - raise ValueError(f"Unknown exchange '{exchange}'") - - completer = SessionCompleter(calendar) - if not fill_gaps: from ml4t.data.sessions import SessionAssigner - assigner = SessionAssigner(calendar) + assigner = SessionAssigner(calendar, exchange=exchange, feed_spec=feed_spec) return assigner.assign_sessions(df) + completer = SessionCompleter(calendar, exchange=exchange, feed_spec=feed_spec) return completer.complete_sessions( df, fill_method=fill_method, diff --git a/src/ml4t/data/sessions/assigner.py b/src/ml4t/data/sessions/assigner.py index b940cf2..1c7c410 100644 --- a/src/ml4t/data/sessions/assigner.py +++ b/src/ml4t/data/sessions/assigner.py @@ -3,11 +3,13 @@ from __future__ import annotations from datetime import date, datetime -from typing import Literal +from typing import Any, Literal import polars as pl import structlog +from ml4t.specs import UNSET_CALENDAR, BarLabel, CalendarUnset, FeedSpec, TimestampSemantics + logger = structlog.get_logger() @@ -41,13 +43,37 @@ class SessionAssigner: The session_date is the date when the session ENDS (4pm date). """ - def __init__(self, calendar_name: str): + def __init__( + self, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, + *, + calendar_name: str | None | CalendarUnset = UNSET_CALENDAR, + exchange: str | None = None, + feed_spec: Any | None = None, + ) -> None: """Initialize session assigner. Args: - calendar_name: Name of calendar from pandas_market_calendars - (e.g., "CME_Globex_Crypto", "NYSE", "NASDAQ") + calendar: Explicit name; None disables session handling. + calendar_name: Compatibility alias for calendar. + exchange: Compatibility selector, not inferred from descriptive metadata. + feed_spec: Shared calendar, timezone and timestamp declarations. """ + selectors = [calendar_name] + if exchange is not None: + if exchange.upper() not in EXCHANGE_CALENDARS: + raise ValueError(f"Unknown exchange '{exchange}'") + selectors.append(EXCHANGE_CALENDARS[exchange.upper()]) + for alias in selectors: + if alias is not UNSET_CALENDAR: + if calendar is not UNSET_CALENDAR and calendar != alias: + raise ValueError("conflicting calendar selectors") + calendar = alias + self.feed_spec = FeedSpec.from_any(feed_spec) + self.calendar_name = self.feed_spec.resolve_calendar(calendar=calendar) + self.calendar = None + if self.calendar_name is None: + return try: import pandas_market_calendars as mcal except ImportError: @@ -56,13 +82,12 @@ def __init__(self, calendar_name: str): "Install with: pip install pandas-market-calendars" ) - self.calendar_name = calendar_name try: - self.calendar = mcal.get_calendar(calendar_name) + self.calendar = mcal.get_calendar(self.calendar_name) except Exception as e: - raise ValueError(f"Unknown calendar '{calendar_name}': {e}") + raise ValueError(f"Unknown calendar '{self.calendar_name}': {e}") - logger.info(f"Initialized SessionAssigner with calendar: {calendar_name}") + logger.info(f"Initialized SessionAssigner with calendar: {self.calendar_name}") @classmethod def from_exchange(cls, exchange: str) -> SessionAssigner: @@ -109,6 +134,8 @@ def assign_sessions( Raises: ValueError: If timestamp column missing """ + if self.calendar is None: + return df if "timestamp" not in df.columns: raise ValueError("DataFrame must have 'timestamp' column") if outside_session not in {"null", "raise", "drop"}: @@ -121,14 +148,22 @@ def assign_sessions( return df.with_columns(pl.lit(None).cast(pl.Date).alias("session_date")) timestamp_dtype = df.schema["timestamp"] - if not isinstance(timestamp_dtype, pl.Datetime): - raise TypeError("'timestamp' must contain Datetime values") + if timestamp_dtype != pl.Date and not isinstance(timestamp_dtype, pl.Datetime): + raise TypeError("'timestamp' must contain Date or Datetime values") + semantics = self.feed_spec.timestamp_semantics + if timestamp_dtype == pl.Date and ( + semantics in (TimestampSemantics.EVENT_TIME, TimestampSemantics.BAR_CLOSE) + or self.feed_spec.bar_label is not None + ): + raise ValueError("event_time/bar_close timestamps must contain Datetime values") # Auto-detect date range from data if start_date is None: - start_date = _datetime_scalar(df["timestamp"].min(), "timestamp") + start_date = df["timestamp"].min() if end_date is None: - end_date = _datetime_scalar(df["timestamp"].max(), "timestamp") + end_date = df["timestamp"].max() + if start_date is None or end_date is None: + raise TypeError("'timestamp' must contain non-null Datetime or Date values") logger.info( f"Assigning sessions for {len(df)} rows", @@ -179,8 +214,12 @@ def assign_sessions( return result timestamp_expr = pl.col("timestamp") - if timestamp_dtype.time_zone is None: - timestamp_expr = timestamp_expr.dt.replace_time_zone("UTC") + if timestamp_dtype == pl.Date: + timestamp_expr = timestamp_expr.cast(pl.Datetime("us")) + if timestamp_dtype == pl.Date or timestamp_dtype.time_zone is None: + timestamp_expr = timestamp_expr.dt.replace_time_zone( + self.feed_spec.timezone or "UTC" + ) timestamp_expr = timestamp_expr.dt.convert_time_zone("UTC").cast( pl.Datetime("us", "UTC") ) @@ -193,10 +232,25 @@ def assign_sessions( if isinstance(value, datetime) ) daily_labels = bar_frequency == "daily" or (bar_frequency == "auto" and inferred_daily) + if semantics is not None: + daily_labels = semantics == TimestampSemantics.SESSION_LABEL + elif self.feed_spec.bar_label is not None: + daily_labels = False + elif timestamp_dtype == pl.Date: + daily_labels = True if daily_labels: + label_expr = pl.col("timestamp") + if ( + isinstance(timestamp_dtype, pl.Datetime) + and timestamp_dtype.time_zone is not None + and self.feed_spec.timezone is not None + ): + label_expr = label_expr.dt.convert_time_zone(self.feed_spec.timezone) + if isinstance(timestamp_dtype, pl.Datetime): + label_expr = label_expr.dt.date() result = ( df.with_row_index("_session_row_index") - .with_columns(timestamp_expr.dt.date().alias("_session_date_label")) + .with_columns(label_expr.alias("_session_date_label")) .join( session_df.select( pl.col("_assigned_session_date").alias("_calendar_session_date_label"), @@ -234,20 +288,38 @@ def assign_sessions( ) .sort("_session_join_timestamp") ) + close_labeled = ( + semantics == TimestampSemantics.BAR_CLOSE + or self.feed_spec.bar_label == BarLabel.RIGHT + ) + join_boundary = "_calendar_session_end" if close_labeled else "_calendar_session_start" joined = observations.join_asof( - session_df, + session_df.sort(join_boundary), left_on="_session_join_timestamp", - right_on="_calendar_session_end", - strategy="forward", - ) - in_session = (pl.col("_session_timestamp") >= pl.col("_calendar_session_start")) & ( - pl.col("_session_timestamp") < pl.col("_calendar_session_end") + right_on=join_boundary, + strategy="forward" if close_labeled else "backward", ) + if close_labeled: + in_session = (pl.col("_session_timestamp") > pl.col("_calendar_session_start")) & ( + pl.col("_session_timestamp") <= pl.col("_calendar_session_end") + ) + else: + in_session = (pl.col("_session_timestamp") >= pl.col("_calendar_session_start")) & ( + pl.col("_session_timestamp") < pl.col("_calendar_session_end") + ) if "_calendar_break_start" in joined.columns: in_break = ( pl.col("_calendar_break_start").is_not_null() - & (pl.col("_session_timestamp") >= pl.col("_calendar_break_start")) - & (pl.col("_session_timestamp") < pl.col("_calendar_break_end")) + & ( + (pl.col("_session_timestamp") > pl.col("_calendar_break_start")) + if close_labeled + else (pl.col("_session_timestamp") >= pl.col("_calendar_break_start")) + ) + & ( + (pl.col("_session_timestamp") <= pl.col("_calendar_break_end")) + if close_labeled + else (pl.col("_session_timestamp") < pl.col("_calendar_break_end")) + ) ) in_session &= ~in_break result = joined.with_columns( diff --git a/src/ml4t/data/sessions/completer.py b/src/ml4t/data/sessions/completer.py index 2b340f3..a947fd2 100644 --- a/src/ml4t/data/sessions/completer.py +++ b/src/ml4t/data/sessions/completer.py @@ -6,10 +6,14 @@ from __future__ import annotations from datetime import date, datetime, timedelta +from typing import Any import polars as pl import structlog +from ml4t.data.sessions.assigner import SessionAssigner +from ml4t.specs import UNSET_CALENDAR, BarLabel, CalendarUnset, TimestampSemantics + logger = structlog.get_logger() @@ -37,28 +41,28 @@ class SessionCompleter: ``` """ - def __init__(self, calendar_name: str): + def __init__( + self, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, + *, + calendar_name: str | None | CalendarUnset = UNSET_CALENDAR, + exchange: str | None = None, + feed_spec: Any | None = None, + ) -> None: """Initialize with exchange calendar. Args: - calendar_name: Name of pandas_market_calendars calendar - (e.g., "CME_Globex_Crypto", "NYSE", "NASDAQ") + calendar: Explicit schedule; None disables completion. + calendar_name: Compatibility alias. + exchange: Compatibility selector. + feed_spec: Shared temporal declaration. """ - try: - import pandas_market_calendars as mcal - except ImportError: - raise ImportError( - "pandas_market_calendars is required for session completion. " - "Install with: pip install pandas-market-calendars" - ) - - self.calendar_name = calendar_name - try: - self.calendar = mcal.get_calendar(calendar_name) - except Exception as e: - raise ValueError(f"Unknown calendar '{calendar_name}': {e}") - - logger.info(f"Initialized SessionCompleter with calendar: {calendar_name}") + assigner = SessionAssigner( + calendar, calendar_name=calendar_name, exchange=exchange, feed_spec=feed_spec + ) + self.calendar_name = assigner.calendar_name + self.calendar = assigner.calendar + self.feed_spec = assigner.feed_spec def complete_sessions( self, @@ -85,6 +89,15 @@ def complete_sessions( not minute-aligned, multiple symbols share a timestamp, or an observation falls outside a scheduled session or the requested half-open range """ + if self.calendar is None: + return df + if self.feed_spec.timestamp_semantics not in ( + None, + TimestampSemantics.EVENT_TIME, + ) or self.feed_spec.bar_label not in (None, BarLabel.LEFT): + raise NotImplementedError( + "minute completion currently requires open-labeled event_time bars" + ) if "timestamp" not in df.columns: raise ValueError("DataFrame must have 'timestamp' column") if fill_method not in {"forward", "backward", "none"}: @@ -121,11 +134,15 @@ def complete_sessions( ): raise ValueError("start_date and end_date must not be NaT") if request_start.tzinfo is None: - request_start = request_start.tz_localize("UTC") + request_start = request_start.tz_localize( + self.feed_spec.timezone or "UTC" + ).tz_convert("UTC") else: request_start = request_start.tz_convert("UTC") if request_end.tzinfo is None: - request_end = request_end.tz_localize("UTC") + request_end = request_end.tz_localize(self.feed_spec.timezone or "UTC").tz_convert( + "UTC" + ) else: request_end = request_end.tz_convert("UTC") if request_start >= request_end: @@ -192,7 +209,9 @@ def complete_sessions( # Ensure input data has matching timestamp type timestamp_expr = pl.col("timestamp") if timestamp_dtype.time_zone is None: - timestamp_expr = timestamp_expr.dt.replace_time_zone("UTC") + timestamp_expr = timestamp_expr.dt.replace_time_zone( + self.feed_spec.timezone or "UTC" + ) timestamp_expr = timestamp_expr.dt.convert_time_zone("UTC").cast( pl.Datetime("ns", "UTC") ) diff --git a/tests/test_session_calendar_contract.py b/tests/test_session_calendar_contract.py index 31e0e78..b62d60d 100644 --- a/tests/test_session_calendar_contract.py +++ b/tests/test_session_calendar_contract.py @@ -135,3 +135,78 @@ def test_calendar_name_alias_is_compatible_but_conflicts_are_not_silently_ignore def test_calendar_free_low_level_completion_is_not_a_continuous_grid(): frame = provider_frame() assert_frame_equal(SessionCompleter(calendar=None).complete_sessions(frame), frame) + + +def test_declared_bar_side_is_not_inferred_from_event_time(): + frame = provider_frame("hourly", continuous=True).head(1) + feed = FeedSpec( + calendar="24/7", + time_contract_version=2, + timestamp_semantics="event_time", + bar_label="right", + ) + assert SessionAssigner(feed_spec=feed).assign_sessions(frame)["session_date"].to_list() == [ + date(2025, 7, 6) + ] + + +def test_persisted_data_metadata_retains_calendar_disable_and_bar_grid(tmp_path): + import yaml + + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + calendar=None, + time_contract_version=2, + timezone="UTC", + timestamp_semantics="event_time", + bar_grid_anchor="session_open", + bar_label="right", + partial_bar_policy="include", + ) + restored = Metadata.model_validate( + yaml.safe_load(yaml.safe_dump(metadata.model_dump(mode="json"))) + ) + feed = FeedSpec.from_any(restored) + assert feed.resolve_calendar() is None + assert feed.bar_label == "right" + assert feed.bar_grid_anchor == "session_open" + assert feed.partial_bar_policy == "include" + frame = provider_frame("hourly", continuous=True).head(1) + storage = HiveStorage(StorageConfig(base_path=tmp_path)) + storage.write(frame, "equities/hourly/SPY", restored.model_dump()) + with DataManager(storage=storage) as manager: + declared = manager.get_metadata_for_key("equities/hourly/SPY") + assert_frame_equal(manager.assign_sessions(frame, feed_spec=declared), frame) + + +def test_explicit_session_labels_keep_civil_dates_in_non_utc_timezone(): + frame = provider_frame().with_columns( + pl.col("timestamp") + .dt.replace_time_zone(None) + .dt.truncate("1d") + .dt.replace_time_zone("Asia/Tokyo") + ) + feed = FeedSpec(calendar="NYSE", timestamp_semantics="session_label", timezone="Asia/Tokyo") + result = SessionAssigner(feed_spec=feed).assign_sessions(frame, bar_frequency="intraday") + assert result["session_date"].to_list() == [date(2025, 7, 7)] + + +def test_naive_event_time_uses_declared_timezone_and_rejects_ambiguous_clock(): + frame = provider_frame("hourly", continuous=True).filter(pl.col("timestamp").dt.hour() == 14) + local = frame.with_columns( + pl.col("timestamp").dt.convert_time_zone("America/New_York").dt.replace_time_zone(None) + ) + feed = FeedSpec(calendar="NYSE", timestamp_semantics="event_time", timezone="America/New_York") + assert SessionAssigner(feed_spec=feed).assign_sessions(local)["session_date"].to_list() == [ + date(2025, 7, 7) + ] + with SyntheticProvider(calendar_mode="continuous") as provider: + ambiguous = ( + provider.fetch_ohlcv("SPY", "2025-11-02", "2025-11-02", "hourly") + .filter(pl.col("timestamp").dt.hour() == 1) + .with_columns(pl.col("timestamp").dt.replace_time_zone(None)) + ) + with pytest.raises(pl.exceptions.ComputeError, match="ambiguous"): + SessionAssigner(feed_spec=feed).assign_sessions(ambiguous) From ef487c61fd7c18e27e04b8c7800a1fed8d12de6e Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:29:15 -0400 Subject: [PATCH 03/12] test: expose calendar disable loss during provider updates --- tests/test_session_calendar_contract.py | 36 +++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/tests/test_session_calendar_contract.py b/tests/test_session_calendar_contract.py index b62d60d..ba9e6cf 100644 --- a/tests/test_session_calendar_contract.py +++ b/tests/test_session_calendar_contract.py @@ -210,3 +210,39 @@ def test_naive_event_time_uses_declared_timezone_and_rejects_ambiguous_clock(): ) with pytest.raises(pl.exceptions.ComputeError, match="ambiguous"): SessionAssigner(feed_spec=feed).assign_sessions(ambiguous) + + +def test_real_provider_update_preserves_a_stored_explicit_calendar_disable(tmp_path): + frame = provider_frame() + storage = HiveStorage(StorageConfig(base_path=tmp_path)) + storage.write( + frame, + "equities/daily/SPY", + {"provider": "synthetic", "calendar": None, "time_contract_version": 2}, + ) + with DataManager(storage=storage) as manager: + assert ( + FeedSpec.from_mapping( + manager.get_metadata_for_key("equities/daily/SPY") + ).resolve_calendar() + is None + ) + manager.update("SPY", fill_gaps=False) + assert ( + FeedSpec.from_mapping( + manager.get_metadata_for_key("equities/daily/SPY") + ).resolve_calendar() + is None + ) + + +@pytest.mark.parametrize("invalid", [{"calendar": {"bad": True}}, {"time_contract_version": 3}]) +def test_real_provider_update_does_not_silently_drop_invalid_temporal_metadata(tmp_path, invalid): + frame = provider_frame() + storage = HiveStorage(StorageConfig(base_path=tmp_path)) + storage.write(frame, "equities/daily/SPY", {"provider": "synthetic", **invalid}) + before = storage.get_metadata("equities/daily/SPY") + with DataManager(storage=storage) as manager, pytest.raises(ValueError): + manager.update("SPY", fill_gaps=False) + assert storage.get_metadata("equities/daily/SPY") == before + assert_frame_equal(storage.read("equities/daily/SPY").collect(), frame) From 55cf992e730f2d172ced199bd3137555961fba91 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:29:43 -0400 Subject: [PATCH 04/12] fix: retain declared null calendar when updating stored metadata --- src/ml4t/data/managers/storage_manager.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/src/ml4t/data/managers/storage_manager.py b/src/ml4t/data/managers/storage_manager.py index 7eb9925..b53d579 100644 --- a/src/ml4t/data/managers/storage_manager.py +++ b/src/ml4t/data/managers/storage_manager.py @@ -622,7 +622,11 @@ def update( existing_model_values = { name: value for name in Metadata.model_fields - if (value := existing_metadata.get(name)) is not None + if name in existing_metadata + and ( + (value := existing_metadata[name]) is not None + or (name == "calendar" and existing_metadata.get("time_contract_version") == 2) + ) } resolved_provider = ( provider @@ -660,7 +664,18 @@ def update( for detail in error.errors() if (location := detail.get("loc")) and isinstance(location[0], str) } - removable_fields = invalid_fields - {"provider", "symbol", "asset_class"} + removable_fields = invalid_fields - { + "provider", + "symbol", + "asset_class", + "calendar", + "time_contract_version", + "timezone", + "timestamp_semantics", + "bar_grid_anchor", + "bar_label", + "partial_bar_policy", + } if not removable_fields: raise logger.warning( From fb4d3b04e9bf0f30d6fa32c39dbefe881827710d Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:31:56 -0400 Subject: [PATCH 05/12] test: isolate temporal update validation from provider routing --- tests/test_session_calendar_contract.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/test_session_calendar_contract.py b/tests/test_session_calendar_contract.py index ba9e6cf..79faa1c 100644 --- a/tests/test_session_calendar_contract.py +++ b/tests/test_session_calendar_contract.py @@ -227,7 +227,7 @@ def test_real_provider_update_preserves_a_stored_explicit_calendar_disable(tmp_p ).resolve_calendar() is None ) - manager.update("SPY", fill_gaps=False) + manager.update("SPY", fill_gaps=False, provider="synthetic") assert ( FeedSpec.from_mapping( manager.get_metadata_for_key("equities/daily/SPY") @@ -242,7 +242,10 @@ def test_real_provider_update_does_not_silently_drop_invalid_temporal_metadata(t storage = HiveStorage(StorageConfig(base_path=tmp_path)) storage.write(frame, "equities/daily/SPY", {"provider": "synthetic", **invalid}) before = storage.get_metadata("equities/daily/SPY") - with DataManager(storage=storage) as manager, pytest.raises(ValueError): - manager.update("SPY", fill_gaps=False) + with ( + DataManager(storage=storage) as manager, + pytest.raises(ValueError, match="calendar|time_contract_version"), + ): + manager.update("SPY", fill_gaps=False, provider="synthetic") assert storage.get_metadata("equities/daily/SPY") == before assert_frame_equal(storage.read("equities/daily/SPY").collect(), frame) From 6488ae7db956b75274ae8c396fc28329d1bd69a6 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:38:37 -0400 Subject: [PATCH 06/12] test: reproduce invalid timezone declarations falling back to UTC --- tests/test_session_calendar_contract.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/test_session_calendar_contract.py b/tests/test_session_calendar_contract.py index 79faa1c..3b87184 100644 --- a/tests/test_session_calendar_contract.py +++ b/tests/test_session_calendar_contract.py @@ -249,3 +249,18 @@ def test_real_provider_update_does_not_silently_drop_invalid_temporal_metadata(t manager.update("SPY", fill_gaps=False, provider="synthetic") assert storage.get_metadata("equities/daily/SPY") == before assert_frame_equal(storage.read("equities/daily/SPY").collect(), frame) + + +@pytest.mark.parametrize("timezone", ["", "Not/A_Zone"]) +@pytest.mark.parametrize("adapter", [SessionAssigner, SessionCompleter]) +def test_invalid_declared_timezone_does_not_silently_become_utc(timezone, adapter): + frame = provider_frame().with_columns(pl.col("timestamp").dt.replace_time_zone(None)) + feed = FeedSpec.from_mapping({"timezone": timezone, "timestamp_semantics": "event_time"}) + original = frame.clone() + with pytest.raises(ValueError, match="timezone"): + instance = adapter(feed_spec=feed) + if adapter is SessionAssigner: + instance.assign_sessions(frame) + else: + instance.complete_sessions(frame) + assert_frame_equal(frame, original) From 692ab4f1ae79ca9d5c5a82fd9fb167bde7a2e281 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:39:12 -0400 Subject: [PATCH 07/12] fix: reject invalid declared timezones before session processing --- src/ml4t/data/sessions/assigner.py | 8 ++++++++ tests/test_session_calendar_contract.py | 2 +- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/src/ml4t/data/sessions/assigner.py b/src/ml4t/data/sessions/assigner.py index 1c7c410..b35cee7 100644 --- a/src/ml4t/data/sessions/assigner.py +++ b/src/ml4t/data/sessions/assigner.py @@ -74,6 +74,14 @@ def __init__( self.calendar = None if self.calendar_name is None: return + timezone = self.feed_spec.timezone + if timezone is not None: + if not isinstance(timezone, str) or not timezone.strip(): + raise ValueError("timezone must be a non-empty name or None") + try: + pl.Series([], dtype=pl.Datetime("us", "UTC")).dt.convert_time_zone(timezone) + except (pl.exceptions.PolarsError, TypeError) as error: + raise ValueError(f"Unknown timezone '{timezone}'") from error try: import pandas_market_calendars as mcal except ImportError: diff --git a/tests/test_session_calendar_contract.py b/tests/test_session_calendar_contract.py index 3b87184..58d4dfb 100644 --- a/tests/test_session_calendar_contract.py +++ b/tests/test_session_calendar_contract.py @@ -257,7 +257,7 @@ def test_invalid_declared_timezone_does_not_silently_become_utc(timezone, adapte frame = provider_frame().with_columns(pl.col("timestamp").dt.replace_time_zone(None)) feed = FeedSpec.from_mapping({"timezone": timezone, "timestamp_semantics": "event_time"}) original = frame.clone() - with pytest.raises(ValueError, match="timezone"): + with pytest.raises((ValueError, pl.exceptions.ComputeError), match="time.?zone"): instance = adapter(feed_spec=feed) if adapter is SessionAssigner: instance.assign_sessions(frame) From 8c5cfc692440c893ff68013ecc3e44f8cb4b8056 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:49:16 -0400 Subject: [PATCH 08/12] test: expose silently ignored session and completion declarations --- tests/test_session_calendar_contract.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tests/test_session_calendar_contract.py b/tests/test_session_calendar_contract.py index 58d4dfb..dab56df 100644 --- a/tests/test_session_calendar_contract.py +++ b/tests/test_session_calendar_contract.py @@ -264,3 +264,27 @@ def test_invalid_declared_timezone_does_not_silently_become_utc(timezone, adapte else: instance.complete_sessions(frame) assert_frame_equal(frame, original) + + +@pytest.mark.parametrize( + "declaration, message", + [({"bar_type": "volume"}, "time bars"), ({"data_frequency": "hourly"}, "minute")], +) +def test_minute_completion_rejects_a_declared_non_minute_or_event_count_feed(declaration, message): + frame = provider_frame("hourly", continuous=True).filter(pl.col("timestamp").dt.hour() == 14) + with pytest.raises((ValueError, NotImplementedError), match=message): + SessionCompleter(feed_spec=FeedSpec(**declaration)).complete_sessions(frame) + + +def test_session_start_override_is_preserved_but_not_silently_applied_as_native_schedule(): + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + time_contract_version=2, + session_start_time="12:00", + ) + restored = Metadata.model_validate_json(metadata.model_dump_json()) + assert FeedSpec.from_any(restored).session_start_time == "12:00" + with pytest.raises(NotImplementedError, match="session_start_time"): + SessionAssigner(feed_spec=restored) From fdadf93d73db071509c1db5ffde4261f07c1062b Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:52:43 -0400 Subject: [PATCH 09/12] fix: preserve and validate explicit session operation declarations --- docs/user-guide/sessions.md | 5 +++++ src/ml4t/data/core/models.py | 2 ++ src/ml4t/data/sessions/assigner.py | 4 ++++ src/ml4t/data/sessions/completer.py | 6 ++++++ 4 files changed, 17 insertions(+) diff --git a/docs/user-guide/sessions.md b/docs/user-guide/sessions.md index c3582c5..df4d958 100644 --- a/docs/user-guide/sessions.md +++ b/docs/user-guide/sessions.md @@ -62,6 +62,10 @@ include the close. The corresponding boundaries apply to lunch breaks. An explic also selects the declared side; `event_time` alone does not infer a bar side. This assignment does not validate intraday grid completeness. +`session_start_time` is retained in metadata, but these schedule adapters reject a non-null override. +Register and select an authoritative calendar with the required intervals instead of silently +replacing a named exchange's session boundaries. + ```python from ml4t.specs import FeedSpec @@ -134,6 +138,7 @@ map to the exchange calendar. Completion remains a one-minute, open-labeled operation, not a general bar-grid validator. Explicit close-labeled or session-label metadata is rejected. `calendar=None` returns the input unchanged. +Declared event-count bar types and frequencies other than one minute are also rejected. `DataManager.complete_sessions(fill_gaps=False)` performs assignment only and never expands rows, whether or not storage is configured. Feed declarations are accepted through the facade's `feed_spec` keyword with the same precedence as the low-level classes. diff --git a/src/ml4t/data/core/models.py b/src/ml4t/data/core/models.py index a892f69..159f39f 100644 --- a/src/ml4t/data/core/models.py +++ b/src/ml4t/data/core/models.py @@ -86,6 +86,7 @@ class Metadata(BaseModel): time_contract_version: int = 1 timezone: str | None = None timestamp_semantics: TimestampSemantics | str | None = None + session_start_time: str | None = None bar_grid_anchor: BarGridAnchor | str | None = None bar_label: BarLabel | str | None = None partial_bar_policy: PartialBarPolicy | str | None = None @@ -120,6 +121,7 @@ def _time_declaration(self) -> FeedSpec: time_contract_version=self.time_contract_version, timezone=self.timezone, timestamp_semantics=self.timestamp_semantics, + session_start_time=self.session_start_time, bar_grid_anchor=self.bar_grid_anchor, bar_label=self.bar_label, partial_bar_policy=self.partial_bar_policy, diff --git a/src/ml4t/data/sessions/assigner.py b/src/ml4t/data/sessions/assigner.py index b35cee7..6d3eb5a 100644 --- a/src/ml4t/data/sessions/assigner.py +++ b/src/ml4t/data/sessions/assigner.py @@ -74,6 +74,10 @@ def __init__( self.calendar = None if self.calendar_name is None: return + if self.feed_spec.session_start_time is not None: + raise NotImplementedError( + "session_start_time overrides require a registered calendar with that schedule" + ) timezone = self.feed_spec.timezone if timezone is not None: if not isinstance(timezone, str) or not timezone.strip(): diff --git a/src/ml4t/data/sessions/completer.py b/src/ml4t/data/sessions/completer.py index a947fd2..ad69ba0 100644 --- a/src/ml4t/data/sessions/completer.py +++ b/src/ml4t/data/sessions/completer.py @@ -91,6 +91,12 @@ def complete_sessions( """ if self.calendar is None: return df + if self.feed_spec.bar_type not in (None, "time"): + raise NotImplementedError("minute completion requires fixed time bars") + if self.feed_spec.data_frequency is not None and str( + self.feed_spec.data_frequency + ).lower() not in {"minute", "1minute", "1min", "1m"}: + raise ValueError("minute completion requires a declared one-minute frequency") if self.feed_spec.timestamp_semantics not in ( None, TimestampSemantics.EVENT_TIME, From 1f979aa8ee216c62dc9092b7cc2b68209e0304c8 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 10:54:54 -0400 Subject: [PATCH 10/12] fix: reject unsupported explicit policies in native minute completion --- docs/user-guide/sessions.md | 2 ++ src/ml4t/data/sessions/completer.py | 7 +++++++ tests/test_session_calendar_contract.py | 9 +++++++++ 3 files changed, 18 insertions(+) diff --git a/docs/user-guide/sessions.md b/docs/user-guide/sessions.md index df4d958..c43f7f8 100644 --- a/docs/user-guide/sessions.md +++ b/docs/user-guide/sessions.md @@ -139,6 +139,8 @@ map to the exchange calendar. Completion remains a one-minute, open-labeled operation, not a general bar-grid validator. Explicit close-labeled or session-label metadata is rejected. `calendar=None` returns the input unchanged. Declared event-count bar types and frequencies other than one minute are also rejected. +General bar-grid anchor/partial-bar policies are preserved as metadata but explicitly rejected by +this native minute completer; they are not silently approximated by its legacy minute template. `DataManager.complete_sessions(fill_gaps=False)` performs assignment only and never expands rows, whether or not storage is configured. Feed declarations are accepted through the facade's `feed_spec` keyword with the same precedence as the low-level classes. diff --git a/src/ml4t/data/sessions/completer.py b/src/ml4t/data/sessions/completer.py index ad69ba0..fe4fb80 100644 --- a/src/ml4t/data/sessions/completer.py +++ b/src/ml4t/data/sessions/completer.py @@ -91,6 +91,13 @@ def complete_sessions( """ if self.calendar is None: return df + if ( + self.feed_spec.bar_grid_anchor is not None + or self.feed_spec.partial_bar_policy is not None + ): + raise NotImplementedError( + "native minute completion does not support explicit general bar-grid policies" + ) if self.feed_spec.bar_type not in (None, "time"): raise NotImplementedError("minute completion requires fixed time bars") if self.feed_spec.data_frequency is not None and str( diff --git a/tests/test_session_calendar_contract.py b/tests/test_session_calendar_contract.py index dab56df..ce644d4 100644 --- a/tests/test_session_calendar_contract.py +++ b/tests/test_session_calendar_contract.py @@ -288,3 +288,12 @@ def test_session_start_override_is_preserved_but_not_silently_applied_as_native_ assert FeedSpec.from_any(restored).session_start_time == "12:00" with pytest.raises(NotImplementedError, match="session_start_time"): SessionAssigner(feed_spec=restored) + + +def test_native_minute_completion_does_not_silently_ignore_general_grid_policies(): + frame = provider_frame("hourly", continuous=True).filter(pl.col("timestamp").dt.hour() == 14) + feed = FeedSpec( + time_contract_version=2, bar_grid_anchor="session_open", partial_bar_policy="exclude" + ) + with pytest.raises(NotImplementedError, match="bar-grid"): + SessionCompleter(feed_spec=feed).complete_sessions(frame) From 4c1c0db41a9bb3cdca0787e159dea9092b965820 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 11:45:48 -0400 Subject: [PATCH 11/12] test: reproduce stored daily-label timezone drift and unsupported explicit instants --- tests/test_daily_timestamp_declarations.py | 107 +++++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 tests/test_daily_timestamp_declarations.py diff --git a/tests/test_daily_timestamp_declarations.py b/tests/test_daily_timestamp_declarations.py new file mode 100644 index 0000000..a779d04 --- /dev/null +++ b/tests/test_daily_timestamp_declarations.py @@ -0,0 +1,107 @@ +"""Daily consumers must preserve declared label dates and map explicit instants to sessions.""" + +from datetime import date + +import polars as pl +import pytest + +from ml4t.data import DataManager +from ml4t.data.core.models import Metadata +from ml4t.data.providers.synthetic import SyntheticProvider +from ml4t.data.storage.config import StorageConfig +from ml4t.data.storage.flat import FlatStorage +from ml4t.data.storage.hive import HiveStorage +from ml4t.data.update_manager import GapDetector as RangeDetector +from ml4t.data.utils.gaps import GapDetector +from ml4t.specs import FeedSpec + + +def provider_frame(start="2025-07-15", end="2025-07-17"): + with SyntheticProvider() as provider: + return provider.fetch_ohlcv("SPY", start, end, "daily") + + +@pytest.fixture(params=[FlatStorage, HiveStorage]) +def stored_labels(request, tmp_path): + storage = request.param(StorageConfig(base_path=tmp_path)) + frame = provider_frame("2025-07-11", "2025-07-14").with_columns( + pl.col("timestamp").dt.date().cast(pl.Datetime("us")).dt.replace_time_zone("Asia/Tokyo") + ) + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + bar_params={"frequency": "daily"}, + calendar="JPX", + time_contract_version=2, + timezone="Asia/Tokyo", + timestamp_semantics="session_label", + ) + key = "equities/daily/SPY" + storage.write(frame, key, metadata.model_dump()) + with DataManager(storage=storage) as manager: + yield storage.read(key).collect(), manager.get_metadata_for_key(key) + + +@pytest.mark.parametrize("consumer", [GapDetector, RangeDetector]) +def test_real_utc_storage_roundtrip_does_not_turn_tokyo_labels_into_a_false_friday_outage( + stored_labels, consumer +): + frame, metadata = stored_labels + assert frame["timestamp"].dt.date().to_list() == [date(2025, 7, 10), date(2025, 7, 13)] + assert consumer().detect_gaps(frame, feed_spec=metadata) == [] + + +@pytest.mark.parametrize("consumer", [GapDetector, RangeDetector]) +@pytest.mark.parametrize("semantics", ["event_time", "bar_close"]) +def test_explicit_daily_instants_use_their_session_and_keep_a_real_missing_session( + consumer, semantics +): + frame = provider_frame().filter(pl.col("timestamp").dt.date() != date(2025, 7, 16)) + feed = FeedSpec(time_contract_version=2, calendar="NYSE", timestamp_semantics=semantics) + found = consumer().detect_gaps(frame, feed_spec=feed) + count = sum(g.missing_periods if consumer is GapDetector else g["size_days"] for g in found) + assert count == 1 + + +@pytest.mark.parametrize("semantics,expected", [("event_time", [12, 13]), ("bar_close", [11, 12])]) +def test_continuous_midnight_right_labels_change_the_missing_session_dates(semantics, expected): + frame = provider_frame("2025-07-11", "2025-07-14").with_columns( + pl.col("timestamp").dt.date().cast(pl.Datetime("us")).dt.replace_time_zone("UTC") + ) + found = RangeDetector().detect_gaps( + frame, + feed_spec=FeedSpec( + time_contract_version=2, + calendar="24/7", + timestamp_semantics=semantics, + ), + ) + assert len(found) == 1 + assert [found[0]["start"].day, found[0]["end"].day] == expected + + +@pytest.mark.parametrize("consumer", [GapDetector, RangeDetector]) +def test_disabled_daily_event_declaration_does_not_restore_a_schedule(consumer): + frame = provider_frame().gather_every(2) + feed = FeedSpec(time_contract_version=2, calendar=None, timestamp_semantics="event_time") + assert consumer().detect_gaps(frame, feed_spec=feed) == [] + + +def test_declared_outside_session_instant_is_not_reinterpreted_as_a_daily_label(): + with SyntheticProvider(calendar_mode="continuous") as provider: + frame = provider.fetch_ohlcv("SPY", "2025-07-07", "2025-07-07", "hourly").head(2) + feed = FeedSpec(time_contract_version=2, calendar="NYSE", timestamp_semantics="event_time") + with pytest.raises(ValueError, match="outside a trading session"): + GapDetector().detect_gaps(frame, feed_spec=feed) + + +def test_declared_instants_support_the_public_custom_timestamp_column(): + frame = ( + provider_frame() + .filter(pl.col("timestamp").dt.date() != date(2025, 7, 16)) + .rename({"timestamp": "stamp"}) + ) + feed = FeedSpec(time_contract_version=2, calendar="NYSE", timestamp_semantics="event_time") + found = GapDetector().detect_gaps(frame, timestamp_col="stamp", feed_spec=feed) + assert sum(g.missing_periods for g in found) == 1 From 92aeae90cca05da3043f1d7a36b8dcf5a04ced26 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 11:51:27 -0400 Subject: [PATCH 12/12] fix: share declared daily timestamp interpretation across gap and backfill workflows --- docs/user-guide/incremental-updates.md | 21 ++++--- src/ml4t/data/sessions/assigner.py | 39 ++++++++++++ src/ml4t/data/update_manager.py | 33 +++------- src/ml4t/data/utils/gaps.py | 21 ++----- tests/test_daily_session_gaps.py | 5 +- tests/test_daily_timestamp_declarations.py | 73 ++++++++++++++++++++++ 6 files changed, 144 insertions(+), 48 deletions(-) diff --git a/docs/user-guide/incremental-updates.md b/docs/user-guide/incremental-updates.md index cf34391..e12df47 100644 --- a/docs/user-guide/incremental-updates.md +++ b/docs/user-guide/incremental-updates.md @@ -74,15 +74,21 @@ applies, then NYSE. `calendar="24/7"` expects a daily observation on weekends an feed declaration selects a calendar. Version-1 feed metadata with `calendar=None` still means unspecified; version-2 metadata preserves an explicit null as disabled after serialization. -Daily inputs are session labels (Polars Date, or the date component of Datetime values in their -existing timezone). Non-midnight labels remain supported. Exchange holidays and weekends do not +Undeclared daily timestamps and explicit `session_label` inputs are labels (Polars Date, or the +date component of Datetime values). Aware labels restore a declared `timezone` before extracting +dates, including after storage's canonical UTC read. Naive labels remain civil identifiers rather +than localized event clocks. Non-midnight labels remain supported. Exchange holidays and weekends do not count as missing NYSE sessions; a shortened session still requires one daily observation. Gap bounds remain the two observed timestamps, while `missing_periods` counts only scheduled dates strictly between them. Inspect each asset separately; observed dates from other assets must not define the schedule. This operation does not validate off-session observations or discover gaps outside the observed range. -Explicit event-time/bar-close metadata is rejected rather than reinterpreted as session labels. +Explicit `event_time`, `bar_close` or bar-side metadata instead uses native authoritative session +mapping, even with daily frequency or midnight timestamps. Right-labeled continuous midnight belongs +to the preceding session; event time belongs to the new session. Out-of-session instants fail clearly. +Gap classification does not add a new off-session rejection policy for legacy daily labels. The +requested daily frequency cannot silently override a declared intraday frequency. Named calendar/feed declarations for intraday detection are not supported yet; legacy intraday calls retain their existing behavior. Unknown calendars and malformed daily timestamps fail without modifying the input. `fill_gaps()` remains a separate elapsed-grid operation, not calendar-aware @@ -108,10 +114,11 @@ or feed declaration is present. `exclude_weekends=True` now follows NYSE holiday Omission follows NYSE, so callers requiring the old elapsed-day counts should pass `calendar="24/7"`. Intraday/weekly range planning retains its legacy elapsed-time behavior; named calendar/feed declarations for those frequencies are not supported by this planner yet. `calendar=None` disables -classification. Session-label dates retain their existing timezone's civil date. Explicit event-time -or bar-close declarations are rejected for daily planning rather than silently relabeled. Declare -the feed's `timezone` when a non-UTC label must survive storage's canonical UTC reads; the planner -restores that declared timezone before interpreting dates. +classification. Session-label dates use their declared timezone, or their existing timezone when +undeclared. Explicit event-time/bar-close/bar-side declarations map to authoritative session labels +before planning, including for provider-row selection during backfill. Out-of-session instants fail +clearly; original provider timestamps are retained in storage. Declare the feed's `timezone` when a +non-UTC label must survive storage's canonical UTC reads. If a provider's history limit cannot reach the stored dataset, the updater records `provider_history_limited` and skips gap filling across the unavailable interval. diff --git a/src/ml4t/data/sessions/assigner.py b/src/ml4t/data/sessions/assigner.py index 6d3eb5a..9b5ab08 100644 --- a/src/ml4t/data/sessions/assigner.py +++ b/src/ml4t/data/sessions/assigner.py @@ -33,6 +33,45 @@ def _datetime_scalar(value: object, name: str) -> datetime: } +def _daily_session_labels( + df: pl.DataFrame, assigner: SessionAssigner, timestamp_col: str = "timestamp" +) -> pl.Series: + if timestamp_col not in df.columns: + raise ValueError(f"DataFrame must have '{timestamp_col}' column") + feed = assigner.feed_spec + if feed.data_frequency is not None and str(feed.data_frequency).lower() not in { + "daily", + "1day", + "1d", + }: + raise ValueError("daily session projection requires a declared daily frequency") + if df.is_empty(): + return pl.Series(timestamp_col, [], dtype=pl.Date) + dtype = df.schema[timestamp_col] + if dtype != pl.Date and not isinstance(dtype, pl.Datetime): + raise TypeError(f"'{timestamp_col}' must contain Date or Datetime values") + if df[timestamp_col].null_count(): + raise ValueError(f"'{timestamp_col}' must not contain null timestamps") + if feed.timestamp_semantics in ( + TimestampSemantics.EVENT_TIME, + TimestampSemantics.BAR_CLOSE, + ) or ( + feed.timestamp_semantics != TimestampSemantics.SESSION_LABEL and feed.bar_label is not None + ): + assigned = assigner.assign_sessions( + df.select(pl.col(timestamp_col).alias("timestamp")), + outside_session="raise", + bar_frequency="daily", + ) + return assigned["session_date"].rename(timestamp_col) + labels = df[timestamp_col] + if isinstance(dtype, pl.Datetime): + if dtype.time_zone is not None and feed.timezone is not None: + labels = labels.dt.convert_time_zone(feed.timezone) + labels = labels.dt.date() + return labels + + class SessionAssigner: """Assigns session dates to trading data based on exchange calendars. diff --git a/src/ml4t/data/update_manager.py b/src/ml4t/data/update_manager.py index 3b8ab0a..b1a7d53 100644 --- a/src/ml4t/data/update_manager.py +++ b/src/ml4t/data/update_manager.py @@ -13,10 +13,10 @@ from ml4t.data.core.schemas import align_frames_for_concat from ml4t.data.providers.base import BaseProvider -from ml4t.data.sessions.assigner import SessionAssigner +from ml4t.data.sessions.assigner import SessionAssigner, _daily_session_labels from ml4t.data.storage.backend import StorageBackend, normalize_storage_metadata from ml4t.data.storage.metadata_tracker import MetadataTracker, UpdateRecord -from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec, TimestampSemantics +from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec logger = structlog.get_logger() @@ -136,34 +136,20 @@ def _feed_calendar( return feed, feed.resolve_calendar(calendar=calendar) @staticmethod - def _label_frame(df: pl.DataFrame, feed: FeedSpec) -> pl.DataFrame: - dtype = df.schema["timestamp"] - if dtype != pl.Date and not isinstance(dtype, pl.Datetime): - raise TypeError("'timestamp' must contain Date or Datetime values") - if df["timestamp"].null_count(): - raise ValueError("'timestamp' must not contain null timestamps") - if isinstance(dtype, pl.Datetime) and feed.timezone is not None: - timestamp = pl.col("timestamp") - if dtype.time_zone is None: - timestamp = timestamp.dt.replace_time_zone(feed.timezone) - else: - timestamp = timestamp.dt.convert_time_zone(feed.timezone) - return df.with_columns(timestamp) - return df + def _label_frame(df: pl.DataFrame, feed: FeedSpec, calendar: str | None) -> pl.DataFrame: + labels = _daily_session_labels(df, SessionAssigner(calendar, feed_spec=feed)) + return df.with_columns(labels.cast(pl.Datetime("us")).dt.replace_time_zone("UTC")) @staticmethod def _daily_ranges( df: pl.DataFrame, start: datetime, end: datetime, - feed: FeedSpec, calendar: str, tolerance_days: int = 0, ) -> list[dict[str, Any]]: if start.date() > end.date(): raise ValueError("start_date must not be after end_date") - if feed.timestamp_semantics not in (None, TimestampSemantics.SESSION_LABEL): - raise NotImplementedError("daily range planning requires session_label timestamps") assigner = SessionAssigner(calendar) observed = { value.date() if isinstance(value, datetime) else value @@ -222,12 +208,11 @@ def detect_gaps( if df.is_empty() or "timestamp" not in df.columns: return [] if frequency.lower() in {"daily", "1day"}: - df = self._label_frame(df, feed) + df = self._label_frame(df, feed, selected) return self._daily_ranges( df, _session_label(df["timestamp"].min()), _session_label(df["timestamp"].max()), - feed, selected, tolerance_days, ) @@ -312,7 +297,7 @@ def detect_gaps_in_storage( except KeyError: df = pl.DataFrame(schema={"timestamp": pl.Datetime("us", "UTC")}) return self._daily_ranges( - self._label_frame(df, feed), start_date, end_date, feed, selected + self._label_frame(df, feed, selected), start_date, end_date, selected ) start_date = _ensure_datetime(start_date) end_date = _ensure_datetime(end_date) @@ -677,7 +662,9 @@ def apply_strategy( start_date=gap["start"].date(), end_date=gap["end"].date() ) gap_rows = new_data.filter( - self.gap_detector._label_frame(new_data, feed)["timestamp"] + self.gap_detector._label_frame(new_data, feed, selected_calendar)[ + "timestamp" + ] .dt.date() .is_in([label.date() for label in expected.index]) ) diff --git a/src/ml4t/data/utils/gaps.py b/src/ml4t/data/utils/gaps.py index 2baf493..cae0b17 100644 --- a/src/ml4t/data/utils/gaps.py +++ b/src/ml4t/data/utils/gaps.py @@ -10,8 +10,8 @@ import polars as pl import structlog -from ml4t.data.sessions.assigner import SessionAssigner -from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec, TimestampSemantics +from ml4t.data.sessions.assigner import SessionAssigner, _daily_session_labels +from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec logger = structlog.get_logger() @@ -185,21 +185,12 @@ def _detect_daily_gaps( calendar: str, feed: FeedSpec, ) -> list[DataGap]: - assigner = SessionAssigner(calendar) - if timestamp_col not in df.columns: - raise ValueError(f"DataFrame must have '{timestamp_col}' column") + assigner = SessionAssigner(calendar, feed_spec=feed) + ordered = df.sort(timestamp_col) if timestamp_col in df.columns else df + labels = _daily_session_labels(ordered, assigner, timestamp_col).to_list() if df.is_empty(): return [] - dtype = df.schema[timestamp_col] - if dtype != pl.Date and not isinstance(dtype, pl.Datetime): - raise TypeError(f"'{timestamp_col}' must contain Date or Datetime values") - if df[timestamp_col].null_count(): - raise ValueError(f"'{timestamp_col}' must not contain null timestamps") - if feed.timestamp_semantics not in (None, TimestampSemantics.SESSION_LABEL): - raise NotImplementedError("daily gap detection requires session_label timestamps") - - timestamps = df.sort(timestamp_col)[timestamp_col].to_list() - labels = [value.date() if isinstance(value, datetime) else value for value in timestamps] + timestamps = ordered[timestamp_col].to_list() sessions = [ label.date() for label in assigner.calendar.schedule(start_date=labels[0], end_date=labels[-1]).index diff --git a/tests/test_daily_session_gaps.py b/tests/test_daily_session_gaps.py index 2199093..ee5751b 100644 --- a/tests/test_daily_session_gaps.py +++ b/tests/test_daily_session_gaps.py @@ -142,10 +142,9 @@ def test_date_session_labels_have_the_same_schedule_as_datetime_labels() -> None @pytest.mark.parametrize("semantics", ["event_time", "bar_close"]) -def test_event_semantics_are_not_silently_reinterpreted_as_daily_labels(semantics) -> None: +def test_explicit_daily_instants_are_mapped_without_creating_a_weekend_gap(semantics) -> None: frame = provider_frame("2025-07-11", "2025-07-14") - with pytest.raises(NotImplementedError, match="session_label"): - GapDetector().detect_gaps(frame, feed_spec={"timestamp_semantics": semantics}) + assert GapDetector().detect_gaps(frame, feed_spec={"timestamp_semantics": semantics}) == [] def test_null_timestamps_fail_without_input_mutation() -> None: diff --git a/tests/test_daily_timestamp_declarations.py b/tests/test_daily_timestamp_declarations.py index a779d04..66095f7 100644 --- a/tests/test_daily_timestamp_declarations.py +++ b/tests/test_daily_timestamp_declarations.py @@ -12,6 +12,7 @@ from ml4t.data.storage.flat import FlatStorage from ml4t.data.storage.hive import HiveStorage from ml4t.data.update_manager import GapDetector as RangeDetector +from ml4t.data.update_manager import IncrementalUpdater, UpdateStrategy from ml4t.data.utils.gaps import GapDetector from ml4t.specs import FeedSpec @@ -105,3 +106,75 @@ def test_declared_instants_support_the_public_custom_timestamp_column(): feed = FeedSpec(time_contract_version=2, calendar="NYSE", timestamp_semantics="event_time") found = GapDetector().detect_gaps(frame, timestamp_col="stamp", feed_spec=feed) assert sum(g.missing_periods for g in found) == 1 + + +@pytest.mark.parametrize("consumer", [GapDetector, RangeDetector]) +def test_declared_hourly_bars_cannot_silently_be_reinterpreted_as_daily_observations(consumer): + frame = provider_frame() + feed = FeedSpec( + time_contract_version=2, data_frequency="hourly", timestamp_semantics="event_time" + ) + with pytest.raises(ValueError, match="frequency"): + consumer().detect_gaps(frame, frequency="daily", feed_spec=feed) + + +@pytest.mark.parametrize("backend", [FlatStorage, HiveStorage]) +@pytest.mark.parametrize("semantics", ["event_time", "bar_close"]) +def test_actual_backfill_selects_provider_instants_by_their_missing_session( + tmp_path, backend, semantics +): + storage = backend(StorageConfig(base_path=tmp_path)) + complete = provider_frame() + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + bar_params={"frequency": "daily"}, + calendar="NYSE", + time_contract_version=2, + timestamp_semantics=semantics, + ) + key = "equities/daily/SPY" + storage.write( + complete.filter(pl.col("timestamp").dt.date() != date(2025, 7, 16)), + key, + metadata.model_dump(), + ) + result = IncrementalUpdater().apply_strategy(storage, key, complete, UpdateStrategy.BACKFILL) + assert result.success + assert result.rows_added == 1 + assert storage.read(key).collect().equals(complete) + + +@pytest.mark.parametrize("backend", [FlatStorage, HiveStorage]) +def test_actual_backfill_restores_tokyo_labels_without_retimestamping_provider_rows( + tmp_path, backend +): + storage = backend(StorageConfig(base_path=tmp_path)) + complete = provider_frame("2025-07-11", "2025-07-15").with_columns( + pl.col("timestamp").dt.date().cast(pl.Datetime("us")).dt.replace_time_zone("Asia/Tokyo") + ) + metadata = Metadata( + provider="synthetic", + symbol="SPY", + asset_class="equities", + bar_params={"frequency": "daily"}, + calendar="JPX", + time_contract_version=2, + timezone="Asia/Tokyo", + timestamp_semantics="session_label", + ) + key = "equities/daily/SPY" + storage.write( + complete.filter(pl.col("timestamp").dt.date() != date(2025, 7, 14)), + key, + metadata.model_dump(), + ) + result = IncrementalUpdater().apply_strategy(storage, key, complete, UpdateStrategy.BACKFILL) + assert result.success + assert result.rows_added == 1 + assert ( + storage.read(key) + .collect() + .equals(complete.with_columns(pl.col("timestamp").dt.convert_time_zone("UTC"))) + )