ageorge156/quant_engine
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: harden backtest contract boundaries

Browse Source
This commit is contained in:
ao gong
2026-09-01 12:17:16 +08:00
parent e1e10ce293
commit 36d444924a
4 changed files with 301 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/quant_engine/artifact.py
+140 -2
View File
@@ -231,6 +231,7 @@ class BacktestEvidenceManifest:
manifest_id: str
run_id: str
profile: str
artifact_schema_version: str
artifact_available_at: str
qualification: EvidenceQualification
evidence_digest: str
@@ -258,6 +259,7 @@ class BacktestEvidenceManifest:
"manifest_id": self.manifest_id,
"run_id": self.run_id,
"profile": self.profile,
"artifact_schema_version": self.artifact_schema_version,
"artifact_available_at": self.artifact_available_at,
"qualification": self.qualification.value,
"run_reference": self._run_reference_dict(),
@@ -278,12 +280,19 @@ class BacktestEvidenceManifest:
) -> Self:
if type(value) is not dict:
_manifest_fail(BacktestContractErrorCode.TYPE_ERROR, "$", "must be an object")
if any(type(key) is not str for key in value):
_manifest_fail(
BacktestContractErrorCode.TYPE_ERROR,
"$",
"object keys must be strings",
)
expected_fields = {
"contract_name",
"schema_version",
"manifest_id",
"run_id",
"profile",
"artifact_schema_version",
"artifact_available_at",
"qualification",
"run_reference",
@@ -416,6 +425,121 @@ def _frame_records(frame: pd.DataFrame) -> list[dict[str, object]]:
]
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
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, np.generic):
return _evidence_json_value(value.item(), path)
if isinstance(value, pd.Timestamp | datetime | date):
return value.isoformat()
if isinstance(value, Mapping):
normalized: dict[str, object] = {}
for key, item in value.items():
if type(key) is not str:
_manifest_fail(
BacktestContractErrorCode.TYPE_ERROR,
path,
"object cell keys must be strings",
)
normalized[key] = _evidence_json_value(item, f"{path}.{key}")
return dict(sorted(normalized.items()))
if isinstance(value, list | tuple):
return [
_evidence_json_value(item, f"{path}[{index}]")
for index, item in enumerate(value)
]
_manifest_fail(
BacktestContractErrorCode.TYPE_ERROR,
path,
f"unsupported evidence cell type: {type(value).__name__}",
)
def _evidence_cell_kind(value: object, path: str) -> str:
if value is None:
return "null"
if type(value) is bool:
return "boolean"
if type(value) is int:
return "integer"
if type(value) is float:
return "number"
if isinstance(value, np.generic):
return _evidence_cell_kind(value.item(), path)
if type(value) is str:
_evidence_json_value(value, path)
return "text"
if isinstance(value, pd.Timestamp | datetime):
return "datetime"
if isinstance(value, date):
return "date"
if isinstance(value, Mapping):
_evidence_json_value(value, path)
return "object"
if isinstance(value, list | tuple):
_evidence_json_value(value, path)
return "array"
_manifest_fail(
BacktestContractErrorCode.TYPE_ERROR,
path,
f"unsupported evidence cell type: {type(value).__name__}",
)
def _evidence_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"$.artifact.tables.{table_name}",
"table records cannot be serialized",
) from error
records: list[dict[str, object]] = []
for row_index, row in enumerate(raw_records):
records.append(
{
str(key): _evidence_json_value(
value,
f"$.artifact.tables.{table_name}.rows[{row_index}].{key}",
)
for key, value in row.items()
}
)
return records
def _evidence_logical_types(frame: pd.DataFrame, table_name: str) -> list[str]:
logical_types: list[str] = []
for column in frame.columns:
kinds = {
_evidence_cell_kind(
value,
f"$.artifact.tables.{table_name}.columns.{column}",
)
for value in frame[column].tolist()
}
if len(kinds) > 1:
kinds.discard("null")
logical_types.append("|".join(sorted(kinds)) if kinds else "unknown")
return logical_types
_MANIFEST_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
_OFFLINE_RESEARCH_V1: tuple[tuple[str, tuple[str, ...]], ...] = (
("run", ("run",)),
@@ -522,6 +646,12 @@ def _artifact_frames(artifact: ResearchRunArtifact) -> dict[str, pd.DataFrame]:
"$.artifact",
"ResearchRunArtifact is required",
)
if artifact.schema_version != RESEARCH_ARTIFACT_SCHEMA_VERSION:
_manifest_fail(
BacktestContractErrorCode.INVALID_VALUE,
"$.artifact.schema_version",
"unsupported research artifact schema version",
)
frames: dict[str, pd.DataFrame] = {}
for table_name in _ARTIFACT_TABLE_NAMES:
try:
@@ -590,11 +720,15 @@ def _table_evidence(
for table_name in _ARTIFACT_TABLE_NAMES:
frame = frames[table_name]
columns = [str(column) for column in frame.columns]
records = _evidence_frame_records(frame, table_name)
schema = {
"columns": columns,
"dtypes": [str(dtype) for dtype in frame.dtypes.tolist()],
"logical_types": _evidence_logical_types(frame, table_name),
}
content = {
"columns": columns,
"records": records,
}
content = {"columns": columns, "records": _frame_records(frame)}
summary = BacktestEvidenceTable(
logical_name=table_name,
row_count=len(frame),
@@ -715,6 +849,7 @@ def _build_evidence_manifest(
*,
run_id: str,
run_reference: dict[str, object],
artifact_schema_version: str,
artifact_available_at: object,
qualification: EvidenceQualification,
summaries: Mapping[str, BacktestEvidenceTable],
@@ -733,6 +868,7 @@ def _build_evidence_manifest(
"schema_version": BACKTEST_EVIDENCE_SCHEMA_VERSION,
"run_id": run_id,
"profile": "offline_research_v1",
"artifact_schema_version": artifact_schema_version,
"artifact_available_at": available_text,
"qualification": qualification.value,
"run_reference": run_reference,
@@ -820,6 +956,7 @@ def build_backtest_evidence_manifest(
return _build_evidence_manifest(
run_id=backtest_run_ref.run_id,
run_reference=run_reference,
artifact_schema_version=artifact.schema_version,
artifact_available_at=available_text,
qualification=qualification,
summaries=summaries,
@@ -876,6 +1013,7 @@ def build_legacy_backtest_evidence_manifest(
return _build_evidence_manifest(
run_id=legacy_run.run_id,
run_reference=run_reference,
artifact_schema_version=artifact.schema_version,
artifact_available_at=available_text,
qualification=EvidenceQualification.LEGACY_EXPLORATORY,
summaries=summaries,
src/quant_engine/governed_pipeline.py
+70 -5
View File
@@ -37,6 +37,28 @@ from quant_engine.research_pipeline import FactorBacktestResult, run_factor_back
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_PREFIXED_SHA256 = re.compile(r"^sha256:[0-9a-f]{64}$")
_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.-]+)?$"
)
_MAX_SAFE_INTEGER = (1 << 53) - 1
_FORBIDDEN_ID_TOKENS = frozenset(
{
"latest",
"provider",
"broker",
"credential",
"locator",
"uri",
"s3",
"http",
"file",
"postgres",
"mysql",
"clickhouse",
}
)
__all__ = [
"BACKTEST_RUN_REF_SCHEMA_VERSION",
@@ -103,6 +125,12 @@ def _backtest_fail(
def _backtest_object(value: Any, path: str, fields: Sequence[str]) -> dict[str, Any]:
if type(value) is not dict:
_backtest_fail(BacktestContractErrorCode.TYPE_ERROR, path, "must be an object")
if any(type(key) is not str for key in value):
_backtest_fail(
BacktestContractErrorCode.TYPE_ERROR,
path,
"object keys must be strings",
)
required = set(fields)
missing = sorted(required - set(value))
if missing:
@@ -124,6 +152,14 @@ def _backtest_object(value: Any, path: str, fields: Sequence[str]) -> dict[str,
def _backtest_text(value: Any, path: str) -> str:
if type(value) is not str:
_backtest_fail(BacktestContractErrorCode.TYPE_ERROR, path, "must be a string")
try:
value.encode("utf-8")
except UnicodeEncodeError as error:
raise BacktestContractError(
BacktestContractErrorCode.INVALID_FORMAT,
path,
"must be valid UTF-8 text",
) from error
if not value or value != value.strip():
_backtest_fail(
BacktestContractErrorCode.INVALID_FORMAT,
@@ -133,6 +169,29 @@ def _backtest_text(value: Any, path: str) -> str:
return value
def _backtest_logical_id(value: Any, path: str) -> str:
logical_id = _backtest_text(value, path)
tokens = set(re.split(r"[._-]+", logical_id))
if _LOGICAL_ID.fullmatch(logical_id) is None or tokens & _FORBIDDEN_ID_TOKENS:
_backtest_fail(
BacktestContractErrorCode.INVALID_FORMAT,
path,
"must be a stable storage-neutral logical identifier",
)
return logical_id
def _backtest_semver(value: Any, path: str) -> str:
version = _backtest_text(value, path)
if _SEMVER.fullmatch(version) is None:
_backtest_fail(
BacktestContractErrorCode.INVALID_FORMAT,
path,
"must be a canonical semantic version",
)
return version
def _backtest_digest(value: Any, path: str) -> str:
digest = _backtest_text(value, path)
if _PREFIXED_SHA256.fullmatch(digest) is None:
@@ -190,6 +249,12 @@ def _backtest_integer(value: Any, path: str, *, minimum: int = 0) -> int:
path,
f"must be at least {minimum}",
)
if value > _MAX_SAFE_INTEGER:
_backtest_fail(
BacktestContractErrorCode.INVALID_VALUE,
path,
"integer exceeds the canonical safe range",
)
return value
@@ -494,10 +559,10 @@ class BacktestRunRef:
"trading_calendar_digest": _digest_document(list(supplied_calendars)),
"corporate_action_revision_ids": list(supplied_actions),
"corporate_action_digest": _digest_document(list(supplied_actions)),
"strategy_id": _backtest_text(strategy_id, "$.strategy_id"),
"strategy_version": _backtest_text(strategy_version, "$.strategy_version"),
"strategy_id": _backtest_logical_id(strategy_id, "$.strategy_id"),
"strategy_version": _backtest_semver(strategy_version, "$.strategy_version"),
"strategy_digest": _backtest_digest(strategy_digest, "$.strategy_digest"),
"execution_model_version": _backtest_text(
"execution_model_version": _backtest_semver(
execution_model_version,
"$.execution_model_version",
),
@@ -505,7 +570,7 @@ class BacktestRunRef:
execution_model_digest,
"$.execution_model_digest",
),
"cost_model_version": _backtest_text(cost_model_version, "$.cost_model_version"),
"cost_model_version": _backtest_semver(cost_model_version, "$.cost_model_version"),
"cost_model_digest": _backtest_digest(
cost_model_digest,
"$.cost_model_digest",
@@ -546,7 +611,7 @@ class BacktestRunRef:
"$.parent",
"BacktestRunRef parent is required",
)
normalized_reason = _backtest_text(replay_reason, "$.replay_reason")
normalized_reason = _backtest_logical_id(replay_reason, "$.replay_reason")
if replay_count != parent.replay_attempt + 1:
_backtest_fail(
BacktestContractErrorCode.LINEAGE_VIOLATION,
tests/fixtures/backtest-evidence-v1.golden.json
Vendored
+2 -2
View File
@@ -1,8 +1,8 @@
{
"run_id": "rhbacktestrunv1:sha256:5036c771c44a2adade9ea590eee0d8cf824ff8f519fadd8ba424e5d914856386",
"replay_spec_digest": "sha256:20f07fcb526bc38b4ba3d71d6c3b00a8c63ad96cab0fecd869326503ab42d98b",
"manifest_id": "rhbacktestevidencev1:sha256:47e86c53e696672a8931e1cb496b6a48344a4ffbf84bbec4818ea2671fee8d42",
"evidence_digest": "sha256:907775054f4316bcb2f559a3068493210cebaa97bb90e462ba39ae0762eea98e",
"manifest_id": "rhbacktestevidencev1:sha256:f94733e849433f62f3da1e1ec8999891b93d49c7d657832d64e64bef9e211117",
"evidence_digest": "sha256:f3913894d032c389c64eef59058b3cbc694cb9cc14ce9cee2699f0068200b650",
"table_content_digests": {
"run": "sha256:7d947ec93f714641669cbb14bd69dd8cf30387aabb7f3a086918fc2e878ab05c",
"signals": "sha256:72dc15064cc45d7c51d2dd4b8c3a6d8c7d4155232d5d70d3c9e7697fab70ce50",
tests/test_backtest_contracts.py
+89
View File
@@ -16,6 +16,7 @@ import pytest
from quant_engine.artifact import (
BacktestEvidenceManifest,
EvidenceQualification,
RESEARCH_ARTIFACT_SCHEMA_VERSION,
ResearchRunArtifact,
build_backtest_evidence_manifest,
build_legacy_backtest_evidence_manifest,
@@ -310,6 +311,50 @@ def test_backtest_run_ref_rejects_unclosed_upstream_and_unsafe_scalars() -> None
_assert_error(factor_type, BacktestContractErrorCode.TYPE_ERROR, "$.factor_set")
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, "$")
def test_replay_lineage_is_acyclic_and_cannot_claim_changed_inputs() -> None:
parent = _run_ref()
replay = _run_ref(
@@ -462,6 +507,50 @@ def test_manifest_rejects_missing_mismatched_or_duplicate_evidence() -> None:
)
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
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",
)
def test_legacy_bridge_is_explicit_and_cannot_be_contract_qualified() -> None:
run_ref = _run_ref()
legacy_run = BacktestRun(