This commit is contained in:
ao gong
@@ -0,0 +1,837 @@
|
||||
"""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]
|
||||
|
||||
|
||||
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",
|
||||
)
|
||||
|
||||
|
||||
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]")
|
||||
Reference in New Issue
Block a user