89 lines
3.5 KiB
Python
89 lines
3.5 KiB
Python
"""Frozen synthetic interoperability vector, not end-to-end source provenance."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from quant_engine.artifact import _evidence_frame_records
|
|
from quant_engine.retrospective_artifact_contracts import build_retrospective_performance_evidence
|
|
from quant_engine.retrospective_portfolio_risk_contracts import (
|
|
assess_retrospective_portfolio_risk,
|
|
)
|
|
from test_retrospective_factor_contracts import factor_arguments
|
|
from test_retrospective_portfolio_risk_contracts import portfolio_arguments, risk_arguments
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
VECTOR = ROOT / "tests" / "fixtures" / "retrospective-computation-v2.golden.json"
|
|
|
|
|
|
def build_vector() -> dict[str, Any]:
|
|
portfolio = portfolio_arguments()
|
|
risk = risk_arguments(portfolio)
|
|
run = portfolio["backtest_run_ref"]
|
|
manifest = portfolio["manifest"]
|
|
artifact = manifest._artifact
|
|
factor = factor_arguments()
|
|
return {
|
|
"fixture_kind": "synthetic_retrospective_contract_vector",
|
|
"artifact_data_provenance": "envelope_test_only_not_end_to_end",
|
|
"source_authenticity": "not_established",
|
|
"factor_definitions": [item.to_dict() for item in factor["definitions"]],
|
|
"dataset_chunks": factor["dataset_chunks"],
|
|
"resolved_view_schema": {"synthetic_schema": "neutral_close_v2"},
|
|
"factor_output_schema": json.loads(factor["output_schema_bytes"]),
|
|
"factor_output_records": json.loads(factor["output_content_bytes"]),
|
|
"factor_set": run._factor_set.to_dict(),
|
|
"backtest_run_ref": run.to_dict(),
|
|
"artifact_tables": {
|
|
name: _evidence_frame_records(frame, name)
|
|
for name, frame in artifact.table_frames().items()
|
|
},
|
|
"backtest_evidence_manifest": manifest.to_dict(),
|
|
"performance_evidence": build_retrospective_performance_evidence(
|
|
artifact, run, manifest
|
|
).to_dict(),
|
|
"portfolio_target": portfolio["target"].to_dict(),
|
|
"portfolio_decision": risk["portfolio_decision"].to_dict(),
|
|
"risk_assessment": assess_retrospective_portfolio_risk(**risk).to_dict(),
|
|
"covariance_matrix": risk["covariance"].covariance.to_dict(),
|
|
}
|
|
|
|
|
|
def test_synthetic_vector_matches_all_current_owner_serializers() -> None:
|
|
expected = VECTOR.read_text(encoding="utf-8")
|
|
actual = (
|
|
json.dumps(build_vector(), ensure_ascii=False, sort_keys=True, indent=2, allow_nan=False)
|
|
+ "\n"
|
|
)
|
|
assert actual == expected
|
|
|
|
|
|
def test_v2_modules_do_not_import_data_owners_publishers_or_execution_authority() -> None:
|
|
modules = sorted((ROOT / "src" / "quant_engine").glob("retrospective_*_contracts.py"))
|
|
assert len(modules) == 5
|
|
for path in modules:
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
imports = {
|
|
alias.name
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Import)
|
|
for alias in node.names
|
|
} | {node.module or "" for node in ast.walk(tree) if isinstance(node, ast.ImportFrom)}
|
|
assert not any(
|
|
name.startswith(("research_results", "research_platform", "edb_data_core"))
|
|
for name in imports
|
|
)
|
|
called = {
|
|
node.func.id
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
|
|
}
|
|
assert not called & {
|
|
"create_paper_order_intent",
|
|
"run_governed_factor_slice",
|
|
"evaluate_portfolio_risk",
|
|
}
|