diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 52994eb..4737ba9 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -82,6 +82,7 @@ ReplaySkipReason, derive_data_requirements, ) +from .calendar import CalendarCoverageError from .contracts import build_backtest_result_event from .data_availability import AvailabilityAssessment from .detail_object_manifest import ( @@ -1377,6 +1378,43 @@ def evaluation_window(manifest: Mapping[str, Any], plan: BasicCompiledPlan) -> t return evaluation_from, coverage_end +def require_compatible_execution_window( + policy: ExecutionPolicy, + manifest: Mapping[str, Any], + calendar: SessionCalendar, +) -> None: + """Reject immutable execution inputs that cannot describe one run. + + The worker must not silently shrink a policy or manifest period to make an + incompatible request executable. This check runs during binding, before a + ``RUNNING`` event is published, because redelivery cannot change any input. + """ + problems: list[str] = [] + manifest_start = _parse_instant(manifest.get("period_start")) + manifest_end = _parse_instant(manifest.get("period_end")) + if manifest_start != policy.period_start or manifest_end != policy.period_end: + problems.append( + "dataset manifest period " + f"{manifest_start.isoformat()}..{manifest_end.isoformat()} does not match " + f"execution policy {policy.version} period " + f"{policy.period_start.isoformat()}..{policy.period_end.isoformat()}" + ) + + zone = ZoneInfo(policy.timezone) + first = policy.period_start.astimezone(zone).date() + last = (policy.period_end - timedelta(microseconds=1)).astimezone(zone).date() + try: + calendar.session_schedule(first, last) + except CalendarCoverageError as exc: + problems.append(f"session calendar does not cover the execution policy period: {exc}") + + if problems: + raise JobNotSatisfiable( + "execution inputs are mutually incompatible: " + "; ".join(problems), + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + + def _parse_instant(value: object) -> datetime: if not isinstance(value, str): # pragma: no cover - the schema requires a string raise JobNotSatisfiable( @@ -1599,6 +1637,7 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: reason_code="REQUIRED_INPUT_UNAVAILABLE", ) manifest = primary[0][1] + require_compatible_execution_window(policy, manifest, self._calendar) try: plan = self._runtime.load(plan_document, compiled_plan_checksum=plan_checksum) except BasicPlanCompatibilityError as exc: diff --git a/tests/test_feature_outputs.py b/tests/test_feature_outputs.py index 97afddf..b143b9c 100644 --- a/tests/test_feature_outputs.py +++ b/tests/test_feature_outputs.py @@ -4,6 +4,7 @@ import hashlib import json import uuid +from dataclasses import replace from datetime import UTC, datetime, timedelta from decimal import Decimal from typing import Any @@ -770,6 +771,57 @@ def publish(self, event: Any, *, delivery_attempt: int) -> None: assert sink.events[0]["retryable"] is False +def test_incompatible_development_windows_are_one_terminal_binding_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The exact INT03 policy/manifest/calendar tuple cannot improve on redelivery.""" + policy = replace( + E2E_EXECUTION_POLICY, + version="development-official-backtest-2026-q3-v1", + release_quarter="2026-Q3", + period_start=datetime(2016, 7, 1, 4, tzinfo=UTC), + period_end=datetime(2026, 7, 1, 4, tzinfo=UTC), + ) + market_bytes = market_bars_parquet() + manifest = dataset_manifest( + hashlib.sha256(market_bytes).hexdigest(), + row_count=len(CLOSES), + coverage_end=EVALUATION_THROUGH, + ) + manifest["period_start"] = "2024-01-01T05:00:00Z" + manifest["period_end"] = "2024-02-01T05:00:00Z" + handler = _handler(Source({}), Reader(b"")) + handler._policies = ExecutionPolicyCatalog([policy]) + handler._manifests = StaticDatasetManifestSource({DATASET_MANIFEST_ID: manifest}) + envelope = replace(_envelope(pins=[]), execution_policy_version=policy.version) + + class Sink: + def __init__(self) -> None: + self.events: list[dict[str, Any]] = [] + + def publish(self, event: Any, *, delivery_attempt: int) -> None: + self.events.append(dict(event) | {"deliveryAttempt": delivery_attempt}) + + sink = Sink() + handler._sink = sink + monkeypatch.setattr(JobEnvelope, "parse", classmethod(lambda _cls, _job: envelope)) + + with pytest.raises(JobNotSatisfiable) as failure: + handler.bind(envelope, _context()) + assert "dataset manifest period" in str(failure.value) + assert "2016-07-01 is outside the pinned XNYS coverage" in str(failure.value) + + outcome = handler({}, _context()) + + assert outcome.result is JobResult.PERMANENT_FAILURE + assert outcome.reason_code == "REQUIRED_INPUT_UNAVAILABLE" + assert len(sink.events) == 1 + assert sink.events[0]["status"] == "FAILED" + assert sink.events[0]["failureCode"] == "REQUIRED_INPUT_UNAVAILABLE" + assert sink.events[0]["retryable"] is False + assert sink.events[0]["deliveryAttempt"] == 1 + + def test_job_binding_attaches_only_fully_verified_feature_series() -> None: body = _parquet() record = _record(body)