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

feat: add explicit retrospective v2 computation contracts (#20)
CI / lite (push) Successful in 19s
Details

Browse Source
This commit was merged in pull request #20.
This commit is contained in:
ageorge156
2026-09-09 01:31:46 +08:00
parent 68dd68392a
commit 861c1e97a8
19 changed files with 7688 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_retrospective_factor_contracts.py
+367
View File
@@ -0,0 +1,367 @@
"""Synthetic v2 computation boundaries; never source authentication."""
from __future__ import annotations
from copy import deepcopy
from typing import Any
import pytest
from quant_engine.factor_contracts import (
ActorIdentity,
FactorContractError,
FactorDefinition,
FactorInput,
FactorSetRef,
OutputArtifactRef,
OutputCoverage,
OutputQuality,
OutputQualityCheck,
PayloadValidation,
ProducerIdentity,
canonical_json_bytes,
factor_input_schema_digest,
)
from quant_engine.retrospective_data_contracts import (
RetrospectiveFoundationEnvelope,
RetrospectiveSnapshotEnvelope,
)
from quant_engine.retrospective_factor_contracts import (
ResolvedRetrospectiveView,
RetrospectiveCausation,
RetrospectiveFactorSetRef,
RetrospectiveInputBinding,
RetrospectiveViewAvailability,
)
from test_retrospective_data_contracts import (
digest,
golden,
identify,
records,
replace_at,
seal_foundation,
)
def factor_arguments() -> dict[str, Any]:
snapshot = RetrospectiveSnapshotEnvelope.from_dict(golden("dataset-snapshot"))
foundation = RetrospectiveFoundationEnvelope.from_dict(
golden("data-foundation"), snapshot=snapshot
)
view = next(iter(foundation.views.values()))
factor_inputs = (FactorInput("market", view.schema_digest, ("value",)),)
definition = FactorDefinition.create(
factor_id="neutral_close",
version="1.0.0",
formula="value",
parameters={},
implementation_digest=digest({"synthetic_formula": "identity"}),
input_schema_digest=factor_input_schema_digest(factor_inputs),
inputs=factor_inputs,
valid_from="2026-01-01T00:00:00Z",
valid_until="2027-01-01T00:00:00Z",
warmup_sessions=0,
lag_sessions=1,
producer=ProducerIdentity("quant_engine", "0.1.0"),
code_revision="c" * 40,
)
schema = {"fields": ["instrument_id", "value"]}
output = [{"instrument_id": row["instrument_id"], "value": row["value"]} for row in records()]
return {
"definitions": (definition,),
"dataset_snapshot": snapshot,
"foundation": foundation,
"selected_view_ref_ids": (view.view_ref_id,),
"input_bindings": (
RetrospectiveInputBinding(
definition.definition_id, "market", view.view_ref_id, view.schema_digest
),
),
"view_availability": (
RetrospectiveViewAvailability(
view.view_ref_id, view.available_at, digest({"synthetic_view_receipt": 1})
),
),
"dataset_chunks": [records()],
"resolved_views": (
ResolvedRetrospectiveView(
view.view_ref_id,
canonical_json_bytes({"synthetic_schema": "neutral_close_v2"}),
canonical_json_bytes(sorted(records(), key=canonical_json_bytes)),
),
),
"output_quality": OutputQuality(
"passed",
(OutputQualityCheck("finite_values", "passed", digest({"synthetic_quality": 1})),),
),
"output_coverage": OutputCoverage(
"complete", 2, 2, "records", "synthetic_market", digest({"synthetic_coverage": 1})
),
"output_schema_bytes": canonical_json_bytes(schema),
"output_content_bytes": canonical_json_bytes(output),
"output_artifact_ref": OutputArtifactRef.create(
schema_digest=digest(schema), content_digest=digest(output)
),
"evaluation_at": "2026-09-08T01:06:00Z",
"computed_at": "2026-09-08T01:07:00Z",
"artifact_available_at": "2026-09-08T01:08:00Z",
"producer": ProducerIdentity("quant_engine", "0.1.0"),
"code_revision": "d" * 40,
"actor": ActorIdentity("service", "synthetic.research"),
"correlation_id": "synthetic.retrospective",
"causation": RetrospectiveCausation("foundation", foundation.foundation_id),
"evidence_scope": "synthetic_fixture",
"decision_eligible": False,
}
def decoding_arguments(arguments: dict[str, Any]) -> dict[str, Any]:
return {key: arguments[key] for key in ("definitions", "dataset_snapshot", "foundation")}
def test_factor_result_closes_v2_inputs_and_preserves_v1_definition() -> None:
arguments = factor_arguments()
result = RetrospectiveFactorSetRef.create(**arguments)
wire = result.to_dict()
assert result.schema_version == "2.0.0"
assert result.factor_set_id.startswith("rhfactorsetv2:sha256:")
assert result.definition_ids == (arguments["definitions"][0].definition_id,)
assert result.definition_ids[0].startswith("rhfactorv1:")
assert wire["usage"] == "retrospective_research"
assert wire["availability_mode"] == "retrospective_replay"
assert wire["historical_availability"] == "not_established"
assert wire["observation_cutoff"] == arguments["foundation"].observation_cutoff
assert wire["decision_eligible"] is False
assert "pit_cutoff" not in wire
assert result.payload_validation is PayloadValidation.PAYLOAD_REVALIDATED
assert result.input_payload_validation is PayloadValidation.PAYLOAD_REVALIDATED
restored = RetrospectiveFactorSetRef.from_json(
result.to_json(), **decoding_arguments(arguments)
)
assert restored.to_dict() == wire
assert restored.payload_validation is PayloadValidation.REFERENCE_ONLY
assert restored.input_payload_validation is PayloadValidation.REFERENCE_ONLY
@pytest.mark.parametrize(
("path", "value"),
[
("schema_version", "1.0.0"),
("schema_version", "2.1.0"),
("contract_name", "researchhub.dataset-snapshot"),
("dataset_snapshot_id", "rhdsv2:sha256:" + "0" * 64),
("foundation_id", "rhdfv2:sha256:" + "0" * 64),
("observation_cutoff", "2018-01-02T07:00:00Z"),
("pit_cutoff", "2018-01-02T07:00:00Z"),
("selected_view_ref_ids", []),
("selected_view_ref_ids", ["rhviewrefv2:sha256:" + "0" * 64]),
("definition_ids", ["rhfactorv1:sha256:" + "0" * 64]),
("input_bindings", []),
("input_bindings.0.input_name", "volume"),
("input_bindings.0.schema_digest", "sha256:" + "0" * 64),
("input_bindings.0.view_ref_id", "rhviewrefv2:sha256:" + "0" * 64),
("view_availability", []),
("view_availability.0.available_at", "2026-09-08T01:03:30Z"),
("view_availability.0.available_at", "2026-09-08T01:04:00.0000001Z"),
("upstream_evidence.quality.checks.0.status", "failed"),
("upstream_evidence.qualification.evaluated_at", "2018-01-02T07:00:00Z"),
("upstream_evidence.time_semantics.earliest_external_knowledge", {"status": "evidenced"}),
("evidence_scope", "real_data"),
("output_quality.status", "failed"),
("output_quality.checks.0.status", "failed"),
("output_coverage.status", "incomplete"),
("output_coverage.observed_count", 1),
("output_schema_digest", "sha256:" + "0" * 64),
("output_artifact_ref.artifact_id", "rhfactoroutputv1:sha256:" + "0" * 64),
("availability_mode", "as_available"),
("usage", "paper_trading"),
("historical_availability", "declared_as_available"),
("decision_eligible", True),
("decision_eligible", 0),
("evaluation_at", "2018-01-02T07:00:00Z"),
("evaluation_at", "2026-09-08T01:04:00Z"),
("computed_at", "2026-09-08T01:05:00Z"),
("artifact_available_at", "2026-09-08T01:06:00Z"),
("producer.id", "research_platform"),
("code_revision", "unknown"),
("actor.id", "https://private/a"),
("causation.id", "rhdfv2:sha256:" + "0" * 64),
],
)
def test_factor_rejects_reidentified_semantic_forgery(path: str, value: Any) -> None:
arguments = factor_arguments()
row = RetrospectiveFactorSetRef.create(**arguments).to_dict()
replace_at(row, path, value)
identify(row, "factor_set_id", "rhfactorsetv2:")
with pytest.raises(FactorContractError):
RetrospectiveFactorSetRef.from_dict(row, **decoding_arguments(arguments))
def test_payload_validation_is_never_inherited_from_serialization() -> None:
arguments = factor_arguments()
result = RetrospectiveFactorSetRef.create(**arguments)
kwargs = decoding_arguments(arguments)
reference = RetrospectiveFactorSetRef.from_dict(result.to_dict(), **kwargs)
with pytest.raises(FactorContractError):
reference.require_payloads_revalidated()
checked = RetrospectiveFactorSetRef.from_dict(
result.to_dict(),
**kwargs,
**{
key: arguments[key]
for key in (
"output_schema_bytes",
"output_content_bytes",
"dataset_chunks",
"resolved_views",
)
},
)
checked.require_payloads_revalidated()
assert checked == result
for extra in (
{"output_schema_bytes": arguments["output_schema_bytes"]},
{"dataset_chunks": arguments["dataset_chunks"]},
{"resolved_views": arguments["resolved_views"]},
{"output_schema_bytes": arguments["output_schema_bytes"], "output_content_bytes": b"[]"},
{"dataset_chunks": arguments["dataset_chunks"], "resolved_views": []},
):
with pytest.raises(FactorContractError):
RetrospectiveFactorSetRef.from_dict(result.to_dict(), **kwargs, **extra)
def test_create_verifies_actual_snapshot_and_each_resolved_view() -> None:
for mutation in (
"content",
"schema",
"snapshot",
"duplicate_view",
"noncanonical",
"unknown_view",
):
arguments = factor_arguments()
view = arguments["resolved_views"][0]
if mutation == "content":
arguments["resolved_views"] = (
ResolvedRetrospectiveView(view.view_ref_id, view.schema_bytes, b"[]"),
)
elif mutation == "schema":
arguments["resolved_views"] = (
ResolvedRetrospectiveView(view.view_ref_id, b"{}", view.content_bytes),
)
elif mutation == "snapshot":
arguments["dataset_chunks"][0][0]["value"] = "0"
elif mutation == "duplicate_view":
arguments["resolved_views"] = (view, view)
elif mutation == "unknown_view":
arguments["resolved_views"] = (
ResolvedRetrospectiveView(
"rhviewrefv2:sha256:" + "0" * 64, view.schema_bytes, view.content_bytes
),
)
else:
arguments["output_content_bytes"] += b"\n"
with pytest.raises(FactorContractError):
RetrospectiveFactorSetRef.create(**arguments)
def test_parent_requires_exact_correlation_scope_and_actual_availability() -> None:
arguments = factor_arguments()
parent = RetrospectiveFactorSetRef.create(**arguments)
child_args = {
**arguments,
"parent": parent,
"causation": RetrospectiveCausation("factor_set", parent.factor_set_id),
"evaluation_at": "2026-09-08T01:09:00Z",
"computed_at": "2026-09-08T01:10:00Z",
"artifact_available_at": "2026-09-08T01:11:00Z",
}
child = RetrospectiveFactorSetRef.create(**child_args)
assert child.factor_set_id != parent.factor_set_id
assert (
RetrospectiveFactorSetRef.from_json(
child.to_json(), **decoding_arguments(arguments), parent=parent
)
== child
)
for changes in (
{"parent": None},
{"correlation_id": "different.correlation"},
{"causation": RetrospectiveCausation("factor_set", "rhfactorsetv2:sha256:" + "0" * 64)},
{"evaluation_at": "2026-09-08T01:07:59Z"},
{"causation": arguments["causation"]},
):
with pytest.raises(FactorContractError):
RetrospectiveFactorSetRef.create(**{**child_args, **changes})
def test_definition_validity_is_checked_at_actual_evaluation() -> None:
arguments = factor_arguments()
arguments.update(
evaluation_at="2027-01-01T00:00:00Z",
computed_at="2027-01-01T00:01:00Z",
artifact_available_at="2027-01-01T00:02:00Z",
)
with pytest.raises(FactorContractError):
RetrospectiveFactorSetRef.create(**arguments)
def test_factor_contract_is_immutable_and_v1_does_not_accept_it() -> None:
arguments = factor_arguments()
result = RetrospectiveFactorSetRef.create(**arguments)
exported = result.to_dict()
exported["upstream_evidence"]["quality"]["status"] = "failed"
assert result.upstream_evidence["quality"]["status"] == "passed"
with pytest.raises(TypeError):
result.upstream_evidence["quality"]["status"] = "failed"
with pytest.raises(FactorContractError):
FactorSetRef.from_dict(result.to_dict(), **decoding_arguments(arguments))
with pytest.raises(FactorContractError):
RetrospectiveFactorSetRef.from_json(
result.to_json() + "\n", **decoding_arguments(arguments)
)
with pytest.raises(FactorContractError):
RetrospectiveInputBinding(
arguments["definitions"][0].definition_id,
"market",
"rhviewrefv1:sha256:" + "0" * 64,
"sha256:" + "0" * 64,
)
def test_missing_synthetic_or_real_readiness_cannot_be_relabelled() -> None:
arguments = factor_arguments()
snapshot_row = arguments["dataset_snapshot"].to_dict()
snapshot_row["evidence_scope"] = "real_data"
identify(snapshot_row, "snapshot_id", "rhdsv2:")
snapshot = RetrospectiveSnapshotEnvelope.from_dict(snapshot_row)
foundation_row = arguments["foundation"].to_dict()
foundation_row["dataset_snapshot_id"] = snapshot.snapshot_id
foundation_row["readiness"]["evidence_scope"] = "real_data"
for view in foundation_row["standardized_views"]:
view["dataset_snapshot_id"] = snapshot.snapshot_id
seal_foundation(foundation_row)
foundation = RetrospectiveFoundationEnvelope.from_dict(foundation_row, snapshot=snapshot)
view = next(iter(foundation.views.values()))
arguments.update(
dataset_snapshot=snapshot,
foundation=foundation,
evidence_scope="real_data",
selected_view_ref_ids=(view.view_ref_id,),
input_bindings=(
RetrospectiveInputBinding(
arguments["definitions"][0].definition_id,
"market",
view.view_ref_id,
view.schema_digest,
),
),
view_availability=(
RetrospectiveViewAvailability(
view.view_ref_id, view.available_at, digest({"synthetic_view": 1})
),
),
causation=RetrospectiveCausation("foundation", foundation.foundation_id),
)
with pytest.raises(FactorContractError, match="real-data"):
RetrospectiveFactorSetRef.create(**arguments)