637 lines
26 KiB
Python
637 lines
26 KiB
Python
"""QE-owned decoding tests against RP's synthetic public v2 golden vectors."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from copy import deepcopy
|
|
from dataclasses import FrozenInstanceError
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from quant_engine.factor_contracts import (
|
|
DataFoundationEnvelope,
|
|
DatasetSnapshotEnvelope,
|
|
FactorContractError,
|
|
canonical_json_bytes,
|
|
)
|
|
from quant_engine.retrospective_data_contracts import (
|
|
RetrospectiveFoundationEnvelope,
|
|
RetrospectiveSnapshotEnvelope,
|
|
)
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def golden(kind: str) -> dict[str, Any]:
|
|
return json.loads((FIXTURES / f"retrospective-{kind}-v2.golden.json").read_text())
|
|
|
|
|
|
def digest(value: Any) -> str:
|
|
return "sha256:" + hashlib.sha256(canonical_json_bytes(value)).hexdigest()
|
|
|
|
|
|
def identify(value: dict[str, Any], field: str, prefix: str) -> None:
|
|
value[field] = prefix + digest({key: item for key, item in value.items() if key != field})
|
|
|
|
|
|
def records() -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"effective_time": "2018-01-02T07:00:00Z",
|
|
"instrument_id": "rhinstrument:" + "1" * 32,
|
|
"metric": "close",
|
|
"value": "101.25",
|
|
},
|
|
{
|
|
"effective_time": "2018-01-02T07:00:00Z",
|
|
"instrument_id": "rhinstrument:" + "2" * 32,
|
|
"metric": "close",
|
|
"value": "87.50",
|
|
},
|
|
]
|
|
|
|
|
|
def bind_records(source: dict[str, Any], chunks: list[list[dict[str, Any]]]) -> None:
|
|
def records_digest(rows: list[dict[str, Any]]) -> str:
|
|
data = b"[" + b",".join(sorted(canonical_json_bytes(row) for row in rows)) + b"]"
|
|
return "sha256:" + hashlib.sha256(data).hexdigest()
|
|
|
|
manifest = {
|
|
"record_count": sum(len(rows) for rows in chunks),
|
|
"chunks": [
|
|
{
|
|
"chunk_index": index,
|
|
"content_digest": records_digest(rows),
|
|
"record_count": len(rows),
|
|
}
|
|
for index, rows in enumerate(chunks)
|
|
],
|
|
}
|
|
source["descriptor"]["content"].update(
|
|
{
|
|
"record_count": manifest["record_count"],
|
|
"logical_manifest": manifest,
|
|
"manifest_digest": digest(manifest),
|
|
"content_digest": records_digest([row for chunk in chunks for row in chunk]),
|
|
}
|
|
)
|
|
source["descriptor"]["observation_manifest"]["batches"] = [
|
|
{
|
|
**chunk,
|
|
"observation_kind": "observed_by",
|
|
"observed_by": "2026-09-08T01:00:00Z",
|
|
"evidence_digest": digest({"synthetic_receipt": index}),
|
|
}
|
|
for index, chunk in enumerate(manifest["chunks"])
|
|
]
|
|
identify(source, "snapshot_id", "rhdsv2:")
|
|
|
|
|
|
def replace_at(source: dict[str, Any], path: str, value: Any) -> None:
|
|
target: Any = source
|
|
keys = path.split(".")
|
|
for key in keys[:-1]:
|
|
target = target[int(key)] if isinstance(target, list) else target[key]
|
|
target[int(keys[-1]) if isinstance(target, list) else keys[-1]] = value
|
|
|
|
|
|
COLLECTIONS = (
|
|
(
|
|
"instrument_routes",
|
|
"route_revision_id",
|
|
"rhroutev2:",
|
|
"instrument_route",
|
|
"instrument_route_revision_ids",
|
|
),
|
|
(
|
|
"trading_calendar_revisions",
|
|
"calendar_revision_id",
|
|
"rhcalv2:",
|
|
"trading_calendar",
|
|
"trading_calendar_revision_ids",
|
|
),
|
|
(
|
|
"corporate_action_revisions",
|
|
"action_revision_id",
|
|
"rhcav2:",
|
|
"corporate_action",
|
|
"corporate_action_revision_ids",
|
|
),
|
|
)
|
|
|
|
|
|
def seal_foundation(source: dict[str, Any], *, rebuild_lineage: bool = True) -> None:
|
|
lineage = []
|
|
for name, key, prefix, kind, view_key in COLLECTIONS:
|
|
replacements = {}
|
|
for row in sorted(source[name], key=lambda row: row["observation_sequence"]):
|
|
old = row[key]
|
|
if "supersedes_observation_id" in row:
|
|
row["supersedes_observation_id"] = replacements.get(
|
|
row["supersedes_observation_id"], row["supersedes_observation_id"]
|
|
)
|
|
identify(row, key, prefix)
|
|
replacements[old] = row[key]
|
|
lineage.append(
|
|
{
|
|
"revision_kind": kind,
|
|
"revision_id": row[key],
|
|
**{
|
|
field: row[field]
|
|
for field in (
|
|
"observation_sequence",
|
|
"observed_by",
|
|
"earliest_external_knowledge",
|
|
"history_completeness",
|
|
"evidence_digest",
|
|
"supersedes_observation_id",
|
|
)
|
|
if field in row
|
|
},
|
|
}
|
|
)
|
|
for view in source["standardized_views"]:
|
|
view[view_key] = [replacements.get(item, item) for item in view[view_key]]
|
|
if rebuild_lineage:
|
|
source["observation_lineage"] = lineage
|
|
for view in source["standardized_views"]:
|
|
identify(view, "view_ref_id", "rhviewrefv2:")
|
|
identify(source, "foundation_id", "rhdfv2:")
|
|
|
|
|
|
def parse_foundation(source: dict[str, Any]) -> RetrospectiveFoundationEnvelope:
|
|
return RetrospectiveFoundationEnvelope.from_dict(
|
|
source, snapshot=RetrospectiveSnapshotEnvelope.from_dict(golden("dataset-snapshot"))
|
|
)
|
|
|
|
|
|
def test_public_snapshot_golden_is_an_explicit_observation_contract() -> None:
|
|
source = golden("dataset-snapshot")
|
|
snapshot = RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
assert snapshot.to_dict() == source
|
|
assert snapshot.snapshot_id == source["snapshot_id"]
|
|
assert snapshot.observation_cutoff == "2026-09-08T01:01:00Z"
|
|
assert snapshot.evidence_scope == "synthetic_fixture"
|
|
assert snapshot.earliest_external_knowledge == {"status": "unknown"}
|
|
assert not hasattr(snapshot, "pit_cutoff")
|
|
assert not hasattr(snapshot, "knowledge_time")
|
|
snapshot.require_qualified()
|
|
assert RetrospectiveSnapshotEnvelope.from_json(snapshot.to_json()) == snapshot
|
|
|
|
|
|
def test_public_foundation_golden_binds_exact_typed_snapshot() -> None:
|
|
source = golden("data-foundation")
|
|
snapshot = RetrospectiveSnapshotEnvelope.from_dict(golden("dataset-snapshot"))
|
|
foundation = RetrospectiveFoundationEnvelope.from_dict(source, snapshot=snapshot)
|
|
assert foundation.to_dict() == source
|
|
assert foundation.dataset_snapshot_id == snapshot.snapshot_id
|
|
assert foundation.observation_cutoff == snapshot.observation_cutoff
|
|
assert foundation.evidence_scope == snapshot.evidence_scope
|
|
assert foundation.real_data_validation_status == "not_validated"
|
|
assert not hasattr(foundation, "pit_cutoff")
|
|
assert (
|
|
RetrospectiveFoundationEnvelope.from_json(foundation.to_json(), snapshot=snapshot)
|
|
== foundation
|
|
)
|
|
|
|
|
|
def test_empty_logical_dimension_is_rejected_after_rebinding_all_bytes() -> None:
|
|
source = golden("dataset-snapshot")
|
|
rows = records()
|
|
rows[0]["instrument_id"] = ""
|
|
bind_records(source, [rows])
|
|
snapshot = RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
with pytest.raises(FactorContractError, match="dimension"):
|
|
snapshot.verify_materialized_records([rows])
|
|
|
|
|
|
def test_boolean_lineage_sequence_does_not_equal_integer_one() -> None:
|
|
source = golden("data-foundation")
|
|
source["observation_lineage"][0]["observation_sequence"] = True
|
|
identify(source, "foundation_id", "rhdfv2:")
|
|
with pytest.raises(FactorContractError):
|
|
parse_foundation(source)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "value"),
|
|
[
|
|
("schema_version", "1.0.0"),
|
|
("schema_version", "2.1.0"),
|
|
("descriptor.time_semantics.pit_cutoff", "2018-01-02T07:00:00Z"),
|
|
("descriptor.time_semantics.knowledge_time", {"start_inclusive": "2018-01-02T07:00:00Z"}),
|
|
("descriptor.time_semantics.historical_availability", "established"),
|
|
("descriptor.time_semantics.observation_cutoff", "2018-01-02T07:00:00Z"),
|
|
("descriptor.time_semantics.observation_cutoff", "2026-09-08T01:01:00.0000001Z"),
|
|
("descriptor.time_semantics.observation_cutoff", "2026-09-08T09:01:00+08:00"),
|
|
(
|
|
"descriptor.time_semantics.earliest_external_knowledge",
|
|
{"status": "unknown", "evidence_digest": "sha256:" + "0" * 64},
|
|
),
|
|
("descriptor.time_semantics.earliest_external_knowledge", {"status": "evidenced"}),
|
|
("descriptor.published_at", "2026-02-30T00:00:00Z"),
|
|
("descriptor.published_at", "2026-09-08T01:00:00Z"),
|
|
("descriptor.qualification.evaluated_at", "2018-01-02T07:00:00Z"),
|
|
("descriptor.qualification.usage", "as_available"),
|
|
("descriptor.qualification.policy_version", "1.0.0"),
|
|
("descriptor.quality.checks.0.severity", "advisory"),
|
|
("descriptor.quality.checks.0.status", "failed"),
|
|
("descriptor.quality.checks.0.check_id", "schema_conformance"),
|
|
("descriptor.quality.status", "failed"),
|
|
("descriptor.content.record_count", True),
|
|
("descriptor.content.record_count", 9007199254740992),
|
|
("descriptor.content.record_count", 2.0),
|
|
("descriptor.content.content_digest", "bad"),
|
|
("descriptor.content.manifest_digest", "sha256:" + "0" * 64),
|
|
("descriptor.observation_manifest.batches", []),
|
|
("descriptor.observation_manifest.batches.0.record_count", 1),
|
|
("descriptor.observation_manifest.batches.0.chunk_index", True),
|
|
("descriptor.observation_manifest.batches.0.observed_by", "2026-09-08T01:02:00Z"),
|
|
("descriptor.observation_manifest.batches.0.observation_kind", "first_published_at"),
|
|
("descriptor.lineage.transformation.id", "rhtransform:private"),
|
|
("descriptor.dataset.dimensions", ["instrument_id", "knowledge_time"]),
|
|
("descriptor.dataset.dataset_id", "rhdataset:macroeconomic:" + "4" * 32),
|
|
],
|
|
)
|
|
def test_snapshot_rejects_reidentified_invalid_declarations(path: str, value: Any) -> None:
|
|
source = golden("dataset-snapshot")
|
|
replace_at(source, path, value)
|
|
# Noncanonical numbers are rejected before identity formation.
|
|
if type(value) is not float and value != 9007199254740992:
|
|
identify(source, "snapshot_id", "rhdsv2:")
|
|
with pytest.raises(FactorContractError):
|
|
RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
|
|
|
|
def test_snapshot_materialized_records_bind_the_public_golden_and_chunks() -> None:
|
|
source = golden("dataset-snapshot")
|
|
snapshot = RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
snapshot.verify_materialized_records([records()])
|
|
snapshot.verify_materialized_records([list(reversed(records()))])
|
|
chunks = [[records()[0]], [records()[1]]]
|
|
bind_records(source, chunks)
|
|
RetrospectiveSnapshotEnvelope.from_dict(source).verify_materialized_records(chunks)
|
|
with pytest.raises(FactorContractError):
|
|
snapshot.verify_materialized_records(chunks)
|
|
rows = records()
|
|
rows[0]["value"] = "0"
|
|
with pytest.raises(FactorContractError):
|
|
snapshot.verify_materialized_records([rows])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutation", ["duplicate", "legacy", "range", "location", "missing_effective"]
|
|
)
|
|
def test_materialized_bad_records_rejected_even_with_matching_digests(mutation: str) -> None:
|
|
source = golden("dataset-snapshot")
|
|
rows = records()
|
|
if mutation == "duplicate":
|
|
rows.append(deepcopy(rows[0]))
|
|
elif mutation == "legacy":
|
|
rows[0]["knowledge_time"] = "2026-09-08T01:00:00Z"
|
|
elif mutation == "range":
|
|
rows[0]["effective_time"] = "2018-01-01T07:00:00Z"
|
|
elif mutation == "location":
|
|
rows[0]["value"] = "/private/records.csv"
|
|
else:
|
|
del rows[0]["effective_time"]
|
|
bind_records(source, [rows])
|
|
with pytest.raises(FactorContractError):
|
|
RetrospectiveSnapshotEnvelope.from_dict(source).verify_materialized_records([rows])
|
|
|
|
|
|
def test_unknown_or_evidenced_knowledge_never_changes_usage() -> None:
|
|
source = golden("dataset-snapshot")
|
|
source["descriptor"]["time_semantics"]["earliest_external_knowledge"] = {
|
|
"status": "evidenced",
|
|
"range": {
|
|
"start_inclusive": "2018-01-02T07:00:00Z",
|
|
"end_inclusive": "2018-01-02T07:00:00Z",
|
|
},
|
|
"evidence_digest": digest({"synthetic_earliest": True}),
|
|
}
|
|
identify(source, "snapshot_id", "rhdsv2:")
|
|
parsed = RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
assert (
|
|
parsed.to_dict()["descriptor"]["time_semantics"]["historical_availability"]
|
|
== "not_established"
|
|
)
|
|
source["descriptor"]["time_semantics"]["earliest_external_knowledge"]["range"][
|
|
"end_inclusive"
|
|
] = "2026-09-08T01:01:00Z"
|
|
identify(source, "snapshot_id", "rhdsv2:")
|
|
with pytest.raises(FactorContractError):
|
|
RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
|
|
|
|
def test_rejected_snapshot_remains_readable_but_cannot_support_foundation() -> None:
|
|
source = golden("dataset-snapshot")
|
|
source["descriptor"]["qualification"]["status"] = "rejected"
|
|
identify(source, "snapshot_id", "rhdsv2:")
|
|
snapshot = RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
with pytest.raises(FactorContractError):
|
|
snapshot.require_qualified()
|
|
foundation = golden("data-foundation")
|
|
foundation["dataset_snapshot_id"] = snapshot.snapshot_id
|
|
for view in foundation["standardized_views"]:
|
|
view["dataset_snapshot_id"] = snapshot.snapshot_id
|
|
seal_foundation(foundation)
|
|
with pytest.raises(FactorContractError):
|
|
RetrospectiveFoundationEnvelope.from_dict(foundation, snapshot=snapshot)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "value"),
|
|
[
|
|
("schema_version", "1.0.0"),
|
|
("dataset_snapshot_id", "rhdsv2:sha256:" + "0" * 64),
|
|
("observation_cutoff", "2026-09-08T01:00:00Z"),
|
|
("published_at", "2026-09-08T01:03:00Z"),
|
|
("usage", "paper_trading"),
|
|
("historical_availability", "established"),
|
|
("instrument_routes.0.observation_sequence", 2),
|
|
("instrument_routes.0.supersedes_observation_id", "rhroutev2:sha256:" + "0" * 64),
|
|
("instrument_routes.0.observed_by", "2026-09-08T01:02:00Z"),
|
|
("instrument_routes.0.history_completeness", "complete"),
|
|
(
|
|
"instrument_routes.0.earliest_external_knowledge",
|
|
{"status": "unknown", "earliest_at": "2018-01-01T00:00:00Z"},
|
|
),
|
|
("instrument_routes.0.instrument_type", "index"),
|
|
("instrument_routes.0.symbol", "WIND.TEST"),
|
|
("instrument_routes.0.symbol", "A" * 33),
|
|
("instrument_routes.0.calendar_id", "rhcalendar:" + "0" * 32),
|
|
("trading_calendar_revisions.0.status", "closed"),
|
|
("trading_calendar_revisions.0.sessions", []),
|
|
("trading_calendar_revisions.0.sessions.0.closes_at", "2018-01-02T00:00:00Z"),
|
|
("trading_calendar_revisions.0.session_date", "2018-02-30"),
|
|
("standardized_views.0.instrument_route_revision_ids", []),
|
|
("standardized_views.0.trading_calendar_revision_ids", []),
|
|
("standardized_views.0.corporate_action_revision_ids", ["rhcav2:sha256:" + "0" * 64]),
|
|
("standardized_views.0.observation_cutoff", "2018-01-02T07:00:00Z"),
|
|
("standardized_views.0.available_at", "2026-09-08T01:02:00Z"),
|
|
("standardized_views.0.available_at", "2026-09-08T01:06:00Z"),
|
|
("standardized_views.0.usage", "as_available"),
|
|
("corporate_action_coverage", []),
|
|
("corporate_action_coverage.0.instrument_id", "rhinstrument:" + "0" * 32),
|
|
("corporate_action_coverage.0.effective_time.end_inclusive", "2018-01-01T07:00:00Z"),
|
|
("corporate_action_coverage.0.effective_time.start_inclusive", "2018-01-02T07:00:01Z"),
|
|
("corporate_action_coverage.0.observed_by", "2026-09-08T01:02:00Z"),
|
|
("corporate_action_coverage.0.evidence_digests", []),
|
|
("readiness.evidence_scope", "real_data"),
|
|
("readiness.contract_validation.evidence_digests", []),
|
|
(
|
|
"readiness.real_data_validation",
|
|
{"status": "validated", "evidence_digests": ["sha256:" + "0" * 64]},
|
|
),
|
|
(
|
|
"readiness.production_validation",
|
|
{"status": "validated", "evidence_digests": ["sha256:" + "0" * 64]},
|
|
),
|
|
(
|
|
"readiness.live_validation",
|
|
{"status": "validated", "evidence_digests": ["sha256:" + "0" * 64]},
|
|
),
|
|
],
|
|
)
|
|
def test_foundation_rejects_semantic_forgery_after_reidentification(path: str, value: Any) -> None:
|
|
source = golden("data-foundation")
|
|
replace_at(source, path, value)
|
|
seal_foundation(source)
|
|
with pytest.raises(FactorContractError):
|
|
parse_foundation(source)
|
|
|
|
|
|
def test_v1_and_v2_never_coerce_each_other() -> None:
|
|
with pytest.raises(FactorContractError):
|
|
DatasetSnapshotEnvelope.from_dict(golden("dataset-snapshot"))
|
|
with pytest.raises(FactorContractError):
|
|
DataFoundationEnvelope.from_dict(golden("data-foundation"))
|
|
old = json.loads((FIXTURES / "factor-contracts-v1.golden.json").read_text())
|
|
with pytest.raises(FactorContractError):
|
|
RetrospectiveSnapshotEnvelope.from_dict(old["dataset_snapshot"])
|
|
with pytest.raises(FactorContractError):
|
|
parse_foundation(old["data_foundation"])
|
|
with pytest.raises(FactorContractError):
|
|
RetrospectiveFoundationEnvelope.from_dict(
|
|
golden("data-foundation"),
|
|
snapshot=DatasetSnapshotEnvelope.from_dict(old["dataset_snapshot"]),
|
|
)
|
|
|
|
|
|
def test_deep_immutability_and_strict_canonical_json() -> None:
|
|
source = golden("dataset-snapshot")
|
|
snapshot = RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
source["descriptor"]["quality"]["status"] = "failed"
|
|
snapshot.to_dict()["descriptor"]["qualification"]["status"] = "rejected"
|
|
snapshot.require_qualified()
|
|
with pytest.raises(FrozenInstanceError):
|
|
snapshot._payload = {}
|
|
with pytest.raises(TypeError):
|
|
snapshot.earliest_external_knowledge["status"] = "evidenced"
|
|
foundation = parse_foundation(golden("data-foundation"))
|
|
with pytest.raises(TypeError):
|
|
foundation.views["new"] = next(iter(foundation.views.values()))
|
|
for decoder, document in (
|
|
(RetrospectiveSnapshotEnvelope.from_json, golden("dataset-snapshot")),
|
|
(
|
|
lambda value: RetrospectiveFoundationEnvelope.from_json(value, snapshot=snapshot),
|
|
golden("data-foundation"),
|
|
),
|
|
):
|
|
wire = canonical_json_bytes(document)
|
|
with pytest.raises(FactorContractError):
|
|
decoder(wire + b"\n")
|
|
with pytest.raises(FactorContractError):
|
|
decoder(b'{"schema_version":"2.0.0",' + wire[1:])
|
|
|
|
|
|
def test_macro_period_is_not_inferred_as_an_effective_instant() -> None:
|
|
source = golden("dataset-snapshot")
|
|
source["descriptor"]["dataset"].update(
|
|
dataset_id="rhdataset:macroeconomic:" + "4" * 32,
|
|
dataset_kind="macroeconomic",
|
|
dimensions=["series_id", "observation_period"],
|
|
)
|
|
rows = [
|
|
{
|
|
"series_id": "cpi",
|
|
"observation_period": "2018-01",
|
|
"effective_time": "2018-01-02T07:00:00Z",
|
|
"value": "2.1",
|
|
}
|
|
]
|
|
bind_records(source, [rows])
|
|
RetrospectiveSnapshotEnvelope.from_dict(source).verify_materialized_records([rows])
|
|
del rows[0]["effective_time"]
|
|
bind_records(source, [rows])
|
|
with pytest.raises(FactorContractError):
|
|
RetrospectiveSnapshotEnvelope.from_dict(source).verify_materialized_records([rows])
|
|
|
|
|
|
def with_successor() -> dict[str, Any]:
|
|
source = golden("data-foundation")
|
|
previous = source["instrument_routes"][0]
|
|
successor = deepcopy(previous)
|
|
successor.update(
|
|
observation_sequence=2,
|
|
observed_by="2026-09-08T01:00:30Z",
|
|
symbol="SIM0B",
|
|
supersedes_observation_id=previous["route_revision_id"],
|
|
)
|
|
identify(successor, "route_revision_id", "rhroutev2:")
|
|
source["instrument_routes"].append(successor)
|
|
source["standardized_views"][0]["instrument_route_revision_ids"].append(
|
|
successor["route_revision_id"]
|
|
)
|
|
seal_foundation(source)
|
|
return source
|
|
|
|
|
|
def test_retained_successor_requires_parent_time_and_view_ancestry() -> None:
|
|
source = with_successor()
|
|
parsed = parse_foundation(source)
|
|
assert parsed.foundation_id == source["foundation_id"]
|
|
assert parsed.published_at.isoformat() == "2026-09-08T01:05:00+00:00"
|
|
assert parsed.contract_evidence_digests
|
|
assert len(next(iter(parsed.views.values())).instrument_route_revision_ids) == 3
|
|
for mutation in (
|
|
"missing_parent",
|
|
"equal_time",
|
|
"omitted_ancestor",
|
|
"duplicate_sequence",
|
|
"wrong_lineage",
|
|
):
|
|
forged = deepcopy(source)
|
|
if mutation == "missing_parent":
|
|
forged["instrument_routes"][-1]["supersedes_observation_id"] = (
|
|
"rhroutev2:sha256:" + "0" * 64
|
|
)
|
|
elif mutation == "equal_time":
|
|
forged["instrument_routes"][-1]["observed_by"] = forged["instrument_routes"][0][
|
|
"observed_by"
|
|
]
|
|
elif mutation == "omitted_ancestor":
|
|
forged["standardized_views"][0]["instrument_route_revision_ids"].remove(
|
|
forged["instrument_routes"][0]["route_revision_id"]
|
|
)
|
|
elif mutation == "duplicate_sequence":
|
|
forged["instrument_routes"][-1]["observation_sequence"] = 1
|
|
else:
|
|
forged["observation_lineage"][0]["evidence_digest"] = "sha256:" + "0" * 64
|
|
seal_foundation(forged, rebuild_lineage=mutation != "wrong_lineage")
|
|
with pytest.raises(FactorContractError):
|
|
parse_foundation(forged)
|
|
|
|
|
|
def test_index_and_closed_calendar_are_explicit_non_execution_data() -> None:
|
|
source = golden("data-foundation")
|
|
source["instrument_routes"][0].update(asset_class="index", instrument_type="index")
|
|
source["trading_calendar_revisions"][0].update(status="closed", sessions=[])
|
|
seal_foundation(source)
|
|
parsed = parse_foundation(source)
|
|
assert parsed.to_dict()["instrument_routes"][0]["instrument_type"] == "index"
|
|
assert parsed.to_dict()["readiness"]["live_validation"]["status"] == "not_validated"
|
|
|
|
|
|
def test_real_scope_is_still_a_declaration_with_separate_evidence_and_coverage() -> None:
|
|
snapshot_source = golden("dataset-snapshot")
|
|
snapshot_source["evidence_scope"] = "real_data"
|
|
identify(snapshot_source, "snapshot_id", "rhdsv2:")
|
|
snapshot = RetrospectiveSnapshotEnvelope.from_dict(snapshot_source)
|
|
source = golden("data-foundation")
|
|
source["dataset_snapshot_id"] = snapshot.snapshot_id
|
|
for view in source["standardized_views"]:
|
|
view["dataset_snapshot_id"] = snapshot.snapshot_id
|
|
source["readiness"]["evidence_scope"] = "real_data"
|
|
source["readiness"]["real_data_validation"] = {
|
|
"status": "validated",
|
|
"evidence_digests": [digest({"synthetic_real_claim_test": 1})],
|
|
}
|
|
seal_foundation(source)
|
|
parsed = RetrospectiveFoundationEnvelope.from_dict(source, snapshot=snapshot)
|
|
assert parsed.real_data_validation_status == "validated" # NOT actual real-data evidence.
|
|
for mutation in ("coverage", "reuse"):
|
|
forged = deepcopy(source)
|
|
if mutation == "coverage":
|
|
forged["corporate_action_coverage"][0].update(
|
|
status="not_validated", evidence_digests=[]
|
|
)
|
|
else:
|
|
forged["readiness"]["real_data_validation"] = deepcopy(
|
|
forged["readiness"]["contract_validation"]
|
|
)
|
|
seal_foundation(forged)
|
|
with pytest.raises(FactorContractError):
|
|
RetrospectiveFoundationEnvelope.from_dict(forged, snapshot=snapshot)
|
|
synthetic = golden("data-foundation")
|
|
synthetic["corporate_action_coverage"][0].update(status="not_validated", evidence_digests=[])
|
|
seal_foundation(synthetic)
|
|
assert parse_foundation(synthetic).real_data_validation_status == "not_validated"
|
|
|
|
|
|
def test_action_must_belong_to_view_selected_instrument() -> None:
|
|
source = golden("data-foundation")
|
|
route = source["instrument_routes"][0]
|
|
action = {
|
|
"action_id": "rhaction:" + "7" * 32,
|
|
"instrument_id": route["instrument_id"],
|
|
"observation_sequence": 1,
|
|
"observed_by": route["observed_by"],
|
|
"earliest_external_knowledge": {
|
|
"status": "evidenced",
|
|
"earliest_at": "2018-01-01T00:00:00Z",
|
|
"evidence_digest": digest({"synthetic_action_earliest": 1}),
|
|
},
|
|
"history_completeness": "not_established",
|
|
"evidence_digest": digest({"synthetic_action": 1}),
|
|
"action_type": "cash_dividend",
|
|
"status": "confirmed",
|
|
"effective_time": "2018-01-02T07:00:00Z",
|
|
"terms_digest": digest({"synthetic_terms": 1}),
|
|
}
|
|
identify(action, "action_revision_id", "rhcav2:")
|
|
source["corporate_action_revisions"] = [action]
|
|
source["standardized_views"][0]["corporate_action_revision_ids"] = [
|
|
action["action_revision_id"]
|
|
]
|
|
seal_foundation(source)
|
|
assert len(parse_foundation(source).to_dict()["corporate_action_revisions"]) == 1
|
|
source["standardized_views"][0]["instrument_route_revision_ids"].remove(
|
|
route["route_revision_id"]
|
|
)
|
|
seal_foundation(source)
|
|
with pytest.raises(FactorContractError):
|
|
parse_foundation(source)
|
|
|
|
|
|
def test_views_cannot_borrow_calendar_selection_from_each_other() -> None:
|
|
source = golden("data-foundation")
|
|
calendar = deepcopy(source["trading_calendar_revisions"][0])
|
|
calendar["calendar_id"] = "rhcalendar:" + "9" * 32
|
|
identify(calendar, "calendar_revision_id", "rhcalv2:")
|
|
source["trading_calendar_revisions"].append(calendar)
|
|
view = deepcopy(source["standardized_views"][0])
|
|
view["view_id"] = "rhview:" + "8" * 32
|
|
view["trading_calendar_revision_ids"] = [calendar["calendar_revision_id"]]
|
|
source["standardized_views"].append(view)
|
|
seal_foundation(source)
|
|
with pytest.raises(FactorContractError):
|
|
parse_foundation(source)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"location", ["/private/a", "./a", "../a", "C:\\data\\a", "\\\\host\\a", "s3://private/a"]
|
|
)
|
|
def test_materialized_values_cannot_carry_physical_locations(location: str) -> None:
|
|
source = golden("dataset-snapshot")
|
|
rows = records()
|
|
rows[0]["value"] = location
|
|
bind_records(source, [rows])
|
|
snapshot = RetrospectiveSnapshotEnvelope.from_dict(source)
|
|
with pytest.raises(FactorContractError):
|
|
snapshot.verify_materialized_records([rows])
|