Execute all pinned official backtest datasets

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": [

src/backtest_engine/contracts.py

@@ -533,8 +533,13 @@ def compute_message_idempotency_key(
 
 
 def official_backtest_operation_key(request: Mapping[str, Any]) -> str:
+    datasets = request.get("datasets")
+    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|{request['datasetManifestId']}|"
+        f"OFFICIAL_BACKTEST|{dataset_identity}|"
         f"{request['assumptionsVersion']}"
     )
 
@@ -605,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

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,25 +516,29 @@ 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 = (
             TimeInterval(bar.starts_at, bar.ends_at)
             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,
             instrument_id=requirement.instrument_id,
             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(

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

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]
 
 

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"},

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

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",

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"
 
 
@@ -306,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],