From 6a7de4b1f3e7b0a53a17d701c69fa77e50c26dce Mon Sep 17 00:00:00 2001 From: ao gong <41768719+ageorge156@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:34:12 +0800 Subject: [PATCH] fix: close final backtest contract boundaries --- src/quant_engine/artifact.py | 261 ++++++++++++++++++++++---- src/quant_engine/governed_pipeline.py | 29 ++- tests/test_backtest_contracts.py | 161 +++++++++++++++- 3 files changed, 411 insertions(+), 40 deletions(-) diff --git a/src/quant_engine/artifact.py b/src/quant_engine/artifact.py index 801da06..94f1b90 100644 --- a/src/quant_engine/artifact.py +++ b/src/quant_engine/artifact.py @@ -32,6 +32,7 @@ from quant_engine.risk import CovarianceSnapshot, labeled_component_risk RESEARCH_ARTIFACT_SCHEMA_VERSION = "1.1.0" BACKTEST_EVIDENCE_SCHEMA_VERSION = "1.0.0" +_MAX_SAFE_INTEGER = (1 << 53) - 1 RISK_COLUMNS = [ "run_id", @@ -136,7 +137,7 @@ class ResearchRunArtifact: payload = { "schema_version": self.schema_version, "tables": { - name: _frame_records(frame) + name: _frame_records(frame, name) for name, frame in self._internal_table_frames().items() }, } @@ -275,8 +276,8 @@ class BacktestEvidenceManifest: cls, value: Any, *, - backtest_run_ref: BacktestRunRef, artifact: ResearchRunArtifact, + backtest_run_ref: BacktestRunRef | None = None, ) -> Self: if type(value) is not dict: _manifest_fail(BacktestContractErrorCode.TYPE_ERROR, "$", "must be an object") @@ -350,12 +351,59 @@ class BacktestEvidenceManifest: "$.qualification", "unsupported evidence qualification", ) from error - rebuilt = build_backtest_evidence_manifest( - backtest_run_ref, - artifact, - artifact_available_at=value["artifact_available_at"], - qualification=qualification, - ) + raw_run_reference = value["run_reference"] + if type(raw_run_reference) is not dict: + _manifest_fail( + BacktestContractErrorCode.TYPE_ERROR, + "$.run_reference", + "must be an object", + ) + if any(type(key) is not str for key in raw_run_reference): + _manifest_fail( + BacktestContractErrorCode.TYPE_ERROR, + "$.run_reference", + "object keys must be strings", + ) + reference_fields = {"kind", "value"} + if set(raw_run_reference) != reference_fields: + _manifest_fail( + BacktestContractErrorCode.INVALID_VALUE, + "$.run_reference", + "must contain exactly kind and value", + ) + reference_kind = raw_run_reference["kind"] + if reference_kind == "legacy_backtest_run": + if qualification is not EvidenceQualification.LEGACY_EXPLORATORY: + _manifest_fail( + BacktestContractErrorCode.READINESS_ESCALATION, + "$.qualification", + "legacy evidence cannot be promoted", + ) + legacy_run = _legacy_run_from_dict(raw_run_reference["value"]) + rebuilt = build_legacy_backtest_evidence_manifest( + legacy_run, + artifact, + artifact_available_at=value["artifact_available_at"], + ) + elif reference_kind == "backtest_run_ref": + if backtest_run_ref is None: + _manifest_fail( + BacktestContractErrorCode.TYPE_ERROR, + "$.backtest_run_ref", + "complete BacktestRunRef is required", + ) + rebuilt = build_backtest_evidence_manifest( + backtest_run_ref, + artifact, + artifact_available_at=value["artifact_available_at"], + qualification=qualification, + ) + else: + _manifest_fail( + BacktestContractErrorCode.INVALID_VALUE, + "$.run_reference.kind", + "unsupported run-reference kind", + ) if value != rebuilt.to_dict(): _manifest_fail( BacktestContractErrorCode.IDENTITY_MISMATCH, @@ -384,13 +432,39 @@ def _aware_timestamp(value: str | pd.Timestamp, name: str) -> pd.Timestamp: return timestamp -def _json_value(value: object) -> object: - if value is None or isinstance(value, str | bool | int): +def _json_value(value: object, path: str = "$") -> object: + if value is None or type(value) is bool: + return value + if type(value) is int: + if abs(value) > _MAX_SAFE_INTEGER: + _manifest_fail( + BacktestContractErrorCode.INVALID_VALUE, + path, + "integer exceeds the canonical safe range", + ) + return value + if type(value) is str: + try: + value.encode("utf-8") + except UnicodeEncodeError as error: + raise BacktestContractError( + BacktestContractErrorCode.INVALID_FORMAT, + path, + "text must be valid UTF-8", + ) from error + return value + if type(value) is float: + if math.isnan(value): + return {"non_finite_float": "nan"} + if math.isinf(value): + return { + "non_finite_float": ( + "positive_infinity" if value > 0 else "negative_infinity" + ) + } return value - if isinstance(value, float): - return value if math.isfinite(value) else None if isinstance(value, np.generic): - return _json_value(value.item()) + return _json_value(value.item(), path) if isinstance(value, pd.Timestamp): return value.isoformat() if isinstance(value, datetime): @@ -398,13 +472,34 @@ def _json_value(value: object) -> object: if isinstance(value, date): return value.isoformat() if isinstance(value, Mapping): - return { - str(key): _json_value(item) - for key, item in sorted(value.items(), key=lambda pair: str(pair[0])) - } + normalized: dict[str, object] = {} + for key, item in value.items(): + if type(key) is not str: + _manifest_fail( + BacktestContractErrorCode.TYPE_ERROR, + f"{path}.keys", + "object keys must be strings", + ) + try: + key.encode("utf-8") + except UnicodeEncodeError as error: + raise BacktestContractError( + BacktestContractErrorCode.INVALID_FORMAT, + f"{path}.keys", + "object keys must be valid UTF-8", + ) from error + normalized[key] = _json_value(item, f"{path}.{key}") + return dict(sorted(normalized.items())) if isinstance(value, list | tuple): - return [_json_value(item) for item in value] - raise TypeError(f"value of type {type(value).__name__} is not JSON serializable") + return [ + _json_value(item, f"{path}[{index}]") + for index, item in enumerate(value) + ] + _manifest_fail( + BacktestContractErrorCode.TYPE_ERROR, + path, + f"unsupported canonical value type: {type(value).__name__}", + ) def _canonical_mapping_json(values: Mapping[str, object]) -> str: @@ -418,24 +513,47 @@ def _canonical_mapping_json(values: Mapping[str, object]) -> str: ) -def _frame_records(frame: pd.DataFrame) -> list[dict[str, object]]: +def _frame_records(frame: pd.DataFrame, table_name: str) -> list[dict[str, object]]: + try: + raw_records = frame.to_dict(orient="records") + except (TypeError, ValueError) as error: + raise BacktestContractError( + BacktestContractErrorCode.TYPE_ERROR, + f"$.tables.{table_name}", + "table records cannot be serialized", + ) from error return [ - {str(key): _json_value(value) for key, value in row.items()} - for row in frame.to_dict(orient="records") + { + str(key): _evidence_json_value( + value, + f"$.tables.{table_name}.rows[{row_index}].{key}", + ) + for key, value in row.items() + } + for row_index, row in enumerate(raw_records) ] def _evidence_json_value(value: object, path: str) -> object: - if value is None or type(value) in {str, bool, int}: - if type(value) is str: - try: - value.encode("utf-8") - except UnicodeEncodeError as error: - raise BacktestContractError( - BacktestContractErrorCode.INVALID_FORMAT, - path, - "text must be valid UTF-8", - ) from error + if value is None or type(value) is bool: + return value + if type(value) is int: + if abs(value) > _MAX_SAFE_INTEGER: + _manifest_fail( + BacktestContractErrorCode.INVALID_VALUE, + path, + "integer exceeds the canonical safe range", + ) + return value + if type(value) is str: + try: + value.encode("utf-8") + except UnicodeEncodeError as error: + raise BacktestContractError( + BacktestContractErrorCode.INVALID_FORMAT, + path, + "text must be valid UTF-8", + ) from error return value if type(value) is float: if math.isnan(value): @@ -453,11 +571,22 @@ def _evidence_json_value(value: object, path: str) -> object: if type(key) is not str: _manifest_fail( BacktestContractErrorCode.TYPE_ERROR, - path, + f"{path}.keys", "object cell keys must be strings", ) + try: + key.encode("utf-8") + except UnicodeEncodeError as error: + raise BacktestContractError( + BacktestContractErrorCode.INVALID_FORMAT, + f"{path}.keys", + "object cell keys must be valid UTF-8", + ) from error normalized[key] = _evidence_json_value(item, f"{path}.{key}") - return dict(sorted(normalized.items())) + return { + "canonical_value_type": "object", + "value": dict(sorted(normalized.items())), + } if isinstance(value, list | tuple): return [ _evidence_json_value(item, f"{path}[{index}]") @@ -476,6 +605,7 @@ def _evidence_cell_kind(value: object, path: str) -> str: if type(value) is bool: return "boolean" if type(value) is int: + _evidence_json_value(value, path) return "integer" if type(value) is float: return "number" @@ -639,6 +769,69 @@ def _legacy_run_dict(run: BacktestRun) -> dict[str, object]: } +def _legacy_run_from_dict(value: object) -> BacktestRun: + path = "$.run_reference.value" + if type(value) is not dict: + _manifest_fail(BacktestContractErrorCode.TYPE_ERROR, path, "must be an object") + if any(type(key) is not str for key in value): + _manifest_fail( + BacktestContractErrorCode.TYPE_ERROR, + path, + "object keys must be strings", + ) + fields = { + "run_id", + "dataset_snapshot_id", + "factor_version_id", + "strategy_version_id", + "code_revision", + "config_hash", + "created_at", + } + if set(value) != fields: + _manifest_fail( + BacktestContractErrorCode.INVALID_VALUE, + path, + "legacy run reference has incomplete or unknown fields", + ) + + def required_text(field: str) -> str: + item = value[field] + item_path = f"{path}.{field}" + if type(item) is not str: + _manifest_fail( + BacktestContractErrorCode.TYPE_ERROR, + item_path, + "must be a string", + ) + try: + item.encode("utf-8") + except UnicodeEncodeError as error: + raise BacktestContractError( + BacktestContractErrorCode.INVALID_FORMAT, + item_path, + "must be valid UTF-8 text", + ) from error + if not item or item != item.strip(): + _manifest_fail( + BacktestContractErrorCode.INVALID_FORMAT, + item_path, + "must be non-empty canonical text", + ) + return item + + _, created_at = _manifest_instant(value["created_at"], f"{path}.created_at") + return BacktestRun( + run_id=required_text("run_id"), + dataset_snapshot_id=required_text("dataset_snapshot_id"), + factor_version_id=required_text("factor_version_id"), + strategy_version_id=required_text("strategy_version_id"), + code_revision=required_text("code_revision"), + config_hash=required_text("config_hash"), + created_at=created_at, + ) + + def _artifact_frames(artifact: ResearchRunArtifact) -> dict[str, pd.DataFrame]: if not isinstance(artifact, ResearchRunArtifact): _manifest_fail( diff --git a/src/quant_engine/governed_pipeline.py b/src/quant_engine/governed_pipeline.py index 88634e3..ef969c3 100644 --- a/src/quant_engine/governed_pipeline.py +++ b/src/quant_engine/governed_pipeline.py @@ -40,13 +40,17 @@ _GIT_SHA = re.compile(r"^[0-9a-f]{40}$") _LOGICAL_ID = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$") _SEMVER = re.compile( r"^(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\." - r"(?:0|[1-9][0-9]*)(?:-[0-9a-z.-]+)?$" + r"(?:0|[1-9][0-9]*)" + r"(?:-(?:0|[1-9][0-9]*|[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*)" + r"(?:\.(?:0|[1-9][0-9]*|[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*))*)?" + r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$" ) _MAX_SAFE_INTEGER = (1 << 53) - 1 _FORBIDDEN_ID_TOKENS = frozenset( { "latest", "provider", + "source", "broker", "credential", "locator", @@ -57,6 +61,19 @@ _FORBIDDEN_ID_TOKENS = frozenset( "postgres", "mysql", "clickhouse", + "table", + "sql", + "tushare", + "wind", + "bloomberg", + "qtdb", + "edb", + "db", + "database", + "schema", + "bucket", + "path", + "endpoint", } ) @@ -528,6 +545,16 @@ class BacktestRunRef: factor_set.artifact_available_at, "$.factor_set.artifact_available_at", ) + _, factor_evaluation = _backtest_instant( + factor_set.evaluation_at, + "$.factor_set.evaluation_at", + ) + if evaluation_time < factor_evaluation or evaluation_time < factor_available: + _backtest_fail( + BacktestContractErrorCode.TIME_ORDER_VIOLATION, + "$.evaluation_at", + "run evaluation must not precede FactorSet evaluation or availability", + ) if computed_time < evaluation_time or computed_time < factor_available: _backtest_fail( BacktestContractErrorCode.TIME_ORDER_VIOLATION, diff --git a/tests/test_backtest_contracts.py b/tests/test_backtest_contracts.py index cd819a3..78066fd 100644 --- a/tests/test_backtest_contracts.py +++ b/tests/test_backtest_contracts.py @@ -65,7 +65,13 @@ def _sha256(value: bytes) -> str: return f"sha256:{hashlib.sha256(value).hexdigest()}" -def _accepted_authorities() -> tuple[ +def _accepted_authorities( + *, + factor_evaluation_at: str = "2026-01-03T11:00:00Z", + factor_computed_at: str = "2026-01-03T10:15:00Z", + factor_artifact_available_at: str = "2026-01-03T10:20:00Z", + factor_availability_mode: AvailabilityMode = AvailabilityMode.AS_AVAILABLE, +) -> tuple[ DatasetSnapshotEnvelope, DataFoundationEnvelope, FactorSetRef, @@ -115,10 +121,10 @@ def _accepted_authorities() -> tuple[ output_schema_bytes=output_schema_bytes, output_content_bytes=output_content_bytes, output_artifact_ref=artifact_ref, - availability_mode=AvailabilityMode.AS_AVAILABLE, - evaluation_at="2026-01-03T11:00:00Z", - computed_at="2026-01-03T10:15:00Z", - artifact_available_at="2026-01-03T10:20:00Z", + availability_mode=factor_availability_mode, + evaluation_at=factor_evaluation_at, + computed_at=factor_computed_at, + artifact_available_at=factor_artifact_available_at, producer=ProducerIdentity("quant_engine", "1.0.0"), code_revision="c" * 40, actor=ActorIdentity("service", "factor_worker_v1"), @@ -311,6 +317,48 @@ def test_backtest_run_ref_rejects_unclosed_upstream_and_unsafe_scalars() -> None _assert_error(factor_type, BacktestContractErrorCode.TYPE_ERROR, "$.factor_set") +@pytest.mark.parametrize( + ("factor_times", "expected_path"), + [ + ( + { + "factor_evaluation_at": "2026-01-08T01:00:01Z", + "factor_computed_at": "2026-01-08T00:59:59Z", + "factor_artifact_available_at": "2026-01-08T01:00:00Z", + }, + "$.evaluation_at", + ), + ( + { + "factor_evaluation_at": "2026-01-03T11:00:00Z", + "factor_computed_at": "2026-01-08T01:00:00Z", + "factor_artifact_available_at": "2026-01-08T01:00:01Z", + "factor_availability_mode": AvailabilityMode.RETROSPECTIVE_REPLAY, + }, + "$.evaluation_at", + ), + ], +) +def test_run_ref_evaluation_closes_factor_pit( + factor_times: dict[str, Any], + expected_path: str, +) -> None: + snapshot, foundation, factor_set = _accepted_authorities(**factor_times) + + with pytest.raises(BacktestContractError) as lookahead: + _run_ref( + dataset_snapshot=snapshot, + foundation=foundation, + factor_set=factor_set, + ) + + _assert_error( + lookahead, + BacktestContractErrorCode.TIME_ORDER_VIOLATION, + expected_path, + ) + + def test_run_ref_rejects_aliases_locators_unsafe_integers_and_invalid_text() -> None: with pytest.raises(BacktestContractError) as mutable_alias: _run_ref(strategy_id="latest") @@ -355,6 +403,33 @@ def test_run_ref_rejects_aliases_locators_unsafe_integers_and_invalid_text() -> _assert_error(invalid_key, BacktestContractErrorCode.TYPE_ERROR, "$") +@pytest.mark.parametrize( + "physical_id", + ["db.table", "source_alpha", "wind.model", "qtdb_view", "bloomberg-signal"], +) +def test_run_ref_rejects_physical_terms_in_logical_ids(physical_id: str) -> None: + with pytest.raises(BacktestContractError) as physical: + _run_ref(strategy_id=physical_id) + _assert_error( + physical, + BacktestContractErrorCode.INVALID_FORMAT, + "$.strategy_id", + ) + + +@pytest.mark.parametrize("version", ["1.0.0-.", "1.0.0-foo..bar", "1.0.0-01"]) +def test_run_ref_requires_strict_semver_prerelease_identifiers(version: str) -> None: + with pytest.raises(BacktestContractError) as invalid: + _run_ref(strategy_version=version) + _assert_error( + invalid, + BacktestContractErrorCode.INVALID_FORMAT, + "$.strategy_version", + ) + + assert _run_ref(strategy_version="1.0.0-alpha.1").strategy_version == "1.0.0-alpha.1" + + def test_replay_lineage_is_acyclic_and_cannot_claim_changed_inputs() -> None: parent = _run_ref() replay = _run_ref( @@ -536,6 +611,21 @@ def test_manifest_binds_supported_schema_and_has_collision_free_cell_encoding() identities.add(manifest.manifest_id) assert len(identities) == 3 + content_digests: set[str] = set() + for value in (float("nan"), {"non_finite_float": "nan"}): + performance = artifact.performance.astype(object) + performance.at[0, "alpha"] = value + manifest = build_backtest_evidence_manifest( + run_ref, + replace(artifact, _performance=performance), + artifact_available_at="2026-01-08T02:05:00Z", + ) + performance_entry = next( + entry for entry in manifest.evidence if entry.category == "performance" + ) + content_digests.add(performance_entry.tables[0].content_digest) + assert len(content_digests) == 2 + unsupported = artifact.performance.astype(object) unsupported.loc[0, "alpha"] = object() with pytest.raises(BacktestContractError) as unsupported_cell: @@ -550,6 +640,63 @@ def test_manifest_binds_supported_schema_and_has_collision_free_cell_encoding() "$.artifact.tables.performance.rows[0].alpha", ) + invalid_nested_key = artifact.performance.astype(object) + invalid_nested_key.at[0, "alpha"] = {"\ud800": "value"} + with pytest.raises(BacktestContractError) as invalid_utf8: + build_backtest_evidence_manifest( + run_ref, + replace(artifact, _performance=invalid_nested_key), + artifact_available_at="2026-01-08T02:05:00Z", + ) + _assert_error( + invalid_utf8, + BacktestContractErrorCode.INVALID_FORMAT, + "$.artifact.tables.performance.rows[0].alpha.keys", + ) + + unsafe_integer = artifact.performance.astype(object) + unsafe_integer.loc[0, "alpha"] = 10**5000 + with pytest.raises(BacktestContractError) as unsafe_cell: + build_backtest_evidence_manifest( + run_ref, + replace(artifact, _performance=unsafe_integer), + artifact_available_at="2026-01-08T02:05:00Z", + ) + _assert_error( + unsafe_cell, + BacktestContractErrorCode.INVALID_VALUE, + "$.artifact.tables.performance.rows[0].alpha", + ) + + +def test_artifact_canonical_content_has_typed_collision_free_cell_encoding() -> None: + run_ref = _run_ref() + artifact = _artifact(run_ref) + + content_hashes: set[str] = set() + for value in ( + float("nan"), + float("inf"), + float("-inf"), + {"non_finite_float": "nan"}, + ): + performance = artifact.performance.astype(object) + performance.at[0, "alpha"] = value + mutated = replace(artifact, _performance=performance) + content_hashes.add(mutated.content_sha256) + assert "non_finite_float" in mutated.canonical_json() + assert len(content_hashes) == 4 + + unsupported = artifact.performance.astype(object) + unsupported.loc[0, "alpha"] = object() + with pytest.raises(BacktestContractError) as unsupported_cell: + replace(artifact, _performance=unsupported).canonical_json() + _assert_error( + unsupported_cell, + BacktestContractErrorCode.TYPE_ERROR, + "$.tables.performance.rows[0].alpha", + ) + def test_legacy_bridge_is_explicit_and_cannot_be_contract_qualified() -> None: run_ref = _run_ref() @@ -573,6 +720,10 @@ def test_legacy_bridge_is_explicit_and_cannot_be_contract_qualified() -> None: assert manifest.run_id == legacy_run.run_id assert manifest.backtest_run_ref is None assert manifest.to_dict()["run_reference"]["kind"] == "legacy_backtest_run" + assert BacktestEvidenceManifest.from_dict( + manifest.to_dict(), + artifact=artifact, + ) == manifest with pytest.raises(BacktestContractError) as implicit_promotion: build_backtest_evidence_manifest( # type: ignore[arg-type] legacy_run,