368 lines
15 KiB
Python
368 lines
15 KiB
Python
"""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)
|