ageorge156/quant_engine
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
main
quant_engine/tests/test_backtest_contracts.py
T
ageorge156 78d65b4db0
CI / lite (push) Successful in 8s
Details
fix: close final backtest contract boundaries (#17)
2026-09-01 12:40:04 +08:00

774 lines
27 KiB
Python
Raw Permalink Blame History

"""Backtest run-reference and closed-evidence contract conformance."""
from __future__ import annotations
import copy
import hashlib
import json
from dataclasses import replace
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
import pandas as pd
import pytest
from quant_engine.artifact import (
BacktestEvidenceManifest,
EvidenceQualification,
RESEARCH_ARTIFACT_SCHEMA_VERSION,
ResearchRunArtifact,
build_backtest_evidence_manifest,
build_legacy_backtest_evidence_manifest,
build_research_run_artifact,
)
from quant_engine.execution import ExecutionConfig
from quant_engine.factor_contracts import (
ActorIdentity,
AvailabilityMode,
Causation,
DataFoundationEnvelope,
DatasetSnapshotEnvelope,
FactorInput,
FactorSetRef,
InputBinding,
OutputArtifactRef,
OutputCoverage,
OutputQuality,
OutputQualityCheck,
ProducerIdentity,
ViewAvailability,
canonical_json_bytes,
factor_definition_from_alpha158,
factor_input_schema_digest,
)
from quant_engine.governed_pipeline import (
BacktestContractError,
BacktestContractErrorCode,
BacktestRun,
BacktestRunRef,
)
from quant_engine.research_pipeline import FactorBacktestResult, run_factor_backtest_research
ROOT = Path(__file__).resolve().parents[1]
FACTOR_FIXTURE = ROOT / "tests" / "fixtures" / "factor-contracts-v1.golden.json"
BACKTEST_FIXTURE = ROOT / "tests" / "fixtures" / "backtest-evidence-v1.golden.json"
VIEW_REF_ID = "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c"
VIEW_SCHEMA_DIGEST = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
CALENDAR_REVISION_ID = "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078"
ACTION_REVISION_ID = "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275"
PARAMETERS = {"lag_sessions": 1, "top_k": 1}
def _sha256(value: bytes) -> str:
return f"sha256:{hashlib.sha256(value).hexdigest()}"
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,
]:
fixture = json.loads(FACTOR_FIXTURE.read_text(encoding="utf-8"))
snapshot = DatasetSnapshotEnvelope.from_dict(fixture["dataset_snapshot"])
foundation = DataFoundationEnvelope.from_dict(fixture["data_foundation"])
factor_input = FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume"))
definition = factor_definition_from_alpha158(
"alpha_005",
version="1.0.0",
parameters={},
inputs=(factor_input,),
implementation_digest="sha256:" + "1" * 64,
input_schema_digest=factor_input_schema_digest((factor_input,)),
valid_from="2026-01-01T00:00:00.000000Z",
valid_until="2027-01-01T00:00:00Z",
warmup_sessions=10,
lag_sessions=1,
producer=ProducerIdentity("quant_engine", "1.0.0"),
code_revision="c" * 40,
)
output_schema_bytes = canonical_json_bytes(fixture["output_schema"])
output_content_bytes = canonical_json_bytes(fixture["output_content"])
artifact_ref = OutputArtifactRef.create(
schema_digest=_sha256(output_schema_bytes),
content_digest=_sha256(output_content_bytes),
)
factor_set = FactorSetRef.create(
definitions=(definition,),
dataset_snapshot=snapshot,
foundation=foundation,
selected_view_ref_ids=(VIEW_REF_ID,),
input_bindings=(
InputBinding(definition.definition_id, "market", VIEW_REF_ID, VIEW_SCHEMA_DIGEST),
),
view_availability=(
ViewAvailability(VIEW_REF_ID, "2026-01-02T23:50:00Z", "sha256:" + "2" * 64),
),
output_quality=OutputQuality(
"passed",
(OutputQualityCheck("finite_values", "passed", "sha256:" + "3" * 64),),
),
output_coverage=OutputCoverage(
"complete", 1, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64
),
output_schema_bytes=output_schema_bytes,
output_content_bytes=output_content_bytes,
output_artifact_ref=artifact_ref,
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"),
correlation_id="research_run_001",
causation=Causation("foundation", foundation.foundation_id),
evidence_scope="synthetic_fixture",
decision_eligible=False,
)
return snapshot, foundation, factor_set
def _config_digest(parameters: dict[str, object] | None = None) -> str:
encoded = json.dumps(
PARAMETERS if parameters is None else parameters,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
return _sha256(encoded)
def _run_ref(**overrides: Any) -> BacktestRunRef:
snapshot, foundation, factor_set = _accepted_authorities()
arguments: dict[str, Any] = {
"dataset_snapshot": snapshot,
"foundation": foundation,
"factor_set": factor_set,
"universe_digest": "sha256:" + "5" * 64,
"trading_calendar_revision_ids": (CALENDAR_REVISION_ID,),
"corporate_action_revision_ids": (ACTION_REVISION_ID,),
"strategy_id": "alpha-top1",
"strategy_version": "1.0.0",
"strategy_digest": "sha256:" + "6" * 64,
"execution_model_version": "1.0.0",
"execution_model_digest": "sha256:" + "7" * 64,
"cost_model_version": "1.0.0",
"cost_model_digest": "sha256:" + "8" * 64,
"random_seed": 7,
"code_revision": "d" * 40,
"environment_lock_digest": "sha256:" + "9" * 64,
"configuration_digest": _config_digest(),
"evaluation_at": "2026-01-08T01:00:00Z",
"computed_at": "2026-01-08T02:00:00Z",
}
arguments.update(overrides)
return BacktestRunRef.create(**arguments)
def _backtest_result() -> FactorBacktestResult:
dates = pd.date_range("2026-01-05", periods=4, freq="B")
scores = pd.DataFrame({"A": [2.0, 0.0], "B": [1.0, 3.0]}, index=dates[:2])
opens = pd.DataFrame(
{"A": [10.0, 10.0, 15.0, 15.0], "B": [20.0, 20.0, 20.0, 21.0]},
index=dates,
)
closes = pd.DataFrame(
{"A": [10.0, 12.0, 15.0, 15.0], "B": [20.0, 20.0, 18.0, 21.0]},
index=dates,
)
return run_factor_backtest_research(
scores,
opens,
closes,
top_k=1,
execution_price_field="open",
valuation_price_field="close",
initial_cash=1_000.0,
config=ExecutionConfig(
commission_bps=0,
stamp_tax_bps=0,
slippage_bps=0,
min_trade_amount=0,
),
)
def _artifact(run_ref: BacktestRunRef, *, run_id: str | None = None) -> ResearchRunArtifact:
result = _backtest_result()
benchmark = pd.Series(
[0.0, 0.01, -0.01, 0.02],
index=result.returns.index,
name="benchmark_return",
)
return build_research_run_artifact(
result,
run_id=run_ref.run_id if run_id is None else run_id,
strategy_id=run_ref.strategy_id,
strategy_name="Alpha Top 1",
strategy_version=run_ref.strategy_version,
engine_version="1.2.0",
code_revision=run_ref.code_revision,
data_snapshot_id=run_ref.dataset_snapshot_id,
calendar="CN-A",
timezone="Asia/Shanghai",
started_at="2026-01-08T10:00:00+08:00",
finished_at="2026-01-08T10:01:00+08:00",
parameters=PARAMETERS,
benchmark_id="000300.SH",
benchmark_returns=benchmark,
)
def _assert_error(
error: pytest.ExceptionInfo[BacktestContractError],
code: BacktestContractErrorCode,
path: str,
) -> None:
assert error.value.code is code
assert error.value.path == path
def test_backtest_run_ref_is_deterministic_and_binds_only_opaque_authorities() -> None:
first = _run_ref()
second = _run_ref()
assert first == second
assert first.run_id.startswith("rhbacktestrunv1:sha256:")
assert first.replay_spec_digest.startswith("sha256:")
assert first.dataset_snapshot_id.startswith("rhdsv1:sha256:")
assert first.foundation_id.startswith("rhdfv1:sha256:")
assert first.factor_set_id.startswith("rhfactorsetv1:sha256:")
assert first.trading_calendar_revision_ids == (CALENDAR_REVISION_ID,)
assert first.corporate_action_revision_ids == (ACTION_REVISION_ID,)
assert first.replay_parent_run_id is None
assert first.replay_attempt == 0
assert first.replay_ancestor_run_ids == ()
snapshot, foundation, factor_set = _accepted_authorities()
assert BacktestRunRef.from_dict(
first.to_dict(),
dataset_snapshot=snapshot,
foundation=foundation,
factor_set=factor_set,
) == first
forbidden = ("latest", "locator", "uri", "credential", "provider", "broker")
assert not any(token in first.to_json().lower() for token in forbidden)
@pytest.mark.parametrize(
("field", "value"),
[
("universe_digest", "sha256:" + "a" * 64),
("strategy_digest", "sha256:" + "b" * 64),
("execution_model_digest", "sha256:" + "c" * 64),
("cost_model_digest", "sha256:" + "e" * 64),
("random_seed", 8),
("code_revision", "e" * 40),
("environment_lock_digest", "sha256:" + "f" * 64),
("configuration_digest", "sha256:" + "0" * 64),
("evaluation_at", "2026-01-08T01:00:01Z"),
("computed_at", "2026-01-08T02:00:01Z"),
],
)
def test_every_governed_run_input_mutation_changes_run_identity(
field: str,
value: object,
) -> None:
assert _run_ref(**{field: value}).run_id != _run_ref().run_id
def test_backtest_run_ref_rejects_unclosed_upstream_and_unsafe_scalars() -> None:
with pytest.raises(BacktestContractError) as wrong_calendar:
_run_ref(trading_calendar_revision_ids=())
_assert_error(
wrong_calendar,
BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
"$.trading_calendar_revision_ids",
)
with pytest.raises(BacktestContractError) as wrong_action:
_run_ref(corporate_action_revision_ids=())
_assert_error(
wrong_action,
BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
"$.corporate_action_revision_ids",
)
with pytest.raises(BacktestContractError) as bool_seed:
_run_ref(random_seed=True)
_assert_error(bool_seed, BacktestContractErrorCode.TYPE_ERROR, "$.random_seed")
with pytest.raises(BacktestContractError) as bad_revision:
_run_ref(code_revision="abc")
_assert_error(bad_revision, BacktestContractErrorCode.INVALID_FORMAT, "$.code_revision")
with pytest.raises(BacktestContractError) as bad_digest:
_run_ref(universe_digest="5" * 64)
_assert_error(bad_digest, BacktestContractErrorCode.INVALID_FORMAT, "$.universe_digest")
with pytest.raises(BacktestContractError) as lookahead:
_run_ref(computed_at="2026-01-08T00:59:59Z")
_assert_error(lookahead, BacktestContractErrorCode.TIME_ORDER_VIOLATION, "$.computed_at")
with pytest.raises(BacktestContractError) as factor_type:
_run_ref(factor_set="rhfactorsetv1:sha256:" + "0" * 64)
_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")
_assert_error(
mutable_alias,
BacktestContractErrorCode.INVALID_FORMAT,
"$.strategy_id",
)
with pytest.raises(BacktestContractError) as physical_uri:
_run_ref(execution_model_version="s3://model-bucket/current")
_assert_error(
physical_uri,
BacktestContractErrorCode.INVALID_FORMAT,
"$.execution_model_version",
)
with pytest.raises(BacktestContractError) as unsafe_seed:
_run_ref(random_seed=2**53)
_assert_error(
unsafe_seed,
BacktestContractErrorCode.INVALID_VALUE,
"$.random_seed",
)
with pytest.raises(BacktestContractError) as invalid_unicode:
_run_ref(strategy_id="\ud800")
_assert_error(
invalid_unicode,
BacktestContractErrorCode.INVALID_FORMAT,
"$.strategy_id",
)
run_ref = _run_ref()
mixed_keys = run_ref.to_dict()
mixed_keys[1] = "not-a-contract-key" # type: ignore[index]
snapshot, foundation, factor_set = _accepted_authorities()
with pytest.raises(BacktestContractError) as invalid_key:
BacktestRunRef.from_dict(
mixed_keys,
dataset_snapshot=snapshot,
foundation=foundation,
factor_set=factor_set,
)
_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(
computed_at="2026-01-08T03:00:00Z",
parent=parent,
replay_reason="deterministic_reproduction",
replay_attempt=1,
)
assert replay.run_id != parent.run_id
assert replay.replay_spec_digest == parent.replay_spec_digest
assert replay.replay_parent_run_id == parent.run_id
assert replay.replay_ancestor_run_ids == (parent.run_id,)
with pytest.raises(BacktestContractError) as changed_input:
_run_ref(
universe_digest="sha256:" + "a" * 64,
computed_at="2026-01-08T03:00:00Z",
parent=parent,
replay_reason="changed_universe",
replay_attempt=1,
)
_assert_error(
changed_input,
BacktestContractErrorCode.LINEAGE_VIOLATION,
"$.replay_spec_digest",
)
with pytest.raises(BacktestContractError) as skipped_attempt:
_run_ref(
computed_at="2026-01-08T03:00:00Z",
parent=parent,
replay_reason="skipped_attempt",
replay_attempt=2,
)
_assert_error(
skipped_attempt,
BacktestContractErrorCode.LINEAGE_VIOLATION,
"$.replay_attempt",
)
def test_offline_research_manifest_closes_exact_existing_evidence_mapping() -> None:
run_ref = _run_ref()
artifact = _artifact(run_ref)
first = build_backtest_evidence_manifest(
run_ref,
artifact,
artifact_available_at="2026-01-08T02:05:00Z",
qualification=EvidenceQualification.CONTRACT_QUALIFIED,
)
second = build_backtest_evidence_manifest(
run_ref,
artifact,
artifact_available_at="2026-01-08T02:05:00Z",
qualification=EvidenceQualification.CONTRACT_QUALIFIED,
)
assert first == second
assert first.manifest_id.startswith("rhbacktestevidencev1:sha256:")
assert first.run_id == run_ref.run_id
assert first.profile == "offline_research_v1"
assert first.qualification is EvidenceQualification.CONTRACT_QUALIFIED
mapping = {
item.category: tuple(table.logical_name for table in item.tables)
for item in first.evidence
}
assert mapping == {
"run": ("run",),
"signal": ("signals",),
"fill": ("trades",),
"position_nav": ("positions", "nav"),
"performance": ("performance",),
"attribution": ("attribution", "attribution_daily"),
"risk_snapshot": ("risk",),
"replay": (),
}
assert "order" not in mapping
assert "rejection" not in mapping
risk = next(item for item in first.evidence if item.category == "risk_snapshot")
assert risk.tables[0].row_count == 0
assert risk.tables[0].schema_digest.startswith("sha256:")
changed_performance = artifact.performance
changed_performance.loc[0, "n_days"] += 1
changed_artifact = replace(artifact, _performance=changed_performance)
changed = build_backtest_evidence_manifest(
run_ref,
changed_artifact,
artifact_available_at="2026-01-08T02:05:00Z",
)
assert changed.manifest_id != first.manifest_id
assert run_ref.run_id == first.run_id == changed.run_id
def test_manifest_rejects_missing_mismatched_or_duplicate_evidence() -> None:
run_ref = _run_ref()
artifact = _artifact(run_ref)
with pytest.raises(BacktestContractError) as wrong_run:
build_backtest_evidence_manifest(
run_ref,
_artifact(run_ref, run_id="different-run"),
artifact_available_at="2026-01-08T02:05:00Z",
)
_assert_error(
wrong_run,
BacktestContractErrorCode.IDENTITY_MISMATCH,
"$.artifact.tables.run.run_id",
)
missing_signals = replace(artifact, _signals=None) # type: ignore[arg-type]
with pytest.raises(BacktestContractError) as missing_table:
build_backtest_evidence_manifest(
run_ref,
missing_signals,
artifact_available_at="2026-01-08T02:05:00Z",
)
_assert_error(
missing_table,
BacktestContractErrorCode.TYPE_ERROR,
"$.artifact.tables.signals",
)
with pytest.raises(BacktestContractError) as digest_mismatch:
build_backtest_evidence_manifest(
run_ref,
artifact,
artifact_available_at="2026-01-08T02:05:00Z",
expected_table_digests={"performance": "sha256:" + "0" * 64},
)
_assert_error(
digest_mismatch,
BacktestContractErrorCode.EVIDENCE_MISMATCH,
"$.artifact.tables.performance.content_digest",
)
manifest = build_backtest_evidence_manifest(
run_ref,
artifact,
artifact_available_at="2026-01-08T02:05:00Z",
)
duplicate = manifest.to_dict()
duplicate["evidence"].append(copy.deepcopy(duplicate["evidence"][0]))
with pytest.raises(BacktestContractError) as duplicate_category:
BacktestEvidenceManifest.from_dict(
duplicate,
backtest_run_ref=run_ref,
artifact=artifact,
)
_assert_error(
duplicate_category,
BacktestContractErrorCode.INVALID_VALUE,
"$.evidence[8].category",
)
def test_manifest_binds_supported_schema_and_has_collision_free_cell_encoding() -> None:
run_ref = _run_ref()
artifact = _artifact(run_ref)
with pytest.raises(BacktestContractError) as unsupported_schema:
build_backtest_evidence_manifest(
run_ref,
replace(artifact, schema_version="999.0.0"),
artifact_available_at="2026-01-08T02:05:00Z",
)
_assert_error(
unsupported_schema,
BacktestContractErrorCode.INVALID_VALUE,
"$.artifact.schema_version",
)
identities: set[str] = set()
for value in (float("nan"), float("inf"), float("-inf")):
performance = artifact.performance
performance.loc[0, "alpha"] = value
manifest = build_backtest_evidence_manifest(
run_ref,
replace(artifact, _performance=performance),
artifact_available_at="2026-01-08T02:05:00Z",
)
assert manifest.artifact_schema_version == RESEARCH_ARTIFACT_SCHEMA_VERSION
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:
build_backtest_evidence_manifest(
run_ref,
replace(artifact, _performance=unsupported),
artifact_available_at="2026-01-08T02:05:00Z",
)
_assert_error(
unsupported_cell,
BacktestContractErrorCode.TYPE_ERROR,
"$.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()
legacy_run = BacktestRun(
run_id="legacy-run-001",
dataset_snapshot_id=run_ref.dataset_snapshot_id,
factor_version_id="alpha_005@1.0.0",
strategy_version_id="alpha-top1@1.0.0",
code_revision=run_ref.code_revision,
config_hash=_config_digest().removeprefix("sha256:"),
created_at=datetime(2026, 1, 8, 2, 0, tzinfo=UTC),
)
artifact = _artifact(run_ref, run_id=legacy_run.run_id)
manifest = build_legacy_backtest_evidence_manifest(
legacy_run,
artifact,
artifact_available_at="2026-01-08T02:05:00Z",
)
assert manifest.qualification is EvidenceQualification.LEGACY_EXPLORATORY
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,
artifact,
artifact_available_at="2026-01-08T02:05:00Z",
qualification=EvidenceQualification.CONTRACT_QUALIFIED,
)
_assert_error(
implicit_promotion,
BacktestContractErrorCode.TYPE_ERROR,
"$.backtest_run_ref",
)
def test_golden_contract_and_architecture_boundary() -> None:
run_ref = _run_ref()
artifact = _artifact(run_ref)
manifest = build_backtest_evidence_manifest(
run_ref,
artifact,
artifact_available_at="2026-01-08T02:05:00Z",
)
golden = json.loads(BACKTEST_FIXTURE.read_text(encoding="utf-8"))
table_digests = {
table.logical_name: table.content_digest
for item in manifest.evidence
for table in item.tables
}
assert golden == {
"run_id": run_ref.run_id,
"replay_spec_digest": run_ref.replay_spec_digest,
"manifest_id": manifest.manifest_id,
"evidence_digest": manifest.evidence_digest,
"table_content_digests": table_digests,
}
governed_source = (ROOT / "src" / "quant_engine" / "governed_pipeline.py").read_text(
encoding="utf-8"
)
artifact_source = (ROOT / "src" / "quant_engine" / "artifact.py").read_text(
encoding="utf-8"
)
assert "from quant_engine.artifact" not in governed_source
assert "BacktestRunRef" in governed_source
assert "BacktestEvidenceManifest" not in governed_source
assert "BacktestEvidenceManifest" in artifact_source
assert not (ROOT / "src" / "quant_engine" / "backtest_contracts.py").exists()
Reference in New Issue View Git Blame Copy Permalink