1035 lines
36 KiB
Python
1035 lines
36 KiB
Python
"""S4.1 portfolio-decision and risk-assessment contract conformance."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
from datetime import UTC, date, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
import quant_engine.portfolio_risk_contracts as contracts_module
|
|
from quant_engine.artifact import (
|
|
BacktestEvidenceManifest,
|
|
EvidenceQualification,
|
|
ResearchRunArtifact,
|
|
build_backtest_evidence_manifest,
|
|
build_research_run_artifact,
|
|
)
|
|
from quant_engine.execution import ExecutionConfig
|
|
from quant_engine.factor_contracts import (
|
|
ActorIdentity,
|
|
Causation,
|
|
DataFoundationEnvelope,
|
|
DatasetSnapshotEnvelope,
|
|
FactorInput,
|
|
FactorSetRef,
|
|
InputBinding,
|
|
OutputArtifactRef,
|
|
OutputCoverage,
|
|
OutputQuality,
|
|
OutputQualityCheck,
|
|
ProducerIdentity,
|
|
ViewAvailability,
|
|
canonical_json_bytes,
|
|
factor_definition_from_alpha158,
|
|
factor_input_schema_digest,
|
|
)
|
|
from quant_engine.governed_pipeline import (
|
|
BacktestRun,
|
|
BacktestRunRef,
|
|
PortfolioTarget,
|
|
RiskDecision,
|
|
RiskDecisionStatus,
|
|
RiskPolicy,
|
|
create_paper_order_intent,
|
|
evaluate_portfolio_risk,
|
|
)
|
|
from quant_engine.portfolio_risk_contracts import (
|
|
ComputationReceipt,
|
|
ConstraintSetV1,
|
|
FreshnessPolicy,
|
|
PortfolioDecision,
|
|
PortfolioRiskContractError,
|
|
PortfolioRiskContractErrorCode,
|
|
ReceiptStatus,
|
|
RiskAssessment,
|
|
RiskAssessmentStatus,
|
|
RiskFindingCode,
|
|
assess_portfolio_risk,
|
|
build_portfolio_decision,
|
|
compute_portfolio_receipt_digests,
|
|
)
|
|
from quant_engine.research_pipeline import FactorBacktestResult, run_factor_backtest_research
|
|
from quant_engine.risk import ComponentRiskResult, CovarianceSnapshot, labeled_component_risk
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
FACTOR_FIXTURE = ROOT / "tests" / "fixtures" / "factor-contracts-v1.golden.json"
|
|
GOLDEN_FIXTURE = ROOT / "tests" / "fixtures" / "portfolio-risk-computation-v1.golden.json"
|
|
VIEW_REF_ID = "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c"
|
|
VIEW_SCHEMA_DIGEST = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
|
|
CALENDAR_REVISION_ID = "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078"
|
|
ACTION_REVISION_ID = "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275"
|
|
PARAMETERS = {"lag_sessions": 1, "top_k": 1}
|
|
DECISION_FIELDS: dict[str, str] = {
|
|
"objective_name": "long_only_allocation",
|
|
"objective_version": "1.0.0",
|
|
"objective_digest": "sha256:" + "b" * 64,
|
|
"model_name": "deterministic_weights",
|
|
"model_version": "1.0.0",
|
|
"model_digest": "sha256:" + "c" * 64,
|
|
"expected_return_digest": "sha256:" + "d" * 64,
|
|
"covariance_digest": "sha256:" + "a" * 64,
|
|
"scenario_digest": "sha256:" + "e" * 64,
|
|
}
|
|
|
|
|
|
def _sha256(value: bytes) -> str:
|
|
return f"sha256:{hashlib.sha256(value).hexdigest()}"
|
|
|
|
|
|
def _accepted_authorities() -> tuple[
|
|
DatasetSnapshotEnvelope,
|
|
DataFoundationEnvelope,
|
|
FactorSetRef,
|
|
]:
|
|
fixture = json.loads(FACTOR_FIXTURE.read_text(encoding="utf-8"))
|
|
snapshot = DatasetSnapshotEnvelope.from_dict(fixture["dataset_snapshot"])
|
|
foundation = DataFoundationEnvelope.from_dict(fixture["data_foundation"])
|
|
factor_input = FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume"))
|
|
definition = 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,
|
|
)
|
|
output_schema_bytes = canonical_json_bytes(fixture["output_schema"])
|
|
output_content_bytes = canonical_json_bytes(fixture["output_content"])
|
|
artifact_ref = OutputArtifactRef.create(
|
|
schema_digest=_sha256(output_schema_bytes),
|
|
content_digest=_sha256(output_content_bytes),
|
|
)
|
|
factor_set = FactorSetRef.create(
|
|
definitions=(definition,),
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
selected_view_ref_ids=(VIEW_REF_ID,),
|
|
input_bindings=(
|
|
InputBinding(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_ref,
|
|
availability_mode="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,
|
|
)
|
|
return snapshot, foundation, factor_set
|
|
|
|
|
|
def _config_digest() -> str:
|
|
encoded = json.dumps(
|
|
PARAMETERS,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
allow_nan=False,
|
|
).encode("utf-8")
|
|
return _sha256(encoded)
|
|
|
|
|
|
def _run_ref(*, universe_character: str = "5") -> BacktestRunRef:
|
|
snapshot, foundation, factor_set = _accepted_authorities()
|
|
return BacktestRunRef.create(
|
|
dataset_snapshot=snapshot,
|
|
foundation=foundation,
|
|
factor_set=factor_set,
|
|
universe_digest="sha256:" + universe_character * 64,
|
|
trading_calendar_revision_ids=(CALENDAR_REVISION_ID,),
|
|
corporate_action_revision_ids=(ACTION_REVISION_ID,),
|
|
strategy_id="alpha-top1",
|
|
strategy_version="1.0.0",
|
|
strategy_digest="sha256:" + "6" * 64,
|
|
execution_model_version="1.0.0",
|
|
execution_model_digest="sha256:" + "7" * 64,
|
|
cost_model_version="1.0.0",
|
|
cost_model_digest="sha256:" + "8" * 64,
|
|
random_seed=7,
|
|
code_revision="d" * 40,
|
|
environment_lock_digest="sha256:" + "9" * 64,
|
|
configuration_digest=_config_digest(),
|
|
evaluation_at="2026-01-08T01:00:00Z",
|
|
computed_at="2026-01-08T02:00:00Z",
|
|
)
|
|
|
|
|
|
def _backtest_result() -> FactorBacktestResult:
|
|
dates = pd.date_range("2026-01-05", periods=4, freq="B")
|
|
scores = pd.DataFrame({"A": [2.0, 0.0], "B": [1.0, 3.0]}, index=dates[:2])
|
|
opens = pd.DataFrame(
|
|
{"A": [10.0, 10.0, 15.0, 15.0], "B": [20.0, 20.0, 20.0, 21.0]},
|
|
index=dates,
|
|
)
|
|
closes = pd.DataFrame(
|
|
{"A": [10.0, 12.0, 15.0, 15.0], "B": [20.0, 20.0, 18.0, 21.0]},
|
|
index=dates,
|
|
)
|
|
return run_factor_backtest_research(
|
|
scores,
|
|
opens,
|
|
closes,
|
|
top_k=1,
|
|
execution_price_field="open",
|
|
valuation_price_field="close",
|
|
initial_cash=1_000.0,
|
|
config=ExecutionConfig(
|
|
commission_bps=0,
|
|
stamp_tax_bps=0,
|
|
slippage_bps=0,
|
|
min_trade_amount=0,
|
|
),
|
|
)
|
|
|
|
|
|
def _artifact(run_ref: BacktestRunRef) -> ResearchRunArtifact:
|
|
result = _backtest_result()
|
|
benchmark = pd.Series(
|
|
[0.0, 0.01, -0.01, 0.02],
|
|
index=result.returns.index,
|
|
name="benchmark_return",
|
|
)
|
|
return build_research_run_artifact(
|
|
result,
|
|
run_id=run_ref.run_id,
|
|
strategy_id=run_ref.strategy_id,
|
|
strategy_name="Alpha Top 1",
|
|
strategy_version=run_ref.strategy_version,
|
|
engine_version="1.2.0",
|
|
code_revision=run_ref.code_revision,
|
|
data_snapshot_id=run_ref.dataset_snapshot_id,
|
|
calendar="CN-A",
|
|
timezone="Asia/Shanghai",
|
|
started_at="2026-01-08T10:00:00+08:00",
|
|
finished_at="2026-01-08T10:01:00+08:00",
|
|
parameters=PARAMETERS,
|
|
benchmark_id="000300.SH",
|
|
benchmark_returns=benchmark,
|
|
)
|
|
|
|
|
|
def _covariance(
|
|
context: dict[str, Any],
|
|
*,
|
|
matrix: pd.DataFrame | None = None,
|
|
data_snapshot_id: str | None = None,
|
|
as_of_date: date = date(2026, 1, 8),
|
|
input_sha256: str = "a" * 64,
|
|
) -> CovarianceSnapshot:
|
|
return CovarianceSnapshot(
|
|
snapshot_id="covariance:synthetic-v1",
|
|
as_of_date=as_of_date,
|
|
covariance=(
|
|
pd.DataFrame([[0.04, 0.01], [0.01, 0.09]], index=["A", "B"], columns=["A", "B"])
|
|
if matrix is None
|
|
else matrix
|
|
),
|
|
return_frequency="1d",
|
|
periods_per_year=252,
|
|
method="provided",
|
|
window_start_date=date(2026, 1, 1),
|
|
window_end_date=as_of_date,
|
|
observations=6,
|
|
lookback_sessions=6,
|
|
missing_policy="complete_case",
|
|
data_snapshot_id=(
|
|
context["run_ref"].dataset_snapshot_id
|
|
if data_snapshot_id is None
|
|
else data_snapshot_id
|
|
),
|
|
input_sha256=input_sha256,
|
|
)
|
|
|
|
|
|
def _context(
|
|
*,
|
|
universe_character: str = "5",
|
|
artifact_available_at: str = "2026-01-08T02:01:00Z",
|
|
) -> dict[str, Any]:
|
|
run_ref = _run_ref(universe_character=universe_character)
|
|
artifact = _artifact(run_ref)
|
|
manifest = build_backtest_evidence_manifest(
|
|
run_ref,
|
|
artifact,
|
|
artifact_available_at=artifact_available_at,
|
|
)
|
|
target = PortfolioTarget(
|
|
target_id="portfolio-target:synthetic-v1",
|
|
backtest_run_id=run_ref.run_id,
|
|
dataset_snapshot_id=run_ref.dataset_snapshot_id,
|
|
weights={"B": 0.4, "A": 0.6},
|
|
created_at=datetime(2026, 1, 8, 3, 0, tzinfo=UTC),
|
|
)
|
|
result: dict[str, Any] = {
|
|
"run_ref": run_ref,
|
|
"artifact": artifact,
|
|
"manifest": manifest,
|
|
"target": target,
|
|
"constraints": ConstraintSetV1(
|
|
gross_exposure_max=1.0,
|
|
net_exposure_min=1.0,
|
|
net_exposure_max=1.0,
|
|
single_asset_min=0.2,
|
|
single_asset_max=0.7,
|
|
position_count_max=2,
|
|
turnover_max=0.2,
|
|
),
|
|
"freshness_policy": FreshnessPolicy(
|
|
max_manifest_age_seconds=3_600,
|
|
max_covariance_age_days=0,
|
|
),
|
|
"prior_weights": {"A": 0.5, "B": 0.5},
|
|
"computed_at": datetime(2026, 1, 8, 3, 1, tzinfo=UTC),
|
|
}
|
|
result["covariance"] = _covariance(result)
|
|
return result
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def context() -> dict[str, Any]:
|
|
return _context()
|
|
|
|
|
|
def _decision_values(context: dict[str, Any], **overrides: Any) -> dict[str, Any]:
|
|
values: dict[str, Any] = {
|
|
"backtest_run_ref": context["run_ref"],
|
|
"manifest": context["manifest"],
|
|
"target": context["target"],
|
|
**DECISION_FIELDS,
|
|
"constraints": context["constraints"],
|
|
"freshness_policy": context["freshness_policy"],
|
|
"computed_at": context["computed_at"],
|
|
"prior_weights": context["prior_weights"],
|
|
}
|
|
values.update(overrides)
|
|
return values
|
|
|
|
|
|
def _receipt(
|
|
values: dict[str, Any],
|
|
*,
|
|
solver_required: bool = False,
|
|
status: ReceiptStatus = ReceiptStatus.COMPLETED,
|
|
overrides: dict[str, Any] | None = None,
|
|
) -> ComputationReceipt:
|
|
digest_arguments = {
|
|
key: value
|
|
for key, value in values.items()
|
|
if key not in {"computed_at", "receipt"}
|
|
}
|
|
digests = compute_portfolio_receipt_digests(**digest_arguments)
|
|
receipt_values: dict[str, Any] = {
|
|
"algorithm": "bounded_allocation",
|
|
"algorithm_version": "1.0.0",
|
|
"implementation_digest": "sha256:" + "f" * 64,
|
|
"parameter_digest": "sha256:" + "0" * 64,
|
|
"input_digest": digests["input_digest"],
|
|
"constraint_digest": digests["constraint_digest"],
|
|
"output_digest": digests["output_digest"],
|
|
"status": status,
|
|
"solver_required": solver_required,
|
|
"solver_name": "slsqp" if solver_required else None,
|
|
"solver_version": "1.0.0" if solver_required else None,
|
|
"solver_config_digest": "sha256:" + "1" * 64 if solver_required else None,
|
|
"iterations": 4 if solver_required else None,
|
|
"objective_value": 0.25 if solver_required else None,
|
|
"max_constraint_residual": digests["max_constraint_residual"],
|
|
"tolerance": 1.0 if solver_required else 1e-12,
|
|
"computed_at": values["computed_at"],
|
|
}
|
|
if overrides:
|
|
receipt_values.update(overrides)
|
|
return ComputationReceipt(**receipt_values)
|
|
|
|
|
|
def _decision(context: dict[str, Any], **overrides: Any) -> PortfolioDecision:
|
|
values = _decision_values(context, **overrides)
|
|
receipt = values.pop("receipt", None)
|
|
if receipt is None:
|
|
receipt = _receipt(values)
|
|
return build_portfolio_decision(**values, receipt=receipt)
|
|
|
|
|
|
def _assessment(
|
|
context: dict[str, Any],
|
|
*,
|
|
decision: PortfolioDecision | None = None,
|
|
covariance: CovarianceSnapshot | None = None,
|
|
**overrides: Any,
|
|
) -> RiskAssessment:
|
|
values: dict[str, Any] = {
|
|
"portfolio_decision": _decision(context) if decision is None else decision,
|
|
"backtest_run_ref": context["run_ref"],
|
|
"manifest": context["manifest"],
|
|
"covariance": context["covariance"] if covariance is None else covariance,
|
|
"risk_model_name": "euler_volatility",
|
|
"risk_model_version": "1.0.0",
|
|
"risk_model_digest": "sha256:" + "2" * 64,
|
|
"risk_budget": {"A": 0.8, "B": 0.8},
|
|
"portfolio_volatility_limit": 10.0,
|
|
"groups": {"A": "equity", "B": "fixed_income"},
|
|
}
|
|
values.update(overrides)
|
|
return assess_portfolio_risk(**values)
|
|
|
|
|
|
def _assert_error(
|
|
error: pytest.ExceptionInfo[PortfolioRiskContractError],
|
|
code: PortfolioRiskContractErrorCode,
|
|
path: str,
|
|
) -> None:
|
|
assert error.value.code is code
|
|
assert error.value.path == path
|
|
|
|
|
|
def test_s41_public_contract_surface_exists() -> None:
|
|
assert all(
|
|
symbol is not None
|
|
for symbol in (
|
|
ComputationReceipt,
|
|
ConstraintSetV1,
|
|
FreshnessPolicy,
|
|
PortfolioDecision,
|
|
RiskAssessment,
|
|
assess_portfolio_risk,
|
|
build_portfolio_decision,
|
|
)
|
|
)
|
|
|
|
|
|
def test_positive_decision_and_assessment_match_canonical_golden(
|
|
context: dict[str, Any],
|
|
) -> None:
|
|
decision = _decision(context)
|
|
assessment = _assessment(context, decision=decision)
|
|
|
|
assert decision.run_ref_document_sha256 == hashlib.sha256(
|
|
context["run_ref"].to_json().encode("utf-8")
|
|
).hexdigest()
|
|
assert decision.manifest_document_sha256 == hashlib.sha256(
|
|
context["manifest"].to_json().encode("utf-8")
|
|
).hexdigest()
|
|
assert decision.source_universe_digest == context["run_ref"].universe_digest
|
|
assert decision.source_universe_digest != decision.portfolio_asset_set_digest
|
|
assert decision.gross_exposure == decision.net_exposure == 1.0
|
|
assert decision.turnover_l1 == pytest.approx(0.2)
|
|
assert assessment.status is RiskAssessmentStatus.READY
|
|
assert assessment.qualified is True
|
|
assert assessment.findings == ()
|
|
assert sum(assessment.component_risk.values()) == pytest.approx(
|
|
assessment.portfolio_volatility
|
|
)
|
|
assert sum(assessment.percentage_risk.values()) == pytest.approx(1.0)
|
|
actual = {
|
|
"portfolio_decision": decision.to_dict(),
|
|
"risk_assessment": assessment.to_dict(),
|
|
}
|
|
expected = json.loads(GOLDEN_FIXTURE.read_text(encoding="utf-8"))
|
|
assert actual == expected
|
|
assert decision.to_json() == json.dumps(
|
|
decision.to_dict(), ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
|
)
|
|
|
|
|
|
def test_container_order_is_identity_neutral(context: dict[str, Any]) -> None:
|
|
target = PortfolioTarget(
|
|
target_id=context["target"].target_id,
|
|
backtest_run_id=context["target"].backtest_run_id,
|
|
dataset_snapshot_id=context["target"].dataset_snapshot_id,
|
|
weights={"A": 0.6, "B": 0.4},
|
|
created_at=context["target"].created_at,
|
|
)
|
|
reordered = _decision(context, target=target, prior_weights={"B": 0.5, "A": 0.5})
|
|
assert reordered.decision_id == _decision(context).decision_id
|
|
|
|
|
|
def test_run_manifest_target_and_model_mutations_change_identity(
|
|
context: dict[str, Any],
|
|
) -> None:
|
|
baseline = _decision(context)
|
|
other_run_context = _context(universe_character="4")
|
|
other_run = _decision(other_run_context)
|
|
later_manifest_context = _context(artifact_available_at="2026-01-08T02:02:00Z")
|
|
later_manifest = _decision(later_manifest_context)
|
|
changed_target = PortfolioTarget(
|
|
target_id="portfolio-target:synthetic-v2",
|
|
backtest_run_id=context["run_ref"].run_id,
|
|
dataset_snapshot_id=context["run_ref"].dataset_snapshot_id,
|
|
weights={"A": 0.5, "B": 0.5},
|
|
created_at=context["target"].created_at,
|
|
)
|
|
changed_weights = _decision(context, target=changed_target)
|
|
changed_assets_target = PortfolioTarget(
|
|
target_id="portfolio-target:synthetic-v3",
|
|
backtest_run_id=context["run_ref"].run_id,
|
|
dataset_snapshot_id=context["run_ref"].dataset_snapshot_id,
|
|
weights={"A": 0.6, "C": 0.4},
|
|
created_at=context["target"].created_at,
|
|
)
|
|
changed_assets = _decision(
|
|
context,
|
|
target=changed_assets_target,
|
|
prior_weights={"A": 0.5, "C": 0.5},
|
|
)
|
|
changed_model = _decision(
|
|
context,
|
|
model_version="1.0.1",
|
|
model_digest="sha256:" + "3" * 64,
|
|
)
|
|
|
|
assert other_run.run_ref_document_sha256 != baseline.run_ref_document_sha256
|
|
assert other_run.source_universe_digest != baseline.source_universe_digest
|
|
assert later_manifest.manifest_document_sha256 != baseline.manifest_document_sha256
|
|
assert changed_weights.portfolio_asset_set_digest == baseline.portfolio_asset_set_digest
|
|
assert changed_weights.output_digest != baseline.output_digest
|
|
assert changed_assets.portfolio_asset_set_digest != baseline.portfolio_asset_set_digest
|
|
assert len(
|
|
{
|
|
baseline.decision_id,
|
|
other_run.decision_id,
|
|
later_manifest.decision_id,
|
|
changed_weights.decision_id,
|
|
changed_assets.decision_id,
|
|
changed_model.decision_id,
|
|
}
|
|
) == 6
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("payload", "path"),
|
|
[
|
|
({"schema_version": "1.0.0", "max_covariance_age_days": 1}, "$.max_manifest_age_seconds"),
|
|
(
|
|
{
|
|
"schema_version": "1.0.0",
|
|
"max_manifest_age_seconds": 1,
|
|
"max_covariance_age_days": 1,
|
|
"extra": 1,
|
|
},
|
|
"$.extra",
|
|
),
|
|
(
|
|
{
|
|
"schema_version": "1.0.0",
|
|
"max_manifest_age_seconds": -1,
|
|
"max_covariance_age_days": 1,
|
|
},
|
|
"$.max_manifest_age_seconds",
|
|
),
|
|
(
|
|
{
|
|
"schema_version": "1.0.0",
|
|
"max_manifest_age_seconds": True,
|
|
"max_covariance_age_days": 1,
|
|
},
|
|
"$.max_manifest_age_seconds",
|
|
),
|
|
(
|
|
{
|
|
"schema_version": "1.0.0",
|
|
"max_manifest_age_seconds": 1.0,
|
|
"max_covariance_age_days": 1,
|
|
},
|
|
"$.max_manifest_age_seconds",
|
|
),
|
|
(
|
|
{
|
|
"schema_version": "1.0.0",
|
|
"max_manifest_age_seconds": 2**53,
|
|
"max_covariance_age_days": 1,
|
|
},
|
|
"$.max_manifest_age_seconds",
|
|
),
|
|
],
|
|
)
|
|
def test_freshness_policy_has_strict_schema_and_safe_thresholds(
|
|
payload: dict[str, object], path: str
|
|
) -> None:
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
FreshnessPolicy.from_dict(payload)
|
|
_assert_error(error, PortfolioRiskContractErrorCode.INVALID_FRESHNESS_POLICY, path)
|
|
|
|
|
|
def test_manifest_freshness_boundary_and_timezone_normalization(
|
|
context: dict[str, Any],
|
|
) -> None:
|
|
equal_boundary = _decision(
|
|
context,
|
|
computed_at="2026-01-08T11:01:00+08:00",
|
|
)
|
|
assert equal_boundary.computed_at == "2026-01-08T03:01:00Z"
|
|
stale_values = _decision_values(
|
|
context,
|
|
computed_at="2026-01-08T03:01:01Z",
|
|
)
|
|
stale_receipt = _receipt(stale_values)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
build_portfolio_decision(**stale_values, receipt=stale_receipt)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.MANIFEST_STALE,
|
|
"$.manifest.artifact_available_at",
|
|
)
|
|
naive_values = _decision_values(
|
|
context,
|
|
computed_at=datetime(2026, 1, 8, 3, 1),
|
|
)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
_receipt(naive_values)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION,
|
|
"$.computed_at",
|
|
)
|
|
|
|
|
|
def test_constraint_set_rejects_unknown_nonfinite_and_unsafe_fields() -> None:
|
|
payload = ConstraintSetV1().to_dict()
|
|
payload["custom_constraint"] = 1.0
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
ConstraintSetV1.from_dict(payload)
|
|
_assert_error(error, PortfolioRiskContractErrorCode.UNKNOWN_FIELD, "$.custom_constraint")
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
ConstraintSetV1(gross_exposure_max=float("nan"))
|
|
_assert_error(error, PortfolioRiskContractErrorCode.INVALID_VALUE, "$.gross_exposure_max")
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
ConstraintSetV1(position_count_max=True)
|
|
_assert_error(error, PortfolioRiskContractErrorCode.TYPE_ERROR, "$.position_count_max")
|
|
|
|
|
|
def test_s3_qualification_embedded_run_and_dataset_mismatch_fail_closed(
|
|
context: dict[str, Any],
|
|
) -> None:
|
|
exploratory = build_backtest_evidence_manifest(
|
|
context["run_ref"],
|
|
context["artifact"],
|
|
artifact_available_at="2026-01-08T02:01:00Z",
|
|
qualification=EvidenceQualification.EXPLORATORY,
|
|
)
|
|
values = _decision_values(context, manifest=exploratory)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
build_portfolio_decision(**values, receipt=_receipt(values))
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.QUALIFICATION_REJECTED,
|
|
"$.manifest.qualification",
|
|
)
|
|
|
|
other_run = _run_ref(universe_character="4")
|
|
manifest = context["manifest"]
|
|
mismatched = object.__new__(BacktestEvidenceManifest)
|
|
for field_name in manifest.__dataclass_fields__:
|
|
object.__setattr__(
|
|
mismatched,
|
|
field_name,
|
|
other_run if field_name == "backtest_run_ref" else getattr(manifest, field_name),
|
|
)
|
|
mismatch_values = _decision_values(context, manifest=mismatched)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
build_portfolio_decision(**mismatch_values, receipt=_receipt(mismatch_values))
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.IDENTITY_MISMATCH,
|
|
"$.manifest.run_reference.value",
|
|
)
|
|
|
|
target = PortfolioTarget(
|
|
target_id="portfolio-target:dataset-drift",
|
|
backtest_run_id=context["run_ref"].run_id,
|
|
dataset_snapshot_id="dataset:other",
|
|
weights={"A": 0.6, "B": 0.4},
|
|
created_at=context["target"].created_at,
|
|
)
|
|
drift_values = _decision_values(context, target=target)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
build_portfolio_decision(**drift_values, receipt=_receipt(drift_values))
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.DATASET_IDENTITY_MISMATCH,
|
|
"$.portfolio_target.dataset_snapshot_id",
|
|
)
|
|
|
|
|
|
def test_explicit_legacy_manifest_cannot_form_a_decision(context: dict[str, Any]) -> None:
|
|
manifest = context["manifest"]
|
|
legacy_run = BacktestRun(
|
|
run_id=context["run_ref"].run_id,
|
|
dataset_snapshot_id=context["run_ref"].dataset_snapshot_id,
|
|
factor_version_id="legacy-factor@1.0.0",
|
|
strategy_version_id="alpha-top1@1.0.0",
|
|
code_revision=context["run_ref"].code_revision,
|
|
config_hash=context["run_ref"].configuration_digest.removeprefix("sha256:"),
|
|
created_at=datetime(2026, 1, 8, 2, tzinfo=UTC),
|
|
)
|
|
legacy = object.__new__(BacktestEvidenceManifest)
|
|
overrides = {
|
|
"qualification": EvidenceQualification.LEGACY_EXPLORATORY,
|
|
"backtest_run_ref": None,
|
|
"legacy_backtest_run": legacy_run,
|
|
}
|
|
for field_name in manifest.__dataclass_fields__:
|
|
object.__setattr__(
|
|
legacy,
|
|
field_name,
|
|
overrides.get(field_name, getattr(manifest, field_name)),
|
|
)
|
|
values = _decision_values(context, manifest=legacy)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
build_portfolio_decision(**values, receipt=_receipt(values))
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.QUALIFICATION_REJECTED,
|
|
"$.manifest.qualification",
|
|
)
|
|
|
|
|
|
def test_receipt_field_matrix_and_digest_recomputation(context: dict[str, Any]) -> None:
|
|
values = _decision_values(context)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
_receipt(values, overrides={"algorithm_version": "v1"})
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.INVALID_FORMAT,
|
|
"$.algorithm_version",
|
|
)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
_receipt(values, overrides={"solver_name": "unexpected"})
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.INVALID_VALUE,
|
|
"$.solver_name",
|
|
)
|
|
bad_receipt = _receipt(values, overrides={"input_digest": "sha256:" + "9" * 64})
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
build_portfolio_decision(**values, receipt=bad_receipt)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.RECEIPT_MISMATCH,
|
|
"$.receipt.input_digest",
|
|
)
|
|
for status in (ReceiptStatus.FAILED, ReceiptStatus.FALLBACK):
|
|
rejected = _receipt(
|
|
values,
|
|
solver_required=True,
|
|
status=status,
|
|
)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
build_portfolio_decision(**values, receipt=rejected)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.QUALIFICATION_REJECTED,
|
|
"$.receipt.status",
|
|
)
|
|
|
|
|
|
def test_constraints_are_independently_recomputed(context: dict[str, Any]) -> None:
|
|
target = PortfolioTarget(
|
|
target_id="portfolio-target:violating",
|
|
backtest_run_id=context["run_ref"].run_id,
|
|
dataset_snapshot_id=context["run_ref"].dataset_snapshot_id,
|
|
weights={"A": 0.8, "B": 0.2},
|
|
created_at=context["target"].created_at,
|
|
)
|
|
values = _decision_values(context, target=target)
|
|
receipt = _receipt(values, solver_required=True, status=ReceiptStatus.CONVERGED)
|
|
assert receipt.max_constraint_residual == pytest.approx(0.4)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
build_portfolio_decision(**values, receipt=receipt)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.CONSTRAINT_VIOLATION,
|
|
"$.constraints",
|
|
)
|
|
|
|
|
|
def test_covariance_target_and_run_dataset_identity_is_three_way_closed(
|
|
context: dict[str, Any],
|
|
) -> None:
|
|
covariance = _covariance(context, data_snapshot_id="dataset:other")
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
_assessment(context, covariance=covariance)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.DATASET_IDENTITY_MISMATCH,
|
|
"$.covariance.data_snapshot_id",
|
|
)
|
|
wrong_digest = _covariance(context, input_sha256="b" * 64)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
_assessment(context, covariance=wrong_digest)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.IDENTITY_MISMATCH,
|
|
"$.covariance.input_sha256",
|
|
)
|
|
|
|
|
|
def test_covariance_pit_and_freshness_are_enforced(context: dict[str, Any]) -> None:
|
|
future = _covariance(context, as_of_date=date(2026, 1, 9))
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
_assessment(context, covariance=future)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION,
|
|
"$.covariance.as_of_date",
|
|
)
|
|
stale = _covariance(context, as_of_date=date(2026, 1, 7))
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
_assessment(context, covariance=stale)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.COVARIANCE_STALE,
|
|
"$.covariance.as_of_date",
|
|
)
|
|
|
|
|
|
def test_risk_decomposition_is_delegated_exactly_once(
|
|
context: dict[str, Any], monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
calls = 0
|
|
|
|
def wrapped(weights: pd.Series[Any], covariance: pd.DataFrame) -> ComponentRiskResult:
|
|
nonlocal calls
|
|
calls += 1
|
|
return labeled_component_risk(weights, covariance)
|
|
|
|
monkeypatch.setattr(contracts_module, "labeled_component_risk", wrapped)
|
|
assessment = _assessment(context)
|
|
assert calls == 1
|
|
assert assessment.qualified is True
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("matrix", "finding"),
|
|
[
|
|
(
|
|
pd.DataFrame([[1.0, 2.0], [2.0, 1.0]], index=["A", "B"], columns=["A", "B"]),
|
|
RiskFindingCode.COVARIANCE_NOT_PSD,
|
|
),
|
|
(
|
|
pd.DataFrame([[0.0, 0.0], [0.0, 0.0]], index=["A", "B"], columns=["A", "B"]),
|
|
RiskFindingCode.PORTFOLIO_VARIANCE_NON_POSITIVE,
|
|
),
|
|
],
|
|
)
|
|
def test_named_numerical_failures_return_stable_unavailable_findings(
|
|
context: dict[str, Any], matrix: pd.DataFrame, finding: RiskFindingCode
|
|
) -> None:
|
|
result = _assessment(context, covariance=_covariance(context, matrix=matrix))
|
|
assert result.status is RiskAssessmentStatus.UNAVAILABLE
|
|
assert result.qualified is False
|
|
assert result.findings == (finding,)
|
|
assert result.portfolio_volatility is None
|
|
assert result.marginal_risk == result.component_risk == result.percentage_risk == {}
|
|
|
|
|
|
def test_unknown_numerical_error_is_sanitized(
|
|
context: dict[str, Any], monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
def fail_unknown(weights: pd.Series[Any], covariance: pd.DataFrame) -> ComponentRiskResult:
|
|
raise ValueError("sensitive lower-level details")
|
|
|
|
monkeypatch.setattr(contracts_module, "labeled_component_risk", fail_unknown)
|
|
with pytest.raises(PortfolioRiskContractError) as error:
|
|
_assessment(context)
|
|
_assert_error(
|
|
error,
|
|
PortfolioRiskContractErrorCode.COMPUTATION_FAILURE,
|
|
"$.covariance",
|
|
)
|
|
assert "sensitive" not in str(error.value)
|
|
|
|
|
|
def test_non_closed_risk_and_budget_breach_have_distinct_semantics(
|
|
context: dict[str, Any], monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
def non_closed(weights: pd.Series[Any], covariance: pd.DataFrame) -> ComponentRiskResult:
|
|
return ComponentRiskResult(
|
|
portfolio_volatility=1.0,
|
|
marginal=pd.Series({"A": 0.1, "B": 0.1}),
|
|
component=pd.Series({"A": 0.1, "B": 0.1}),
|
|
percentage=pd.Series({"A": 0.5, "B": 0.4}),
|
|
)
|
|
|
|
monkeypatch.setattr(contracts_module, "labeled_component_risk", non_closed)
|
|
unavailable = _assessment(context)
|
|
assert unavailable.status is RiskAssessmentStatus.UNAVAILABLE
|
|
assert unavailable.findings == (RiskFindingCode.RISK_CONTRIBUTION_NOT_CLOSED,)
|
|
monkeypatch.setattr(contracts_module, "labeled_component_risk", labeled_component_risk)
|
|
breached = _assessment(context, risk_budget={"A": 0.0})
|
|
assert breached.status is RiskAssessmentStatus.READY
|
|
assert breached.qualified is False
|
|
assert breached.findings == (RiskFindingCode.RISK_BUDGET_BREACH,)
|
|
assert breached.component_risk
|
|
|
|
|
|
def test_existing_target_risk_decision_artifact_and_paper_slice_remain_legacy_only(
|
|
context: dict[str, Any],
|
|
) -> None:
|
|
target = context["target"]
|
|
legacy_decision = evaluate_portfolio_risk(
|
|
target,
|
|
RiskPolicy(
|
|
policy_id="legacy-paper-policy",
|
|
max_gross_exposure=1.0,
|
|
max_single_asset_weight=0.7,
|
|
max_positions=2,
|
|
),
|
|
created_at=target.created_at,
|
|
)
|
|
intent = create_paper_order_intent(target, legacy_decision)
|
|
assert isinstance(legacy_decision, RiskDecision)
|
|
assert legacy_decision.status is RiskDecisionStatus.APPROVED
|
|
assert intent.environment == "paper"
|
|
assert isinstance(context["artifact"], ResearchRunArtifact)
|
|
assert not isinstance(target, PortfolioDecision)
|
|
assert not isinstance(legacy_decision, RiskAssessment)
|
|
|
|
|
|
def test_architecture_dependency_no_copy_and_authority_boundaries() -> None:
|
|
new_path = ROOT / "src" / "quant_engine" / "portfolio_risk_contracts.py"
|
|
source = new_path.read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
called = {
|
|
node.func.id
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
|
|
}
|
|
assert not called & {
|
|
"evaluate_portfolio_risk",
|
|
"create_paper_order_intent",
|
|
"run_governed_factor_slice",
|
|
}
|
|
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")) for name in imports)
|
|
for candidate in ("riskfolio", "pyp", "skfolio", "cvxportfolio"):
|
|
assert candidate not in source.lower()
|
|
artifact_source = (ROOT / "src" / "quant_engine" / "artifact.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
artifact_tree = ast.parse(artifact_source)
|
|
forbidden_artifact_authority_symbols = {
|
|
"PortfolioDecision",
|
|
"RiskAssessment",
|
|
"build_portfolio_decision",
|
|
"assess_portfolio_risk",
|
|
}
|
|
artifact_imports = {
|
|
alias.name
|
|
for node in ast.walk(artifact_tree)
|
|
if isinstance(node, ast.Import)
|
|
for alias in node.names
|
|
} | {
|
|
node.module or ""
|
|
for node in ast.walk(artifact_tree)
|
|
if isinstance(node, ast.ImportFrom)
|
|
}
|
|
artifact_names = {
|
|
node.id for node in ast.walk(artifact_tree) if isinstance(node, ast.Name)
|
|
} | {
|
|
node.attr for node in ast.walk(artifact_tree) if isinstance(node, ast.Attribute)
|
|
}
|
|
assert "quant_engine.portfolio_risk_contracts" not in artifact_imports
|
|
assert not forbidden_artifact_authority_symbols & artifact_names
|
|
assert all(
|
|
token not in artifact_source
|
|
for token in {"portfolio_risk_contracts", *forbidden_artifact_authority_symbols}
|
|
)
|
|
for owner_path in (
|
|
ROOT / "src" / "quant_engine" / "governed_pipeline.py",
|
|
ROOT / "src" / "quant_engine" / "artifact.py",
|
|
ROOT / "src" / "quant_engine" / "portfolio_construction.py",
|
|
ROOT / "src" / "quant_engine" / "portfolio_decomp.py",
|
|
ROOT / "src" / "quant_engine" / "risk.py",
|
|
):
|
|
assert "portfolio_risk_contracts" not in owner_path.read_text(encoding="utf-8")
|
|
forbidden_fields = {
|
|
"approved",
|
|
"approval",
|
|
"maker",
|
|
"checker",
|
|
"owner",
|
|
"publication",
|
|
"order",
|
|
"broker",
|
|
"environment",
|
|
"paper",
|
|
"live",
|
|
"uri",
|
|
"path",
|
|
"locator",
|
|
}
|
|
for contract_type in (
|
|
FreshnessPolicy,
|
|
ConstraintSetV1,
|
|
ComputationReceipt,
|
|
PortfolioDecision,
|
|
RiskAssessment,
|
|
):
|
|
assert not forbidden_fields & set(contract_type.__dataclass_fields__)
|
|
|
|
|
|
def test_read_only_owner_dependency_lock_and_ci_hashes_match_baseline() -> None:
|
|
expected = {
|
|
"src/quant_engine/governed_pipeline.py": "3b334f340898db78ed869375ab532f156e8c1fee595a8318f544bebdd391049d",
|
|
"src/quant_engine/portfolio_construction.py": "e93d71da8d61b2047c19d4b99dace934a8cbc96d8d2b150ad62a9ceebd9163d4",
|
|
"src/quant_engine/portfolio_decomp.py": "1a4f9f9aac2c46bf6ed2826d1b0d1723f06e3ce7c4b3f6479e098cbbd135bea6",
|
|
"src/quant_engine/risk.py": "4a66c312d517d40f6f67bb71f438523e135645624ae78d49fa9c0fda2c02074e",
|
|
"pyproject.toml": "9340a25f3710778945765f14a02c91844aa66bd47b77c55d8fa4bd36849b4bd5",
|
|
"uv.lock": "076a16a01bc363109174909999652e81055954200a3ab93eab5f76df2249ee1a",
|
|
"ci-profile.yml": "dfe1b7c2820747fa28eaad409153623c468ee3e10f38f4fd94241731c5a55803",
|
|
}
|
|
actual = {
|
|
path: hashlib.sha256((ROOT / path).read_bytes()).hexdigest()
|
|
for path in expected
|
|
}
|
|
assert actual == expected
|