From 4440e4a69f24a4a0bd6f87d5ca8eaf7cc93716c7 Mon Sep 17 00:00:00 2001 From: HJ <16863475+hjcud@users.noreply.github.com> Date: Mon, 10 Aug 2026 03:26:04 +0900 Subject: [PATCH 1/2] feat: execute all pinned backtest datasets --- src/backtest_engine/api.py | 8 +-- src/backtest_engine/contracts.py | 8 ++- src/backtest_engine/orchestrator.py | 49 ++++++++++--- src/backtest_engine/request_dispatch.py | 32 +++++++-- src/backtest_engine/result_query.py | 4 +- .../v1/official-backtest-request.schema.json | 15 ++++ src/backtest_engine/wiring.py | 50 ++++++++++++-- .../v1/official-backtest-request.valid.json | 7 ++ tests/test_contracts.py | 34 +++++++++ tests/test_feature_outputs.py | 38 ++++++++++ tests/test_orchestrator.py | 69 +++++++++++++++++++ tests/test_request_dispatch.py | 34 +++++++++ tests/test_result_query.py | 19 +++++ 13 files changed, 338 insertions(+), 29 deletions(-) diff --git a/src/backtest_engine/api.py b/src/backtest_engine/api.py index d301581..b005917 100644 --- a/src/backtest_engine/api.py +++ b/src/backtest_engine/api.py @@ -302,10 +302,10 @@ def _input_model_payload(view: InputModelView, overview: BacktestOverview) -> di plausible-looking default. """ market_bars = tuple(item for item in view.datasets if item.purpose_code == "MARKET_BARS") - if len(market_bars) != 1: + if not market_bars: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="canonical input bundle must contain exactly one MARKET_BARS dataset", + detail="canonical input bundle must contain at least one MARKET_BARS dataset", ) primary = market_bars[0] return { @@ -314,8 +314,8 @@ def _input_model_payload(view: InputModelView, overview: BacktestOverview) -> di "status": overview.status, "strategySnapshotHash": view.strategy_snapshot_hash, "compiledPlanChecksum": view.compiled_plan_checksum, - # Kept for client compatibility; both values are projections of the one - # canonical MARKET_BARS child, never duplicate pin-row columns. + # Kept for client compatibility as a deterministic representative; the + # complete immutable input set is always returned in datasets. "datasetManifestId": primary.dataset_manifest_id, "datasetHash": primary.locked_dataset_hash, "datasets": [ diff --git a/src/backtest_engine/contracts.py b/src/backtest_engine/contracts.py index 2e124e3..3622222 100644 --- a/src/backtest_engine/contracts.py +++ b/src/backtest_engine/contracts.py @@ -533,8 +533,14 @@ def compute_message_idempotency_key( def official_backtest_operation_key(request: Mapping[str, Any]) -> str: + datasets = request.get("datasets") + dataset_identity = ( + ",".join(str(item["datasetManifestId"]) for item in datasets) + if isinstance(datasets, list) and datasets + else str(request["datasetManifestId"]) + ) return ( - f"OFFICIAL_BACKTEST|{request['datasetManifestId']}|" + f"OFFICIAL_BACKTEST|{dataset_identity}|" f"{request['assumptionsVersion']}" ) diff --git a/src/backtest_engine/orchestrator.py b/src/backtest_engine/orchestrator.py index 4b03a4f..33cd319 100644 --- a/src/backtest_engine/orchestrator.py +++ b/src/backtest_engine/orchestrator.py @@ -368,6 +368,7 @@ class BacktestJob: data_kind: str resolution: str initial_cash: Decimal + manifests: tuple[Mapping[str, Any], ...] = () def __post_init__(self) -> None: if not self.run_id: @@ -380,6 +381,10 @@ def __post_init__(self) -> None: raise OrchestratorError("data_kind and resolution must be pinned") if self.initial_cash < 0: raise OrchestratorError("initial_cash must not be negative") + manifests = tuple(self.manifests) or (self.manifest,) + if any(not isinstance(item, Mapping) for item in manifests): + raise OrchestratorError("manifests must contain dataset manifest mappings") + object.__setattr__(self, "manifests", manifests) # One source of truth for the bar period: the element catalog's # resolution table. A separately configured interval could disagree # with the resolution the plan actually reads. @@ -511,7 +516,7 @@ def observations_from_events( tuple, which the assessor turns into ``REQUIRED_SERIES_ABSENT`` rather than into a silently empty replay. """ - by_instrument: dict[str, list[TimeInterval]] = {} + by_series: dict[tuple[str, str, str], list[TimeInterval]] = {} for event in events: bar = event.payload.get("bar") interval = ( @@ -519,9 +524,9 @@ def observations_from_events( if isinstance(bar, SeriesBar) else TimeInterval(event.occurred_at - bar_interval, event.occurred_at) ) - by_instrument.setdefault(event.instrument_id, []).append( - interval - ) + data_kind = str(event.payload.get("dataKind", "")) + resolution = str(event.payload.get("resolution", "")) + by_series.setdefault((event.instrument_id, data_kind, resolution), []).append(interval) return tuple( DataObservation( requirement_id=requirement.requirement_id, @@ -529,7 +534,11 @@ def observations_from_events( data_kind=requirement.data_kind, resolution=requirement.resolution, available_intervals=( - tuple(by_instrument.get(requirement.instrument_id, ())) + tuple(by_series.get(( + requirement.instrument_id, + requirement.data_kind, + requirement.resolution, + ), ())) + ( _closed_market_intervals(schedule, requirement.required_interval) if schedule is not None @@ -669,12 +678,30 @@ def run( ) -> ReplayOutcome: schedule = self._schedule(job.execution_policy) try: - events = bar_events_from_batches( - self._reader.iter_batches(job.manifest, job.execution_policy), - data_kind=job.data_kind, - resolution=job.resolution, - publication_lag=self._publication_lag, - schedule=schedule, + combined_events: list[MarketDataEvent] = [] + for manifest in job.manifests: + resolution = str(manifest.get("resolution") or ( + job.resolution if manifest is job.manifest else "" + )) + if not resolution: + raise MarketDataValidationError( + "every pinned dataset manifest must declare its resolution" + ) + manifest_events = bar_events_from_batches( + self._reader.iter_batches(manifest, job.execution_policy), + data_kind=job.data_kind, + resolution=resolution, + publication_lag=self._publication_lag, + schedule=schedule, + ) + combined_events.extend( + replace(event, event_id=f"{event.event_id}:{resolution}") + if len(job.manifests) > 1 else event + for event in manifest_events + ) + events = tuple( + replace(event, source_sequence=index) + for index, event in enumerate(combined_events, start=1) ) except MarketDataValidationError: return self._abort( diff --git a/src/backtest_engine/request_dispatch.py b/src/backtest_engine/request_dispatch.py index d2a2a3f..2bdfc9a 100644 --- a/src/backtest_engine/request_dispatch.py +++ b/src/backtest_engine/request_dispatch.py @@ -82,14 +82,24 @@ def _request_period( ) -> tuple[date, date, tuple[PinnedDataset, ...], tuple[PinnedFeatureMaterialization, ...]]: try: if lane is RequestLane.BASIC: + raw_datasets = request.get("datasets") or ( + { + "datasetManifestId": request["datasetManifestId"], + "purposeCode": "MARKET_BARS", + "expectedDatasetHash": request["expectedDatasetHash"], + }, + ) return ( date.fromisoformat(str(request["periodStart"])), date.fromisoformat(str(request["periodEnd"])), - ( + tuple( + sorted( PinnedDataset( - uuid.UUID(str(request["datasetManifestId"])), - "MARKET_BARS", - str(request["expectedDatasetHash"]), + uuid.UUID(str(item["datasetManifestId"])), + str(item["purposeCode"]), + str(item["expectedDatasetHash"]), + ) + for item in raw_datasets ), ), tuple( @@ -182,9 +192,19 @@ def __call__(self, request: Mapping[str, Any], lane: RequestLane) -> None: stored_datasets = tuple(sorted(run.datasets)) stored_features = tuple(sorted(run.feature_materializations)) market_bars = tuple(item for item in stored_datasets if item.purpose_code == "MARKET_BARS") - if len(market_bars) != 1: + if not market_bars or (lane is not RequestLane.BASIC and len(market_bars) != 1): + raise RequestProcessingError("MARKET_BARS_DATASET_INVALID", retryable=False) + representative_id = ( + market_bars[0].dataset_manifest_id + if lane is RequestLane.COMPETITION + else uuid.UUID(str(request["datasetManifestId"])) + ) + representatives = tuple( + item for item in market_bars if item.dataset_manifest_id == representative_id + ) + if len(representatives) != 1: raise RequestProcessingError("MARKET_BARS_DATASET_INVALID", retryable=False) - primary = market_bars[0] + primary = representatives[0] if lane is RequestLane.BASIC: start, end, requested_datasets, requested_features = _request_period( request, lane diff --git a/src/backtest_engine/result_query.py b/src/backtest_engine/result_query.py index 74a7d5a..344be73 100644 --- a/src/backtest_engine/result_query.py +++ b/src/backtest_engine/result_query.py @@ -261,8 +261,8 @@ def __post_init__(self) -> None: @property def market_bars(self) -> RunDatasetInput: matches = tuple(item for item in self.datasets if item.purpose_code == "MARKET_BARS") - if len(matches) != 1: - raise QueryValidationError("inputs must contain exactly one MARKET_BARS dataset") + if not matches: + raise QueryValidationError("inputs must contain at least one MARKET_BARS dataset") return matches[0] diff --git a/src/backtest_engine/schemas/strategy-bot/v1/official-backtest-request.schema.json b/src/backtest_engine/schemas/strategy-bot/v1/official-backtest-request.schema.json index 3b631b7..a5b3e33 100644 --- a/src/backtest_engine/schemas/strategy-bot/v1/official-backtest-request.schema.json +++ b/src/backtest_engine/schemas/strategy-bot/v1/official-backtest-request.schema.json @@ -37,6 +37,21 @@ "compiledPlanChecksum": {"$ref": "https://contracts.idea2strategy.io/common/v1/primitives.schema.json#/$defs/sha256Prefixed"}, "datasetManifestId": {"$ref": "https://contracts.idea2strategy.io/common/v1/primitives.schema.json#/$defs/uuid"}, "expectedDatasetHash": {"$ref": "https://contracts.idea2strategy.io/common/v1/primitives.schema.json#/$defs/sha256Prefixed"}, + "datasets": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": { + "type": "object", + "required": ["datasetManifestId", "purposeCode", "expectedDatasetHash"], + "properties": { + "datasetManifestId": {"$ref": "https://contracts.idea2strategy.io/common/v1/primitives.schema.json#/$defs/uuid"}, + "purposeCode": {"const": "MARKET_BARS"}, + "expectedDatasetHash": {"$ref": "https://contracts.idea2strategy.io/common/v1/primitives.schema.json#/$defs/sha256Prefixed"} + }, + "additionalProperties": false + } + }, "periodStart": {"type": "string", "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}$"}, "periodEnd": {"type": "string", "pattern": "^[0-9]{4}-[0-9]{2}-[0-9]{2}$"}, "assumptionsVersion": {"$ref": "https://contracts.idea2strategy.io/common/v1/primitives.schema.json#/$defs/nonEmptyString"}, diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 735d9fa..5966db2 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -1650,14 +1650,25 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: reason_code="REQUIRED_INPUT_UNAVAILABLE", ) resolved_manifests.append((pin, resolved)) - primary = [item for item in resolved_manifests if item[0].manifest_id == envelope.dataset_manifest_id] - if len(primary) != 1: + for _pin, resolved in resolved_manifests: + require_compatible_execution_window(policy, resolved, self._calendar) + declared_resolutions = [ + str(resolved.get("resolution", "")) for _pin, resolved in resolved_manifests + ] + if len(resolved_manifests) > 1 and ( + any(not resolution for resolution in declared_resolutions) + or len(set(declared_resolutions)) != len(declared_resolutions) + ): + raise JobNotSatisfiable( + "multiple pinned market datasets must declare unique resolutions", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + coverage_windows = {dataset_coverage(resolved) for _pin, resolved in resolved_manifests} + if len(coverage_windows) != 1: raise JobNotSatisfiable( - "the representative dataset is not pinned exactly once", + "all pinned market datasets must share one evaluation period", 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: @@ -1669,6 +1680,13 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: # deterministic producer/consumer mismatch to exhaustion. raise JobNotSatisfiable(str(exc), reason_code=exc.failure.value) from exc if plan.reference_series[1] == "$DATASET": + primary = [item for item in resolved_manifests if item[0].manifest_id == envelope.dataset_manifest_id] + if len(primary) != 1: + raise JobNotSatisfiable( + "the representative dataset is not pinned exactly once", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + manifest = primary[0][1] dataset_resolution = str(manifest.get("resolution", "")) if dataset_resolution not in {"30m", "1h", "4h", "1d"}: raise JobNotSatisfiable( @@ -1678,6 +1696,27 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: plan = replace( plan, reference_series=(plan.reference_series[0], dataset_resolution) ) + else: + reference_resolution = plan.reference_series[1] + reference_manifests = [ + resolved for _pin, resolved in resolved_manifests + if str(resolved.get("resolution", "")) == reference_resolution + ] + if ( + not reference_manifests + and len(resolved_manifests) == 1 + and not str(resolved_manifests[0][1].get("resolution", "")) + ): + # Legacy single-dataset fixtures and messages predate the + # manifest-level resolution field. Multiple datasets never + # receive this fallback because their binding must be explicit. + reference_manifests = [resolved_manifests[0][1]] + if len(reference_manifests) != 1: + raise JobNotSatisfiable( + f"the plan reference resolution {reference_resolution} must match exactly one pinned dataset", + reason_code="REQUIRED_INPUT_UNAVAILABLE", + ) + manifest = reference_manifests[0] evaluation_from, evaluation_through = evaluation_window(manifest, plan) feature_series: tuple[PinnedFeatureSeries, ...] = () if self._feature_materializations is not None or envelope.feature_materializations: @@ -1732,6 +1771,7 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding: data_kind=data_kind, resolution=resolution, initial_cash=plan.initial_cash, + manifests=tuple(resolved for _pin, resolved in resolved_manifests), ) except OrchestratorError as exc: raise JobNotSatisfiable(str(exc), reason_code="REQUIRED_INPUT_UNAVAILABLE") from exc diff --git a/tests/fixtures/contracts/strategy-bot/v1/official-backtest-request.valid.json b/tests/fixtures/contracts/strategy-bot/v1/official-backtest-request.valid.json index 1b754da..a1ce602 100644 --- a/tests/fixtures/contracts/strategy-bot/v1/official-backtest-request.valid.json +++ b/tests/fixtures/contracts/strategy-bot/v1/official-backtest-request.valid.json @@ -15,6 +15,13 @@ "compiledPlanChecksum": "sha256:88d61198d46dce161c2a929702a7fd1cee5c9b044c470d2590b96f3825fcacb3", "datasetManifestId": "00000000-0000-4000-8000-000000000203", "expectedDatasetHash": "sha256:1111111111111111111111111111111111111111111111111111111111111111", + "datasets": [ + { + "datasetManifestId": "00000000-0000-4000-8000-000000000203", + "purposeCode": "MARKET_BARS", + "expectedDatasetHash": "sha256:1111111111111111111111111111111111111111111111111111111111111111" + } + ], "periodStart": "2025-01-01", "periodEnd": "2025-12-31", "assumptionsVersion": "accounting:1.0.0", diff --git a/tests/test_contracts.py b/tests/test_contracts.py index 063eb18..6609e75 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -269,6 +269,40 @@ def test_consumer_accepts_bs_official_backtest_request_verbatim( assert accepted["compiledPlanChecksum"] == B_PLAN_CHECKSUM assert accepted["expectedSnapshotHash"] == B_SNAPSHOT_HASH assert accepted["datasetManifestId"] == B_DATASET_MANIFEST_ID + + +def test_official_request_identity_covers_every_server_selected_dataset( + official_request: dict[str, Any], +) -> None: + second_id = "00000000-0000-4000-8000-000000000204" + official_request["datasets"] = [ + { + "datasetManifestId": official_request["datasetManifestId"], + "purposeCode": "MARKET_BARS", + "expectedDatasetHash": official_request["expectedDatasetHash"], + }, + { + "datasetManifestId": second_id, + "purposeCode": "MARKET_BARS", + "expectedDatasetHash": "sha256:" + "6" * 64, + }, + ] + operation_key = official_backtest_operation_key(official_request) + official_request["metadata"]["idempotencyKey"] = compute_message_idempotency_key( + contract_version=STRATEGY_BOT_CONTRACT_VERSION, + message_type="OFFICIAL_BACKTEST_REQUESTED", + aggregate_id=B_BOT_ID, + snapshot_hash=B_SNAPSHOT_HASH, + operation_key=operation_key, + ) + + accepted = validate_official_backtest_request(official_request) + + assert operation_key == ( + "OFFICIAL_BACKTEST|00000000-0000-4000-8000-000000000203," + f"{second_id}|accounting:1.0.0" + ) + assert len(accepted["datasets"]) == 2 assert accepted["requestReason"] == "STRATEGY_RELEASE" diff --git a/tests/test_feature_outputs.py b/tests/test_feature_outputs.py index b143b9c..4174fac 100644 --- a/tests/test_feature_outputs.py +++ b/tests/test_feature_outputs.py @@ -35,6 +35,7 @@ from backtest_engine.lifecycle import StaticCompiledPlanSource, StaticDatasetManifestSource from backtest_engine.production import S3VersionedFeatureObjectReader from backtest_engine.wiring import ( + DatasetPin, FeatureMaterializationPin, JobEnvelope, JobNotSatisfiable, @@ -843,6 +844,43 @@ def test_job_binding_attaches_only_fully_verified_feature_series() -> None: assert binding.feature_series[0].value_at(EVALUATION_FROM) == Decimal("0.00000000") +def test_job_binding_preserves_every_dataset_and_uses_the_plan_reference_resolution() -> None: + body = _parquet() + record = _record(body) + handler = _handler(Source({MATERIALIZATION_ID: record}), Reader(body)) + primary_id = DATASET_MANIFEST_ID + secondary_id = uuid.UUID("40000000-0000-4000-8000-000000000099") + market_bytes = market_bars_parquet() + primary = dataset_manifest( + hashlib.sha256(market_bytes).hexdigest(), + row_count=len(CLOSES), + coverage_end=EVALUATION_THROUGH, + ) + primary["resolution"] = "1m" + secondary = dict(primary) + secondary["resolution"] = "30m" + secondary["dataset_hash"] = "f" * 64 + handler._manifests = StaticDatasetManifestSource({ + primary_id: primary, + secondary_id: secondary, + }) + envelope = replace( + _envelope(pins=[{ + "featureMaterializationId": str(MATERIALIZATION_ID), + "lockedResultHash": "sha256:" + str(record["result_hash"]), + }]), + datasets=( + DatasetPin(primary_id, "MARKET_BARS", "sha256:" + str(primary["dataset_hash"])), + DatasetPin(secondary_id, "MARKET_BARS", "sha256:" + str(secondary["dataset_hash"])), + ), + ) + + binding = handler.bind(envelope, _context()) + + assert binding.manifest["resolution"] == "1m" + assert {item["resolution"] for item in binding.job.manifests} == {"1m", "30m"} + + def test_production_feature_binding_refuses_a_job_with_no_required_pin() -> None: handler = _handler(Source({}), Reader(b"")) diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py index 479eb84..b7445d4 100644 --- a/tests/test_orchestrator.py +++ b/tests/test_orchestrator.py @@ -541,6 +541,75 @@ def materialized_read_is_forbidden(*_args: object, **_kwargs: object) -> object: assert outcome.status is ReplayStatus.COMPLETED +def test_orchestrator_reads_every_pinned_resolution_into_one_replay_clock() -> None: + rows_15m = pa.Table.from_pylist( + [_bar_row(_utc(14, 30), date(2024, 1, 2)), + _bar_row(_utc(14, 45), date(2024, 1, 2))], + schema=_SCHEMA, + ) + rows_30m = pa.Table.from_pylist( + [_bar_row(_utc(14, 30), date(2024, 1, 2))], + schema=_SCHEMA, + ) + manifests = ({"resolution": "15m"}, {"resolution": "30m"}) + + class Reader: + def __init__(self) -> None: + self.seen: list[str] = [] + + def iter_batches(self, manifest: Mapping[str, Any], _policy: Any) -> Any: + resolution = str(manifest["resolution"]) + self.seen.append(resolution) + return (rows_15m if resolution == "15m" else rows_30m).to_batches() + + reader = Reader() + runtime = StubRuntime(buy_at=None) + harness = Harness(Path(), runtime, RecordingEngine(), RecordingPublisher()) + orchestrator = BacktestOrchestrator( + reader=reader, + calendar=XNYS_CALENDAR, + replay_factory=harness.factory, + engine=harness.engine, + publisher=harness.publisher, + wall_clock=WallClock(), + ) + requirements = ( + DataRequirement( + requirement_id="aapl-15m", instrument_id=AAPL, data_kind=DATA_KIND, + resolution="15m", warmup_from=_utc(14, 30), evaluation_from=_utc(14, 30), + evaluation_through=_utc(15, 0), + ), + DataRequirement( + requirement_id="aapl-30m", instrument_id=AAPL, data_kind=DATA_KIND, + resolution="30m", warmup_from=_utc(14, 30), evaluation_from=_utc(14, 30), + evaluation_through=_utc(15, 0), + ), + ) + coordinator = AttemptCoordinator(RUN_ID, _policy(), WALL_T0) + outcome = orchestrator.run( + BacktestJob( + run_id=RUN_ID, + idempotency_key="OFFICIAL_BACKTEST:multi-resolution", + worker_execution_key=f"BACKTEST_RUN:{RUN_ID}:multi-resolution", + manifest=manifests[0], + execution_policy=D17_EXECUTION_POLICY_FIXTURE, + requirements=requirements, + data_kind=DATA_KIND, + resolution="15m", + initial_cash=Decimal("10000"), + manifests=manifests, + ), + coordinator=coordinator, + lease=coordinator.acquire("multi-resolution-worker", WALL_T0), + monitor=FixedMonitor(), + ) + + assert outcome.status is ReplayStatus.COMPLETED + assert reader.seen == ["15m", "30m"] + visible = runtime.inputs_by_instant[_utc(15, 0)][AAPL] + assert {series.resolution for series in visible.series} == {"15m", "30m"} + + # -------------------------------------------------------------------------- # The replay loop is genuinely driven by the event clock. # -------------------------------------------------------------------------- diff --git a/tests/test_request_dispatch.py b/tests/test_request_dispatch.py index ddaa005..83f8fa0 100644 --- a/tests/test_request_dispatch.py +++ b/tests/test_request_dispatch.py @@ -285,6 +285,40 @@ def test_basic_request_is_dispatched_through_the_same_pinned_two_stage_boundary( ] +def test_basic_request_preserves_every_server_selected_market_dataset() -> None: + request = basic_request() + second_id = uuid.UUID("95000000-0000-4000-8000-000000000002") + request["datasets"] = [ + { + "datasetManifestId": request["datasetManifestId"], + "purposeCode": "MARKET_BARS", + "expectedDatasetHash": request["expectedDatasetHash"], + }, + { + "datasetManifestId": str(second_id), + "purposeCode": "MARKET_BARS", + "expectedDatasetHash": "sha256:" + "8" * 64, + }, + ] + run = replace( + projection(request, RequestLane.BASIC), + datasets=( + PinnedDataset( + uuid.UUID(request["datasetManifestId"]), + "MARKET_BARS", + request["expectedDatasetHash"], + ), + PinnedDataset(second_id, "MARKET_BARS", "sha256:" + "8" * 64), + ), + ) + queue = Queue() + + BacktestRequestJobPublisher(Source(run), queue)(request, RequestLane.BASIC) + + assert queue.jobs[0][1]["datasetManifestId"] == request["datasetManifestId"] + assert queue.jobs[0][1]["datasets"] == request["datasets"] + + def test_changed_feature_output_is_rejected_before_execution() -> None: feature_id = uuid.uuid4() diff --git a/tests/test_result_query.py b/tests/test_result_query.py index 3d48278..7f8cfeb 100644 --- a/tests/test_result_query.py +++ b/tests/test_result_query.py @@ -355,6 +355,25 @@ def test_inputs_and_models_preserve_locked_reproducibility_identity() -> None: assert view.execution_model_version == "execution-v5" +def test_run_inputs_accept_multiple_market_bar_dataset_pins() -> None: + inputs = RunInputs( + compiled_plan_checksum="sha256:" + "b" * 64, + strategy_snapshot_hash="sha256:" + "a" * 64, + input_bundle_fingerprint=FINGERPRINT, + input_contract_version="backtest-request.v1", + datasets=( + RunDatasetInput(DATASET_ID, "MARKET_BARS", "c" * 64), + RunDatasetInput("00000000-0000-4000-8000-000000000099", "MARKET_BARS", "f" * 64), + ), + feature_materializations=(), + execution_policy_version="official-policy-v4", + precision_rules_version="precision:1.0.0", + ) + + assert len(inputs.datasets) == 2 + assert inputs.market_bars.dataset_manifest_id == DATASET_ID + + def test_complete_query_returns_performance_and_et_monthly_judgments() -> None: service, store = _service() run, result, details, monthly = _completed_artifacts() From 2abd4b7cf278cb13e4529b32ac67ec5edc8e31f1 Mon Sep 17 00:00:00 2001 From: HJ <16863475+hjcud@users.noreply.github.com> Date: Mon, 10 Aug 2026 03:33:40 +0900 Subject: [PATCH 2/2] fix: bind legacy dataset representative --- src/backtest_engine/contracts.py | 22 +++++++++++++++++----- tests/test_contracts.py | 20 ++++++++++++++++++++ 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/src/backtest_engine/contracts.py b/src/backtest_engine/contracts.py index 3622222..21972fe 100644 --- a/src/backtest_engine/contracts.py +++ b/src/backtest_engine/contracts.py @@ -534,11 +534,10 @@ def compute_message_idempotency_key( def official_backtest_operation_key(request: Mapping[str, Any]) -> str: datasets = request.get("datasets") - dataset_identity = ( - ",".join(str(item["datasetManifestId"]) for item in datasets) - if isinstance(datasets, list) and datasets - else str(request["datasetManifestId"]) - ) + dataset_ids = [str(request["datasetManifestId"])] + if isinstance(datasets, list) and datasets: + dataset_ids.extend(str(item["datasetManifestId"]) for item in datasets[1:]) + dataset_identity = ",".join(dataset_ids) return ( f"OFFICIAL_BACKTEST|{dataset_identity}|" f"{request['assumptionsVersion']}" @@ -611,6 +610,19 @@ def validate_official_backtest_request( f"canonical material: declared {declared_key}, computed {computed_key}" ) + datasets = request.get("datasets") + if isinstance(datasets, list) and datasets: + representative = datasets[0] + if ( + representative["datasetManifestId"] != request["datasetManifestId"] + or representative["expectedDatasetHash"] + != request["expectedDatasetHash"] + ): + raise ContractValidationError( + "official_backtest_request.datasets[0] must match the legacy " + "datasetManifestId and expectedDatasetHash representative" + ) + if compiled_plan is not None: cross_check_request_against_plan(request, validate_basic_compiled_plan(compiled_plan)) return request diff --git a/tests/test_contracts.py b/tests/test_contracts.py index 6609e75..0eb78a6 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -340,6 +340,26 @@ def test_request_whose_dataset_manifest_was_swapped_fails_its_idempotency_key( validate_official_backtest_request(swapped) +@pytest.mark.parametrize( + ("field", "replacement"), + [ + ("datasetManifestId", "00000000-0000-4000-8000-000000000999"), + ("expectedDatasetHash", "sha256:" + "9" * 64), + ], +) +def test_dataset_array_representative_must_match_legacy_fields( + field: str, + replacement: str, +) -> None: + request = _load( + STRATEGY_BOT_FIXTURES / "official-backtest-request.valid.json" + ) + request["datasets"][0][field] = replacement + + with pytest.raises(ContractValidationError, match=r"datasets\[0\]"): + validate_official_backtest_request(request) + + def test_request_is_rejected_when_it_names_a_different_compiled_plan( official_request: dict[str, Any], compiled_plan: dict[str, Any],