From b93bdc6203600520eef062f13472ae095d9326a0 Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 11:34:05 -0400 Subject: [PATCH 1/2] test: reproduce calendar and outage errors through continuous futures construction --- tests/futures/test_scheduled_roll_workflow.py | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 tests/futures/test_scheduled_roll_workflow.py diff --git a/tests/futures/test_scheduled_roll_workflow.py b/tests/futures/test_scheduled_roll_workflow.py new file mode 100644 index 00000000..d42a3785 --- /dev/null +++ b/tests/futures/test_scheduled_roll_workflow.py @@ -0,0 +1,53 @@ +"""Provider-generated contract files must use scheduled, not observed, expiry offsets.""" + +from datetime import date + +import polars as pl + +from ml4t.data.futures.adjustment import NoAdjustment +from ml4t.data.futures.continuous import ContinuousContractBuilder +from ml4t.data.futures.roll import TimeBasedRoll +from ml4t.data.providers.synthetic import SyntheticProvider + + +def write_contract_files(root, expiration, *, missing=None): + with SyntheticProvider() as provider: + frame = provider.fetch_ohlcv("TEST", "2025-07-01", "2025-07-11", "daily") + frame = frame.filter(pl.col("timestamp").dt.date() != date(2025, 7, 4)) + if missing is not None: + frame = frame.filter(pl.col("timestamp").dt.date() != missing) + frames = [] + for symbol in ("TESTN2025", "TESTU2025"): + frames.append(frame.with_columns(pl.lit(symbol).alias("symbol"))) + ohlcv = root / "ohlcv_1d" / "product=TEST" + definitions = root / "definition" / "product=TEST" + ohlcv.mkdir(parents=True) + definitions.mkdir(parents=True) + pl.concat(frames).write_parquet(ohlcv / "ohlcv_1d.parquet") + pl.DataFrame( + { + "raw_symbol": ["TESTN2025", "TESTU2025"], + "expiration": [expiration, date(2025, 9, 19)], + } + ).write_parquet(definitions / "definition.parquet") + + +def build(root, strategy=None): + return ContinuousContractBuilder( + roll_strategy=strategy or TimeBasedRoll(5), + adjustment_method=NoAdjustment(), + ).build("TEST", data_source="databento", storage_path=root) + + +def test_continuous_workflow_counts_the_july4_closure_in_the_expiry_offset(tmp_path): + write_contract_files(tmp_path, date(2025, 7, 10)) + result = build(tmp_path) + assert result.filter(pl.col("date") == date(2025, 7, 2))["symbol"].to_list() == ["TESTU2025"] + assert result.filter(pl.col("is_roll_date"))["date"].to_list() == [date(2025, 7, 2)] + + +def test_missing_scheduled_roll_session_cannot_move_the_roll_before_the_outage(tmp_path): + write_contract_files(tmp_path, date(2025, 7, 11), missing=date(2025, 7, 3)) + result = build(tmp_path) + assert result.filter(pl.col("date") == date(2025, 7, 2))["symbol"].to_list() == ["TESTN2025"] + assert result.filter(pl.col("is_roll_date"))["date"].to_list() == [date(2025, 7, 7)] From 720cf76f6f537ad58816d07f8580ef2c812cd24a Mon Sep 17 00:00:00 2001 From: Stefan Jansen Date: Sat, 17 Oct 2026 11:37:24 -0400 Subject: [PATCH 2/2] fix: derive futures expiry roll triggers from authoritative sessions --- docs/api/index.md | 41 +++++++++- src/ml4t/data/futures/roll.py | 78 ++++++++++++++----- tests/futures/test_scheduled_roll_workflow.py | 52 ++++++++++++- 3 files changed, 148 insertions(+), 23 deletions(-) diff --git a/docs/api/index.md b/docs/api/index.md index 41af145c..4387e59a 100644 --- a/docs/api/index.md +++ b/docs/api/index.md @@ -1,6 +1,6 @@ # API Reference -Complete API documentation for the `ml4t-data` library, auto-generated from +API documentation for the `ml4t-data` library, generated from source docstrings via [mkdocstrings](https://mkdocstrings.github.io/). --- @@ -420,3 +420,42 @@ ML4TDataError options: show_source: false heading_level: 4 + +## Futures Expiry Rolls + +`TimeBasedRoll` selects the nearest available contract until its configured pre-expiry trigger. +`FirstNoticeDateRoll` delegates the same calendar selection while retaining its existing +first-notice offset convention and five-session cash-settled fallback. + +| Mode | Offset units | Calendar behavior | +|------|--------------|-------------------| +| `use_business_days=True` (default) | Authoritative sessions strictly before expiration | Explicit `calendar`, then `feed_spec`, then NYSE | +| `use_business_days=False` | Elapsed civil days | No session counting; an explicitly disabled calendar is allowed | +| Session-count mode with `calendar=None` | Unsupported observation-based expiry offset | Raises clearly; does not substitute NYSE or 24/7 | + +The legacy `use_business_days` name now counts the selected exchange's sessions rather than +weekdays. `calendar="24/7"` counts a real continuous schedule. Contract exchange identity does not +select a calendar. Pass the feed declaration explicitly when constructing a roll strategy from +stored metadata. Named calendars are validated before selection. + +An offset of zero uses expiration itself. A missing scheduled trigger does not move the trigger +backward to an observed date: selection switches at the first eligible later observation, subject +to the existing contract availability and paired-price checks. Explicit elapsed-day mode also +stops snapping its target backward. Neither mode reinterprets missing data as a closure or changes +elapsed days into row counts. + +::: ml4t.data.futures.roll.TimeBasedRoll + options: + show_source: false + heading_level: 3 + members: + - __init__ + - select_contracts + +::: ml4t.data.futures.roll.FirstNoticeDateRoll + options: + show_source: false + heading_level: 3 + members: + - __init__ + - select_contracts diff --git a/src/ml4t/data/futures/roll.py b/src/ml4t/data/futures/roll.py index 08ab0176..2908ea69 100644 --- a/src/ml4t/data/futures/roll.py +++ b/src/ml4t/data/futures/roll.py @@ -6,10 +6,13 @@ from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import date, timedelta +from typing import Any import polars as pl from ml4t.data.futures.schema import ContractSpec +from ml4t.data.sessions.assigner import SessionAssigner +from ml4t.specs import UNSET_CALENDAR, CalendarUnset, FeedSpec @dataclass(frozen=True) @@ -99,17 +102,34 @@ def select_contracts( class TimeBasedRoll(RollStrategy): - """Select the nearest contract until its configured pre-expiry roll date.""" + """Select the nearest contract until its configured pre-expiry roll date. + + ``use_business_days=True`` counts authoritative sessions strictly before expiration. + Explicit ``calendar`` overrides ``feed_spec``, then NYSE; an exchange identity does not + select a calendar. A missing target session never moves the trigger to an earlier observation. + ``use_business_days=False`` selects elapsed civil days, not an observation count. + Calendar-free row-based expiry offsets are unsupported in session-count mode. + """ def __init__( self, days_before_expiration: int = 5, use_business_days: bool = True, - ): + *, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, + feed_spec: Any | None = None, + ) -> None: if days_before_expiration < 0: raise ValueError("days_before_expiration cannot be negative") self.days_before_expiration = days_before_expiration self.use_business_days = use_business_days + self.feed_spec = FeedSpec.from_any(feed_spec) + self.calendar = self.feed_spec.resolve_calendar(calendar=calendar) + if self.calendar is None and use_business_days: + raise ValueError("calendar-free row-based expiry rolls are not supported") + self._schedule_calendar = ( + SessionAssigner(self.calendar).calendar if self.calendar is not None else None + ) def select_contracts( self, @@ -122,7 +142,7 @@ def select_contracts( return _empty_selections() contracts = dict(_contract_expirations(data)) roll_dates = { - symbol: self._calculate_roll_date(expiration, dates) + symbol: self._calculate_roll_date(expiration) for symbol, expiration in contracts.items() } available_by_date = _available_symbols_by_date(data) @@ -139,38 +159,54 @@ def select_contracts( ) return _selection_frame(records) - def _calculate_roll_date(self, expiration: date, trading_dates: list[date]) -> date: - """Calculate the last configured roll date on the observed calendar.""" - target = expiration - if self.use_business_days: - days_counted = 0 - while days_counted < self.days_before_expiration: - target -= timedelta(days=1) - if target.weekday() < 5: - days_counted += 1 - else: - target -= timedelta(days=self.days_before_expiration) - if target >= trading_dates[-1] or target < trading_dates[0]: - return target - observed = [trading_date for trading_date in trading_dates if trading_date <= target] - return max(observed) if observed else target + def _calculate_roll_date(self, expiration: date) -> date: + """Calculate a schedule or explicit elapsed-day trigger independently of observations.""" + if not self.use_business_days: + return expiration - timedelta(days=self.days_before_expiration) + if self.days_before_expiration == 0: + return expiration + span = max(14, self.days_before_expiration * 2) + while True: + try: + start = expiration - timedelta(days=span) + except OverflowError as error: + raise ValueError( + "not enough calendar history for the requested expiry offset" + ) from error + schedule = self._schedule_calendar.schedule( + start_date=start, end_date=expiration - timedelta(days=1) + ) + if len(schedule) >= self.days_before_expiration: + return schedule.index[-self.days_before_expiration].date() + span *= 2 class FirstNoticeDateRoll(RollStrategy): """Select physical contracts before the configured first-notice interval.""" - def __init__(self, days_before_first_notice: int = 1): + def __init__( + self, + days_before_first_notice: int = 1, + *, + calendar: str | None | CalendarUnset = UNSET_CALENDAR, + feed_spec: Any | None = None, + ) -> None: if days_before_first_notice < 0: raise ValueError("days_before_first_notice cannot be negative") self.days_before_first_notice = days_before_first_notice + self._time_roll = TimeBasedRoll(calendar=calendar, feed_spec=feed_spec) def select_contracts( self, data: pl.DataFrame, contract_spec: ContractSpec | None = None ) -> pl.DataFrame: if contract_spec is None or contract_spec.is_cash_settled: - return TimeBasedRoll(days_before_expiration=5).select_contracts(data, contract_spec) + return self._time_roll.select_contracts(data, contract_spec) days = (contract_spec.first_notice_days or 25) + self.days_before_first_notice - return TimeBasedRoll(days_before_expiration=days).select_contracts(data, contract_spec) + return TimeBasedRoll( + days_before_expiration=days, + calendar=self._time_roll.calendar, + feed_spec=self._time_roll.feed_spec, + ).select_contracts(data, contract_spec) class CalendarRoll(RollStrategy): diff --git a/tests/futures/test_scheduled_roll_workflow.py b/tests/futures/test_scheduled_roll_workflow.py index d42a3785..bcdaca8d 100644 --- a/tests/futures/test_scheduled_roll_workflow.py +++ b/tests/futures/test_scheduled_roll_workflow.py @@ -3,11 +3,13 @@ from datetime import date import polars as pl +import pytest from ml4t.data.futures.adjustment import NoAdjustment from ml4t.data.futures.continuous import ContinuousContractBuilder -from ml4t.data.futures.roll import TimeBasedRoll +from ml4t.data.futures.roll import FirstNoticeDateRoll, TimeBasedRoll from ml4t.data.providers.synthetic import SyntheticProvider +from ml4t.specs import FeedSpec def write_contract_files(root, expiration, *, missing=None): @@ -51,3 +53,51 @@ def test_missing_scheduled_roll_session_cannot_move_the_roll_before_the_outage(t result = build(tmp_path) assert result.filter(pl.col("date") == date(2025, 7, 2))["symbol"].to_list() == ["TESTN2025"] assert result.filter(pl.col("is_roll_date"))["date"].to_list() == [date(2025, 7, 7)] + + +@pytest.mark.parametrize("strategy", [TimeBasedRoll, FirstNoticeDateRoll]) +@pytest.mark.parametrize("explicit,expected", [("NYSE", "TESTU2025"), ("24/7", "TESTN2025")]) +def test_roll_and_first_notice_delegate_preserve_calendar_precedence( + tmp_path, strategy, explicit, expected +): + write_contract_files(tmp_path, date(2025, 7, 10)) + feed = FeedSpec(time_contract_version=2, calendar="24/7") + result = build(tmp_path, strategy(calendar=explicit, feed_spec=feed)) + assert result.filter(pl.col("date") == date(2025, 7, 3))["symbol"].to_list() == [expected] + + +@pytest.mark.parametrize("strategy", [TimeBasedRoll, FirstNoticeDateRoll]) +def test_declared_continuous_schedule_is_not_replaced_by_default_nyse(tmp_path, strategy): + write_contract_files(tmp_path, date(2025, 7, 10)) + result = build(tmp_path, strategy(feed_spec=FeedSpec(time_contract_version=2, calendar="24/7"))) + assert result.filter(pl.col("is_roll_date"))["date"].to_list() == [date(2025, 7, 7)] + + +def test_elapsed_day_mode_does_not_snap_the_trigger_before_a_missing_weekend(tmp_path): + write_contract_files(tmp_path, date(2025, 7, 10)) + result = build(tmp_path, TimeBasedRoll(5, False)) + assert result.filter(pl.col("date") == date(2025, 7, 3))["symbol"].to_list() == ["TESTN2025"] + assert result.filter(pl.col("is_roll_date"))["date"].to_list() == [date(2025, 7, 7)] + + +def test_explicit_elapsed_units_need_no_calendar_but_do_not_become_row_counts(tmp_path): + write_contract_files(tmp_path, date(2025, 7, 10)) + result = build(tmp_path, TimeBasedRoll(5, False, calendar=None)) + assert result.filter(pl.col("is_roll_date"))["date"].to_list() == [date(2025, 7, 7)] + + +@pytest.mark.parametrize("strategy", [TimeBasedRoll, FirstNoticeDateRoll]) +def test_disabled_session_count_roll_fails_instead_of_substituting_a_continuous_schedule(strategy): + with pytest.raises(ValueError, match="row-based"): + strategy(calendar=None) + with pytest.raises(ValueError, match="row-based"): + strategy(feed_spec=FeedSpec(time_contract_version=2, calendar=None)) + + +def test_invalid_calendar_does_not_change_provider_contract_files(tmp_path): + write_contract_files(tmp_path, date(2025, 7, 10)) + source = tmp_path / "ohlcv_1d" / "product=TEST" / "ohlcv_1d.parquet" + before = source.read_bytes() + with pytest.raises(ValueError, match="Unknown calendar"): + build(tmp_path, TimeBasedRoll(calendar="not-a-calendar")) + assert source.read_bytes() == before