939 lines
40 KiB
Python
939 lines
40 KiB
Python
"""Versioned factor-definition and factor-set contract conformance."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from dataclasses import FrozenInstanceError
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
import pytest
|
|
|
|
from quant_engine.factor_contracts import (
|
|
ActorIdentity,
|
|
AvailabilityMode,
|
|
ContractErrorCode,
|
|
Causation,
|
|
DataFoundationEnvelope,
|
|
DatasetSnapshotEnvelope,
|
|
FactorContractError,
|
|
FactorDefinition,
|
|
FactorInput,
|
|
FactorSetRef,
|
|
HistoricalAvailability,
|
|
InputBinding,
|
|
LegacyFactorBinding,
|
|
OutputArtifactRef,
|
|
OutputCoverage,
|
|
OutputQuality,
|
|
OutputQualityCheck,
|
|
PayloadValidation,
|
|
ProducerIdentity,
|
|
TypedParameter,
|
|
ViewAvailability,
|
|
canonical_json_bytes,
|
|
factor_definition_from_alpha158,
|
|
factor_input_schema_digest,
|
|
validate_factor_catalog,
|
|
)
|
|
from quant_engine.governed_pipeline import (
|
|
FactorVersion,
|
|
bind_legacy_factor,
|
|
project_legacy_factor,
|
|
)
|
|
|
|
|
|
FIXTURE_PATH = Path(__file__).parent / "fixtures" / "factor-contracts-v1.golden.json"
|
|
VIEW_REF_ID = "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c"
|
|
VIEW_SCHEMA_DIGEST = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
|
|
|
|
|
|
def _golden() -> dict[str, Any]:
|
|
loaded = json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))
|
|
assert isinstance(loaded, dict)
|
|
return loaded
|
|
|
|
|
|
def _sha256(value: bytes) -> str:
|
|
return f"sha256:{hashlib.sha256(value).hexdigest()}"
|
|
|
|
|
|
def _reidentify(item: dict[str, Any], field: str, prefix: str) -> None:
|
|
payload = {key: value for key, value in item.items() if key != field}
|
|
item[field] = f"{prefix}{hashlib.sha256(canonical_json_bytes(payload)).hexdigest()}"
|
|
|
|
|
|
def _snapshot_and_foundation(
|
|
fixture: dict[str, Any] | None = None,
|
|
) -> tuple[DatasetSnapshotEnvelope, DataFoundationEnvelope]:
|
|
source = _golden() if fixture is None else fixture
|
|
return (
|
|
DatasetSnapshotEnvelope.from_dict(source["dataset_snapshot"]),
|
|
DataFoundationEnvelope.from_dict(source["data_foundation"]),
|
|
)
|
|
|
|
|
|
def _definition(
|
|
*,
|
|
inputs: tuple[FactorInput, ...] | None = None,
|
|
**overrides: Any,
|
|
) -> FactorDefinition:
|
|
factor_inputs = inputs or (
|
|
FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume")),
|
|
)
|
|
arguments: dict[str, Any] = {
|
|
"factor_id": "alpha_005",
|
|
"version": "1.0.0",
|
|
"formula": "correlation(close, volume, 10)",
|
|
"parameters": {},
|
|
"implementation_digest": "sha256:" + "1" * 64,
|
|
"input_schema_digest": factor_input_schema_digest(factor_inputs),
|
|
"inputs": factor_inputs,
|
|
"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,
|
|
}
|
|
arguments.update(overrides)
|
|
return FactorDefinition.create(**arguments)
|
|
|
|
|
|
def _golden_definition() -> FactorDefinition:
|
|
factor_input = FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume"))
|
|
return 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,
|
|
)
|
|
|
|
|
|
def _factor_set_arguments(
|
|
*,
|
|
fixture: dict[str, Any] | None = None,
|
|
snapshot: DatasetSnapshotEnvelope | None = None,
|
|
foundation: DataFoundationEnvelope | None = None,
|
|
definition: FactorDefinition | None = None,
|
|
) -> dict[str, Any]:
|
|
source = _golden() if fixture is None else fixture
|
|
if snapshot is None or foundation is None:
|
|
parsed_snapshot, parsed_foundation = _snapshot_and_foundation(source)
|
|
snapshot = snapshot or parsed_snapshot
|
|
foundation = foundation or parsed_foundation
|
|
selected_definition = definition or _golden_definition()
|
|
output_schema_bytes = canonical_json_bytes(source["output_schema"])
|
|
output_content_bytes = canonical_json_bytes(source["output_content"])
|
|
artifact = OutputArtifactRef.create(
|
|
schema_digest=_sha256(output_schema_bytes),
|
|
content_digest=_sha256(output_content_bytes),
|
|
)
|
|
return {
|
|
"definitions": (selected_definition,),
|
|
"dataset_snapshot": snapshot,
|
|
"foundation": foundation,
|
|
"selected_view_ref_ids": (VIEW_REF_ID,),
|
|
"input_bindings": (
|
|
InputBinding(
|
|
selected_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,
|
|
"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",
|
|
"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,
|
|
}
|
|
|
|
|
|
def _factor_set(**overrides: Any) -> FactorSetRef:
|
|
arguments = _factor_set_arguments()
|
|
arguments.update(overrides)
|
|
return FactorSetRef.create(**arguments)
|
|
|
|
|
|
def _assert_error(
|
|
error: pytest.ExceptionInfo[FactorContractError],
|
|
code: ContractErrorCode,
|
|
path: str,
|
|
) -> None:
|
|
assert error.value.code is code
|
|
assert error.value.path == path
|
|
|
|
|
|
def _mutate_artifact_schema_binding(value: dict[str, Any]) -> None:
|
|
artifact = value["output_artifact_ref"]
|
|
artifact["schema_digest"] = "sha256:" + "0" * 64
|
|
_reidentify(artifact, "artifact_id", "rhfactoroutputv1:sha256:")
|
|
|
|
|
|
def test_golden_contracts_are_content_addressed_round_trippable_and_deeply_immutable() -> None:
|
|
fixture = _golden()
|
|
original_snapshot = copy.deepcopy(fixture["dataset_snapshot"])
|
|
original_foundation = copy.deepcopy(fixture["data_foundation"])
|
|
snapshot, foundation = _snapshot_and_foundation(fixture)
|
|
definition = _golden_definition()
|
|
factor_set = FactorSetRef.create(**_factor_set_arguments(fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition))
|
|
binding = LegacyFactorBinding.create(
|
|
definition=definition,
|
|
legacy_factor_id="factor:demo-momentum",
|
|
legacy_version="1.0.0",
|
|
legacy_definition_sha256="b" * 64,
|
|
legacy_dataset_schema_version="1.0.0",
|
|
canonical_input_schema_digest=definition.input_schema_digest,
|
|
correspondence_evidence_digest="sha256:" + "5" * 64,
|
|
)
|
|
|
|
assert snapshot.pit_cutoff == "2026-01-02T07:01:00Z"
|
|
assert foundation.pit_cutoff == factor_set.pit_cutoff == "2026-01-03T00:00:00Z"
|
|
assert snapshot.pit_cutoff != foundation.pit_cutoff
|
|
assert definition.definition_id == fixture["expected"]["definition_id"]
|
|
assert definition.input_schema_digest == fixture["expected"]["input_schema_digest"]
|
|
assert factor_set.factor_set_id == fixture["expected"]["factor_set_id"]
|
|
assert factor_set.output_artifact_ref.artifact_id == fixture["expected"]["output_artifact_id"]
|
|
assert binding.binding_id == fixture["expected"]["legacy_binding_id"]
|
|
assert not definition.to_json().endswith("\n")
|
|
assert not factor_set.to_json().endswith("\n")
|
|
assert FactorDefinition.from_json(definition.to_json()) == definition
|
|
|
|
reparsed = FactorSetRef.from_json(
|
|
factor_set.to_json(),
|
|
definitions=(definition,),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
output_schema_bytes=canonical_json_bytes(fixture["output_schema"]),
|
|
output_content_bytes=canonical_json_bytes(fixture["output_content"]),
|
|
)
|
|
reference_only = FactorSetRef.from_json(
|
|
factor_set.to_json(),
|
|
definitions=(definition,),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
assert reparsed.factor_set_id == factor_set.factor_set_id
|
|
assert reparsed.payload_validation is PayloadValidation.PAYLOAD_REVALIDATED
|
|
assert reference_only.payload_validation is PayloadValidation.REFERENCE_ONLY
|
|
|
|
fixture["dataset_snapshot"]["descriptor"]["dataset"]["dimensions"].append("forbidden")
|
|
fixture["data_foundation"]["standardized_views"][0]["schema_digest"] = "sha256:" + "0" * 64
|
|
assert snapshot.to_dict() == original_snapshot
|
|
assert foundation.to_dict() == original_foundation
|
|
returned = snapshot.to_dict()
|
|
returned["descriptor"]["dataset"]["dimensions"].append("also_forbidden")
|
|
assert snapshot.to_dict() == original_snapshot
|
|
with pytest.raises(FrozenInstanceError):
|
|
snapshot.snapshot_id = "rhdsv1:sha256:" + "0" * 64 # type: ignore[misc]
|
|
|
|
|
|
@pytest.mark.parametrize("variant", ["whitespace", "key_order"])
|
|
def test_contract_decoders_reject_non_canonical_json(variant: str) -> None:
|
|
fixture = _golden()
|
|
snapshot, foundation = _snapshot_and_foundation(fixture)
|
|
definition = _golden_definition()
|
|
factor_set = FactorSetRef.create(
|
|
**_factor_set_arguments(
|
|
fixture=fixture,
|
|
snapshot=snapshot,
|
|
foundation=foundation,
|
|
definition=definition,
|
|
)
|
|
)
|
|
binding = LegacyFactorBinding.create(
|
|
definition=definition,
|
|
legacy_factor_id="factor:demo-momentum",
|
|
legacy_version="1.0.0",
|
|
legacy_definition_sha256="b" * 64,
|
|
legacy_dataset_schema_version="1.0.0",
|
|
canonical_input_schema_digest=definition.input_schema_digest,
|
|
correspondence_evidence_digest="sha256:" + "5" * 64,
|
|
)
|
|
|
|
def non_canonical(value: str) -> str:
|
|
if variant == "whitespace":
|
|
return value + "\n"
|
|
loaded = json.loads(value)
|
|
reversed_items = dict(reversed(tuple(loaded.items())))
|
|
return json.dumps(reversed_items, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
decoders = (
|
|
lambda value: FactorDefinition.from_json(value),
|
|
lambda value: FactorSetRef.from_json(
|
|
value,
|
|
definitions=(definition,),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
),
|
|
lambda value: LegacyFactorBinding.from_json(value, definition=definition),
|
|
)
|
|
for decoder, encoded in zip(
|
|
decoders,
|
|
(definition.to_json(), factor_set.to_json(), binding.to_json()),
|
|
strict=True,
|
|
):
|
|
with pytest.raises(FactorContractError) as exc_info:
|
|
decoder(non_canonical(encoded))
|
|
assert exc_info.value.code is ContractErrorCode.INVALID_FORMAT
|
|
assert exc_info.value.path == "$"
|
|
|
|
|
|
def test_factor_definition_identity_is_order_independent_where_semantics_are_unordered() -> None:
|
|
first_input = FactorInput("prices", "sha256:" + "6" * 64, ("close",))
|
|
second_input = FactorInput("volumes", "sha256:" + "7" * 64, ("volume",))
|
|
inputs = (first_input, second_input)
|
|
parameters_a = {
|
|
"window": TypedParameter("integer", 10),
|
|
"weights": TypedParameter("json", {"fast": [1, 2], "slow": [3, 4]}),
|
|
}
|
|
parameters_b = {
|
|
"weights": TypedParameter("json", {"slow": [3, 4], "fast": [1, 2]}),
|
|
"window": TypedParameter("integer", 10),
|
|
}
|
|
first = _definition(
|
|
inputs=inputs,
|
|
parameters=parameters_a,
|
|
input_schema_digest=factor_input_schema_digest(inputs),
|
|
)
|
|
second = _definition(
|
|
inputs=tuple(reversed(inputs)),
|
|
parameters=parameters_b,
|
|
input_schema_digest=factor_input_schema_digest(tuple(reversed(inputs))),
|
|
)
|
|
assert first.definition_id == second.definition_id
|
|
assert first.to_json() == second.to_json()
|
|
|
|
semantic_changes = (
|
|
_definition(factor_id="alpha_006"),
|
|
_definition(version="1.0.1"),
|
|
_definition(formula="correlation(close, volume, 11)"),
|
|
_definition(parameters={"window": TypedParameter("integer", 10)}),
|
|
_definition(implementation_digest="sha256:" + "9" * 64),
|
|
_definition(valid_until="2027-01-02T00:00:00Z"),
|
|
_definition(warmup_sessions=11),
|
|
_definition(lag_sessions=2),
|
|
_definition(producer=ProducerIdentity("quant_engine", "1.0.1")),
|
|
_definition(code_revision="d" * 40),
|
|
)
|
|
assert all(changed.definition_id != _golden_definition().definition_id for changed in semantic_changes)
|
|
assert len({changed.definition_id for changed in semantic_changes}) == len(semantic_changes)
|
|
|
|
|
|
def test_parameter_types_decimal_profile_and_detached_nested_values_are_strict() -> None:
|
|
nested = {"ordered": [1, {"flag": True}]}
|
|
parameter = TypedParameter("json", nested)
|
|
nested["ordered"].append(2)
|
|
definition = _definition(parameters={"payload": parameter})
|
|
assert definition.to_dict()["parameters"]["payload"]["value"] == {
|
|
"ordered": [1, {"flag": True}]
|
|
}
|
|
integer_definition = _definition(parameters={"value": TypedParameter("integer", 1)})
|
|
string_definition = _definition(parameters={"value": TypedParameter("string", "1")})
|
|
assert integer_definition.definition_id != string_definition.definition_id
|
|
|
|
for parameter_type, value, code in (
|
|
("decimal", "1.0", ContractErrorCode.INVALID_FORMAT),
|
|
("decimal", "1e3", ContractErrorCode.INVALID_FORMAT),
|
|
("decimal", "-0", ContractErrorCode.INVALID_FORMAT),
|
|
("integer", True, ContractErrorCode.TYPE_ERROR),
|
|
("json", 1.5, ContractErrorCode.TYPE_ERROR),
|
|
("json", {"é": "bad-key"}, ContractErrorCode.INVALID_FORMAT),
|
|
("json", 9_007_199_254_740_992, ContractErrorCode.INVALID_VALUE),
|
|
):
|
|
with pytest.raises(FactorContractError) as error:
|
|
TypedParameter(parameter_type, value)
|
|
assert error.value.code is code
|
|
assert TypedParameter("decimal", "10.25").to_dict()["value"] == "10.25"
|
|
|
|
|
|
def test_catalog_rejects_duplicate_and_overlapping_logical_validity_but_allows_adjacency() -> None:
|
|
base = _golden_definition()
|
|
adjacent = _definition(valid_from="2027-01-01T00:00:00Z", valid_until="2028-01-01T00:00:00Z")
|
|
assert len(validate_factor_catalog((adjacent, base))) == 2
|
|
with pytest.raises(FactorContractError) as duplicate:
|
|
validate_factor_catalog((base, base))
|
|
_assert_error(duplicate, ContractErrorCode.INVALID_VALUE, "$.definitions")
|
|
overlapping = _definition(valid_from="2026-06-01T00:00:00Z", valid_until="2028-01-01T00:00:00Z")
|
|
with pytest.raises(FactorContractError) as overlap:
|
|
validate_factor_catalog((base, overlapping))
|
|
_assert_error(overlap, ContractErrorCode.TIME_ORDER_VIOLATION, "$.definitions")
|
|
|
|
|
|
def test_upstream_contracts_reject_unknown_fields_identity_forgery_and_unqualified_input() -> None:
|
|
unknown = _golden()["dataset_snapshot"]
|
|
unknown["provider"] = "forbidden"
|
|
with pytest.raises(FactorContractError) as unknown_error:
|
|
DatasetSnapshotEnvelope.from_dict(unknown)
|
|
_assert_error(unknown_error, ContractErrorCode.UNKNOWN_FIELD, "$.provider")
|
|
|
|
forged = _golden()["data_foundation"]
|
|
forged["standardized_views"][0]["schema_digest"] = "sha256:" + "0" * 64
|
|
with pytest.raises(FactorContractError) as forged_error:
|
|
DataFoundationEnvelope.from_dict(forged)
|
|
assert forged_error.value.code is ContractErrorCode.IDENTITY_MISMATCH
|
|
assert forged_error.value.path.endswith("view_ref_id")
|
|
|
|
rejected_source = _golden()
|
|
rejected_source["dataset_snapshot"]["descriptor"]["qualification"]["status"] = "rejected"
|
|
_reidentify(rejected_source["dataset_snapshot"], "snapshot_id", "rhdsv1:sha256:")
|
|
rejected_snapshot = DatasetSnapshotEnvelope.from_dict(rejected_source["dataset_snapshot"])
|
|
_, foundation = _snapshot_and_foundation()
|
|
with pytest.raises(FactorContractError) as rejected_error:
|
|
FactorSetRef.create(
|
|
**_factor_set_arguments(snapshot=rejected_snapshot, foundation=foundation)
|
|
)
|
|
_assert_error(
|
|
rejected_error,
|
|
ContractErrorCode.QUALIFICATION_REJECTED,
|
|
"$.dataset_snapshot.descriptor.qualification",
|
|
)
|
|
|
|
|
|
def test_foundation_rejects_future_knowledge_and_per_view_calendar_borrowing() -> None:
|
|
future = _golden()["data_foundation"]
|
|
action = future["corporate_action_revisions"][0]
|
|
old_action_id = action["action_revision_id"]
|
|
action["knowledge_time"] = "2026-01-03T00:00:01Z"
|
|
_reidentify(action, "action_revision_id", "rhcav1:sha256:")
|
|
future["standardized_views"][0]["corporate_action_revision_ids"] = [action["action_revision_id"]]
|
|
lineage = next(item for item in future["revision_lineage"] if item["revision_id"] == old_action_id)
|
|
lineage["revision_id"] = action["action_revision_id"]
|
|
lineage["knowledge_time"] = action["knowledge_time"]
|
|
_reidentify(future["standardized_views"][0], "view_ref_id", "rhviewrefv1:sha256:")
|
|
_reidentify(future, "foundation_id", "rhdfv1:sha256:")
|
|
with pytest.raises(FactorContractError) as future_error:
|
|
DataFoundationEnvelope.from_dict(future)
|
|
_assert_error(
|
|
future_error,
|
|
ContractErrorCode.TIME_ORDER_VIOLATION,
|
|
"$.revision_lineage.knowledge_time",
|
|
)
|
|
|
|
uncovered = _golden()["data_foundation"]
|
|
original_route_id = uncovered["instrument_routes"][0]["route_revision_id"]
|
|
second_calendar = copy.deepcopy(uncovered["trading_calendar_revisions"][0])
|
|
second_calendar["calendar_id"] = "rhcalendar:99990000111122223333444455556666"
|
|
_reidentify(second_calendar, "calendar_revision_id", "rhcalv1:sha256:")
|
|
uncovered["trading_calendar_revisions"].append(second_calendar)
|
|
route = uncovered["instrument_routes"][0]
|
|
route["calendar_id"] = second_calendar["calendar_id"]
|
|
_reidentify(route, "route_revision_id", "rhroutev1:sha256:")
|
|
route_lineage = next(item for item in uncovered["revision_lineage"] if item["revision_id"] == original_route_id)
|
|
route_lineage["revision_id"] = route["route_revision_id"]
|
|
uncovered["revision_lineage"].append(
|
|
{
|
|
"revision_kind": "trading_calendar",
|
|
"revision_id": second_calendar["calendar_revision_id"],
|
|
"revision_number": 1,
|
|
"knowledge_time": second_calendar["knowledge_time"],
|
|
"evidence_digest": second_calendar["evidence_digest"],
|
|
}
|
|
)
|
|
view = uncovered["standardized_views"][0]
|
|
view["instrument_route_revision_ids"] = [route["route_revision_id"]]
|
|
_reidentify(view, "view_ref_id", "rhviewrefv1:sha256:")
|
|
_reidentify(uncovered, "foundation_id", "rhdfv1:sha256:")
|
|
with pytest.raises(FactorContractError) as calendar_error:
|
|
DataFoundationEnvelope.from_dict(uncovered)
|
|
assert calendar_error.value.code is ContractErrorCode.INPUT_CLOSURE_VIOLATION
|
|
assert "selected route calendar" in calendar_error.value.detail
|
|
|
|
|
|
def _replay_fixture() -> dict[str, Any]:
|
|
fixture = _golden()
|
|
snapshot = fixture["dataset_snapshot"]
|
|
snapshot["descriptor"]["published_at"] = "2026-01-04T00:00:00Z"
|
|
_reidentify(snapshot, "snapshot_id", "rhdsv1:sha256:")
|
|
foundation = fixture["data_foundation"]
|
|
foundation["dataset_snapshot_id"] = snapshot["snapshot_id"]
|
|
for view in foundation["standardized_views"]:
|
|
view["dataset_snapshot_id"] = snapshot["snapshot_id"]
|
|
_reidentify(view, "view_ref_id", "rhviewrefv1:sha256:")
|
|
_reidentify(foundation, "foundation_id", "rhdfv1:sha256:")
|
|
return fixture
|
|
|
|
|
|
def test_as_available_and_retrospective_replay_keep_distinct_time_claims() -> None:
|
|
as_available = _factor_set()
|
|
assert as_available.historical_availability is HistoricalAvailability.DECLARED_AS_AVAILABLE
|
|
|
|
replay_source = _replay_fixture()
|
|
snapshot, foundation = _snapshot_and_foundation(replay_source)
|
|
replay_view_id = next(iter(foundation.views))
|
|
arguments = _factor_set_arguments(
|
|
fixture=replay_source,
|
|
snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
arguments.update(
|
|
selected_view_ref_ids=(replay_view_id,),
|
|
input_bindings=(
|
|
InputBinding(
|
|
arguments["definitions"][0].definition_id,
|
|
"market",
|
|
replay_view_id,
|
|
VIEW_SCHEMA_DIGEST,
|
|
),
|
|
),
|
|
view_availability=(
|
|
ViewAvailability(replay_view_id, "2026-01-04T00:10:00Z", "sha256:" + "2" * 64),
|
|
),
|
|
availability_mode=AvailabilityMode.RETROSPECTIVE_REPLAY,
|
|
computed_at="2026-01-04T00:20:00Z",
|
|
artifact_available_at="2026-01-04T00:25:00Z",
|
|
causation=Causation("foundation", foundation.foundation_id),
|
|
)
|
|
replay = FactorSetRef.create(**arguments)
|
|
assert replay.evaluation_at == "2026-01-03T11:00:00Z"
|
|
assert replay.computed_at == "2026-01-04T00:20:00Z"
|
|
assert replay.historical_availability is HistoricalAvailability.NOT_ESTABLISHED
|
|
|
|
replay_source_args = _factor_set_arguments(
|
|
fixture=replay_source,
|
|
snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
replay_source_args.update(
|
|
selected_view_ref_ids=(replay_view_id,),
|
|
input_bindings=(
|
|
InputBinding(
|
|
replay_source_args["definitions"][0].definition_id,
|
|
"market",
|
|
replay_view_id,
|
|
VIEW_SCHEMA_DIGEST,
|
|
),
|
|
),
|
|
view_availability=(
|
|
ViewAvailability(replay_view_id, "2026-01-02T23:50:00Z", "sha256:" + "2" * 64),
|
|
),
|
|
causation=Causation("foundation", foundation.foundation_id),
|
|
)
|
|
with pytest.raises(FactorContractError) as late_publication:
|
|
FactorSetRef.create(**replay_source_args)
|
|
_assert_error(
|
|
late_publication,
|
|
ContractErrorCode.TIME_ORDER_VIOLATION,
|
|
"$.dataset_snapshot.descriptor.published_at",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("overrides", "path"),
|
|
[
|
|
({"view_availability": (ViewAvailability(VIEW_REF_ID, "2026-01-03T00:00:01Z", "sha256:" + "2" * 64),)}, "$.view_availability"),
|
|
({"computed_at": "2026-01-02T23:40:00Z"}, "$.computed_at"),
|
|
({"artifact_available_at": "2026-01-03T10:14:00Z"}, "$.artifact_available_at"),
|
|
({"artifact_available_at": "2026-01-03T11:00:01Z"}, "$.artifact_available_at"),
|
|
({"evaluation_at": "2026-01-03T11:00:00"}, "$.evaluation_at"),
|
|
],
|
|
)
|
|
def test_as_available_time_failures_are_typed(overrides: dict[str, Any], path: str) -> None:
|
|
with pytest.raises(FactorContractError) as error:
|
|
_factor_set(**overrides)
|
|
assert error.value.code in {
|
|
ContractErrorCode.INVALID_FORMAT,
|
|
ContractErrorCode.TIME_ORDER_VIOLATION,
|
|
}
|
|
assert error.value.path == path
|
|
|
|
|
|
def test_replay_rejects_backdating_and_historical_availability_promotion() -> None:
|
|
source = _replay_fixture()
|
|
snapshot, foundation = _snapshot_and_foundation(source)
|
|
view_id = next(iter(foundation.views))
|
|
arguments = _factor_set_arguments(fixture=source, snapshot=snapshot, foundation=foundation)
|
|
definition = arguments["definitions"][0]
|
|
arguments.update(
|
|
selected_view_ref_ids=(view_id,),
|
|
input_bindings=(InputBinding(definition.definition_id, "market", view_id, VIEW_SCHEMA_DIGEST),),
|
|
view_availability=(ViewAvailability(view_id, "2026-01-04T00:10:00Z", "sha256:" + "2" * 64),),
|
|
availability_mode=AvailabilityMode.RETROSPECTIVE_REPLAY,
|
|
computed_at="2026-01-04T00:20:00Z",
|
|
artifact_available_at="2026-01-04T00:25:00Z",
|
|
causation=Causation("foundation", foundation.foundation_id),
|
|
)
|
|
replay = FactorSetRef.create(**arguments)
|
|
promoted = replay.to_dict()
|
|
promoted["historical_availability"] = "declared_as_available"
|
|
_reidentify(promoted, "factor_set_id", "rhfactorsetv1:sha256:")
|
|
with pytest.raises(FactorContractError) as promotion_error:
|
|
FactorSetRef.from_dict(
|
|
promoted,
|
|
definitions=(definition,),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
_assert_error(
|
|
promotion_error,
|
|
ContractErrorCode.READINESS_ESCALATION,
|
|
"$.historical_availability",
|
|
)
|
|
arguments["computed_at"] = "2026-01-03T11:30:00Z"
|
|
with pytest.raises(FactorContractError) as backdated_error:
|
|
FactorSetRef.create(**arguments)
|
|
_assert_error(backdated_error, ContractErrorCode.TIME_ORDER_VIOLATION, "$.computed_at")
|
|
|
|
|
|
def _multi_view_fixture() -> tuple[dict[str, Any], str]:
|
|
fixture = _golden()
|
|
foundation = fixture["data_foundation"]
|
|
second = copy.deepcopy(foundation["standardized_views"][0])
|
|
second["view_id"] = "rhview:11111111222222223333333344444444"
|
|
second["schema_digest"] = "sha256:" + "6" * 64
|
|
second["content_digest"] = "sha256:" + "7" * 64
|
|
second["transformation_digest"] = "sha256:" + "8" * 64
|
|
_reidentify(second, "view_ref_id", "rhviewrefv1:sha256:")
|
|
foundation["standardized_views"].append(second)
|
|
_reidentify(foundation, "foundation_id", "rhdfv1:sha256:")
|
|
return fixture, second["view_ref_id"]
|
|
|
|
|
|
def test_multi_input_mapping_requires_exact_consumption_closure_and_is_order_independent() -> None:
|
|
fixture, second_view_id = _multi_view_fixture()
|
|
snapshot, foundation = _snapshot_and_foundation(fixture)
|
|
inputs = (
|
|
FactorInput("prices", VIEW_SCHEMA_DIGEST, ("close",)),
|
|
FactorInput("volumes", "sha256:" + "6" * 64, ("volume",)),
|
|
)
|
|
definition = _definition(
|
|
inputs=inputs,
|
|
formula="correlation(close, volume, 10)",
|
|
input_schema_digest=factor_input_schema_digest(inputs),
|
|
)
|
|
first_binding = InputBinding(definition.definition_id, "prices", VIEW_REF_ID, VIEW_SCHEMA_DIGEST)
|
|
second_binding = InputBinding(definition.definition_id, "volumes", second_view_id, "sha256:" + "6" * 64)
|
|
first_availability = ViewAvailability(VIEW_REF_ID, "2026-01-02T23:40:00Z", "sha256:" + "2" * 64)
|
|
second_availability = ViewAvailability(second_view_id, "2026-01-02T23:50:00Z", "sha256:" + "6" * 64)
|
|
base = _factor_set_arguments(fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition)
|
|
base.update(
|
|
selected_view_ref_ids=(VIEW_REF_ID, second_view_id),
|
|
input_bindings=(first_binding, second_binding),
|
|
view_availability=(first_availability, second_availability),
|
|
causation=Causation("foundation", foundation.foundation_id),
|
|
)
|
|
first = FactorSetRef.create(**base)
|
|
reordered = dict(base)
|
|
reordered.update(
|
|
selected_view_ref_ids=(second_view_id, VIEW_REF_ID),
|
|
input_bindings=(second_binding, first_binding),
|
|
view_availability=(second_availability, first_availability),
|
|
)
|
|
assert FactorSetRef.create(**reordered).factor_set_id == first.factor_set_id
|
|
|
|
for invalid_bindings, invalid_views in (
|
|
((first_binding,), (VIEW_REF_ID, second_view_id)),
|
|
((first_binding, second_binding), (VIEW_REF_ID,)),
|
|
((first_binding, second_binding), (VIEW_REF_ID, second_view_id, VIEW_REF_ID)),
|
|
):
|
|
invalid = dict(base)
|
|
invalid.update(input_bindings=invalid_bindings, selected_view_ref_ids=invalid_views)
|
|
with pytest.raises(FactorContractError) as error:
|
|
FactorSetRef.create(**invalid)
|
|
assert error.value.code in {
|
|
ContractErrorCode.INPUT_CLOSURE_VIOLATION,
|
|
ContractErrorCode.INVALID_VALUE,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutate", "code", "path"),
|
|
[
|
|
(lambda value: value["producer"].pop("id"), ContractErrorCode.MISSING_FIELD, "$.producer.id"),
|
|
(lambda value: value["producer"].pop("version"), ContractErrorCode.MISSING_FIELD, "$.producer.version"),
|
|
(lambda value: value["producer"].update(id="other_engine"), ContractErrorCode.LINEAGE_VIOLATION, "$.producer.id"),
|
|
(lambda value: value["producer"].update(version="latest"), ContractErrorCode.INVALID_FORMAT, "$.producer.version"),
|
|
(lambda value: value.update(code_revision="bad"), ContractErrorCode.INVALID_FORMAT, "$.code_revision"),
|
|
(lambda value: value["actor"].pop("kind"), ContractErrorCode.MISSING_FIELD, "$.actor.kind"),
|
|
(lambda value: value["actor"].pop("id"), ContractErrorCode.MISSING_FIELD, "$.actor.id"),
|
|
(lambda value: value["actor"].update(kind="robot"), ContractErrorCode.INVALID_VALUE, "$.actor.kind"),
|
|
(lambda value: value["actor"].update(id="latest"), ContractErrorCode.INVALID_VALUE, "$.actor.id"),
|
|
(lambda value: value.pop("correlation_id"), ContractErrorCode.MISSING_FIELD, "$.correlation_id"),
|
|
(lambda value: value.update(correlation_id="latest"), ContractErrorCode.INVALID_VALUE, "$.correlation_id"),
|
|
(lambda value: value["causation"].pop("kind"), ContractErrorCode.MISSING_FIELD, "$.causation.kind"),
|
|
(lambda value: value["causation"].update(kind="run"), ContractErrorCode.INVALID_VALUE, "$.causation.kind"),
|
|
(lambda value: value["causation"].update(id="rhdfv1:sha256:" + "0" * 64), ContractErrorCode.LINEAGE_VIOLATION, "$.causation.id"),
|
|
(lambda value: value.pop("output_artifact_ref"), ContractErrorCode.MISSING_FIELD, "$.output_artifact_ref"),
|
|
(lambda value: value["output_artifact_ref"].update(artifact_id="rhfactoroutputv1:sha256:" + "0" * 64), ContractErrorCode.IDENTITY_MISMATCH, "$.output_artifact_ref.artifact_id"),
|
|
(_mutate_artifact_schema_binding, ContractErrorCode.ARTIFACT_MISMATCH, "$.output_artifact_ref"),
|
|
(lambda value: value.update(availability_mode="implicit_fallback"), ContractErrorCode.INVALID_VALUE, "$.availability_mode"),
|
|
(lambda value: value.pop("computed_at"), ContractErrorCode.MISSING_FIELD, "$.computed_at"),
|
|
(lambda value: value.update(decision_eligible=True), ContractErrorCode.READINESS_ESCALATION, "$.decision_eligible"),
|
|
(lambda value: value.update(evidence_scope="real_data"), ContractErrorCode.READINESS_ESCALATION, "$.evidence_scope"),
|
|
(lambda value: value["upstream_evidence"].update(qualification_evidence_digest="sha256:" + "0" * 64), ContractErrorCode.IDENTITY_MISMATCH, "$.upstream_evidence"),
|
|
],
|
|
)
|
|
def test_lineage_artifact_and_readiness_fields_have_independent_typed_negatives(
|
|
mutate: Callable[[dict[str, Any]], Any],
|
|
code: ContractErrorCode,
|
|
path: str,
|
|
) -> None:
|
|
factor_set = _factor_set()
|
|
value = factor_set.to_dict()
|
|
mutate(value)
|
|
if "factor_set_id" in value:
|
|
_reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
|
|
snapshot, foundation = _snapshot_and_foundation()
|
|
with pytest.raises(FactorContractError) as error:
|
|
FactorSetRef.from_dict(
|
|
value,
|
|
definitions=(_golden_definition(),),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
_assert_error(error, code, path)
|
|
|
|
|
|
def test_output_schema_content_bytes_cannot_be_swapped_or_forged() -> None:
|
|
factor_set = _factor_set()
|
|
fixture = _golden()
|
|
snapshot, foundation = _snapshot_and_foundation()
|
|
schema_bytes = canonical_json_bytes(fixture["output_schema"])
|
|
content_bytes = canonical_json_bytes(fixture["output_content"])
|
|
with pytest.raises(FactorContractError) as swapped:
|
|
FactorSetRef.from_dict(
|
|
factor_set.to_dict(),
|
|
definitions=(_golden_definition(),),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
output_schema_bytes=content_bytes,
|
|
output_content_bytes=schema_bytes,
|
|
)
|
|
_assert_error(swapped, ContractErrorCode.ARTIFACT_MISMATCH, "$.output_artifact_ref")
|
|
with pytest.raises(FactorContractError) as noncanonical:
|
|
FactorSetRef.create(
|
|
**{
|
|
**_factor_set_arguments(),
|
|
"output_schema_bytes": json.dumps(fixture["output_schema"], indent=2).encode(),
|
|
}
|
|
)
|
|
_assert_error(noncanonical, ContractErrorCode.INVALID_FORMAT, "$.output_schema_bytes")
|
|
|
|
|
|
def test_unsuccessful_output_quality_or_coverage_cannot_form_a_factor_set() -> None:
|
|
with pytest.raises(FactorContractError) as failed_quality:
|
|
_factor_set(
|
|
output_quality=OutputQuality(
|
|
"failed",
|
|
(OutputQualityCheck("finite_values", "failed", "sha256:" + "3" * 64),),
|
|
)
|
|
)
|
|
_assert_error(failed_quality, ContractErrorCode.INVALID_VALUE, "$.output_quality")
|
|
|
|
for coverage in (
|
|
OutputCoverage("incomplete", 2, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64),
|
|
OutputCoverage("complete", 2, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64),
|
|
):
|
|
with pytest.raises(FactorContractError) as incomplete:
|
|
_factor_set(output_coverage=coverage)
|
|
_assert_error(incomplete, ContractErrorCode.INVALID_VALUE, "$.output_coverage")
|
|
|
|
|
|
def test_external_snapshot_definition_and_view_references_cannot_be_substituted() -> None:
|
|
factor_set = _factor_set()
|
|
snapshot, foundation = _snapshot_and_foundation()
|
|
value = factor_set.to_dict()
|
|
value["dataset_snapshot_id"] = "rhdsv1:sha256:" + "0" * 64
|
|
_reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
|
|
with pytest.raises(FactorContractError) as snapshot_error:
|
|
FactorSetRef.from_dict(
|
|
value,
|
|
definitions=(_golden_definition(),),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
_assert_error(
|
|
snapshot_error,
|
|
ContractErrorCode.INPUT_CLOSURE_VIOLATION,
|
|
"$.dataset_snapshot_id",
|
|
)
|
|
|
|
value = factor_set.to_dict()
|
|
value["definition_ids"] = ["rhfactorv1:sha256:" + "0" * 64]
|
|
_reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
|
|
with pytest.raises(FactorContractError) as definition_error:
|
|
FactorSetRef.from_dict(
|
|
value,
|
|
definitions=(_golden_definition(),),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
_assert_error(
|
|
definition_error,
|
|
ContractErrorCode.INPUT_CLOSURE_VIOLATION,
|
|
"$.definition_ids",
|
|
)
|
|
|
|
arguments = _factor_set_arguments()
|
|
arguments["selected_view_ref_ids"] = ("rhviewrefv1:sha256:" + "0" * 64,)
|
|
with pytest.raises(FactorContractError) as view_error:
|
|
FactorSetRef.create(**arguments)
|
|
_assert_error(
|
|
view_error,
|
|
ContractErrorCode.INPUT_CLOSURE_VIOLATION,
|
|
"$.selected_view_ref_ids",
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"invalid_definition_id",
|
|
[
|
|
{"unexpected": "object"},
|
|
["array"],
|
|
42,
|
|
True,
|
|
None,
|
|
],
|
|
)
|
|
def test_factor_set_ref_definition_ids_reject_non_string_types(
|
|
invalid_definition_id: Any,
|
|
) -> None:
|
|
factor_set = _factor_set()
|
|
definition = _golden_definition()
|
|
value = factor_set.to_dict()
|
|
value["definition_ids"] = [definition.definition_id, invalid_definition_id]
|
|
_reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
|
|
snapshot, foundation = _snapshot_and_foundation()
|
|
|
|
with pytest.raises(FactorContractError) as error:
|
|
FactorSetRef.from_json(
|
|
canonical_json_bytes(value),
|
|
definitions=(definition,),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
|
|
_assert_error(error, ContractErrorCode.TYPE_ERROR, "$.definition_ids[1]")
|
|
|
|
|
|
def test_factor_set_ref_definition_ids_still_reject_duplicate_strings() -> None:
|
|
factor_set = _factor_set()
|
|
definition = _golden_definition()
|
|
value = factor_set.to_dict()
|
|
value["definition_ids"] = [definition.definition_id, definition.definition_id]
|
|
_reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
|
|
snapshot, foundation = _snapshot_and_foundation()
|
|
|
|
with pytest.raises(FactorContractError) as error:
|
|
FactorSetRef.from_json(
|
|
canonical_json_bytes(value),
|
|
definitions=(definition,),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
|
|
_assert_error(error, ContractErrorCode.INVALID_VALUE, "$.definition_ids")
|
|
|
|
|
|
def test_factor_set_parent_requires_exact_identity_and_correlation() -> None:
|
|
parent = _factor_set()
|
|
child_arguments = _factor_set_arguments()
|
|
child_arguments.update(
|
|
output_content_bytes=canonical_json_bytes({"rows": [{"value": "0.250"}]}),
|
|
causation=Causation("factor_set", parent.factor_set_id),
|
|
parent=parent,
|
|
)
|
|
child_arguments["output_artifact_ref"] = OutputArtifactRef.create(
|
|
schema_digest=_sha256(child_arguments["output_schema_bytes"]),
|
|
content_digest=_sha256(child_arguments["output_content_bytes"]),
|
|
)
|
|
child = FactorSetRef.create(**child_arguments)
|
|
assert child.causation.id == parent.factor_set_id
|
|
missing_parent = child.to_dict()
|
|
snapshot, foundation = _snapshot_and_foundation()
|
|
with pytest.raises(FactorContractError) as missing_error:
|
|
FactorSetRef.from_dict(
|
|
missing_parent,
|
|
definitions=(_golden_definition(),),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
)
|
|
_assert_error(missing_error, ContractErrorCode.LINEAGE_VIOLATION, "$.causation")
|
|
wrong_correlation = dict(child_arguments)
|
|
wrong_correlation["correlation_id"] = "different_run"
|
|
with pytest.raises(FactorContractError) as correlation_error:
|
|
FactorSetRef.create(**wrong_correlation)
|
|
_assert_error(correlation_error, ContractErrorCode.LINEAGE_VIOLATION, "$.correlation_id")
|
|
|
|
|
|
def test_legacy_bridge_is_explicit_lossy_and_preserves_all_four_historical_fields() -> None:
|
|
definition = _golden_definition()
|
|
legacy = FactorVersion(
|
|
factor_id="factor:demo-momentum",
|
|
version="1.0.0",
|
|
definition_sha256="b" * 64,
|
|
dataset_schema_version="1.0.0",
|
|
)
|
|
binding = LegacyFactorBinding.create(
|
|
definition=definition,
|
|
legacy_factor_id=legacy.factor_id,
|
|
legacy_version=legacy.version,
|
|
legacy_definition_sha256=legacy.definition_sha256,
|
|
legacy_dataset_schema_version=legacy.dataset_schema_version,
|
|
canonical_input_schema_digest=definition.input_schema_digest,
|
|
correspondence_evidence_digest="sha256:" + "5" * 64,
|
|
)
|
|
assert bind_legacy_factor(legacy, definition, binding) is definition
|
|
assert project_legacy_factor(definition, binding) == legacy
|
|
assert legacy.version_id == "factor:demo-momentum@1.0.0"
|
|
assert legacy.definition_sha256 != definition.definition_id.rsplit(":", maxsplit=1)[-1]
|
|
assert LegacyFactorBinding.from_json(binding.to_json(), definition=definition) == binding
|
|
|
|
mismatched = FactorVersion(
|
|
factor_id="factor:different",
|
|
version=legacy.version,
|
|
definition_sha256=legacy.definition_sha256,
|
|
dataset_schema_version=legacy.dataset_schema_version,
|
|
)
|
|
with pytest.raises(FactorContractError) as mismatch_error:
|
|
bind_legacy_factor(mismatched, definition, binding)
|
|
_assert_error(mismatch_error, ContractErrorCode.LEGACY_BINDING_MISMATCH, "$.binding")
|
|
|
|
|
|
def test_bare_legacy_factor_or_id_cannot_enter_factor_set_contract() -> None:
|
|
legacy = FactorVersion("factor:demo-momentum", "1.0.0", "b" * 64, "1.0.0")
|
|
arguments = _factor_set_arguments()
|
|
arguments["definitions"] = (legacy,)
|
|
with pytest.raises(FactorContractError) as legacy_error:
|
|
FactorSetRef.create(**arguments)
|
|
_assert_error(legacy_error, ContractErrorCode.TYPE_ERROR, "$.definitions[0]")
|
|
arguments["definitions"] = (legacy.version_id,)
|
|
with pytest.raises(FactorContractError) as id_error:
|
|
FactorSetRef.create(**arguments)
|
|
_assert_error(id_error, ContractErrorCode.TYPE_ERROR, "$.definitions[0]")
|