diff --git a/AGENTS.md b/AGENTS.md index cf2b874..236272e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -61,3 +61,23 @@ uv build uv run python -c "import ml4t.engineer" uv run mkdocs build --strict ``` + +## Maintainer-directed change batching + +For our own fixes and agent work directed by the maintainer, use a bounded, coherent PR rather +than one PR per small defect or implementation step. One PR may close several related issues +when the changes serve the same user workflow, subsystem repair or compatibility objective and +can be reviewed and reverted together. Keep implementation, regressions, documentation and +follow-up repairs together; use focused commits and retain evidence for each defect. Do not +create separate issues or PRs merely for individual files, tests, assertions or platform variants. + +Split unrelated changes, independently risky or breaking changes, separate ownership boundaries, +and work that needs independent release or rollback decisions. A general maintenance label alone +does not make a batch coherent. Reuse or safely consolidate existing owned submissions instead of +opening overlapping PRs; preserve contributor attribution, recoverable history and other owners' +active work. Do not rewrite or close someone else's work merely to reduce the queue count. + +Run focused checks while iterating, then qualify the complete final revision with every applicable +required gate. Group related work before expensive CI; do not weaken checks, bypass review or +claim old-head checks qualify changed code. This maintainer-directed batching rule takes precedence +over a literal one-issue-per-PR reading of contributor intake guidance. diff --git a/docs/user-guide/features.md b/docs/user-guide/features.md index de5ce9c..a4f0271 100644 --- a/docs/user-guide/features.md +++ b/docs/user-guide/features.md @@ -69,8 +69,9 @@ Common asset columns such as `asset_id`, `symbol`, and `ticker` are detected aut and every rolling or lagged feature is evaluated independently for each asset. The examples explicitly select `calendar=None`: their periods count observations and input order is preserved, even when timestamps are irregular. Sort prices yourself before -deriving returns if chronological order is intended. Without timestamps, also pass -`assume_sorted=True` to assert that the supplied row order is correct. +deriving returns if chronological order is intended. Row mode does not require timestamps +or an additional `assume_sorted=True` assertion. A disabled calendar in shared feed metadata +has the same behavior, even when the feed's timestamp column is absent. Use `output` when requesting more than one configuration of the same feature: @@ -87,23 +88,52 @@ raise `ValueError` before feature execution. ### Calendar and bar grid The calendar-aware default targets the next minor version (0.2), separate from the 0.1.7 patch. -Development and qualification currently pin the merged Specs v2 implementation at -`701bde68744035050ede386efce498234c661c21`. The same immutable Git dependency is declared -in built package metadata; published Specs 0.1.5 does not provide this API. Before a PyPI -release, replace that source dependency with the verified published Specs version containing -the v2 contract and repeat installed-package qualification. +The shared temporal contract requires published `ml4t-specs>=0.2.0,<0.3.0`, also declared +in built package metadata. Specs 0.1.5 does not provide this API. Engineer's development +branch uses the published dependency; the temporal migration is not part of the 0.1.7 patch. Calendar selection uses the explicit `calendar` argument, then declared feed metadata, then NYSE. Explicit `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. Temporal feature windows below use the +same completeness checks; choosing a unit does not permit missing scheduled observations. + + +```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"]) +``` `CryptoCalendar` also changes in the next minor version. Install the `calendars` extra to use its authoritative `24/7` schedule. It enumerates UTC daily session opens, not @@ -174,6 +204,56 @@ This does not change legacy Specs bridges or the numerical feature kernels. Neither means sessions or elapsed hours. A shortened final bar still counts as one expected bar when `partial_bar_policy="include"`. Annualization remains a separate convention: for example, `realized_volatility(trading_periods=252)` retains its documented numerical default. + +### Session and elapsed feature windows + +`sma`, `stddev` and `realized_volatility` accept `window_unit="session"` or `"elapsed"`. +They retain original bars and row counts. They do not first aggregate to session closes. +For sessions, `params.period=N` includes the current session's observations through the +current row and the preceding N-1 scheduled sessions. Earlier rows of the first N-1 +observed sessions are null. July 4 holidays and lunch breaks do not create extra sessions. +A one-session average of intraday prices100/110 is therefore100 at the first row and105 +at the second row, not the session's eventual closing price110 at both rows. + +Elapsed periods use a positive duration such as `params.period="2h"` and the physical +UTC interval `(timestamp - duration, timestamp]`, including through DST changes. +Outputs remain null until observed coverage spans the requested duration. A half-hour +final bar can put three observations inside two elapsed hours. Daily `session_label` +identifiers cannot define elapsed windows; they do support session-count windows. +Windows with insufficient samples for the numerical statistic also remain null. +Annualization parameters and all other supported feature parameters retain their +numerical definitions. Recursive EMA/RSI and other unqualified window definitions +reject temporal units explicitly rather than silently resetting or reseeding. + + +```python +from datetime import date, datetime, timedelta, timezone + +import polars as pl +from ml4t.engineer import compute_features +from ml4t.specs import FeedSpec + +daily = FeedSpec(time_contract_version=2, calendar="NYSE", data_frequency="1d", + timestamp_semantics="session_label", bar_type="time") +values = [100.0, 101.0, 104.0] +prices = pl.DataFrame({ + "timestamp": [date(2025, 7, day) for day in (2, 3, 7)], + "open": values, "high": values, "low": values, "close": values, "volume": 1.0, +}) +result = compute_features(prices, [{"name": "sma", "params": {"period": 2}}], + feed_spec=daily, window_unit="session") +assert result["sma"].to_list() == [None, 100.5, 102.5] + +hourly = FeedSpec(time_contract_version=2, calendar="24/7", data_frequency="1h", + timestamp_semantics="event_time", timezone="UTC", bar_type="time", + bar_grid_anchor="session_open", bar_label="left", partial_bar_policy="include") +prices = prices.with_columns(pl.Series("timestamp", [ + datetime(2025, 7, 5, tzinfo=timezone.utc) + timedelta(hours=i) for i in range(3) +])) +result = compute_features(prices, [{"name": "sma", "params": {"period": "2h"}}], + feed_spec=hourly, window_unit="elapsed") +assert result["sma"].to_list() == [None, None, 102.5] +``` No calendar-derived annualization factor is silently substituted. ### Equity schedule helpers @@ -501,7 +581,7 @@ Most features expect a DataFrame with standardized column names (lowercase): | Column | Type | Required By | |--------|------|-------------| -| `timestamp` | datetime | Temporal ordering, unless `assume_sorted=True` | +| `timestamp` | datetime | Required for calendar-aware grids; optional in observation-row mode | | `asset_id` | string | Optional panel grouping, auto-detected when present | | `open` | float | OHLCV, OHLC features | | `high` | float | OHLCV, OHLC, HLC, HL features | diff --git a/docs/user-guide/labeling.md b/docs/user-guide/labeling.md index da9d314..8d866c1 100644 --- a/docs/user-guide/labeling.md +++ b/docs/user-guide/labeling.md @@ -414,6 +414,12 @@ result = apply_meta_model( For futures and other instruments with defined trading sessions, calendar-aware labeling respects session boundaries. A label that would span an overnight gap is handled correctly. +In the next minor version, `calendar_aware_labels` uses the shared temporal contract. +An explicit `calendar` argument wins over a declared `LabelingConfig.calendar`, then +`FeedSpec` metadata, then NYSE. Omitting a selector differs from explicit `None`. +Config JSON/YAML output preserves that distinction and emits only explicitly declared +column overrides, so a round trip does not replace feed column mappings with defaults. + ```python from ml4t.engineer.labeling import calendar_aware_labels @@ -433,6 +439,112 @@ former pause because trading is continuous until 4:00 PM. See the [CME exchange notice](https://www.cmegroup.com/notices/electronic-trading/2021/06/20210621.html) for the affected equity-index contracts. + + +### Observation rows, expected bars, sessions and elapsed durations + +Use `calendar=None` for observation-row labeling, including timestamp-indexed data: + +```python +row_config = LabelingConfig.triple_barrier( + upper_barrier=0.02, + lower_barrier=0.02, + max_holding_period=10, + calendar=None, + window_unit="observation", +) +labels = calendar_aware_labels(df, row_config) +``` + +Row mode retains input order, processes each asset independently, does not require +timestamps and never loads a schedule provider. Its `label_bars` and `label_time` are +observation counts and per-asset row offsets; `label_duration` is unset. Timestamp +spacing and weekend placement do not alter its numerical labels. Duration holding +periods are incompatible with this mode. + +Named calendars segment authoritative sessions and breaks. New shared-feed calls and +omitted selectors default count windows to `"expected_bar"`; an undeclared or incomplete +grid fails before labeling. Supply the complete shared grid +declarations described in the [feature guide](features.md#calendar-and-bar-grid) to reject +missing scheduled bars before labeling. Named `"24/7"` still has daily schedule boundaries +and is not equivalent to row mode. Duration holding periods select `"elapsed"` and remain truncated at session +boundaries, rather than counting sessions or observed dates. + +Pass `feed_spec=FeedSpec(...)` directly or through `LabelingConfig.feed_spec`. Shared +`timestamp_semantics` must distinguish intraday instants from `session_label` identifiers. +Naive feed timestamps require an explicit timezone, and ambiguous/nonexistent local +times fail. Right-labeled/bar-close observations at a break boundary end the preceding +tradable interval. Input timestamp representations are retained in the output. +Instant horizons and `label_duration` use physical elapsed time, including across +daylight-saving transitions. Shared `session_label` identifiers do not identify instants; +their `label_time` retains the session identifier and `label_duration` is unset. + +Legacy calls without FeedSpec keep naive-UTC handling. Custom protocol objects and +explicit `calendar="auto"` remain compatibility modes without shared feed metadata. +An explicit legacy calendar selector without FeedSpec retains observation-count horizons. +New calls can also select `window_unit="observation"` explicitly to count received rows +within each authoritative segment, without claiming scheduled-bar completeness. +Auto-gap segmentation does not establish market holidays, session completeness or a +bar grid. Standalone labeling kernels retain their documented row-count or elapsed-time +definitions and chronological sorting; they do not acquire calendar interpretation from +timestamps. Use this wrapper when selecting the shared calendar/row contract. +Standalone kernels reject configs declaring calendar/feed/window semantics instead +of silently ignoring those declarations. +Standalone kernels retain date-valued label times and calendar-day differences in +`label_duration` for legacy `Date` inputs. These standalone calls do not validate +the dates against an exchange schedule. + +For `window_unit="session"`, an integer `max_holding_period=N` expires at the final +bar of the Nth scheduled session after the entry session. A positive integral count +column can select N per event. July 3, 2025 on NYSE therefore reaches July 7 for N=1, +not the July 4 holiday. Intraday entries reach the following session's last bar, +including an early close; lunch and overnight breaks do not truncate this horizon. +Profit, stop and trailing barriers still use the existing triple-barrier calculation. +The full declared grid is checked before any labels are computed, including whole-session +outages. No prices are filled or rows inserted. + +With `partial_bar_policy="exclude"`, a shortened session can have no bars. It still +counts as a scheduled session. The exit uses the last available bar no later than the +deadline, even if that is the entry itself with zero holding bars. The calculation +does not extend the deadline into a later session to obtain another observation. + +Daily input uses `FeedSpec(time_contract_version=2, calendar="NYSE", +timestamp_semantics="session_label", data_frequency="1d", bar_type="time")`. +Do not supply intraday anchor/label/partial-bar fields. Date-valued and midnight datetime +identifiers keep their displayed session date; they are not converted to UTC instants. +Daily observation and expected-bar horizons scan across session rows rather than +truncating every label to its entry row. Elapsed horizons reject these identifiers. + +As with existing count horizons, a deadline beyond the available dataset is censored +at the last available bar. `label_bars` reports the actual observed holding length and +`label_time` the actual exit, not an inferred future observation. These changes target +the next 0.2 development release, not the historical 0.1.7 patch. + + +```python +from datetime import date +from io import BytesIO + +import polars as pl +from ml4t.specs import FeedSpec +from ml4t.engineer.config import LabelingConfig +from ml4t.engineer.labeling import calendar_aware_labels + +prices = pl.DataFrame({ + "timestamp": [date(2025, 7, day) for day in (3, 7, 8)], + "close": [100.0, 101.0, 103.0], +}) +config = LabelingConfig.triple_barrier( + upper_barrier=None, lower_barrier=None, max_holding_period=1, window_unit="session", + feed_spec=FeedSpec(time_contract_version=2, calendar="NYSE", data_frequency="1d", + timestamp_semantics="session_label", bar_type="time"), +) +labels = calendar_aware_labels(prices, config) +assert labels["label_time"].to_list() == [date(2025, 7, 7), date(2025, 7, 8), date(2025, 7, 8)] +assert labels["label_bars"].to_list() == [1, 1, 0] +labels.write_parquet(BytesIO()) +labels.head(0).write_parquet(BytesIO()) +``` ## Sample Weighting AFML Chapter 4 addresses the problem of overlapping labels creating correlated samples. The library provides the full toolkit: diff --git a/pyproject.toml b/pyproject.toml index 0dd9f84..93321da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,10 +8,6 @@ build-backend = "hatchling.build" [tool.hatch.version] source = "vcs" -[tool.hatch.metadata] -# Qualification pin until the shared time-contract v2 API is published on PyPI. -allow-direct-references = true - [tool.hatch.build.targets.wheel] packages = ["src/ml4t"] namespaces = true @@ -76,7 +72,7 @@ requires-python = ">=3.12,<3.15" # Core dependencies dependencies = [ - "ml4t-specs @ git+https://github.com/ml4t/specs.git@701bde68744035050ede386efce498234c661c21", + "ml4t-specs>=0.2.0,<0.3.0", # Data processing (Polars-first) "polars>=0.20.0", diff --git a/src/ml4t/engineer/api.py b/src/ml4t/engineer/api.py index 8aed198..8bec54d 100644 --- a/src/ml4t/engineer/api.py +++ b/src/ml4t/engineer/api.py @@ -21,7 +21,7 @@ from typing import Any, TypedDict import polars as pl -from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec, WindowUnit +from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec, TimestampSemantics, WindowUnit try: import yaml @@ -112,18 +112,21 @@ def compute_features( Calendar-aware input is sorted by group and timestamp. Row mode retains input order within every group. assume_sorted : bool, default False - Permit row-order execution when no timestamp column exists. This is an - explicit assertion by the caller that rows are already in the required - temporal order. + Retained legacy ordering argument. Explicit calendar=None already selects + input observation order and needs no timestamp or additional sort assertion. + Calendar-aware execution always requires a declared datetime grid. calendar : str | None, optional Explicit selection overrides declared feed metadata, then defaults to NYSE. Explicit None selects observation rows in input order, without schedule checks. feed_spec : FeedSpec | Mapping | None Shared input schema and temporal declarations. Calendar-aware execution requires - version 2, fixed intraday frequency, timestamp semantics and an explicit bar grid. + version 2, declared frequency and timestamp semantics. Intraday feeds also + require an explicit bar grid; daily feeds identify session labels. window_unit : WindowUnit | str | None - Defaults to observation in row mode or expected_bar with a calendar. Session - and elapsed windows are not implemented by these row-based numerical kernels. + Defaults to observation in row mode or expected_bar with a calendar. For sma, + stddev and realized_volatility, session periods count scheduled sessions and + elapsed periods are positive duration strings/timedeltas over original bars. + Recursive and multi-window features reject these temporal units explicitly. Returns ------- @@ -182,7 +185,7 @@ def compute_features( - Calendar-aware calls require a declared complete grid before any feature runs. Empty feature lists can validate and sort prices before returns are derived. - Explicit ``calendar=None`` preserves input row order without schedule checks. - Without a timestamp, callers must still set ``assume_sorted=True`` explicitly. + Row mode does not require a timestamp or an additional sort assertion. """ from ml4t.engineer.core.bar_grid import validate_feature_grid from ml4t.engineer.core.schemas import validate_ohlcv_schema @@ -212,13 +215,26 @@ def compute_features( ) if resolved_calendar is None and unit != WindowUnit.OBSERVATION: raise ValueError("calendar=None requires window_unit='observation'") - if resolved_calendar is not None and unit != WindowUnit.EXPECTED_BAR: - raise ValueError("Calendar-aware features support only window_unit='expected_bar'") + if resolved_calendar is not None and unit == WindowUnit.OBSERVATION: + raise ValueError("Calendar-aware features require expected_bar, session or elapsed windows") + temporal = unit in {WindowUnit.SESSION, WindowUnit.ELAPSED} + if ( + unit == WindowUnit.ELAPSED + and contract.timestamp_semantics == TimestampSemantics.SESSION_LABEL + ): + raise ValueError("session_label identifiers cannot define elapsed feature windows") + if temporal: + from ml4t.engineer.core.temporal_windows import temporal_period + + for spec in feature_specs: + temporal_period(spec["name"], spec["params"], unit) if not feature_specs and resolved_calendar is None: return data if feed_spec is not None: - if timestamp_col is None: + if timestamp_col is None and ( + resolved_calendar is not None or contract.timestamp_col in schema + ): timestamp_col = contract.timestamp_col if group_col is None and contract.entity_col is not None: group_col = ( @@ -231,7 +247,7 @@ def compute_features( data=data, group_col=group_col, timestamp_col=timestamp_col, - assume_sorted=assume_sorted, + assume_sorted=assume_sorted or resolved_calendar is None, ) # Resolve dependencies and get execution order @@ -253,13 +269,29 @@ def compute_features( # Execute features in order result = data.sort(sort_cols) if resolved_calendar is not None and sort_cols else data for spec in execution_order: - result = _execute_feature( - result, - feature_name=spec["name"], - params=spec["params"], - output_name=spec["output"], - group_cols=group_cols, - ) + if temporal: + from ml4t.engineer.core.temporal_windows import execute_temporal_feature + + assert resolved_timestamp_col is not None and resolved_calendar is not None + result = execute_temporal_feature( + result, + feature_name=spec["name"], + params=spec["params"], + output_name=spec["output"], + group_cols=group_cols, + timestamp_col=resolved_timestamp_col, + feed=contract, + calendar=resolved_calendar, + period=temporal_period(spec["name"], spec["params"], unit), + ) + else: + result = _execute_feature( + result, + feature_name=spec["name"], + params=spec["params"], + output_name=spec["output"], + group_cols=group_cols, + ) return result diff --git a/src/ml4t/engineer/config/labeling.py b/src/ml4t/engineer/config/labeling.py index b67941b..b8197f2 100644 --- a/src/ml4t/engineer/config/labeling.py +++ b/src/ml4t/engineer/config/labeling.py @@ -28,7 +28,16 @@ from datetime import timedelta from typing import Any, Literal -from pydantic import AliasChoices, Field, field_serializer, field_validator +from ml4t.specs import FeedSpec, WindowUnit +from pydantic import ( + AliasChoices, + Field, + SerializationInfo, + SerializerFunctionWrapHandler, + field_serializer, + field_validator, + model_serializer, +) from ml4t.engineer.config.base import ( BaseConfig, @@ -145,6 +154,49 @@ class LabelingConfig(BaseConfig): validation_alias=AliasChoices("data_contract", "contract"), description="Optional shared dataframe column contract for cross-library workflows", ) + calendar: str | None = Field( + None, + description="Declared calendar selector; omitted inherits feed metadata, explicit null disables", + ) + feed_spec: dict[str, Any] | None = Field( + None, description="Shared FeedSpec declarations for calendar-aware labeling" + ) + window_unit: WindowUnit | None = Field( + None, + description="Observation, expected_bar, session or elapsed; consumer validates support", + ) + + def __eq__(self, other: object) -> bool: + if not isinstance(other, LabelingConfig): + return NotImplemented + return self.to_dict() == other.to_dict() + + @field_validator("calendar", mode="before") + @classmethod + def validate_calendar(cls, value: Any) -> str | None: + return FeedSpec(time_contract_version=2).resolve_calendar(calendar=value) + + @field_validator("feed_spec", mode="before") + @classmethod + def validate_feed_spec(cls, value: Any) -> dict[str, Any] | None: + return None if value is None else FeedSpec.from_any(value).to_dict() + + @model_serializer(mode="wrap") + def serialize_declarations( + self, handler: SerializerFunctionWrapHandler, info: SerializationInfo + ) -> dict[str, Any]: + payload = handler(self) + for name in ("calendar", "price_col", "timestamp_col", "group_col", "data_contract"): + if name not in self.model_fields_set: + payload.pop(name, None) + if ( + "calendar" in self.model_fields_set + and self.calendar is None + and (info.include is None or "calendar" in info.include) + and (info.exclude is None or "calendar" not in info.exclude) + ): + payload["calendar"] = None + return payload # Triple barrier parameters upper_barrier: float | str | None = Field( diff --git a/src/ml4t/engineer/core/bar_grid.py b/src/ml4t/engineer/core/bar_grid.py index 5155e65..f8ac87d 100644 --- a/src/ml4t/engineer/core/bar_grid.py +++ b/src/ml4t/engineer/core/bar_grid.py @@ -6,6 +6,8 @@ from typing import TYPE_CHECKING, Literal, cast from zoneinfo import ZoneInfo +import numpy as np +import numpy.typing as npt import polars as pl from ml4t.specs import FeedSpec, TimestampSemantics @@ -13,7 +15,7 @@ from ml4t.engineer.labeling.calendar import PandasMarketCalendar -def validate_feature_grid( +def validate_declared_grid( data: pl.DataFrame | pl.LazyFrame, *, feed_spec: FeedSpec, @@ -21,15 +23,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 +43,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 +55,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 +63,51 @@ 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( + 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", + ) + + +def validate_feature_grid( + data: pl.DataFrame | pl.LazyFrame, + *, + feed_spec: FeedSpec, + calendar: str, + timestamp_col: str | None, + group_cols: list[str], +) -> None: + """Validate the declared grid and supplied returns against adjacent prices.""" + validate_declared_grid( data, + feed_spec=feed_spec, 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", ) + schema = data.collect_schema() if isinstance(data, pl.LazyFrame) else data.schema 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 +126,90 @@ 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 scheduled_session_indices( + data: pl.DataFrame, + *, + feed_spec: FeedSpec, + calendar: str, + timestamp_col: str, +) -> npt.NDArray[np.int64]: + """Map a validated, sorted single-asset grid to actual schedule ordinals.""" + from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + if data.is_empty(): + return np.empty(0, dtype=np.int64) + labels = feed_spec.timestamp_semantics == TimestampSemantics.SESSION_LABEL + values = data[timestamp_col] + if labels: + days = values.dt.date().to_list() + else: + dtype = values.dtype + if not isinstance(dtype, pl.Datetime): + raise ValueError("Session windows require datetime instants or declared session labels") + if dtype.time_zone is None: + values = values.dt.replace_time_zone( + feed_spec.timezone, ambiguous="raise", non_existent="raise" + ) + values = values.dt.convert_time_zone("UTC") + days = values.dt.date().to_list() + adapter = PandasMarketCalendar(calendar) + schedule = adapter._calendar.schedule( + start_date=min(days) - timedelta(days=7), end_date=max(days) + timedelta(days=7) + ) + if labels: + indices = {label.date(): index for index, label in enumerate(schedule.index)} + return np.array([indices[day] for day in days], dtype=np.int64) + timestamps = values.dt.epoch("us").to_numpy() + opened = np.array([stamp.value // 1000 for stamp in schedule["market_open"]], dtype=np.int64) + side = ( + "left" + if feed_spec.bar_label == "right" + or feed_spec.timestamp_semantics == TimestampSemantics.BAR_CLOSE + else "right" + ) + return np.asarray(np.searchsorted(opened, timestamps, side=side) - 1, dtype=np.int64) + + def validate_bar_grid( data: pl.DataFrame | pl.LazyFrame, *, diff --git a/src/ml4t/engineer/core/calendars/AGENTS.md b/src/ml4t/engineer/core/calendars/AGENTS.md index 5dc3ca9..39a283d 100644 --- a/src/ml4t/engineer/core/calendars/AGENTS.md +++ b/src/ml4t/engineer/core/calendars/AGENTS.md @@ -20,3 +20,5 @@ Calendar utilities for session-aware workflows. This does not imply that all labeling APIs already consume the shared temporal contract. - Calendar-free feature calculations use `compute_features(..., calendar=None)`. A named continuous calendar is not calendar-free execution. +- `CryptoCalendar` uses UTC daily sessions, not hourly observation points. Maintenance + reopens do not create extra sessions; keep membership and daily boundary methods distinct. diff --git a/src/ml4t/engineer/core/calendars/__init__.py b/src/ml4t/engineer/core/calendars/__init__.py index 8bb72b0..54f0b77 100644 --- a/src/ml4t/engineer/core/calendars/__init__.py +++ b/src/ml4t/engineer/core/calendars/__init__.py @@ -1,11 +1,10 @@ """ -Exchange calendars for time-aware feature engineering. +Exchange schedule helpers for membership, boundaries and dataframe annotations. -This module provides calendar-aware functionality for: -- Rolling windows over trading sessions (not calendar days) -- Holiday-aware feature calculations -- Market-hours filtering -- Cross-validation split timing +These helpers do not implement feature windows or cross-validation. CryptoCalendar +enumerates UTC daily sessions; recurring maintenance does not create extra sessions. +Shared calendar/grid consumers use labeling.calendar.PandasMarketCalendar and +FeedSpec declarations. """ from .base import ExchangeCalendar diff --git a/src/ml4t/engineer/core/temporal_windows.py b/src/ml4t/engineer/core/temporal_windows.py new file mode 100644 index 0000000..2199c15 --- /dev/null +++ b/src/ml4t/engineer/core/temporal_windows.py @@ -0,0 +1,155 @@ +"""Causal temporal bounds for finite-window registered features.""" + +import inspect +from datetime import timedelta +from typing import Any + +import numpy as np +import numpy.typing as npt +import polars as pl +from ml4t.specs import FeedSpec, TimestampSemantics, WindowUnit + +from ml4t.engineer.core.bar_grid import scheduled_session_indices +from ml4t.engineer.core.registry import get_registry + +_MINIMUM_PERIODS = {"sma": 1, "stddev": 1, "realized_volatility": 2} + + +def temporal_period(feature_name: str, params: dict[str, Any], unit: WindowUnit) -> int | timedelta: + """Validate temporal support without executing any feature.""" + if feature_name not in _MINIMUM_PERIODS: + raise ValueError( + f"Feature {feature_name!r} does not support temporal windows; " + f"supported finite-window features are {sorted(_MINIMUM_PERIODS)}. " + "Recursive and multi-window definitions are not converted to row counts." + ) + metadata = get_registry().get(feature_name) + assert metadata is not None + period = params.get( + "period", + metadata.parameters.get( + "period", inspect.signature(metadata.func).parameters["period"].default + ), + ) + if unit == WindowUnit.SESSION: + if isinstance(period, bool) or not isinstance(period, int) or period <= 0: + raise ValueError("Session feature period must be a positive integer session count") + return period + from ml4t.engineer.labeling.utils import parse_duration + + if isinstance(period, str): + try: + period = parse_duration(period) + except ValueError as exc: + raise ValueError("Elapsed feature period must be a positive fixed duration") from exc + if not isinstance(period, timedelta) or period <= timedelta(0): + raise ValueError( + "Elapsed feature period must be a positive fixed duration, not a row count" + ) + return period + + +def _window_sizes( + data: pl.DataFrame, + period: int | timedelta, + feed: FeedSpec, + calendar: str, + timestamp_col: str, +) -> tuple[npt.NDArray[np.int64], npt.NDArray[np.bool_]]: + positions = np.arange(data.height, dtype=np.int64) + if data.is_empty(): + return positions, np.empty(0, dtype=np.bool_) + if isinstance(period, int): + sessions = scheduled_session_indices( + data, feed_spec=feed, calendar=calendar, timestamp_col=timestamp_col + ) + # Counts above observed coverage cannot produce a complete window. Bound + # before NumPy integer arithmetic so arbitrarily large Python counts are safe. + if period > int(sessions[-1] - sessions[0]) + 1: + return positions + 1, np.zeros(data.height, dtype=np.bool_) + starts = np.searchsorted(sessions, sessions - period + 1, side="left") + complete = sessions - sessions[0] >= period - 1 + else: + if feed.timestamp_semantics == TimestampSemantics.SESSION_LABEL: + raise ValueError("session_label identifiers cannot define elapsed feature windows") + values = data[timestamp_col] + dtype = values.dtype + assert isinstance(dtype, pl.Datetime) + if dtype.time_zone is None: + assert feed.timezone is not None + values = values.dt.replace_time_zone( + feed.timezone, ambiguous="raise", non_existent="raise" + ) + timestamps = values.dt.convert_time_zone("UTC").dt.epoch("us").to_numpy() + duration = (period.days * 86400 + period.seconds) * 1_000_000 + period.microseconds + if duration > int(timestamps[-1] - timestamps[0]): + return positions + 1, np.zeros(data.height, dtype=np.bool_) + starts = np.searchsorted(timestamps, timestamps - duration, side="right") + complete = timestamps - timestamps[0] >= duration + return np.asarray(positions - starts + 1, dtype=np.int64), complete + + +def execute_temporal_feature( + data: pl.DataFrame | pl.LazyFrame, + *, + feature_name: str, + params: dict[str, Any], + output_name: str, + group_cols: list[str], + timestamp_col: str, + feed: FeedSpec, + calendar: str, + period: int | timedelta, +) -> pl.DataFrame | pl.LazyFrame: + """Reuse finite row kernels once per distinct causal window size, without resampling.""" + from ml4t.engineer.api import _execute_feature + + minimum = _MINIMUM_PERIODS[feature_name] + template = _execute_feature( + data.head(0), feature_name, {**params, "period": minimum}, output_name, [] + ) + schema = template.collect_schema() if isinstance(template, pl.LazyFrame) else template.schema + if isinstance(data, pl.LazyFrame): + + def apply(frame: pl.DataFrame) -> pl.DataFrame: + result = execute_temporal_feature( + frame, + feature_name=feature_name, + params=params, + output_name=output_name, + group_cols=group_cols, + timestamp_col=timestamp_col, + feed=feed, + calendar=calendar, + period=period, + ) + assert isinstance(result, pl.DataFrame) + return result + + return data.map_batches( + apply, + schema=schema, + predicate_pushdown=False, + projection_pushdown=False, + slice_pushdown=False, + streamable=False, + ) + output_columns = [name for name in schema if name not in data.columns] + frames = data.partition_by(group_cols, maintain_order=True) if group_cols else [data] + results = [] + for frame in frames or [data]: + sizes, complete = _window_sizes(frame, period, feed, calendar, timestamp_col) + outputs = { + name: pl.Series(name, [None] * frame.height, dtype=schema[name]) + for name in output_columns + } + for size in np.unique(sizes[complete & (sizes >= minimum)]): + computed = _execute_feature( + frame, feature_name, {**params, "period": int(size)}, output_name, [] + ) + assert isinstance(computed, pl.DataFrame) + indices = np.flatnonzero(complete & (sizes == size)) + for name in output_columns: + outputs[name] = outputs[name].scatter(indices, computed[name].gather(indices)) + results.append(frame.with_columns(list(outputs.values()))) + return pl.concat(results, how="vertical") diff --git a/src/ml4t/engineer/features/AGENTS.md b/src/ml4t/engineer/features/AGENTS.md index 85ecb6f..6435dea 100644 --- a/src/ml4t/engineer/features/AGENTS.md +++ b/src/ml4t/engineer/features/AGENTS.md @@ -55,9 +55,19 @@ result = compute_features( {"name": "sma", "params": {"period": 50}}, {"name": "bollinger_bands", "params": {"period": 20, "std_dev": 2.5}}, ], + calendar=None, ) ``` +This example explicitly counts observation rows. Omitting `calendar` selects shared +feed metadata, then NYSE, and requires a declared complete daily or intraday grid. Use +`FeedSpec` declarations for expected bars; do not infer frequency or holidays from gaps. +Numerical kernels keep their documented annualization conventions independently of +calendar selection. `sma`, `stddev` and `realized_volatility` also support session/elapsed +windows through the public API. Reuse their existing finite kernels over causal original +bars, with schedule ordinals for sessions and open-left physical UTC intervals for elapsed +durations. Do not resample, infer missing sessions or reinterpret recursive indicators. + ## Trust Signals - 120 total features across 11 categories diff --git a/src/ml4t/engineer/labeling/AGENTS.md b/src/ml4t/engineer/labeling/AGENTS.md index 057e8b5..94b2c09 100644 --- a/src/ml4t/engineer/labeling/AGENTS.md +++ b/src/ml4t/engineer/labeling/AGENTS.md @@ -69,3 +69,26 @@ Uniqueness score per label. Legacy module notes: - `core.py`, `barriers.py`, and `barrier_utils.py` are removed compatibility paths that raise import-time migration errors. + +## Temporal contract boundary + +`calendar_aware_labels` consumes shared Specs declarations. Use canonical `calendar`: +explicit argument, declared config selector, feed metadata, then NYSE. Explicit None is +observation-row mode without timestamp sorting or provider loading. Never turn that +opt-out into a continuous schedule. Config serialization must preserve declaration presence. + +Named schedules segment actual sessions/breaks. Observation counts are not expected-bar +counts; expected-bar and session modes validate the declared grid before labeling. +Session-count horizons expire at the last available bar of the Nth following scheduled +session, censored at observed coverage end. They must not be split at lunch or overnight. +Daily session-label rows preserve date identifiers and can span whole-session rows; +elapsed durations need instants and must reject session-label identifiers. +Shared session-label outputs leave physical label_duration unset. For instant inputs, +use the declared timezone for both elapsed-horizon targeting and duration arithmetic, +including DST, while preserving input timestamp and label-time representations. +Right-labeled bars can end at a break start; event timestamps at that instant are off-session. + +Standalone numerical labeling functions retain their legacy chronological row/duration +definitions. Do not infer calendar adoption from a timestamp parameter or change numerical +definitions without explicit migration evidence. Legacy auto-gap mode is not an authoritative +calendar or completeness check. Keep this boundary explicit in task guidance and user docs. diff --git a/src/ml4t/engineer/labeling/calendar.py b/src/ml4t/engineer/labeling/calendar.py index 9f4f7c6..4cbea40 100644 --- a/src/ml4t/engineer/labeling/calendar.py +++ b/src/ml4t/engineer/labeling/calendar.py @@ -8,15 +8,23 @@ from __future__ import annotations +from collections.abc import Mapping from datetime import UTC, date, datetime, timedelta from inspect import signature from typing import TYPE_CHECKING, Any, Protocol +import numpy as np +import numpy.typing as npt import polars as pl +from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec, TimestampSemantics, WindowUnit from ml4t.engineer.core.exceptions import DataValidationError -from ml4t.engineer.labeling.triple_barrier import triple_barrier_labels -from ml4t.engineer.labeling.utils import resolve_labeling_columns +from ml4t.engineer.labeling.triple_barrier import _triple_barrier_labels_single_group +from ml4t.engineer.labeling.utils import ( + is_duration_string, + resolve_labeling_columns, + validate_price_no_nans, +) if TYPE_CHECKING: # pragma: no cover - imports used only by static analysis from ml4t.engineer.config import DataContractConfig, LabelingConfig @@ -108,14 +116,32 @@ class PandasMarketCalendar: Supports 200+ calendars including CME, NYSE, LSE, etc. """ - def __init__(self, calendar_name: str): + def __init__( + self, + calendar_name: str | CalendarUnset = UNSET_CALENDAR, + *, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, + ): """ Parameters ---------- - calendar_name : str - Calendar name (e.g., "CME_Equity", "NYSE", "LSE") - See pandas_market_calendars.get_calendar_names() + calendar_name : str, optional + Legacy positional/keyword selector. Conflicting selectors fail. + calendar : str or None, optional + Canonical named selector, default NYSE. None is rejected because a + provider adapter cannot represent calendar-free observation order. """ + if ( + calendar_name is not UNSET_CALENDAR + and calendar is not UNSET_CALENDAR + and calendar_name != calendar + ): + raise ValueError("calendar and calendar_name selectors conflict") + selected = FeedSpec(time_contract_version=2).resolve_calendar( + calendar=calendar_name if calendar is UNSET_CALENDAR else calendar + ) + if selected is None: + raise ValueError("PandasMarketCalendar requires a named calendar, not calendar=None") try: import pandas_market_calendars as mcal except ImportError as err: @@ -124,8 +150,9 @@ def __init__(self, calendar_name: str): "Install with: pip install pandas-market-calendars" ) from err - self.calendar_name = calendar_name - self._calendar = mcal.get_calendar(calendar_name) + self.calendar = selected + self.calendar_name = selected + self._calendar = mcal.get_calendar(selected) self._supports_interruptions = ( "interruptions" in signature(self._calendar.schedule).parameters ) @@ -219,13 +246,20 @@ def _contains_timestamp( """Check timestamp membership with explicit close-boundary semantics.""" return start <= timestamp <= end if end_inclusive else start <= timestamp < end - def _trading_boundary(self, timestamp: datetime) -> tuple[bool, datetime | None]: + def _trading_boundary( + self, timestamp: datetime, *, bar_close: bool = False + ) -> tuple[bool, datetime | None]: """Return whether a timestamp is trading and its next session boundary.""" normalized = self._normalize_timestamp(timestamp) for start, end, boundary, end_inclusive in self._trading_intervals_around( normalized.date() ): - if self._contains_timestamp(normalized, start, end, end_inclusive): + contains = ( + start < normalized <= end + if bar_close + else self._contains_timestamp(normalized, start, end, end_inclusive) + ) + if contains: return True, boundary return False, None @@ -269,26 +303,39 @@ def _calendar_session_ids( calendar: TradingCalendar, timestamp_col: str, group_cols: list[str], + timezone: str | None = None, + bar_close: bool = False, ) -> tuple[pl.DataFrame, list[int]]: """Sort rows and derive sessions from an explicit calendar.""" _validate_explicit_calendar(calendar) sort_cols = [*group_cols, timestamp_col] sorted_data = data.sort(sort_cols) selected_cols = [*group_cols, timestamp_col] + projection = sorted_data.select(selected_cols) + dtype = projection.schema[timestamp_col] + if isinstance(dtype, pl.Datetime): + if projection.select((pl.col(timestamp_col).dt.nanosecond() % 1000 != 0).any()).item(): + raise ValueError("calendar timestamp has finer than supported microsecond precision") + if dtype.time_zone is None and timezone is not None: + projection = projection.with_columns( + pl.col(timestamp_col) + .dt.replace_time_zone(timezone, ambiguous="raise", non_existent="raise") + .dt.convert_time_zone("UTC") + ) session_ids: list[int] = [] previous_group: tuple[Any, ...] | None = None previous_break: datetime | None = None session_id = -1 - for row in sorted_data.select(selected_cols).iter_rows(named=False): + for row in projection.iter_rows(named=False): group = tuple(row[: len(group_cols)]) timestamp = row[-1] if not isinstance(timestamp, datetime): raise TypeError(f"timestamp column {timestamp_col!r} must contain datetime values") if isinstance(calendar, PandasMarketCalendar): - is_trading, session_break = calendar._trading_boundary(timestamp) + is_trading, session_break = calendar._trading_boundary(timestamp, bar_close=bar_close) else: is_trading = calendar.is_trading_time(timestamp) session_break = calendar.next_session_break(timestamp) if is_trading else None @@ -326,19 +373,149 @@ def _calendar_session_ids( return sorted_data, session_ids +def _calendar_group_labels( + data: pl.DataFrame, + config: LabelingConfig, + price_col: str, + timestamp_col: str | None, + timezone: str | None = None, + *, + max_holding_bars: npt.NDArray[np.int64] | None = None, +) -> pl.DataFrame: + dtype = data.schema[timestamp_col] if timestamp_col is not None else None + local_clock = ( + isinstance(dtype, pl.Datetime) and dtype.time_zone is None and timezone is not None + ) + if local_clock: + assert timestamp_col is not None + data = data.with_columns( + pl.col(timestamp_col) + .dt.replace_time_zone(timezone, ambiguous="raise", non_existent="raise") + .dt.convert_time_zone("UTC") + ) + result = _triple_barrier_labels_single_group( + data, + config, + price_col, + None, + None, + None, + timestamp_col, + False, + "returns_uniqueness", + max_holding_bars=max_holding_bars, + ) + if local_clock: + assert timestamp_col is not None + result = result.with_columns( + pl.col(timestamp_col, "label_time") + .dt.convert_time_zone(timezone) + .dt.replace_time_zone(None) + ) + return result + + +def _segment_labels( + data: pl.DataFrame, + config: LabelingConfig, + price_col: str, + group_cols: list[str], + timestamp_col: str | None = None, + timezone: str | None = None, +) -> pl.DataFrame: + validate_price_no_nans(data, price_col) + row_col = _temporary_column_name(data.columns, "__ml4t_observation_row") + indexed = data.with_row_index(row_col) + frames = indexed.partition_by(group_cols, maintain_order=True) if group_cols else [indexed] + frames = frames or [indexed] + results = [ + _calendar_group_labels(frame, config, price_col, timestamp_col, timezone) + for frame in frames + ] + result = pl.concat(results, how="vertical").sort(row_col).drop(row_col) + if timestamp_col is None: + result = result.with_columns(pl.lit(None).alias("label_duration")) + return result + + +def _session_window_labels( + data: pl.DataFrame, + config: LabelingConfig, + feed: FeedSpec, + calendar: str, + price_col: str, + timestamp_col: str, + group_cols: list[str], +) -> pl.DataFrame: + from ml4t.engineer.core.bar_grid import scheduled_session_indices + + validate_price_no_nans(data, price_col) + sorted_data = data.sort([*group_cols, timestamp_col]) + frames = ( + sorted_data.partition_by(group_cols, maintain_order=True) if group_cols else [sorted_data] + ) + frames = frames or [sorted_data] + results = [] + for frame in frames: + session_ids = scheduled_session_indices( + frame, feed_spec=feed, calendar=calendar, timestamp_col=timestamp_col + ) + horizon = config.max_holding_period + if isinstance(horizon, int): + counts = np.full(frame.height, horizon, dtype=np.float64) + elif isinstance(horizon, str) and horizon in frame.columns: + counts = np.asarray(frame[horizon].to_numpy(), dtype=np.float64) + else: + raise ValueError("Session holding periods require a positive integer or count column") + if ( + not np.isfinite(counts).all() + or np.any(counts <= 0) + or np.any(counts != np.floor(counts)) + ): + raise ValueError("Session holding periods must contain finite positive integral counts") + # A deadline beyond observed coverage is censored at the final available bar, + # as in the existing count-horizon engine. Clip before integer conversion. + bounded_counts = ( + np.minimum(counts, session_ids[-1] + 1).astype(np.int64) + if frame.height + else np.empty(0, dtype=np.int64) + ) + exits = np.searchsorted(session_ids, session_ids + bounded_counts, side="right") - 1 + periods = np.asarray(exits - np.arange(frame.height), dtype=np.int64) + results.append( + _calendar_group_labels( + frame, + config, + price_col, + timestamp_col, + None + if feed.timestamp_semantics == TimestampSemantics.SESSION_LABEL + else feed.timezone, + max_holding_bars=periods, + ) + ) + result = pl.concat(results, how="vertical") + if feed.timestamp_semantics == TimestampSemantics.SESSION_LABEL: + result = result.with_columns(pl.lit(None).alias("label_duration")) + return result + + def calendar_aware_labels( data: pl.DataFrame, config: LabelingConfig, - calendar: str | TradingCalendar, + calendar: str | TradingCalendar | None | CalendarUnset = UNSET_CALENDAR, price_col: str | None = None, timestamp_col: str | None = None, group_col: str | list[str] | None = None, contract: DataContractConfig | None = None, + *, + feed_spec: FeedSpec | Mapping[str, Any] | None = None, + window_unit: WindowUnit | str | None = None, ) -> pl.DataFrame: - """Apply triple-barrier labeling with session awareness. + """Apply triple-barrier labeling with declared calendar and horizon units. - Splits data by trading sessions and applies labeling within each session. - This prevents labels from spanning session gaps (maintenance, overnight, holidays). + Intraday count/elapsed compatibility modes stop at breaks. Session-count + horizons span scheduled sessions, and daily labels count whole-session rows. Parameters ---------- @@ -346,11 +523,10 @@ def calendar_aware_labels( Input data with OHLCV and timestamp config : LabelingConfig Barrier configuration - calendar : str or TradingCalendar - Either: - - Calendar name string (uses pandas_market_calendars) - - TradingCalendar protocol implementation - - "auto" to detect gaps automatically + calendar : str, TradingCalendar or None, optional + Explicit selector overrides config, then FeedSpec metadata, then NYSE. + None selects observation-row labeling without a schedule. Protocol objects + and explicit legacy "auto" gap mode remain available without FeedSpec. price_col : str | None, default None Price column name timestamp_col : str | None, default None @@ -359,6 +535,15 @@ def calendar_aware_labels( Grouping column(s) for panel-aware session labeling. contract : DataContractConfig | None, default None Optional shared dataframe contract. Used after config and before defaults. + feed_spec : FeedSpec or Mapping, optional + Shared column/time declarations, otherwise config.feed_spec. Named-calendar + feeds require timestamp semantics and a timezone for naive timestamps. + window_unit : WindowUnit or str, optional + Count windows default to expected_bar for shared feeds and omitted selectors; + a complete declared grid is required. Explicit observation counts and elapsed + duration horizons remain distinct. Legacy explicit selectors without FeedSpec + retain observation counts. Session counts expire on the final available + bar of the Nth following scheduled session, or at observed coverage end. Returns ------- @@ -399,11 +584,17 @@ def calendar_aware_labels( - Uses pandas_market_calendars for all string calendar names - Supports 200+ global calendars + product-specific futures calendars - See pandas_market_calendars.get_calendar_names() for available calendars - - Labels that would span session breaks are truncated at the break + - Intraday observation/expected-bar and legacy duration windows stop at breaks + - Session-count windows cross breaks and use the same numerical barrier engine - This may result in more timeout labels near session closes - Rows outside an explicit calendar's trading times are rejected - - Naive datetimes passed to PandasMarketCalendar are interpreted as UTC - - For 24/7 markets, use standard triple_barrier_labels instead + - Without FeedSpec, legacy naive calendar timestamps are interpreted as UTC + - FeedSpec naive timestamps use its declared timezone; ambiguous times fail + - Instant horizons and label_duration use physical elapsed time across DST; + session-label identifiers retain label_time but leave label_duration unset + - Explicit None preserves per-asset observation order and global row order; + label_time contains per-asset row offsets and label_duration is unset + - A named 24/7 schedule still segments daily sessions; it is not row mode """ from ml4t.engineer.config import LabelingConfig as _LabelingConfig @@ -414,6 +605,50 @@ def calendar_aware_labels( "use LabelingConfig.triple_barrier(...)." ) + declared_feed = feed_spec if feed_spec is not None else config.feed_spec + legacy_selector = declared_feed is None and calendar is not UNSET_CALENDAR + feed = ( + FeedSpec.from_any(declared_feed) + if declared_feed is not None + else FeedSpec(time_contract_version=2) + ) + if calendar is UNSET_CALENDAR and "calendar" in config.model_fields_set: + calendar = config.calendar + custom_calendar = ( + calendar is not UNSET_CALENDAR and calendar is not None and not isinstance(calendar, str) + ) + selected = calendar if custom_calendar else feed.resolve_calendar(calendar=calendar) + duration_horizon = isinstance(config.max_holding_period, timedelta) or ( + isinstance(config.max_holding_period, str) and is_duration_string(config.max_holding_period) + ) + declared_unit = window_unit if window_unit is not None else config.window_unit + unit = ( + WindowUnit(declared_unit) + if declared_unit is not None + else ( + WindowUnit.ELAPSED + if duration_horizon + else WindowUnit.OBSERVATION + if selected is None or legacy_selector + else WindowUnit.EXPECTED_BAR + ) + ) + if selected is None and (unit != WindowUnit.OBSERVATION or duration_horizon): + raise ValueError( + "calendar=None requires observation-count holding periods and window_unit='observation'" + ) + if (unit == WindowUnit.ELAPSED) != duration_horizon: + raise ValueError( + "elapsed windows require duration holding periods; count windows require observation/expected_bar" + ) + if declared_feed is not None and contract is None and config.data_contract is None: + from ml4t.engineer.config import DataContractConfig as _DataContractConfig + + contract = _DataContractConfig( + timestamp_col=feed.timestamp_col, + price_col=feed.price_col, + symbol_col=feed.entity_col, + ) resolved_price_col, resolved_ts_col, resolved_group_cols = resolve_labeling_columns( data=data, price_col=price_col, @@ -421,20 +656,68 @@ def calendar_aware_labels( group_col=group_col, config=config, contract=contract, - require_timestamp=True, + require_timestamp=selected is not None, + allow_temporal_contract=True, ) if config.method != "triple_barrier": raise ValueError("calendar_aware_labels requires LabelingConfig.method='triple_barrier'.") + if selected is None: + return _segment_labels(data, config, resolved_price_col, resolved_group_cols) + daily_labels = feed.timestamp_semantics == TimestampSemantics.SESSION_LABEL + if daily_labels and unit == WindowUnit.ELAPSED: + raise ValueError("session_label identifiers cannot define an elapsed horizon") + if declared_feed is not None and feed.timestamp_semantics is None: + raise ValueError("Calendar-aware FeedSpec requires declared timestamp_semantics") + if declared_feed is not None and selected == "auto": + raise ValueError( + "auto gap inference is a legacy mode, not an authoritative FeedSpec calendar" + ) + assert resolved_ts_col is not None + dtype = data.schema[resolved_ts_col] + if ( + declared_feed is not None + and not daily_labels + and isinstance(dtype, pl.Datetime) + and dtype.time_zone is None + and feed.timezone is None + ): + raise ValueError("Calendar-aware FeedSpec requires timezone metadata for naive timestamps") + if unit in {WindowUnit.EXPECTED_BAR, WindowUnit.SESSION} or daily_labels: + if not isinstance(selected, str) or selected == "auto": + raise ValueError("Scheduled windows require an authoritative named calendar") + from ml4t.engineer.core.bar_grid import validate_declared_grid + + validate_declared_grid( + data, + feed_spec=feed, + calendar=selected, + timestamp_col=resolved_ts_col, + group_cols=resolved_group_cols, + ) + if unit == WindowUnit.SESSION: + assert isinstance(selected, str) + return _session_window_labels( + data, config, feed, selected, resolved_price_col, resolved_ts_col, resolved_group_cols + ) + if daily_labels: + return _segment_labels( + data.sort([*resolved_group_cols, resolved_ts_col]), + config, + resolved_price_col, + resolved_group_cols, + resolved_ts_col, + ).with_columns(pl.lit(None).alias("label_duration")) + # Create calendar instance if string provided - if isinstance(calendar, str): - if calendar == "auto": + if isinstance(selected, str): + if selected == "auto": cal = SimpleTradingCalendar() else: # Always use pandas_market_calendars for string calendars - cal = PandasMarketCalendar(calendar) + cal = PandasMarketCalendar(selected) else: - cal = calendar + cal = selected session_col = _temporary_column_name(data.columns, "__ml4t_session_id") assert resolved_ts_col is not None @@ -471,6 +754,9 @@ def calendar_aware_labels( cal, resolved_ts_col, resolved_group_cols, + timezone=feed.timezone, + bar_close=feed.timestamp_semantics == TimestampSemantics.BAR_CLOSE + or feed.bar_label == "right", ) data_with_session = sorted_data.with_columns( pl.Series(session_col, session_ids, dtype=pl.Int64) @@ -479,12 +765,13 @@ def calendar_aware_labels( session_group_cols = ( [*resolved_group_cols, session_col] if resolved_group_cols else [session_col] ) - labeled = triple_barrier_labels( - data=data_with_session, - config=config, - price_col=resolved_price_col, - timestamp_col=resolved_ts_col, - group_col=session_group_cols, + labeled = _segment_labels( + data_with_session, + config, + resolved_price_col, + session_group_cols, + resolved_ts_col, + timezone=feed.timezone, ) return labeled.drop(session_col) diff --git a/src/ml4t/engineer/labeling/triple_barrier.py b/src/ml4t/engineer/labeling/triple_barrier.py index 68c28b0..e0e1360 100644 --- a/src/ml4t/engineer/labeling/triple_barrier.py +++ b/src/ml4t/engineer/labeling/triple_barrier.py @@ -263,19 +263,20 @@ def _calculate_time_durations( event_indices: npt.NDArray[np.intp], label_indices: npt.NDArray[np.int64], timestamp_col: str | None, -) -> tuple[list[Any], npt.NDArray[Any]]: +) -> tuple[pl.Series, npt.NDArray[Any]]: """Calculate label times and durations.""" n_events = len(event_indices) if timestamp_col is not None: - timestamps = data[timestamp_col].to_list() - label_times = [timestamps[idx] if idx < len(timestamps) else None for idx in label_indices] + label_times = data[timestamp_col].gather(label_indices) timestamps_array = data[timestamp_col].to_numpy() + if data.schema[timestamp_col] == pl.Date: + timestamps_array = timestamps_array.astype("datetime64[ms]") entry_times = timestamps_array[event_indices] exit_times = timestamps_array[label_indices] time_durations = exit_times - entry_times else: - label_times = list(label_indices) + label_times = pl.Series(label_indices) time_durations = np.array([None] * n_events, dtype=object) return label_times, time_durations @@ -285,7 +286,7 @@ def _build_labeling_result( data: pl.DataFrame, event_indices: npt.NDArray[np.intp], labels: npt.NDArray[np.int32], - label_times: list[Any], + label_times: pl.Series, label_prices: npt.NDArray[np.float64], label_returns: npt.NDArray[np.float64], bar_durations: npt.NDArray[np.int64], @@ -296,6 +297,11 @@ def _build_labeling_result( ) -> pl.DataFrame: """Build the result DataFrame from labeling outputs.""" calc_uniq = uniqueness is not None + duration_series = ( + pl.Series(time_durations.tolist()) + if time_durations.dtype.kind == "O" + else pl.Series(time_durations) + ) if "event_time" in data.columns: label_dict: dict[str, Any] = { @@ -305,7 +311,7 @@ def _build_labeling_result( "label_price": label_prices, "label_return": label_returns, "label_bars": bar_durations, - "label_duration": time_durations, + "label_duration": duration_series, "barrier_hit": barrier_hit, } if calc_uniq: @@ -325,8 +331,8 @@ def _build_labeling_result( "label_price": pl.Series(label_prices, dtype=pl.Float64), "label_return": pl.Series(label_returns, dtype=pl.Float64), "label_bars": pl.Series(bar_durations, dtype=pl.Int64), - "label_duration": pl.Series(time_durations), - "barrier_hit": pl.Series(barrier_hit), + "label_duration": duration_series, + "barrier_hit": pl.Series(barrier_hit, dtype=pl.String), } if calc_uniq: columns_to_add["label_uniqueness"] = pl.Series(uniqueness, dtype=pl.Float64) @@ -349,6 +355,8 @@ def _triple_barrier_labels_single_group( uniqueness_weight_scheme: Literal[ "returns_uniqueness", "uniqueness_only", "returns_only", "equal" ], + *, + max_holding_bars: npt.NDArray[np.int64] | None = None, ) -> pl.DataFrame: """Apply triple barrier labeling to a single asset/group.""" if timestamp_col is not None: @@ -358,15 +366,27 @@ def _triple_barrier_labels_single_group( event_mask = data["event_time"].is_not_null() event_indices = np.where(event_mask.to_numpy())[0] if len(event_indices) == 0: - return data.with_columns( + time_dtype = data.schema[timestamp_col] if timestamp_col is not None else pl.Int64 + duration_dtype = ( + pl.Duration(time_dtype.time_unit if isinstance(time_dtype, pl.Datetime) else "ms") + if timestamp_col is not None + else pl.Null + ) + result = data.with_columns( label=pl.lit(None, dtype=pl.Int32), - label_time=pl.lit(None, dtype=pl.Int64), + label_time=pl.lit(None, dtype=time_dtype), label_price=pl.lit(None, dtype=pl.Float64), label_return=pl.lit(None, dtype=pl.Float64), label_bars=pl.lit(None, dtype=pl.Int64), - label_duration=pl.lit(None, dtype=pl.Utf8), + label_duration=pl.Series([None] * data.height, dtype=duration_dtype), barrier_hit=pl.lit(None, dtype=pl.Utf8), ) + if calculate_uniqueness: + result = result.with_columns( + label_uniqueness=pl.lit(None, dtype=pl.Float64), + sample_weight=pl.lit(None, dtype=pl.Float64), + ) + return result else: event_indices = np.arange(len(data)) @@ -381,6 +401,8 @@ def _triple_barrier_labels_single_group( upper_barriers, lower_barriers, max_periods, sides, trailing_stops = _prepare_barrier_arrays( data, config, event_indices, timestamp_col=timestamp_col ) + if max_holding_bars is not None: + max_periods = max_holding_bars[event_indices] labels, label_indices, label_prices, label_returns, bar_durations = _apply_triple_barrier_nb( closes, opens, @@ -521,7 +543,7 @@ def triple_barrier_labels( if group_cols: sort_cols = group_cols + ([resolved_ts_col] if resolved_ts_col else []) sorted_data = data.sort(sort_cols) - grouped_frames = sorted_data.partition_by(group_cols, maintain_order=True) + grouped_frames = sorted_data.partition_by(group_cols, maintain_order=True) or [sorted_data] grouped_results = [ _triple_barrier_labels_single_group( diff --git a/src/ml4t/engineer/labeling/utils.py b/src/ml4t/engineer/labeling/utils.py index 7ae5a36..c494821 100644 --- a/src/ml4t/engineer/labeling/utils.py +++ b/src/ml4t/engineer/labeling/utils.py @@ -454,11 +454,25 @@ def resolve_labeling_columns( config: LabelingConfig | None = None, contract: DataContractConfig | None = None, require_timestamp: bool = False, + allow_temporal_contract: bool = False, ) -> tuple[str, str | None, list[str]]: """Resolve columns using explicit args > labeling config > shared contract > defaults.""" config_fields_set = ( set(getattr(config, "model_fields_set", set())) if config is not None else set() ) + if ( + config is not None + and not allow_temporal_contract + and ( + "calendar" in config_fields_set + or config.feed_spec is not None + or config.window_unit is not None + ) + ): + raise DataValidationError( + "Declared calendar/feed/window semantics require calendar_aware_labels; " + "this standalone kernel does not interpret the temporal contract" + ) timestamp_is_explicit = timestamp_col is not None if contract is None and config is not None: diff --git a/tests/test_api.py b/tests/test_api.py index aa9a5d2..56c3bfd 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -317,17 +317,17 @@ def test_output_cannot_replace_input_column(sample_ohlcv_data): ) -def test_compute_requires_time_or_explicit_order_assumption(sample_ohlcv_data): - """Row-order semantics must be explicit when no time column exists.""" +def test_explicit_row_mode_needs_no_additional_temporal_order_assertion(sample_ohlcv_data): + """Selecting observation order is sufficient even without timestamp metadata.""" data = sample_ohlcv_data.drop("timestamp") - with pytest.raises(DataSchemaError, match="No time column found"): - compute_features(data, [{"name": "sma", "params": {"period": 2}}], calendar=None) + rows = compute_features(data, [{"name": "sma", "params": {"period": 2}}], calendar=None) result = compute_features( data, [{"name": "sma", "params": {"period": 2}}], assume_sorted=True, calendar=None ) assert "sma" in result.columns + assert rows.equals(result) def test_row_mode_isolates_assets_without_reordering_observations(): diff --git a/tests/test_api_annualization_contract.py b/tests/test_api_annualization_contract.py new file mode 100644 index 0000000..6a3424a --- /dev/null +++ b/tests/test_api_annualization_contract.py @@ -0,0 +1,208 @@ +from datetime import UTC, datetime, timedelta +from math import sqrt +from statistics import mean, stdev + +import numpy as np +import pandas_market_calendars as mcal +import polars as pl +import pytest +import yaml +from ml4t.specs import FeedSpec + +from ml4t.engineer import compute_features + +RETURNS = np.array( + [ + 0.012, + -0.008, + 0.006, + -0.003, + 0.015, + -0.011, + 0.004, + 0.009, + -0.005, + 0.007, + -0.002, + 0.013, + -0.009, + 0.005, + 0.011, + -0.006, + 0.008, + -0.004, + 0.014, + -0.007, + ] +) + + +def prices(times): + frames = [] + for symbol, multiplier in [("A", 1.0), ("B", 1.7)]: + close = 100 * np.cumprod(np.r_[1.0, 1.0 + RETURNS * multiplier]) + frames.append( + pl.DataFrame( + { + "timestamp": times, + "symbol": symbol, + "open": close, + "high": close, + "low": close, + "close": close, + "volume": 1.0, + } + ) + ) + return pl.concat(frames) + + +def feed(calendar): + return FeedSpec( + time_contract_version=2, + calendar=calendar, + entity_col="symbol", + data_frequency="1h", + timestamp_semantics="bar_close", + bar_type="time", + bar_grid_anchor="session_open", + bar_label="right", + partial_bar_policy="include", + ) + + +def actual_times(calendar): + schedule = mcal.get_calendar(calendar).schedule("2025-07-07", "2025-07-17") + return list( + mcal.date_range(schedule, frequency="1h", closed="right", force_close=True)[ + :21 + ].to_pydatetime() + ) + + +def request_file(tmp_path, requests, serialized): + if not serialized: + return requests + path = tmp_path / "features.yaml" + path.write_text(yaml.safe_dump({"features": requests})) + return path + + +def materialize(frame): + return frame.collect() if isinstance(frame, pl.LazyFrame) else frame + + +@pytest.mark.parametrize("calendar", ["NYSE", "24/7"]) +@pytest.mark.parametrize("lazy", [False, True]) +@pytest.mark.parametrize("serialized", [False, True]) +def test_real_price_pipeline_uses_explicit_scale_not_calendar_or_bar_frequency( + tmp_path, calendar, lazy, serialized +): + raw = prices(actual_times(calendar)) + data = raw.lazy() if lazy else raw + checked = compute_features(data, [], feed_spec=feed(calendar)) + returns = checked.with_columns(pl.col("close").pct_change().over("symbol").alias("returns")) + requests = [ + { + "name": "realized_volatility", + "params": {"period": 3, "annualize": False}, + "output": "raw", + }, + {"name": "realized_volatility", "params": {"period": 3}, "output": "legacy_default"}, + ] + requests += [ + { + "name": "realized_volatility", + "params": {"period": 3, "trading_periods": scale}, + "output": f"v{scale}", + } + for scale in (252, 365, 8760) + ] + output = materialize( + compute_features( + returns, request_file(tmp_path, requests, serialized), feed_spec=feed(calendar) + ) + ) + assert output.select(raw.columns).equals(raw) + for symbol in ("A", "B"): + asset = output.filter(pl.col("symbol") == symbol) + values = asset["returns"].drop_nulls().to_list() + expected = stdev(values[-3:]) + assert asset["raw"][-1] == pytest.approx(expected, rel=1e-11) + assert asset["legacy_default"][-1] == pytest.approx(expected * sqrt(252), rel=1e-11) + for scale in (252, 365, 8760): + assert asset[f"v{scale}"][-1] == pytest.approx(expected * sqrt(scale), rel=1e-11) + + +@pytest.mark.parametrize("lazy", [False, True]) +def test_row_mode_annualization_is_invariant_to_irregular_and_reordered_timestamps( + monkeypatch, lazy +): + import sys + + monkeypatch.setitem(sys.modules, "pandas_market_calendars", None) + start = datetime(2025, 7, 11, tzinfo=UTC) + outputs = [] + requests = [ + { + "name": "realized_volatility", + "params": {"period": 3, "trading_periods": scale}, + "output": f"v{scale}", + } + for scale in (252, 365, 8760) + ] + for offsets in [list(range(21)), [i * i for i in range(21)], list(reversed(range(21)))]: + raw = prices([start + timedelta(hours=h) for h in offsets]) + data = raw.lazy() if lazy else raw + returns = data.with_columns(pl.col("close").pct_change().over("symbol").alias("returns")) + result = materialize( + compute_features(returns, requests, feed_spec=feed("NYSE"), calendar=None) + ) + assert result.select(raw.columns).equals(raw) + outputs.append(result.select([f"v{scale}" for scale in (252, 365, 8760)])) + assert outputs[0].equals(outputs[1]) + assert outputs[0].equals(outputs[2]) + + +@pytest.mark.parametrize("scale", [252, 365, 8760]) +@pytest.mark.parametrize("lazy", [False, True]) +@pytest.mark.parametrize("serialized", [False, True]) +def test_public_risk_workflow_retains_explicit_annualization_and_period_risk_free_rate( + tmp_path, scale, lazy, serialized +): + raw = prices(actual_times("24/7")) + data = raw.lazy() if lazy else raw + checked = compute_features(data, [], feed_spec=feed("24/7")) + returns = checked.with_columns(pl.col("close").pct_change().over("symbol").alias("returns")) + requests = [ + { + "name": "risk_adjusted_returns", + "params": {"window": 20, "risk_free_rate": 0.052, "trading_periods": scale}, + "output": "risk", + } + ] + output = materialize( + compute_features( + returns, request_file(tmp_path, requests, serialized), feed_spec=feed("24/7") + ) + ) + assert output.select(raw.columns).equals(raw) + for symbol in ("A", "B"): + asset = output.filter(pl.col("symbol") == symbol) + values = asset["returns"].drop_nulls().to_numpy() + rf = 0.052 / scale + excess = mean(values) - rf + downside = np.minimum(values - rf, 0) + close = asset["close"].to_numpy() + peaks = np.maximum.accumulate(close) + drawdown = abs(np.min((close - peaks) / peaks)) + gains = np.sum(values[values > rf] - rf) + losses = np.sum(rf - values[values <= rf]) + expected = { + "sharpe_ratio": excess / (stdev(values) + 1e-10) * sqrt(scale), + "sortino_ratio": excess / (sqrt(np.mean(downside**2)) + 1e-10) * sqrt(scale), + "calmar_ratio": mean(values) * scale / (drawdown + 1e-10), + "omega_ratio": gains / losses, + } + for metric, value in expected.items(): + assert asset[f"risk_{metric}"][-1] == pytest.approx(value, rel=1e-9) diff --git a/tests/test_api_bar_grid.py b/tests/test_api_bar_grid.py index 0875fda..8a1f00f 100644 --- a/tests/test_api_bar_grid.py +++ b/tests/test_api_bar_grid.py @@ -144,6 +144,29 @@ def test_row_mode_does_not_reorder_values_when_timestamp_order_changes(lazy): assert outputs[0].equals(outputs[1]) +@pytest.mark.parametrize("lazy", [False, True]) +@pytest.mark.parametrize("metadata_disable", [False, True]) +def test_row_mode_does_not_require_timestamps_or_a_temporal_sort_assertion( + monkeypatch, lazy, metadata_disable +): + import sys + + monkeypatch.setitem(sys.modules, "pandas_market_calendars", None) + start = datetime(2025, 7, 11, tzinfo=UTC) + timed = returns(prices([start + timedelta(hours=h) for h in [200, 0, 73, 1, 2, 300, 7, 5]])) + rows = timed.drop("timestamp") + kwargs = ( + {"feed_spec": feed("NYSE").with_overrides(calendar=None)} + if metadata_disable + else {"calendar": None} + ) + output = run(rows, lazy, **kwargs) + control = run(timed, lazy, calendar=None) + assert output.height == rows.height + assert output.select(rows.columns).equals(rows) + assert output["realized_volatility"].equals(control["realized_volatility"]) + + @pytest.mark.parametrize("suffix", [".json", ".yaml"]) @pytest.mark.parametrize("calendar", ["NYSE", None, UNSET_CALENDAR]) def test_serialized_artifact_calendar_intent_reaches_real_feature_workflow( @@ -246,8 +269,13 @@ def unexpected(*_args, **_kwargs): def test_calendar_grid_does_not_reinterpret_other_window_units_as_expected_bars(unit): start = datetime(2025, 7, 11, tzinfo=UTC) data = returns(prices([start + timedelta(hours=h) for h in range(8)])) - with pytest.raises(ValueError, match="window_unit='expected_bar'"): - run(data, False, feed_spec=feed("24/7"), window_unit=unit) + if unit is WindowUnit.SESSION: + result = run(data, False, feed_spec=feed("24/7"), window_unit=unit) + assert result["realized_volatility"].is_null().all() + else: + error = "period" if unit is WindowUnit.ELAPSED else "Calendar-aware" + with pytest.raises(ValueError, match=error): + run(data, False, feed_spec=feed("24/7"), window_unit=unit) if unit is WindowUnit.OBSERVATION: assert run(data, False, calendar=None, window_unit=unit).height == 8 else: 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] diff --git a/tests/test_api_temporal_windows.py b/tests/test_api_temporal_windows.py new file mode 100644 index 0000000..6cf2275 --- /dev/null +++ b/tests/test_api_temporal_windows.py @@ -0,0 +1,384 @@ +from datetime import UTC, date, datetime, timedelta +from statistics import mean, pstdev, stdev + +import polars as pl +import pytest +import yaml +from ml4t.specs import FeedSpec + +from ml4t.engineer import compute_features + + +def _prices(): + from ml4t.engineer.core.bar_grid import _expected_bar_labels + from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + timestamps = list( + _expected_bar_labels( + PandasMarketCalendar("NYSE"), + datetime(2025, 7, 2, tzinfo=UTC), + datetime(2025, 7, 9, tzinfo=UTC), + timedelta(hours=1), + "right", + True, + ) + ) + values = [100.0 + index for index in range(len(timestamps))] + return pl.DataFrame( + { + "timestamp": timestamps, + "symbol": "A", + "open": values, + "high": values, + "low": values, + "close": values, + "volume": 1.0, + } + ) + + +def _feed(): + return FeedSpec( + time_contract_version=2, + calendar="NYSE", + timezone="UTC", + entity_col="symbol", + timestamp_semantics="bar_close", + data_frequency="1h", + bar_type="time", + bar_grid_anchor="session_open", + bar_label="right", + partial_bar_policy="include", + ) + + +def _run(prices, period, unit, lazy=False): + checked = compute_features(prices, [], feed_spec=_feed()) + returns = checked.with_columns(pl.col("close").pct_change().over("symbol").alias("returns")) + features = [ + {"name": "sma", "params": {"period": period}}, + {"name": "stddev", "params": {"period": period}}, + {"name": "realized_volatility", "params": {"period": period, "annualize": False}}, + ] + data = returns.lazy() if lazy else returns + result = compute_features(data, features, feed_spec=_feed(), window_unit=unit) + assert isinstance(result, type(data)) + return result.collect() if lazy else result + + +@pytest.mark.parametrize("lazy", [False, True]) +def test_session_feature_windows_use_original_causal_bars_and_actual_shortened_sessions(lazy): + prices = _prices() + result = _run(prices.reverse(), 2, "session", lazy) + assert result.select(prices.columns).equals(prices) + assert result["sma"][:7].to_list() == [None] * 7 + # First July3 bar includes seven July2 bars and itself, not two rows or two closes. + assert result["sma"][7] == pytest.approx(mean(range(100, 108))) + assert result["stddev"][7] == pytest.approx(pstdev(range(100, 108))) + # July3 has only four bars; July4 is closed. First July7 bar sees July3 + itself. + assert result["sma"][11] == pytest.approx(mean(range(107, 112))) + assert result["stddev"][11] == pytest.approx(pstdev(range(107, 112))) + assert result["realized_volatility"][11] == pytest.approx( + stdev([i / (i - 1) - 1 for i in range(107, 112)]) + ) + + +@pytest.mark.parametrize("lazy", [False, True]) +def test_elapsed_feature_windows_use_open_left_utc_duration_not_rows_or_sessions(lazy): + prices = _prices() + result = _run(prices, "2h", "elapsed", lazy) + assert result["sma"][:2].to_list() == [None, None] + assert result["sma"][2] == pytest.approx(101.5) + # First July3 bar has no bar from July2 inside the elapsed duration. + assert result["sma"][7] == pytest.approx(107.0) + assert result["realized_volatility"][7] is None + # The final half-hour July3 bar puts three observations inside two physical hours. + assert result["sma"][10] == pytest.approx(mean([108, 109, 110])) + assert result["stddev"][10] == pytest.approx(pstdev([108, 109, 110])) + assert result["realized_volatility"][10] == pytest.approx( + stdev([108 / 107 - 1, 109 / 108 - 1, 110 / 109 - 1]) + ) + + +@pytest.mark.parametrize("unit,period", [("session", 2), ("elapsed", "2h")]) +def test_temporal_feature_panels_are_isolated_and_future_prices_do_not_enter_earlier_windows( + unit, period +): + first = _prices() + second = first.with_columns( + *[(pl.col(name) * 2).alias(name) for name in ("open", "high", "low", "close")], + pl.lit("B").alias("symbol"), + ) + result = _run(pl.concat([second, first]).reverse(), period, unit, True) + for frame in (first, second): + symbol = frame["symbol"][0] + assert result.filter(pl.col("symbol") == symbol).equals(_run(frame, period, unit)) + changed = first.with_columns( + *[ + pl.when(pl.col("timestamp").dt.date() >= date(2025, 7, 7)) + .then(pl.col(name) * 4) + .otherwise(pl.col(name)) + .alias(name) + for name in ("open", "high", "low", "close") + ] + ) + before = _run(first, period, unit).head(11) + after = _run(changed, period, unit).head(11) + assert before.equals(after) + + +@pytest.mark.parametrize("unit,period", [("session", 2), ("elapsed", "2h")]) +def test_lazy_filter_projection_and_slice_do_not_remove_temporal_warmup_history(unit, period): + prices = _prices() + features = [{"name": "sma", "params": {"period": period}}] + full = compute_features(prices, features, feed_spec=_feed(), window_unit=unit) + predicate = pl.col("timestamp").dt.date() == date(2025, 7, 7) + lazy = compute_features(prices.lazy(), features, feed_spec=_feed(), window_unit=unit) + assert ( + lazy.filter(predicate).select("sma").collect().equals(full.filter(predicate).select("sma")) + ) + assert lazy.slice(7, 4).collect().equals(full.slice(7, 4)) + + +@pytest.mark.parametrize("unit,period", [("session", 2), ("elapsed", "2h")]) +def test_serialized_existing_feature_requests_keep_explicit_temporal_units_and_parameters( + tmp_path, unit, period +): + features = [{"name": "sma", "params": {"period": period}, "output": "average"}] + path = tmp_path / "features.yaml" + path.write_text(yaml.safe_dump({"features": features})) + expected = compute_features(_prices(), features, feed_spec=_feed(), window_unit=unit) + actual = compute_features(_prices(), path, feed_spec=_feed().to_dict(), window_unit=unit) + assert actual.equals(expected) + + +@pytest.mark.parametrize("unit,period", [("session", 2), ("elapsed", "2h")]) +def test_missing_complete_session_is_rejected_before_temporal_feature_returns_are_interpreted( + unit, period +): + prices = _prices().filter(pl.col("timestamp").dt.date() != date(2025, 7, 3)) + before = prices.clone() + with pytest.raises(ValueError, match="missing scheduled bar.*2025-07-03"): + compute_features( + prices, + [{"name": "sma", "params": {"period": period}}], + feed_spec=_feed(), + window_unit=unit, + ) + assert prices.equals(before) + + +@pytest.mark.parametrize("unit,period", [("session", 2), ("elapsed", "2h")]) +@pytest.mark.parametrize("feature", ["ema", "rsi"]) +def test_temporal_units_do_not_silently_redefine_recursive_indicators(feature, unit, period): + with pytest.raises(ValueError, match=f"{feature}.*temporal"): + compute_features( + _prices(), + [{"name": feature, "params": {"period": period}}], + feed_spec=_feed(), + window_unit=unit, + ) + + +@pytest.mark.parametrize( + "unit,period", + [("session", 0), ("session", 1.5), ("session", True), ("elapsed", 2), ("elapsed", "0h")], +) +def test_temporal_feature_periods_reject_unit_mismatches_instead_of_converting_rows(unit, period): + with pytest.raises(ValueError, match="period"): + compute_features( + _prices(), + [{"name": "sma", "params": {"period": period}}], + feed_spec=_feed(), + window_unit=unit, + ) + + +@pytest.mark.parametrize("unit,period", [("session", 2), ("elapsed", "2h")]) +def test_calendar_none_remains_observation_mode_even_when_temporal_windows_are_requested( + unit, period +): + with pytest.raises(ValueError, match="observation"): + compute_features( + _prices(), + [{"name": "sma", "params": {"period": period}}], + calendar=None, + window_unit=unit, + ) + + +@pytest.mark.parametrize("naive", [False, True]) +def test_elapsed_window_counts_physical_hours_through_spring_dst_in_declared_timezone(naive): + timestamps = pl.Series( + "timestamp", [datetime(2025, 3, 9, hour, tzinfo=UTC) for hour in range(5, 11)] + ).dt.convert_time_zone("America/New_York") + if naive: + timestamps = timestamps.dt.replace_time_zone(None) + values = [100.0 + i for i in range(6)] + prices = pl.DataFrame( + { + "timestamp": timestamps, + "close": values, + "open": values, + "high": values, + "low": values, + "volume": 1.0, + } + ) + feed = FeedSpec( + time_contract_version=2, + calendar="24/7", + timezone="America/New_York", + timestamp_semantics="event_time", + data_frequency="1h", + bar_type="time", + bar_grid_anchor="session_open", + bar_label="left", + partial_bar_policy="include", + ) + result = compute_features( + prices, [{"name": "sma", "params": {"period": "2h"}}], feed_spec=feed, window_unit="elapsed" + ) + assert result["sma"].to_list() == [None, None, 101.5, 102.5, 103.5, 104.5] + + +@pytest.mark.parametrize("lazy", [False, True]) +def test_daily_feature_session_windows_count_sessions_without_inventing_intraday_instants(lazy): + values = [100.0, 101.0, 104.0, 105.0] + prices = pl.DataFrame( + { + "timestamp": [date(2025, 7, day) for day in (2, 3, 7, 8)], + "close": values, + "open": values, + "high": values, + "low": values, + "volume": 1.0, + } + ) + feed = FeedSpec( + time_contract_version=2, + calendar="NYSE", + timestamp_semantics="session_label", + data_frequency="1d", + bar_type="time", + ) + result = compute_features( + prices.lazy() if lazy else prices, + [{"name": "sma", "params": {"period": 2}}], + feed_spec=feed, + window_unit="session", + ) + result = result.collect() if lazy else result + assert result["sma"].to_list() == [None, 100.5, 102.5, 104.5] + with pytest.raises(ValueError, match="session_label.*elapsed"): + compute_features( + prices, + [{"name": "sma", "params": {"period": "2h"}}], + feed_spec=feed, + window_unit="elapsed", + ) + + +@pytest.mark.parametrize("lazy", [False, True]) +def test_temporal_windows_preserve_ddof_scale_annualization_and_input_parameter_values(lazy): + from copy import deepcopy + from math import sqrt + + prices = _prices().with_columns(pl.col("close").pct_change().over("symbol").alias("returns")) + features = [ + {"name": "stddev", "params": {"period": 2, "nbdev": 2.0, "ddof": 1}}, + {"name": "realized_volatility", "params": {"period": 2, "trading_periods": 365}}, + {"name": "sma", "params": {"period": 1}, "output": "current_session"}, + {"name": "sma", "params": {"period": 2}, "output": "two_sessions"}, + ] + original = deepcopy(features) + result = compute_features( + prices.lazy() if lazy else prices, features, feed_spec=_feed(), window_unit="session" + ) + result = result.collect() if lazy else result + assert result["stddev"][11] == pytest.approx(2 * stdev(range(107, 112))) + assert result["realized_volatility"][11] == pytest.approx( + stdev([i / (i - 1) - 1 for i in range(107, 112)]) * sqrt(365) + ) + assert result["current_session"][11] == 111.0 + assert result["two_sessions"][11] == 109.0 + assert features == original + + +@pytest.mark.parametrize("unit,period", [("session", 10**100), ("elapsed", "365d")]) +def test_temporal_window_exceeding_available_coverage_stays_null_without_integer_overflow( + unit, period +): + result = compute_features( + _prices(), + [{"name": "sma", "params": {"period": period}}], + feed_spec=_feed(), + window_unit=unit, + ) + assert result["sma"].is_null().all() + + +@pytest.mark.parametrize("lazy", [False, True]) +def test_empty_temporal_feature_input_preserves_typed_columns_and_no_observations(lazy): + prices = _prices().head(0) + result = compute_features( + prices.lazy() if lazy else prices, + [{"name": "sma", "params": {"period": 1}}], + feed_spec=_feed(), + window_unit="session", + ) + result = result.collect() if lazy else result + assert result.height == 0 + assert result.schema["sma"] == pl.Float64 + assert result.select(prices.columns).equals(prices) + + +@pytest.mark.parametrize("side", ["left", "right"]) +@pytest.mark.parametrize("lazy", [False, True]) +def test_feature_session_windows_count_short_sessions_even_when_partial_policy_excludes_all_bars( + side, lazy +): + from ml4t.engineer.core.bar_grid import _expected_bar_labels + from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + timestamps = list( + _expected_bar_labels( + PandasMarketCalendar("NYSE"), + datetime(2025, 7, 2, tzinfo=UTC), + datetime(2025, 7, 9, tzinfo=UTC), + timedelta(hours=5), + side, + False, + ) + ) + values = [100.0, 101.0, 102.0] + prices = pl.DataFrame( + { + "timestamp": timestamps, + "open": values, + "high": values, + "low": values, + "close": values, + "volume": 1.0, + } + ) + feed = FeedSpec( + time_contract_version=2, + calendar="NYSE", + timezone="UTC", + data_frequency="5h", + timestamp_semantics="bar_close" if side == "right" else "event_time", + bar_type="time", + bar_grid_anchor="session_open", + bar_label=side, + partial_bar_policy="exclude", + ) + result = compute_features( + prices.lazy() if lazy else prices, + [{"name": "sma", "params": {"period": 2}}], + feed_spec=feed, + window_unit="session", + ) + result = result.collect() if lazy else result + # July3 counts, despite having no5h bar. July7 must not include the July2 price. + assert result["sma"].to_list() == [None, 101.0, 101.5] diff --git a/tests/test_labeling_empty_schema.py b/tests/test_labeling_empty_schema.py new file mode 100644 index 0000000..673e05c --- /dev/null +++ b/tests/test_labeling_empty_schema.py @@ -0,0 +1,81 @@ +from datetime import UTC, date, datetime +from io import BytesIO + +import polars as pl +import pytest +from ml4t.specs import FeedSpec + +from ml4t.engineer.config import LabelingConfig +from ml4t.engineer.labeling import calendar_aware_labels, triple_barrier_labels + + +@pytest.mark.parametrize("kind", ["date", "instant"]) +@pytest.mark.parametrize("mode", ["session", "expected_bar", "observation", "standalone"]) +@pytest.mark.parametrize("sparse_events", [False, True]) +def test_filtering_all_prices_still_produces_a_portable_typed_label_table( + kind, mode, sparse_events +): + days = [date(2025, 7, day) for day in (3, 7)] + if kind == "instant": + days = [datetime(2025, 7, day, tzinfo=UTC) for day in (3, 7)] + prices = pl.DataFrame({"timestamp": days, "close": [100.0, 101.0], "symbol": "A"}) + if sparse_events: + prices = prices.with_columns(pl.col("timestamp").alias("event_time")) + empty = prices.filter(pl.col("close") > 1000) + config = LabelingConfig.triple_barrier(max_holding_period=1) + if mode == "standalone": + result = triple_barrier_labels(empty, config, timestamp_col="timestamp") + else: + feed = FeedSpec( + time_contract_version=2, + calendar="NYSE", + timestamp_semantics="session_label", + data_frequency="1d", + bar_type="time", + entity_col="symbol", + ) + result = calendar_aware_labels(empty, config, feed_spec=feed, window_unit=mode) + assert result.height == 0 + assert result.select(empty.columns).schema == empty.schema + assert result.schema["barrier_hit"] == pl.String + assert result.schema["label_time"] == prices.schema["timestamp"] + buffer = BytesIO() + result.write_parquet(buffer) + buffer.seek(0) + assert pl.read_parquet(buffer).schema == result.schema + + +@pytest.mark.parametrize("empty", [False, True]) +@pytest.mark.parametrize("grouped", [False, True]) +@pytest.mark.parametrize("events", ["dense", "sparse", "none"]) +def test_timestamp_free_legacy_labels_keep_unknown_duration_portable_without_changing_holding_bars( + empty, grouped, events +): + prices = pl.DataFrame({"close": [100.0, 101.0, 102.0], "symbol": "A"}) + if not grouped: + prices = prices.drop("symbol") + if events != "dense": + prices = prices.with_row_index("event_time").with_columns( + pl.when(pl.col("event_time") % 2 == 0 if events == "sparse" else pl.col("close") > 1000) + .then(pl.col("event_time")) + .otherwise(None) + .alias("event_time") + ) + if empty: + prices = prices.filter(pl.col("close") > 1000) + config = LabelingConfig.triple_barrier( + upper_barrier=None, lower_barrier=None, max_holding_period=1 + ) + result = triple_barrier_labels(prices, config) + assert result.select(prices.columns).equals(prices) + assert result.schema["label_duration"] == pl.Null + assert result["label_duration"].is_null().all() + if not empty: + expected = {"dense": [1, 1, 0], "sparse": [1, None, 0], "none": [None] * 3} + assert result["label_bars"].to_list() == expected[events] + if events != "none": + assert result["label_return"][0] == pytest.approx(0.01) + buffer = BytesIO() + result.write_parquet(buffer) + buffer.seek(0) + assert pl.read_parquet(buffer).equals(result) diff --git a/tests/test_labeling_session_windows.py b/tests/test_labeling_session_windows.py new file mode 100644 index 0000000..949e9ec --- /dev/null +++ b/tests/test_labeling_session_windows.py @@ -0,0 +1,514 @@ +from datetime import UTC, date, datetime, time, timedelta +from io import BytesIO +from zoneinfo import ZoneInfo + +import polars as pl +import pytest +from ml4t.specs import FeedSpec + +from ml4t.engineer.config import LabelingConfig +from ml4t.engineer.labeling import calendar_aware_labels, triple_barrier_labels + + +def _feed(**kwargs): + return FeedSpec( + time_contract_version=2, + calendar="NYSE", + timestamp_semantics="session_label", + data_frequency="1d", + bar_type="time", + **kwargs, + ) + + +def _config(horizon=1, **kwargs): + values = { + "upper_barrier": None, + "lower_barrier": None, + "max_holding_period": horizon, + "window_unit": "session", + } + values.update(kwargs) + return LabelingConfig.triple_barrier(**values) + + +@pytest.mark.parametrize("representation", ["date", "naive", "UTC", "Asia/Tokyo"]) +@pytest.mark.parametrize("sessions,exits", [(1, [1, 2, 3, 4, 4]), (2, [2, 3, 4, 4, 4])]) +def test_session_labels_count_scheduled_sessions_not_elapsed_holiday_days( + representation, sessions, exits +): + days = [date(2025, 7, day) for day in (2, 3, 7, 8, 9)] + timestamps = days + if representation != "date": + zone = None if representation == "naive" else ZoneInfo(representation) + timestamps = [datetime.combine(day, time(0, 15), tzinfo=zone) for day in days] + prices = pl.DataFrame({"timestamp": timestamps, "close": [100.0, 101.0, 102.0, 103.0, 104.0]}) + buffer = BytesIO() + prices.write_parquet(buffer) + buffer.seek(0) + restored = pl.read_parquet(buffer) + result = calendar_aware_labels(restored.reverse(), _config(sessions), feed_spec=_feed()) + assert result.select(prices.columns).equals(prices) + assert result["label_time"].to_list() == [timestamps[index] for index in exits] + assert result["label_bars"].to_list() == [exit - entry for entry, exit in enumerate(exits)] + assert result["label_return"].to_list() == pytest.approx( + [prices["close"][exit] / prices["close"][entry] - 1 for entry, exit in enumerate(exits)] + ) + assert result["barrier_hit"].to_list() == ["time"] * 5 + + +@pytest.mark.parametrize("unit", ["expected_bar", "observation"]) +def test_daily_count_horizon_crosses_valid_sessions_instead_of_zero_length_segments(unit): + timestamps = [date(2025, 7, day) for day in (3, 7, 8)] + prices = pl.DataFrame({"timestamp": timestamps, "close": [100.0, 101.0, 103.0]}) + result = calendar_aware_labels(prices, _config(window_unit=unit), feed_spec=_feed()) + assert result["label_return"].to_list() == pytest.approx([0.01, 2 / 101, 0.0]) + assert result["label_time"].to_list() == [timestamps[1], timestamps[2], timestamps[2]] + + +@pytest.mark.parametrize("representation", ["date", "naive", "UTC", "Asia/Tokyo"]) +@pytest.mark.parametrize("unit", ["observation", "expected_bar", "session"]) +@pytest.mark.parametrize("empty", [False, True]) +def test_session_identifiers_do_not_fabricate_physical_elapsed_durations( + representation, unit, empty +): + days = [date(2025, 7, day) for day in (3, 7)] + timestamps = days + if representation != "date": + zone = None if representation == "naive" else ZoneInfo(representation) + timestamps = [datetime.combine(day, time(0, 15), tzinfo=zone) for day in days] + prices = pl.DataFrame({"timestamp": timestamps, "close": [100.0, 101.0]}) + if empty: + prices = prices.filter(pl.col("close") > 1000) + result = calendar_aware_labels(prices, _config(window_unit=unit), feed_spec=_feed()) + assert result.select(prices.columns).equals(prices) + assert result["label_duration"].is_null().all() + assert result.schema["label_duration"] != pl.Object + if not empty: + assert result["label_time"].to_list() == [timestamps[1], timestamps[1]] + assert result["label_bars"].to_list() == [1, 0] + assert result["label_return"].to_list() == pytest.approx([0.01, 0.0]) + buffer = BytesIO() + result.write_parquet(buffer) + buffer.seek(0) + assert pl.read_parquet(buffer).equals(result) + + +@pytest.mark.parametrize("suffix", ["json", "yaml"]) +def test_serialized_session_identifiers_keep_elapsed_duration_unset(tmp_path, suffix): + days = [date(2025, 7, day) for day in (3, 7)] + first = pl.DataFrame( + {"timestamp": days, "close": [100.0, 101.0], "symbol": "A", "event_time": [days[0], None]} + ) + second = first.with_columns((pl.col("close") * 2).alias("close"), pl.lit("B").alias("symbol")) + prices = pl.concat([second, first]).sort("timestamp", "symbol") + config = _config(feed_spec=_feed(entity_col="symbol")) + path = tmp_path / f"labels.{suffix}" + getattr(config, f"to_{suffix}")(path) + restored = getattr(LabelingConfig, f"from_{suffix}")(path) + result = calendar_aware_labels(prices, restored) + assert result["label_duration"].is_null().all() + for symbol in ("A", "B"): + group = result.filter(pl.col("symbol") == symbol) + assert group["label_bars"].to_list() == [1, None] + assert group["label_return"].to_list() == pytest.approx([0.01, None]) + + +def test_standalone_date_labels_keep_legacy_calendar_day_differences(): + prices = pl.DataFrame( + {"timestamp": [date(2025, 7, day) for day in (3, 7)], "close": [100.0, 101.0]} + ) + config = LabelingConfig.triple_barrier( + upper_barrier=None, lower_barrier=None, max_holding_period=1 + ) + result = triple_barrier_labels(prices, config) + assert result["label_duration"][0] == timedelta(days=4) + assert result["label_return"][0] == pytest.approx(0.01) + + +def _intraday(): + from ml4t.engineer.core.bar_grid import _expected_bar_labels + from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + timestamps = list( + _expected_bar_labels( + PandasMarketCalendar("NYSE"), + datetime(2025, 7, 2, tzinfo=UTC), + datetime(2025, 7, 9, tzinfo=UTC), + timedelta(hours=1), + "right", + True, + ) + ) + prices = pl.DataFrame( + {"timestamp": timestamps, "close": [100.0 + i for i in range(len(timestamps))]} + ) + feed = FeedSpec( + time_contract_version=2, + calendar="NYSE", + timestamp_semantics="bar_close", + data_frequency="1h", + timezone="UTC", + bar_type="time", + bar_grid_anchor="session_open", + bar_label="right", + partial_bar_policy="include", + ) + return prices, feed + + +@pytest.mark.parametrize( + "unit,expected", + [ + ("observation", timedelta(hours=1)), + ("expected_bar", timedelta(hours=1)), + ("session", timedelta(hours=26, minutes=30)), + ], +) +def test_instant_labeling_keeps_real_elapsed_durations(unit, expected): + prices, feed = _intraday() + result = calendar_aware_labels(prices, _config(window_unit=unit), feed_spec=feed) + assert result["label_duration"][0] == expected + + +@pytest.mark.parametrize("representation", ["UTC", "America/New_York", "naive"]) +@pytest.mark.parametrize("unit", ["observation", "expected_bar", "elapsed", "session"]) +@pytest.mark.parametrize("sparse", [False, True]) +def test_declared_local_clocks_use_physical_elapsed_time_through_spring_dst( + representation, unit, sparse +): + instants = [datetime(2025, 3, 9, 5, tzinfo=UTC) + timedelta(hours=i) for i in range(6)] + timestamps = pl.Series("timestamp", instants) + if representation != "UTC": + timestamps = timestamps.dt.convert_time_zone("America/New_York") + if representation == "naive": + timestamps = timestamps.dt.replace_time_zone(None) + prices = pl.DataFrame({"timestamp": timestamps, "close": [100.0 + i for i in range(6)]}) + if sparse: + prices = prices.with_columns( + pl.when(pl.col("close") == 101).then(pl.col("timestamp")).alias("event_time") + ) + feed = FeedSpec( + time_contract_version=2, + calendar="24/7", + timezone="America/New_York", + timestamp_semantics="event_time", + data_frequency="1h", + bar_type="time", + bar_grid_anchor="session_open", + bar_label="left", + partial_bar_policy="include", + ) + config = _config("2h" if unit == "elapsed" else 1 if unit == "session" else 2, window_unit=unit) + result = calendar_aware_labels(prices, config, feed_spec=feed) + assert result.select(prices.columns).equals(prices) + assert result.schema["label_time"] == timestamps.dtype + exits = [5] * 6 if unit == "session" else [2, 3, 4, 5, 5, 5] + for entry, exit_index in enumerate(exits): + if sparse and entry != 1: + assert result["label_time"][entry] is None + assert result["label_duration"][entry] is None + continue + assert result["label_time"][entry] == timestamps[exit_index] + assert result["label_bars"][entry] == exit_index - entry + assert result["label_return"][entry] == pytest.approx( + (100 + exit_index) / (100 + entry) - 1 + ) + assert result["label_duration"][entry] == instants[exit_index] - instants[entry] + + +@pytest.mark.parametrize("suffix", ["json", "yaml"]) +@pytest.mark.parametrize("time_unit", ["ms", "us", "ns"]) +@pytest.mark.parametrize("empty", [False, True]) +def test_serialized_local_clock_labels_preserve_panel_columns_and_timestamp_precision( + tmp_path, suffix, time_unit, empty +): + instants = [datetime(2025, 3, 9, 5, tzinfo=UTC) + timedelta(hours=i) for i in range(6)] + timestamps = ( + pl.Series("ts", instants, dtype=pl.Datetime(time_unit, "UTC")) + .dt.convert_time_zone("America/New_York") + .dt.replace_time_zone(None) + ) + first = pl.DataFrame({"ts": timestamps, "price": [100.0 + i for i in range(6)], "asset": "A"}) + second = first.with_columns((pl.col("price") * 2).alias("price"), pl.lit("B").alias("asset")) + prices = pl.concat([first, second]).sort("ts", "asset") + if empty: + prices = prices.filter(pl.col("price") > 1000) + config = _config( + "2h", + window_unit="elapsed", + feed_spec=FeedSpec( + time_contract_version=2, + calendar="24/7", + timezone="America/New_York", + timestamp_semantics="event_time", + timestamp_col="ts", + price_col="price", + entity_col="asset", + ), + ) + path = tmp_path / f"labels.{suffix}" + getattr(config, f"to_{suffix}")(path) + restored = getattr(LabelingConfig, f"from_{suffix}")(path) + result = calendar_aware_labels(prices, restored) + assert result.select(prices.columns).equals(prices.sort("asset", "ts")) + assert result.schema["label_time"] == pl.Datetime(time_unit) + assert result.schema["label_duration"] == pl.Duration(time_unit) + if not empty: + for asset in ("A", "B"): + group = result.filter(pl.col("asset") == asset) + assert group["label_time"][1] == datetime(2025, 3, 9, 4) + assert group["label_duration"][0] == timedelta(hours=2) + assert group["label_return"][1] == pytest.approx(2 / 101) + buffer = BytesIO() + result.write_parquet(buffer) + buffer.seek(0) + assert pl.read_parquet(buffer).equals(result) + + +def test_intraday_session_horizon_reaches_next_session_close_not_lunch_or_overnight_break(): + prices, feed = _intraday() + result = calendar_aware_labels(prices, _config(), feed_spec=feed) + # July 3 closes at 17:00 UTC; July 4 is closed. July 7 closes at 20:00 UTC. + july2 = result.filter(pl.col("timestamp").dt.date() == date(2025, 7, 2)) + july3 = result.filter(pl.col("timestamp").dt.date() == date(2025, 7, 3)) + assert july2["label_time"].to_list() == [datetime(2025, 7, 3, 17, tzinfo=UTC)] * 7 + assert july3["label_time"].to_list() == [datetime(2025, 7, 7, 20, tzinfo=UTC)] * 4 + assert july3["label_bars"].to_list() == [10, 9, 8, 7] + assert result["label_return"][0] == pytest.approx(110 / 100 - 1) + + +def test_missing_entire_scheduled_session_is_rejected_before_any_label_or_input_change(): + prices, feed = _intraday() + incomplete = prices.filter(pl.col("timestamp").dt.date() != date(2025, 7, 3)) + before = incomplete.clone() + with pytest.raises(ValueError, match="missing scheduled bar.*2025-07-03"): + calendar_aware_labels(incomplete, _config(), feed_spec=feed) + assert incomplete.equals(before) + daily = pl.DataFrame( + {"timestamp": [date(2025, 7, 2), date(2025, 7, 7)], "close": [100.0, 103.0]} + ) + with pytest.raises(ValueError, match="missing scheduled session.*2025-07-03"): + calendar_aware_labels(daily, _config(), feed_spec=_feed()) + + +@pytest.mark.parametrize("suffix", ["json", "yaml"]) +def test_session_config_handoff_retains_panel_sparse_events_and_existing_barrier_kernel( + tmp_path, suffix +): + days = [date(2025, 7, day) for day in (2, 3, 7, 8)] + first = pl.DataFrame( + { + "timestamp": days, + "close": [100.0, 101.0, 104.0, 103.0], + "symbol": "A", + "event_time": [days[0], None, days[2], None], + } + ) + second = first.with_columns((pl.col("close") * 2).alias("close"), pl.lit("B").alias("symbol")) + config = _config(2, upper_barrier=0.02, feed_spec=_feed()) + path = tmp_path / f"session.{suffix}" + getattr(config, f"to_{suffix}")(path) + loaded = getattr(LabelingConfig, f"from_{suffix}")(path) + result = calendar_aware_labels(pl.concat([first, second]).reverse(), loaded) + for symbol in ("A", "B"): + asset = result.filter(pl.col("symbol") == symbol) + assert asset["label"].to_list() == [1, None, 0, None] + assert asset["label_time"].to_list() == [days[2], None, days[3], None] + assert asset["label_bars"].to_list() == [2, None, 1, None] + assert asset["barrier_hit"].to_list() == ["upper", None, "time", None] + assert loaded.to_dict()["window_unit"] == "session" + + +@pytest.mark.parametrize("unit", ["session", "elapsed"]) +def test_none_never_becomes_a_continuous_calendar_to_satisfy_temporal_labels(unit): + prices = pl.DataFrame({"close": [100.0, 101.0]}) + with pytest.raises(ValueError, match="observation"): + calendar_aware_labels(prices, _config(), calendar=None, window_unit=unit) + + +def test_elapsed_horizon_rejects_session_identifiers_instead_of_inventing_midnight_instants(): + prices = pl.DataFrame( + {"timestamp": [date(2025, 7, 3), date(2025, 7, 7)], "close": [100.0, 101.0]} + ) + config = LabelingConfig.triple_barrier(max_holding_period="1d", window_unit="elapsed") + with pytest.raises(ValueError, match="session_label.*elapsed"): + calendar_aware_labels(prices, config, feed_spec=_feed()) + + +@pytest.mark.parametrize("counts", [[0, 1, 1], [1.5, 1.0, 1.0], [float("nan"), 1.0, 1.0]]) +def test_session_count_columns_reject_non_positive_non_integral_or_missing_counts(counts): + prices = pl.DataFrame( + { + "timestamp": [date(2025, 7, day) for day in (3, 7, 8)], + "close": [100.0, 101.0, 103.0], + "horizon": counts, + } + ) + with pytest.raises(ValueError, match="finite positive integral"): + calendar_aware_labels(prices, _config("horizon"), feed_spec=_feed()) + + +def test_per_event_session_counts_and_temporary_column_collisions_preserve_original_data(): + prices = pl.DataFrame( + { + "timestamp": [date(2025, 7, day) for day in (2, 3, 7, 8)], + "close": [100.0, 101.0, 103.0, 105.0], + "horizon": [2, 1, 1, 1], + "__ml4t_session_holding_bars": [19] * 4, + } + ) + result = calendar_aware_labels(prices, _config("horizon"), feed_spec=_feed()) + assert result["label_bars"].to_list() == [2, 1, 1, 0] + assert result.select(prices.columns).equals(prices) + + +def test_overnight_cme_daily_session_labels_do_not_shift_to_the_prior_civil_date(): + days = [date(2025, 7, day) for day in (7, 8, 9)] + prices = pl.DataFrame({"timestamp": days, "close": [100.0, 101.0, 103.0]}) + result = calendar_aware_labels(prices, _config(), feed_spec=_feed(), calendar="CME_Equity") + assert result["label_time"].to_list() == [days[1], days[2], days[2]] + assert result["label_return"].to_list() == pytest.approx([0.01, 2 / 101, 0.0]) + + +def test_explicit_continuous_session_counts_include_weekends_without_becoming_row_mode(): + from dataclasses import replace + + days = [date(2025, 7, day) for day in (4, 5, 6, 7)] + prices = pl.DataFrame({"timestamp": days, "close": [100.0, 101.0, 103.0, 105.0]}) + result = calendar_aware_labels(prices, _config(2), feed_spec=replace(_feed(), calendar="24/7")) + assert result["label_time"].to_list() == [days[2], days[3], days[3], days[3]] + with pytest.raises(ValueError, match="missing scheduled session"): + calendar_aware_labels( + prices.filter(pl.col("timestamp") != days[1]), + _config(), + calendar="24/7", + feed_spec=_feed(), + ) + + +@pytest.mark.parametrize("side", ["left", "right"]) +@pytest.mark.parametrize("sparse", [False, True]) +@pytest.mark.parametrize("horizon,first_exit", [(1, 0), (2, 1)]) +def test_barless_short_session_never_pushes_a_session_deadline_into_a_later_session( + side, sparse, horizon, first_exit +): + from ml4t.engineer.core.bar_grid import _expected_bar_labels + from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + timestamps = list( + _expected_bar_labels( + PandasMarketCalendar("NYSE"), + datetime(2025, 7, 2, tzinfo=UTC), + datetime(2025, 7, 9, tzinfo=UTC), + timedelta(hours=5), + side, + False, + ) + ) + assert [stamp.date() for stamp in timestamps] == [date(2025, 7, d) for d in (2, 7, 8)] + prices = pl.DataFrame({"timestamp": timestamps, "close": [100.0, 101.0, 102.0]}) + if sparse: + prices = prices.with_columns(pl.Series("event_time", [timestamps[0], None, timestamps[2]])) + feed = FeedSpec( + time_contract_version=2, + calendar="NYSE", + timezone="UTC", + data_frequency="5h", + timestamp_semantics="bar_close" if side == "right" else "event_time", + bar_type="time", + bar_grid_anchor="session_open", + bar_label=side, + partial_bar_policy="exclude", + ) + result = calendar_aware_labels(prices, _config(horizon), feed_spec=feed) + exits = [first_exit, None if sparse else 2, 2] + assert result["label_time"].to_list() == [ + timestamps[index] if index is not None else None for index in exits + ] + assert result["label_bars"].to_list() == [ + index - entry if index is not None else None for entry, index in enumerate(exits) + ] + assert result["label_return"][0] == pytest.approx(first_exit / 100) + + +@pytest.mark.parametrize("representation", ["UTC", "America/New_York", "naive"]) +def test_session_horizons_retain_actual_exchange_close_through_spring_dst(representation): + from ml4t.engineer.core.bar_grid import _expected_bar_labels + from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + timestamps = list( + _expected_bar_labels( + PandasMarketCalendar("NYSE"), + datetime(2025, 3, 7, tzinfo=UTC), + datetime(2025, 3, 12, tzinfo=UTC), + timedelta(hours=1), + "right", + True, + ) + ) + series = pl.Series("timestamp", timestamps) + if representation != "UTC": + series = series.dt.convert_time_zone("America/New_York") + if representation == "naive": + series = series.dt.replace_time_zone(None) + prices = pl.DataFrame( + {"timestamp": series, "close": [100.0 + i for i in range(len(timestamps))]} + ) + feed = FeedSpec( + time_contract_version=2, + calendar="NYSE", + timezone="America/New_York", + timestamp_semantics="bar_close", + data_frequency="1h", + bar_type="time", + bar_grid_anchor="session_open", + bar_label="right", + partial_bar_policy="include", + ) + result = calendar_aware_labels(prices, _config(), feed_spec=feed) + expected = datetime(2025, 3, 10, 20, tzinfo=UTC) + actual = result["label_time"] + if representation == "naive": + actual = actual.dt.replace_time_zone("America/New_York") + assert actual.dt.convert_time_zone("UTC")[0] == expected + assert result["label_bars"][0] == 13 + assert result["label_return"][0] == pytest.approx(0.13) + assert result["label_duration"][0] == timedelta(hours=76, minutes=30) + assert result.select(prices.columns).equals(prices) + + +def test_overnight_cme_session_horizon_counts_exchange_labels_not_utc_civil_dates(): + from ml4t.engineer.core.bar_grid import _expected_bar_labels + from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + timestamps = list( + _expected_bar_labels( + PandasMarketCalendar("CME_Equity"), + datetime(2025, 7, 6, tzinfo=UTC), + datetime(2025, 7, 10, tzinfo=UTC), + timedelta(hours=1), + "right", + True, + ) + ) + assert timestamps[0] == datetime(2025, 7, 6, 23, tzinfo=UTC) + prices = pl.DataFrame( + {"timestamp": timestamps, "close": [100.0 + i for i in range(len(timestamps))]} + ) + feed = FeedSpec( + time_contract_version=2, + calendar="CME_Equity", + timezone="UTC", + timestamp_semantics="bar_close", + data_frequency="1h", + bar_type="time", + bar_grid_anchor="session_open", + bar_label="right", + partial_bar_policy="include", + ) + result = calendar_aware_labels(prices, _config(), feed_spec=feed) + # Sunday evening belongs to Monday's exchange session. One following session ends Tuesday. + assert result["label_time"][0] == datetime(2025, 7, 8, 21, tzinfo=UTC) + assert result["label_bars"][0] == 45 + assert result["label_return"][0] == pytest.approx(0.45) diff --git a/tests/test_labeling_time_contract.py b/tests/test_labeling_time_contract.py new file mode 100644 index 0000000..05d38e7 --- /dev/null +++ b/tests/test_labeling_time_contract.py @@ -0,0 +1,319 @@ +from copy import deepcopy +from datetime import UTC, date, datetime, timedelta +from io import BytesIO + +import polars as pl +import pytest +from ml4t.specs import FeedSpec +from polars.testing import assert_frame_equal + +from ml4t.engineer.config import LabelingConfig +from ml4t.engineer.core.exceptions import DataValidationError +from ml4t.engineer.labeling import calendar_aware_labels, triple_barrier_labels +from ml4t.engineer.labeling.calendar import PandasMarketCalendar + + +def _config(**kwargs): + return LabelingConfig.triple_barrier( + upper_barrier=0.5, lower_barrier=0.5, max_holding_period=1, **kwargs + ) + + +def _prices(timestamps=None): + values = {"row": [0, 1, 2, 3], "close": [100.0, 101.0, 99.0, 102.0]} + if timestamps is not None: + values["timestamp"] = timestamps + return pl.DataFrame(values) + + +@pytest.mark.parametrize( + "timestamps", + [ + None, + [datetime(2025, 7, 4) + timedelta(hours=i) for i in range(4)], + [datetime(2025, 7, day) for day in (14, 4, 31, 7)], + ], +) +def test_explicit_none_labels_ordered_observations_without_loading_calendar( + monkeypatch, timestamps +): + import sys + + monkeypatch.setitem(sys.modules, "pandas_market_calendars", None) + result = calendar_aware_labels(_prices(timestamps), _config(), calendar=None) + assert result["row"].to_list() == [0, 1, 2, 3] + assert result["label_return"].to_list() == pytest.approx([0.01, -2 / 101, 3 / 99, 0.0]) + assert result["label_bars"].to_list() == [1, 1, 1, 0] + + +def test_row_mode_keeps_interleaved_asset_order_and_scans_each_asset_independently(): + data = pl.DataFrame( + { + "symbol": ["A", "B", "A", "B"], + "close": [100.0, 200.0, 101.0, 198.0], + "timestamp": [datetime(2025, 7, d) for d in (14, 11, 4, 7)], + "row": [0, 1, 2, 3], + } + ) + result = calendar_aware_labels(data, _config(), calendar=None) + assert result["row"].to_list() == [0, 1, 2, 3] + assert result["label_return"].to_list() == pytest.approx([0.01, -0.01, 0.0, 0.0]) + + +@pytest.mark.parametrize("suffix", ["json", "yaml"]) +@pytest.mark.parametrize( + "declaration,expected", + [({}, "NYSE"), ({"calendar": None}, None), ({"calendar": "24/7"}, "24/7")], +) +def test_real_config_roundtrip_preserves_omitted_named_and_disabled_calendar( + tmp_path, suffix, declaration, expected +): + config = _config(**declaration) + path = tmp_path / f"labeling.{suffix}" + getattr(config, f"to_{suffix}")(path) + loaded = getattr(LabelingConfig, f"from_{suffix}")(path) + copied = deepcopy(loaded).model_copy() + assert ("calendar" in copied.to_dict()) == ("calendar" in declaration) + saturday = pl.DataFrame({"timestamp": [datetime(2025, 7, 5, 14, tzinfo=UTC)], "close": [100.0]}) + if expected == "NYSE": + with pytest.raises(DataValidationError, match="outside"): + calendar_aware_labels(saturday, copied, window_unit="observation") + else: + assert calendar_aware_labels(saturday, copied, window_unit="observation").height == 1 + + +def test_feed_metadata_precedence_and_declared_null_survive_nested_json_handoff(tmp_path): + feed = FeedSpec( + time_contract_version=2, + calendar=None, + price_col="mid", + timestamp_col="time", + entity_col=None, + ) + config = _config(feed_spec=feed) + path = tmp_path / "metadata.json" + config.to_json(path) + loaded = LabelingConfig.from_json(path) + data = pl.DataFrame( + {"mid": [100.0, 101.0], "time": [datetime(2025, 7, 5), datetime(2025, 7, 4)]} + ) + result = calendar_aware_labels(data, loaded) + assert result["label_return"].to_list() == pytest.approx([0.01, 0.0]) + assert loaded.to_dict()["feed_spec"]["calendar"] is None + with pytest.raises(DataValidationError, match="outside"): + calendar_aware_labels( + data, + loaded, + calendar="NYSE", + feed_spec=FeedSpec( + time_contract_version=2, + timezone="UTC", + timestamp_semantics="event_time", + price_col="mid", + timestamp_col="time", + entity_col=None, + ), + window_unit="observation", + ) + + +def test_explicit_argument_overrides_config_and_metadata_without_mutating_either(): + data = _prices([datetime(2025, 7, 5) + timedelta(hours=i) for i in range(4)]) + config = _config(calendar="NYSE", feed_spec=FeedSpec(time_contract_version=2, calendar="LSE")) + before = config.to_dict() + result = calendar_aware_labels(data, config, calendar=None) + assert result.height == 4 + assert config.to_dict() == before + assert_frame_equal(data, _prices([datetime(2025, 7, 5) + timedelta(hours=i) for i in range(4)])) + + +def test_named_schedule_uses_declared_naive_timezone_instead_of_legacy_utc(): + data = pl.DataFrame( + { + "timestamp": [datetime(2025, 7, 7, 9, 30), datetime(2025, 7, 7, 10)], + "close": [100.0, 101.0], + } + ) + feed = FeedSpec( + time_contract_version=2, + calendar="NYSE", + timezone="America/New_York", + timestamp_semantics="event_time", + ) + result = calendar_aware_labels(data, _config(), feed_spec=feed, window_unit="observation") + assert result["label_return"].to_list() == pytest.approx([0.01, 0.0]) + assert_frame_equal(result.select(data.columns), data) + + +@pytest.mark.parametrize("unit", ["session", "elapsed", "expected_bar"]) +def test_disabled_calendar_rejects_non_observation_units(unit): + with pytest.raises(ValueError, match="observation"): + calendar_aware_labels(_prices(), _config(), calendar=None, window_unit=unit) + + +def test_expected_bar_labels_reject_price_outage_before_horizon_interpretation(): + feed = FeedSpec( + time_contract_version=2, + calendar="24/7", + timezone="UTC", + data_frequency="1h", + timestamp_semantics="bar_close", + bar_type="time", + bar_grid_anchor="session_open", + bar_label="right", + partial_bar_policy="include", + ) + complete = pl.DataFrame( + { + "timestamp": [datetime(2025, 7, 7, i, tzinfo=UTC) for i in range(1, 5)], + "close": [100.0, 101.0, 99.0, 102.0], + } + ) + assert ( + calendar_aware_labels( + complete, _config(), feed_spec=feed, window_unit="expected_bar" + ).height + == 4 + ) + outage = complete.filter(pl.col("timestamp").dt.hour() != 2) + with pytest.raises(ValueError, match="missing scheduled bar"): + calendar_aware_labels(outage, _config(), feed_spec=feed, window_unit="expected_bar") + + +@pytest.mark.parametrize( + "declaration,error", + [ + ({"timestamp_semantics": "session_label"}, "declared bar grid"), + ({"timestamp_semantics": "event_time"}, "timezone"), + ], +) +def test_named_schedule_rejects_ambiguous_timestamp_declaration_atomically(declaration, error): + data = _prices([datetime(2025, 7, 7, 14) + timedelta(minutes=i) for i in range(4)]) + original = data.clone() + with pytest.raises(ValueError, match=error): + calendar_aware_labels( + data, _config(), feed_spec=FeedSpec(time_contract_version=2, **declaration) + ) + assert_frame_equal(data, original) + + +def test_close_labeled_price_at_lunch_break_is_valid_and_does_not_cross_the_break(): + data = pl.DataFrame( + { + "timestamp": [ + datetime(2025, 7, 7, h, m, tzinfo=UTC) for h, m in [(3, 30), (4, 0), (5, 30)] + ], + "close": [100.0, 101.0, 110.0], + } + ) + feed = FeedSpec(time_contract_version=2, calendar="HKEX", timestamp_semantics="bar_close") + result = calendar_aware_labels(data, _config(), feed_spec=feed, window_unit="observation") + assert result["label_return"].to_list() == pytest.approx([0.01, 0.0, 0.0]) + assert_frame_equal(result.select(data.columns), data) + + +def test_named_feed_does_not_infer_timestamp_meaning_from_aware_values(): + data = _prices([datetime(2025, 7, 7, 14, tzinfo=UTC) + timedelta(minutes=i) for i in range(4)]) + with pytest.raises(ValueError, match="timestamp_semantics"): + calendar_aware_labels(data, _config(), feed_spec=FeedSpec(time_contract_version=2)) + + +def test_legacy_kernel_does_not_silently_ignore_a_declared_calendar_opt_out(): + data = _prices([datetime(2025, 7, day) for day in (14, 4, 31, 7)]) + with pytest.raises(DataValidationError, match="calendar_aware_labels"): + triple_barrier_labels(data, _config(calendar=None)) + + +def test_new_default_does_not_treat_an_undeclared_intraday_grid_as_complete(): + data = _prices([datetime(2025, 7, 7, 14, tzinfo=UTC) + timedelta(hours=i) for i in range(4)]) + with pytest.raises(ValueError, match="declared bar grid"): + calendar_aware_labels(data, _config()) + + +def test_config_equality_does_not_collapse_disabled_and_undeclared_calendars(): + inherited = _config() + disabled = inherited.model_copy(update={"calendar": None}) + assert disabled != inherited + assert disabled == deepcopy(disabled) + assert disabled.to_dict()["calendar"] is None + assert "calendar" not in inherited.to_dict() + + +@pytest.mark.parametrize("mode", ["python", "json"]) +def test_declared_null_calendar_does_not_leak_into_explicit_serialization_projection(mode): + config = _config(calendar=None) + assert config.model_dump(mode=mode, include={"method"}) == {"method": "triple_barrier"} + assert "calendar" not in config.model_dump(mode=mode, exclude={"calendar"}, exclude_none=True) + assert config.to_dict(mode=mode, exclude_none=True)["calendar"] is None + + +def test_schedule_adapter_canonical_calendar_preserves_legacy_selector_and_rejects_conflicts(): + assert PandasMarketCalendar(calendar="LSE").calendar_name == "LSE" + assert PandasMarketCalendar(calendar_name="LSE", calendar="LSE").calendar == "LSE" + assert PandasMarketCalendar().calendar == "NYSE" + with pytest.raises(ValueError, match="conflict"): + PandasMarketCalendar(calendar_name="NYSE", calendar="LSE") + with pytest.raises(ValueError, match="named calendar"): + PandasMarketCalendar(calendar=None) + + +@pytest.mark.parametrize("precision", ["us", "ns"]) +@pytest.mark.parametrize("workflow", ["calendar", "row", "standalone"]) +def test_asset_without_selected_events_keeps_compatible_null_label_schema(precision, workflow): + data = pl.DataFrame( + { + "symbol": ["A", "B", "A", "B"], + "timestamp": [datetime(2025, 7, 7, hour, tzinfo=UTC) for hour in (14, 14, 15, 15)], + "close": [100.0, 200.0, 101.0, 201.0], + } + ).with_columns(pl.col("timestamp").cast(pl.Datetime(precision, "UTC"))) + selected = data.with_columns( + pl.when(pl.col("symbol") == "A") + .then(pl.col("timestamp")) + .otherwise(None) + .alias("event_time") + ) + if workflow == "standalone": + result = triple_barrier_labels(selected, _config(), calculate_uniqueness=True) + assert result.filter(pl.col("symbol") == "B")["sample_weight"].is_null().all() + else: + result = calendar_aware_labels( + selected, _config(), calendar=None if workflow == "row" else "NYSE" + ) + assert result.height == data.height + inactive = result.filter(pl.col("symbol") == "B") + assert inactive["label_time"].is_null().all() + assert inactive["label_duration"].is_null().all() + assert result.filter(pl.col("symbol") == "A")["label_return"].to_list() == pytest.approx( + [0.01, 0.0] + ) + assert result["label_time"].dtype == ( + pl.Int64 if workflow == "row" else data["timestamp"].dtype + ) + + +def test_legacy_date_labels_report_calendar_day_difference_without_numpy_resolution_error(): + data = pl.DataFrame( + { + "timestamp": [date(2025, 7, day) for day in (3, 7, 8)], + "close": [100.0, 101.0, 99.0], + } + ) + result = triple_barrier_labels(data, _config()) + assert result["label_time"].dtype == pl.Date + assert result["label_time"].to_list() == [date(2025, 7, 7), date(2025, 7, 8), date(2025, 7, 8)] + assert result["label_duration"].to_list() == [ + timedelta(days=4), + timedelta(days=1), + timedelta(0), + ] + assert result["label_bars"].to_list() == [1, 1, 0] + + +def test_row_labels_persist_without_untyped_python_object_duration_metadata(): + result = calendar_aware_labels(_prices(), _config(), calendar=None) + target = BytesIO() + result.write_parquet(target) + target.seek(0) + assert_frame_equal(pl.read_parquet(target), result) + assert result["label_duration"].is_null().all() diff --git a/tests/test_release_policy.py b/tests/test_release_policy.py index eef78f6..5238b81 100644 --- a/tests/test_release_policy.py +++ b/tests/test_release_policy.py @@ -238,7 +238,7 @@ def test_ci_audits_core_and_complete_locked_environments() -> None: assert "complete.txt" in audit -def test_core_dependency_uses_bounded_specs_or_immutable_source_for_qualification() -> None: +def test_core_dependency_requires_published_time_contract_v2() -> None: project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8")) lock = tomllib.loads((ROOT / "uv.lock").read_text(encoding="utf-8")) @@ -248,17 +248,10 @@ def test_core_dependency_uses_bounded_specs_or_immutable_source_for_qualificatio if Requirement(value).name == "ml4t-specs" ) specs = next(package for package in lock["package"] if package["name"] == "ml4t-specs") - if requirement.url is None: - assert str(requirement.specifier) == "<0.2.0,>=0.1.0" - assert specs["version"] in requirement.specifier - else: - assert re.fullmatch( - r"git\+https://github.com/ml4t/specs\.git@[0-9a-f]{40}", requirement.url - ) - commit = requirement.url.rsplit("@", maxsplit=1)[1] - assert specs["source"] == { - "git": f"https://github.com/ml4t/specs.git?rev={commit}#{commit}" - } + assert requirement.url is None + assert str(requirement.specifier) == "<0.3.0,>=0.2.0" + assert specs["version"] in requirement.specifier + assert specs["source"] == {"registry": "https://pypi.org/simple"} def test_package_supports_python_312_through_314() -> None: diff --git a/uv.lock b/uv.lock index af62738..c5c6d49 100644 --- a/uv.lock +++ b/uv.lock @@ -1651,7 +1651,7 @@ requires-dist = [ { name = "mkdocs-material", marker = "extra == 'docs'", specifier = ">=9.5.0" }, { name = "mkdocstrings", extras = ["python"], marker = "extra == 'all'", specifier = ">=0.24.0" }, { name = "mkdocstrings", extras = ["python"], marker = "extra == 'docs'", specifier = ">=0.24.0" }, - { name = "ml4t-specs", git = "https://github.com/ml4t/specs.git?rev=701bde68744035050ede386efce498234c661c21" }, + { name = "ml4t-specs", specifier = ">=0.2.0,<0.3.0" }, { name = "numba", specifier = ">=0.57.0" }, { name = "numpy", specifier = ">=1.24.0,<2.6" }, { name = "pandas", specifier = ">=2.0.0" }, @@ -1728,11 +1728,15 @@ test = [ [[package]] name = "ml4t-specs" -version = "0.1.5" -source = { git = "https://github.com/ml4t/specs.git?rev=701bde68744035050ede386efce498234c661c21#701bde68744035050ede386efce498234c661c21" } +version = "0.2.0" +source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pyyaml" }, ] +sdist = { url = "https://files.pythonhosted.org/packages/2e/46/c734e5bcac1f214c83423243023b5d313bccabd25563da4e7d5e6b80f9f4/ml4t_specs-0.2.0.tar.gz", hash = "sha256:e5d5d614f71d14223341f992009d0d0bf14f30a8398c50872d2b3fb80fd33d82", size = 59749, upload-time = "2026-10-17T18:21:32.512Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/66/ff/bfc427bb71635f31cb240330aa2b1b486748999f05a5922ee5f58df721fe/ml4t_specs-0.2.0-py3-none-any.whl", hash = "sha256:cfd550d19b469f1b083d6ac3f83b235fe71aee19c28465f2ac2c8620064bae61", size = 36946, upload-time = "2026-10-17T18:21:30.964Z" }, +] [[package]] name = "more-itertools"