fix: derive futures expiry rolls from authoritative sessions

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

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):

tests/futures/test_scheduled_roll_workflow.py

@@ -0,0 +1,103 @@
+"""Provider-generated contract files must use scheduled, not observed, expiry offsets."""
+
+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 FirstNoticeDateRoll, TimeBasedRoll
+from ml4t.data.providers.synthetic import SyntheticProvider
+from ml4t.specs import FeedSpec
+
+
+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)]
+
+
+@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