From 72e856d586e8398f1fdf4076875a41b60c7c291b Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 16:00:34 -0400 Subject: [PATCH] fix(grid): validate daily session labels before feature returns (#76) --- docs/user-guide/features.md | 43 +++++- src/ml4t/engineer/core/bar_grid.py | 97 +++++++++---- tests/test_api_session_grid.py | 220 +++++++++++++++++++++++++++++ 3 files changed, 330 insertions(+), 30 deletions(-) create mode 100644 tests/test_api_session_grid.py diff --git a/docs/user-guide/features.md b/docs/user-guide/features.md index d5c2160..7be1fe6 100644 --- a/docs/user-guide/features.md +++ b/docs/user-guide/features.md @@ -98,12 +98,43 @@ uses the explicit `calendar` argument, then declared feed metadata, then NYSE. E `calendar=None` disables schedule interpretation. A named `"24/7"` calendar instead defines expected bars and detects outages, including on weekends. -Calendar-aware execution requires a shared `FeedSpec(time_contract_version=2)` with an -explicit fixed intraday frequency and bar-grid policy. The initial implementation supports -time bars anchored at each tradable interval's open. It uses real holidays, early closes and -breaks from the selected schedule. Daily/session-label grids and session or elapsed-duration -windows are not supported by this validator; they fail rather than being treated as row counts. -Naive timestamps need a declared timezone. Aware timestamps already declare their timezone. +Calendar-aware execution requires a shared `FeedSpec(time_contract_version=2)` with declared +frequency and timestamp meaning. Fixed intraday time bars require an explicit bar-grid policy, +anchored at each tradable interval's open. The selected schedule supplies holidays, early closes +and breaks. Naive instants need a declared timezone; aware instants declare their timezone. + +Daily bars instead declare `timestamp_semantics="session_label"`, `data_frequency="1d"` and +`bar_type="time"`. Do not supply intraday anchor, label-side or partial-bar fields for session +labels. Date values and datetime representations identify the displayed session date, without +timezone conversion. Validation checks every scheduled session between each asset's first and +last label, including completely missing sessions. A named `24/7` schedule includes weekends; +an NYSE schedule does not. No prices are filled or rows inserted. A daily frequency alone does +not turn `event_time` instants into session labels. Session-count and elapsed-duration feature +windows remain unsupported; they fail rather than being treated as expected-bar counts. + + +```python +from datetime import date + +import polars as pl +from ml4t.engineer import compute_features +from ml4t.specs import FeedSpec + +close = [100.0, 103.0, 106.0, 109.0] +daily_prices = pl.DataFrame({ + "timestamp": [date(2025, 7, day) for day in (3, 7, 8, 9)], + "open": close, "high": close, "low": close, "close": close, "volume": 1.0, +}) +daily_feed = FeedSpec( + time_contract_version=2, calendar="NYSE", data_frequency="1d", + timestamp_semantics="session_label", bar_type="time", +) +checked = compute_features(daily_prices, [], feed_spec=daily_feed) +returns = checked.with_columns(pl.col("close").log().diff().alias("returns")) +features = compute_features(returns, [{"name": "sma", "params": {"period": 2}}], feed_spec=daily_feed) +assert features["sma"].to_list() == [None, 101.5, 104.5, 107.5] +assert features["timestamp"].equals(daily_prices["timestamp"]) +``` Validate prices before deriving returns. An empty feature list performs the same grid validation and chronological sorting without adding feature columns: diff --git a/src/ml4t/engineer/core/bar_grid.py b/src/ml4t/engineer/core/bar_grid.py index 5155e65..ebdfc5a 100644 --- a/src/ml4t/engineer/core/bar_grid.py +++ b/src/ml4t/engineer/core/bar_grid.py @@ -21,15 +21,15 @@ def validate_feature_grid( timestamp_col: str | None, group_cols: list[str], ) -> None: - """Bind shared v2 declarations to the fixed-intraday schedule checker.""" + """Validate shared v2 daily session labels or fixed intraday bar timestamps.""" + daily_labels = feed_spec.timestamp_semantics == TimestampSemantics.SESSION_LABEL required = ( "data_frequency", "timestamp_semantics", "bar_type", - "bar_grid_anchor", - "bar_label", - "partial_bar_policy", ) + if not daily_labels: + required += ("bar_grid_anchor", "bar_label", "partial_bar_policy") missing = [name for name in required if getattr(feed_spec, name) is None] if feed_spec.time_contract_version != 2 or missing or timestamp_col is None: raise ValueError( @@ -41,16 +41,10 @@ def validate_feature_grid( if feed_spec.timestamp_semantics not in { TimestampSemantics.BAR_CLOSE, TimestampSemantics.EVENT_TIME, + TimestampSemantics.SESSION_LABEL, }: - raise ValueError("Fixed intraday bar grids require event_time or bar_close timestamps") + raise ValueError("Bar grids require event_time, bar_close or session_label timestamps") schema = data.collect_schema() if isinstance(data, pl.LazyFrame) else data.schema - dtype = schema[timestamp_col] - timezone = feed_spec.timezone - if timezone is None and isinstance(dtype, pl.Datetime): - timezone = dtype.time_zone - if timezone is None: - raise ValueError("A declared bar grid requires timezone metadata for naive timestamps") - frequency = feed_spec.data_frequency if isinstance(frequency, str): import pandas as pd @@ -59,7 +53,7 @@ def validate_feature_grid( with warnings.catch_warnings(): warnings.simplefilter("error", FutureWarning) warnings.simplefilter("error", DeprecationWarning) - parsed = pd.Timedelta(frequency) + parsed = pd.Timedelta("1D" if frequency == "1d" else frequency) except (ValueError, TypeError, FutureWarning, DeprecationWarning) as exc: raise ValueError("declared grid data_frequency must be a fixed duration") from exc if not isinstance(parsed, pd.Timedelta) or parsed.value % 1000: @@ -67,17 +61,32 @@ def validate_feature_grid( frequency = parsed.to_pytimedelta() if not isinstance(frequency, timedelta): raise ValueError("declared grid data_frequency must be a duration string or timedelta") - validate_bar_grid( - data, - calendar=calendar, - frequency=frequency, - timezone=timezone, - timestamp_col=timestamp_col, - group_cols=group_cols, - anchor=cast(Literal["session_open"], feed_spec.bar_grid_anchor), - label_side=cast(Literal["left", "right"], feed_spec.bar_label), - include_partial=feed_spec.partial_bar_policy == "include", - ) + if daily_labels: + _validate_session_grid( + data, + calendar=calendar, + frequency=frequency, + timestamp_col=timestamp_col, + group_cols=group_cols, + ) + else: + dtype = schema[timestamp_col] + timezone = feed_spec.timezone + if timezone is None and isinstance(dtype, pl.Datetime): + timezone = dtype.time_zone + if timezone is None: + raise ValueError("A declared bar grid requires timezone metadata for naive timestamps") + validate_bar_grid( + data, + calendar=calendar, + frequency=frequency, + timezone=timezone, + timestamp_col=timestamp_col, + group_cols=group_cols, + anchor=cast(Literal["session_open"], feed_spec.bar_grid_anchor), + label_side=cast(Literal["left", "right"], feed_spec.bar_label), + include_partial=feed_spec.partial_bar_policy == "include", + ) if "returns" in schema: ordered = data.sort([*group_cols, timestamp_col]) missing_price = ~pl.col("close").cast(pl.Float64).is_finite().fill_null(False) @@ -96,6 +105,46 @@ def validate_feature_grid( ) +def _validate_session_grid( + data: pl.DataFrame | pl.LazyFrame, + *, + calendar: str, + frequency: timedelta, + timestamp_col: str, + group_cols: list[str], +) -> None: + if frequency != timedelta(days=1): + raise ValueError("Daily session labels require declared data_frequency='1d'") + from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + adapter = PandasMarketCalendar(calendar) + labels = data.select([*group_cols, pl.col(timestamp_col).dt.date()]) + frame = labels.collect() if isinstance(labels, pl.LazyFrame) else labels + if frame[timestamp_col].null_count(): + raise ValueError("declared daily grid contains null session labels") + groups = frame.partition_by(group_cols, as_dict=True) if group_cols else {(): frame} + for group, rows in groups.items(): + actual = sorted(rows[timestamp_col].to_list()) + if not actual: + continue + observed = set(actual) + if len(actual) != len(observed): + raise ValueError(f"duplicate session label in group {group!r}") + schedule = adapter._calendar.schedule(start_date=actual[0], end_date=actual[-1]) + expected = {label.date() for label in schedule.index} + missing = sorted(expected - observed) + if missing: + raise ValueError( + f"missing scheduled session in group {group!r} at {missing[0].isoformat()}; " + "validate daily prices before deriving returns, or explicitly select calendar=None" + ) + off_schedule = sorted(observed - expected) + if off_schedule: + raise ValueError( + f"off-schedule session label in group {group!r} at {off_schedule[0].isoformat()}" + ) + + def validate_bar_grid( data: pl.DataFrame | pl.LazyFrame, *, diff --git a/tests/test_api_session_grid.py b/tests/test_api_session_grid.py new file mode 100644 index 0000000..dc31abe --- /dev/null +++ b/tests/test_api_session_grid.py @@ -0,0 +1,220 @@ +from dataclasses import replace +from datetime import UTC, date, datetime, time, timedelta +from math import log +from statistics import stdev +from zoneinfo import ZoneInfo + +import polars as pl +import pytest +from ml4t.specs import UNSET_CALENDAR, FeedSpec, read_spec_payload, write_spec_payload + +from ml4t.engineer import compute_features +from ml4t.engineer.core.exceptions import DataSchemaError + +REQUEST = [ + {"name": "sma", "params": {"period": 2}}, + {"name": "realized_volatility", "params": {"period": 2, "annualize": False}}, +] + + +def prices(days, symbol="TEST"): + close = [100.0 + 3 * index for index in range(len(days))] + return pl.DataFrame( + { + "timestamp": days, + "symbol": symbol, + "open": close, + "high": close, + "low": close, + "close": close, + "volume": 1.0, + } + ) + + +def contract(calendar=UNSET_CALENDAR): + return FeedSpec( + time_contract_version=2, + calendar=calendar, + entity_col="symbol", + timestamp_semantics="session_label", + data_frequency="1d", + bar_type="time", + ) + + +def materialize(data): + return data.collect() if isinstance(data, pl.LazyFrame) else data + + +def derive_returns(data): + return data.with_columns(pl.col("close").log().diff().over("symbol").alias("returns")) + + +@pytest.mark.parametrize("lazy", [False, True]) +@pytest.mark.parametrize("representation", ["date", "naive", "UTC", "Asia/Tokyo"]) +def test_daily_price_returns_features_preserve_session_identifiers_across_closures( + lazy, representation +): + days = [date(2025, 7, day) for day in (3, 7, 8, 9)] + if representation != "date": + zone = None if representation == "naive" else ZoneInfo(representation) + days = [datetime.combine(day, time(0, 15), tzinfo=zone) for day in days] + frame = prices(days) + original = frame.clone() + data = frame.lazy() if lazy else frame + checked = compute_features(data.reverse(), [], feed_spec=contract()) + assert isinstance(checked, type(data)) + assert materialize(checked).equals(original) + result = materialize(compute_features(derive_returns(checked), REQUEST, feed_spec=contract())) + assert result.select(frame.columns).equals(frame) + assert result["sma"].to_list() == [None, 101.5, 104.5, 107.5] + assert result["realized_volatility"][2] == pytest.approx( + stdev([log(103 / 100), log(106 / 103)]) + ) + assert frame.equals(original) + + +@pytest.mark.parametrize("lazy", [False, True]) +def test_removed_daily_price_cannot_be_interpreted_as_a_one_bar_return(lazy): + complete = prices([date(2025, 7, day) for day in range(14, 19)]) + incomplete = complete.filter(pl.col("timestamp") != date(2025, 7, 16)) + original = incomplete.clone() + for frame, requests in [(incomplete, []), (derive_returns(incomplete), REQUEST)]: + data = frame.lazy() if lazy else frame + with pytest.raises(ValueError, match="missing scheduled session.*2025-07-16"): + compute_features(data, requests, feed_spec=contract("NYSE")) + assert incomplete.equals(original) + + +def test_complete_asset_does_not_hide_another_assets_missing_daily_price(): + complete = prices([date(2025, 7, day) for day in range(14, 19)], "COMPLETE") + missing = complete.with_columns(pl.lit("MISSING").alias("symbol")).filter( + pl.col("timestamp") != date(2025, 7, 16) + ) + with pytest.raises(ValueError, match="missing scheduled session.*MISSING.*2025-07-16"): + compute_features( + derive_returns(pl.concat([complete, missing])), REQUEST, feed_spec=contract() + ) + + +@pytest.mark.parametrize("frequency", ["1d", "1D", "24h", timedelta(days=1)]) +def test_complete_continuous_daily_grid_includes_weekends_without_equity_annualization(frequency): + frame = derive_returns(prices([date(2025, 7, day) for day in range(10, 15)])) + result = compute_features( + frame, REQUEST, feed_spec=replace(contract("24/7"), data_frequency=frequency) + ) + assert result.height == 5 + assert result["realized_volatility"][2] == pytest.approx( + stdev([log(103 / 100), log(106 / 103)]) + ) + with pytest.raises(ValueError, match="off-schedule"): + compute_features(frame, [], feed_spec=contract("NYSE")) + + +def test_reindexing_derived_daily_returns_does_not_hide_a_cross_session_return(): + complete = prices([date(2025, 7, day) for day in range(14, 19)]) + missing = complete.filter(pl.col("timestamp") != date(2025, 7, 16)) + unsafe = ( + complete.select("timestamp") + .join(derive_returns(missing), on="timestamp", how="left") + .with_columns(pl.col("symbol").fill_null("TEST")) + ) + original = unsafe.clone() + with pytest.raises(ValueError, match="Return adjacent to a missing price"): + compute_features(unsafe, REQUEST, feed_spec=contract()) + assert unsafe.equals(original) + safe_prices = ( + complete.select("timestamp") + .join(missing, on="timestamp", how="left") + .with_columns(pl.col("symbol").fill_null("TEST")) + ) + safe = compute_features(derive_returns(safe_prices), REQUEST, feed_spec=contract()) + assert safe.height == 5 + assert safe["returns"][3] is None + assert safe["realized_volatility"][3] is None + + +@pytest.mark.parametrize("count", [0, 1]) +@pytest.mark.parametrize("lazy", [False, True]) +def test_empty_and_single_daily_grids_preserve_declared_schema_and_shape(count, lazy): + frame = prices([date(2025, 7, 3)]).head(count) + data = frame.lazy() if lazy else frame + assert materialize(compute_features(data, [], feed_spec=contract())).equals(frame) + + +@pytest.mark.parametrize("suffix", [".json", ".yaml"]) +def test_serialized_daily_calendar_selection_does_not_turn_continuous_weekends_into_closures( + tmp_path, suffix +): + frame = prices([date(2025, 7, 11), date(2025, 7, 14)]) + path = tmp_path / f"feed{suffix}" + write_spec_payload(contract("24/7").to_dict(), path) + metadata = read_spec_payload(path) + with pytest.raises(ValueError, match="missing scheduled session.*2025-07-12"): + compute_features(frame, [], feed_spec=metadata) + assert compute_features(frame, [], feed_spec=metadata, calendar="NYSE").equals(frame) + assert compute_features(frame, [], feed_spec=metadata, calendar=None).equals(frame) + write_spec_payload(contract().to_dict(), path) + payload = read_spec_payload(path) + assert "calendar" not in payload + assert compute_features(frame, [], feed_spec=payload).equals(frame) + + +@pytest.mark.parametrize("kind", ["duplicate", "closed", "null"]) +def test_invalid_daily_session_labels_fail_without_changing_prices(kind): + days = [date(2025, 7, day) for day in (3, 7, 8)] + if kind == "null": + days[1] = None + else: + days.insert(1, days[0] if kind == "duplicate" else date(2025, 7, 4)) + frame = prices(days) + original = frame.clone() + with pytest.raises((ValueError, DataSchemaError), match="duplicate|off-schedule|null"): + compute_features(frame, [], feed_spec=contract()) + assert frame.equals(original) + + +def test_distinct_instants_with_the_same_session_identifier_are_duplicate_daily_bars(): + frame = prices([datetime(2025, 7, 3, hour, tzinfo=UTC) for hour in (10, 11)]) + with pytest.raises(ValueError, match="duplicate session label"): + compute_features(frame, [], feed_spec=contract()) + + +def test_overnight_exchange_uses_session_labels_not_the_open_instant_date(): + frame = prices([date(2025, 7, day) for day in (7, 8)]) + assert compute_features(frame, [], feed_spec=contract("CME_Equity")).equals(frame) + sunday = prices([date(2025, 7, 6)]) + with pytest.raises(ValueError, match="off-schedule session label"): + compute_features(sunday, [], feed_spec=contract("CME_Equity")) + + +@pytest.mark.parametrize("frequency", [None, "1h", "2d", timedelta(0)]) +def test_daily_labels_require_a_declared_single_session_frequency(frequency): + frame = prices([date(2025, 7, 3)]) + with pytest.raises(ValueError, match="frequency|declared bar grid"): + compute_features(frame, [], feed_spec=replace(contract(), data_frequency=frequency)) + + +def test_daily_frequency_does_not_reclassify_event_instants_as_session_labels(): + frame = prices([datetime(2025, 7, 3, tzinfo=UTC)]) + instant = replace( + contract(), + timestamp_semantics="event_time", + bar_grid_anchor="session_open", + bar_label="left", + partial_bar_policy="include", + timezone="UTC", + ) + with pytest.raises(ValueError, match="frequency.*below one day"): + compute_features(frame, [], feed_spec=instant) + + +def test_disabled_calendar_keeps_daily_observation_order_without_loading_a_schedule(monkeypatch): + import sys + + monkeypatch.setitem(sys.modules, "pandas_market_calendars", None) + frame = derive_returns(prices([date(2025, 7, day) for day in (8, 4, 20, 7)])) + result = compute_features(frame, REQUEST, feed_spec=contract(None)) + assert result.select(frame.columns).equals(frame) + assert result["sma"].to_list() == [None, 101.5, 104.5, 107.5]