From a724e1e57a99d1304a932d01ee836bac56c5c15c Mon Sep 17 00:00:00 2001 From: ageorge156 Date: Tue, 1 Sep 2026 14:08:23 +0800 Subject: [PATCH] feat: add portfolio risk computation contracts (#18) --- MODULE_SPEC.yaml | 8 +- README.md | 106 + src/quant_engine/portfolio_risk_contracts.py | 1784 +++++++++++++++++ .../portfolio-risk-computation-v1.golden.json | 129 ++ tests/governance/test_module_spec.py | 14 +- tests/test_portfolio_risk_contracts.py | 1004 ++++++++++ 6 files changed, 3042 insertions(+), 3 deletions(-) create mode 100644 src/quant_engine/portfolio_risk_contracts.py create mode 100644 tests/fixtures/portfolio-risk-computation-v1.golden.json create mode 100644 tests/test_portfolio_risk_contracts.py diff --git a/MODULE_SPEC.yaml b/MODULE_SPEC.yaml index d6af360..54b4016 100644 --- a/MODULE_SPEC.yaml +++ b/MODULE_SPEC.yaml @@ -1,7 +1,7 @@ { "schema_version": 1, "module_id": "quant_engine", - "authority": {"scope": "module_metadata", "subject": "quant_engine", "owner": "quant-engine-owner", "source": "MODULE_SPEC.yaml", "revision": 3, "effective_from": "2026-09-01T00:00:00+08:00"}, + "authority": {"scope": "module_metadata", "subject": "quant_engine", "owner": "quant-engine-owner", "source": "MODULE_SPEC.yaml", "revision": 4, "effective_from": "2026-09-01T00:00:00+08:00"}, "repository": {"name": "quant_engine", "workspace_id": "researchhub", "type": "research_engine", "maturity": "operational"}, "bounded_context": { "domain": "quantitative-research-engine", @@ -11,6 +11,7 @@ "Submitting live orders, routing trades, managing brokerage accounts, or claiming transaction execution", "Owning market-data source facts, research-result publication, or platform presentation state", "Loading provider credentials, brokerage credentials, or production secrets", + "Granting portfolio approval, maker-checker decisions, publication eligibility, paper execution, or live execution authority", "Changing financial model semantics through module metadata" ] }, @@ -19,6 +20,7 @@ {"id": "execution-simulation", "summary": "Simulate costs, slippage, market constraints, fills, NAV, and PnL without live order routing.", "status": "operational"}, {"id": "portfolio-backtesting", "summary": "Run weight-based backtests and benchmark comparisons.", "status": "operational"}, {"id": "backtest-evidence-contracts", "summary": "Identify governed offline backtest inputs and close existing research artifact evidence without persistence or decision authority.", "status": "operational"}, + {"id": "portfolio-risk-computation-contracts", "summary": "Verify deterministic portfolio-computation receipts and expose S3-bound portfolio decisions and risk assessments without adding algorithms or execution authority.", "status": "operational"}, {"id": "risk-and-performance-analysis", "summary": "Calculate portfolio decomposition, risk contribution, and performance statistics.", "status": "operational"} ], "data": {"owns": [ @@ -30,7 +32,9 @@ {"contract_id": "researchhub.factor-definition", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/factor_contracts.py"}, {"contract_id": "researchhub.factor-set-ref", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/factor_contracts.py"}, {"contract_id": "researchhub.backtest-run-ref", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/governed_pipeline.py"}, - {"contract_id": "researchhub.backtest-evidence-manifest", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/artifact.py"} + {"contract_id": "researchhub.backtest-evidence-manifest", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/artifact.py"}, + {"contract_id": "researchhub.portfolio-decision", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/portfolio_risk_contracts.py"}, + {"contract_id": "researchhub.risk-assessment", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/portfolio_risk_contracts.py"} ], "consumes": [ {"contract_id": "researchhub.dataset-snapshot", "version": "1.0.0", "authority": "researchhub.data", "admission": "qualified_immutable_envelope"}, diff --git a/README.md b/README.md index 060261c..fa98fea 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ - `research_pipeline` — 因子日 → 下一真实交易日 → 显式执行价 → 日末估值 → 成本后绩效（防前视编排） - `governed_pipeline` — 数据快照 → 因子版本 → 策略版本 → 回测运行 → 目标组合 → 风险决策 → Paper 订单意图；同时拥有输入/配置/重放血缘决定的 `BacktestRunRef` - `artifact` — 版本化、确定性、存储中立的完整 research run 事实表，以及只映射现有表的 `BacktestEvidenceManifest` +- `portfolio_risk_contracts` — S3 证据闭合的 `PortfolioDecision` / `RiskAssessment` v1；独立复核 freshness、约束与 computation receipt，并复用既有标签安全风险分解 - `attribution` — 基于实际成交后持仓的隔夜 / 日内 / 交易成本逐日收益归因与闭合审计 - `metrics` — 绝对绩效 + 严格日期对齐的 TE / IR / alpha / beta 基准相对绩效 - `factor_library` — 通用方法（turnover / winsorize / IC / OLS / jb_test） @@ -259,6 +260,111 @@ digest 等价，也不会把旧 run 静默升级为新合同。 `LEGACY_EXPLORATORY`；不能隐式提升为 `CONTRACT_QUALIFIED`。所有资格均只描述离线证据闭合， 不表示投资有效、组合获批、Paper、生产或实盘就绪。 +## 组合决策与风险评估合同 v1 + +`quant_engine.portfolio_risk_contracts` 是现有计算 owner 外围的薄合同层。创建 +`PortfolioDecision` 必须同时提供完整 `BacktestRunRef`、嵌入同一 RunRef 的 +`CONTRACT_QUALIFIED` 非 legacy `BacktestEvidenceManifest`、现有 `PortfolioTarget`、 +`FreshnessPolicy`、`ConstraintSetV1` 与 `ComputationReceipt`。适配器会从权威输入独立重算 +receipt 的 input/constraint/output digest、敞口、持仓数和 L1 turnover 残差；receipt 自报 +成功、fallback 或放宽 tolerance 均不能替代复核。 + +```python +from quant_engine.portfolio_risk_contracts import ( + ComputationReceipt, + ConstraintSetV1, + FreshnessPolicy, + assess_portfolio_risk, + build_portfolio_decision, + compute_portfolio_receipt_digests, +) + +freshness = FreshnessPolicy( + max_manifest_age_seconds=3600, + max_covariance_age_days=5, +) +constraints = ConstraintSetV1( + gross_exposure_max=1.0, + single_asset_max=0.10, + position_count_max=20, + turnover_max=0.30, +) + +# 生产者先形成公开 canonical digest；decision 构建时仍会独立重算。 +expected = compute_portfolio_receipt_digests( + backtest_run_ref=run_ref, + manifest=evidence_manifest, + target=portfolio_target, + objective_name="long_only_allocation", + objective_version="1.0.0", + objective_digest=objective_digest, + model_name="factor_weighting", + model_version="1.0.0", + model_digest=model_digest, + expected_return_digest=expected_return_digest, + covariance_digest=covariance_digest, + scenario_digest=scenario_digest, + constraints=constraints, + freshness_policy=freshness, + prior_weights=prior_weights, +) + +receipt = ComputationReceipt( + algorithm="factor_weighting", + algorithm_version="1.0.0", + implementation_digest=implementation_digest, + parameter_digest=parameter_digest, + input_digest=expected["input_digest"], + constraint_digest=expected["constraint_digest"], + output_digest=expected["output_digest"], + status="completed", + solver_required=False, + solver_name=None, + solver_version=None, + solver_config_digest=None, + iterations=None, + objective_value=None, + max_constraint_residual=expected["max_constraint_residual"], + tolerance=1e-12, + computed_at=computed_at, +) + +decision = build_portfolio_decision( + backtest_run_ref=run_ref, + manifest=evidence_manifest, + target=portfolio_target, + objective_name="long_only_allocation", + objective_version="1.0.0", + objective_digest=objective_digest, + model_name="factor_weighting", + model_version="1.0.0", + model_digest=model_digest, + expected_return_digest=expected_return_digest, + covariance_digest=covariance_digest, + scenario_digest=scenario_digest, + constraints=constraints, + freshness_policy=freshness, + receipt=receipt, + computed_at=computed_at, + prior_weights=prior_weights, +) + +assessment = assess_portfolio_risk( + portfolio_decision=decision, + backtest_run_ref=run_ref, + manifest=evidence_manifest, + covariance=covariance_snapshot, + risk_model_name="euler_volatility", + risk_model_version="1.0.0", + risk_model_digest=risk_model_digest, +) +``` + +`source_universe_digest` 保留 S3 研究 universe 身份，`portfolio_asset_set_digest` 只描述实际 +目标资产标签；二者不会互相冒充成员证明。风险评估在任何数值计算前要求 covariance、target、 +RunRef 的 dataset identity 三方一致，并且只调用一次现有 `labeled_component_risk()`。合同中的 +`qualified` 仅表示 S4.1 计算证据闭合，不授予 maker-checker、发布、订单、Paper、生产或实盘权限。 + ## 治理垂直切片 `governed_pipeline` 不复制因子、回测、组合或执行算法，只编排现有能力并补充版本与风险契约。 diff --git a/src/quant_engine/portfolio_risk_contracts.py b/src/quant_engine/portfolio_risk_contracts.py new file mode 100644 index 0000000..31cec98 --- /dev/null +++ b/src/quant_engine/portfolio_risk_contracts.py @@ -0,0 +1,1784 @@ +"""Deterministic S4.1 portfolio-decision and risk-assessment contracts. + +The module verifies identities, time ordering, supported constraints, and +computation receipts around existing ``quant_engine`` calculation owners. It +does not implement portfolio optimization, covariance estimation, or a second +risk formula. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import UTC, date, datetime +from enum import StrEnum +from types import MappingProxyType +from typing import Any, Never, Self, cast + +import numpy as np +import pandas as pd + +from quant_engine.artifact import ( + BacktestEvidenceManifest, + EvidenceQualification, +) +from quant_engine.governed_pipeline import BacktestRunRef, PortfolioTarget +from quant_engine.risk import CovarianceSnapshot, labeled_component_risk + +PORTFOLIO_DECISION_SCHEMA_VERSION = "1.0.0" +RISK_ASSESSMENT_SCHEMA_VERSION = "1.0.0" +FRESHNESS_POLICY_SCHEMA_VERSION = "1.0.0" +CONSTRAINT_SET_SCHEMA_VERSION = "1.0.0" +COMPUTATION_RECEIPT_SCHEMA_VERSION = "1.0.0" + +_MAX_SAFE_INTEGER = (1 << 53) - 1 +_SHA256 = re.compile(r"^[0-9a-f]{64}$") +_PREFIXED_SHA256 = re.compile(r"^sha256:[0-9a-f]{64}$") +_SEMVER = re.compile( + r"^(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\." + r"(?:0|[1-9][0-9]*)" + r"(?:-(?:0|[1-9][0-9]*|[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*)" + r"(?:\.(?:0|[1-9][0-9]*|[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*))*)?" + r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$" +) +_CONSTRAINT_ATOL = 1e-12 +_CLOSURE_RTOL = 1e-10 +_CLOSURE_ATOL = 1e-12 + +__all__ = [ + "COMPUTATION_RECEIPT_SCHEMA_VERSION", + "CONSTRAINT_SET_SCHEMA_VERSION", + "FRESHNESS_POLICY_SCHEMA_VERSION", + "PORTFOLIO_DECISION_SCHEMA_VERSION", + "RISK_ASSESSMENT_SCHEMA_VERSION", + "ComputationReceipt", + "ConstraintSetV1", + "FreshnessPolicy", + "PortfolioDecision", + "PortfolioRiskContractError", + "PortfolioRiskContractErrorCode", + "ReceiptStatus", + "RiskAssessment", + "RiskAssessmentStatus", + "RiskFindingCode", + "assess_portfolio_risk", + "build_portfolio_decision", + "compute_portfolio_receipt_digests", +] + + +class PortfolioRiskContractErrorCode(StrEnum): + """Stable public rejection categories for the S4.1 contracts.""" + + TYPE_ERROR = "TYPE_ERROR" + MISSING_FIELD = "MISSING_FIELD" + UNKNOWN_FIELD = "UNKNOWN_FIELD" + INVALID_FORMAT = "INVALID_FORMAT" + INVALID_VALUE = "INVALID_VALUE" + UNSUPPORTED_SCHEMA = "UNSUPPORTED_SCHEMA" + INVALID_FRESHNESS_POLICY = "INVALID_FRESHNESS_POLICY" + QUALIFICATION_REJECTED = "QUALIFICATION_REJECTED" + IDENTITY_MISMATCH = "IDENTITY_MISMATCH" + DATASET_IDENTITY_MISMATCH = "DATASET_IDENTITY_MISMATCH" + TIME_ORDER_VIOLATION = "TIME_ORDER_VIOLATION" + MANIFEST_STALE = "MANIFEST_STALE" + COVARIANCE_STALE = "COVARIANCE_STALE" + CONSTRAINT_VIOLATION = "CONSTRAINT_VIOLATION" + RECEIPT_MISMATCH = "RECEIPT_MISMATCH" + COMPUTATION_FAILURE = "COMPUTATION_FAILURE" + + +class PortfolioRiskContractError(ValueError): + """Typed deterministic contract error with an exact JSON path.""" + + def __init__( + self, + code: PortfolioRiskContractErrorCode, + path: str, + detail: str, + ) -> None: + self.code = code + self.path = path + self.detail = detail + super().__init__(f"{code.value} at {path}: {detail}") + + +def _fail( + code: PortfolioRiskContractErrorCode, + path: str, + detail: str, +) -> Never: + raise PortfolioRiskContractError(code, path, detail) + + +def _strict_object( + value: object, + path: str, + fields: Sequence[str], + *, + invalid_code: PortfolioRiskContractErrorCode | None = None, +) -> dict[str, Any]: + error_code = invalid_code or PortfolioRiskContractErrorCode.TYPE_ERROR + if type(value) is not dict: + _fail(error_code, path, "must be an object") + raw = cast(dict[object, object], value) + if any(type(key) is not str for key in raw): + _fail(error_code, path, "object keys must be strings") + item = cast(dict[str, Any], value) + expected = set(fields) + missing = sorted(expected - set(item)) + if missing: + code = invalid_code or PortfolioRiskContractErrorCode.MISSING_FIELD + _fail(code, f"{path}.{missing[0]}", "field is required") + unknown = sorted(set(item) - expected) + if unknown: + code = invalid_code or PortfolioRiskContractErrorCode.UNKNOWN_FIELD + _fail(code, f"{path}.{unknown[0]}", "field is not permitted") + return item + + +def _text(value: object, path: str) -> str: + if type(value) is not str: + _fail(PortfolioRiskContractErrorCode.TYPE_ERROR, path, "must be a string") + normalized = value + try: + normalized.encode("utf-8") + except UnicodeEncodeError as error: + raise PortfolioRiskContractError( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + path, + "must be valid UTF-8 text", + ) from error + if not normalized or normalized != normalized.strip(): + _fail( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + path, + "must be non-empty canonical text", + ) + return normalized + + +def _semver(value: object, path: str) -> str: + version = _text(value, path) + if _SEMVER.fullmatch(version) is None: + _fail( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + path, + "must be a canonical semantic version", + ) + return version + + +def _digest(value: object, path: str) -> str: + digest = _text(value, path) + if _PREFIXED_SHA256.fullmatch(digest) is None: + _fail( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + path, + "must be a lowercase sha256 digest", + ) + return digest + + +def _finite_number( + value: object, + path: str, + *, + non_negative: bool = False, +) -> float: + if type(value) not in {int, float}: + _fail(PortfolioRiskContractErrorCode.TYPE_ERROR, path, "must be a number") + number = float(cast(int | float, value)) + if not math.isfinite(number): + _fail(PortfolioRiskContractErrorCode.INVALID_VALUE, path, "must be finite") + if non_negative and number < 0: + _fail(PortfolioRiskContractErrorCode.INVALID_VALUE, path, "must be non-negative") + return 0.0 if number == 0 else number + + +def _safe_integer(value: object, path: str, *, minimum: int = 0) -> int: + if type(value) is not int: + _fail(PortfolioRiskContractErrorCode.TYPE_ERROR, path, "must be an integer") + integer = value + if integer < minimum: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + path, + f"must be at least {minimum}", + ) + if integer > _MAX_SAFE_INTEGER: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + path, + "integer exceeds the canonical safe range", + ) + return integer + + +def _instant(value: object, path: str) -> tuple[str, datetime]: + if isinstance(value, datetime): + parsed = value + elif type(value) is str: + text = value + try: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError as error: + raise PortfolioRiskContractError( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + path, + "must be a timezone-aware instant", + ) from error + else: + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + path, + "must be a timezone-aware instant", + ) + if parsed.tzinfo is None or parsed.utcoffset() is None: + _fail( + PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION, + path, + "must be a timezone-aware instant", + ) + normalized = parsed.astimezone(UTC) + return normalized.isoformat().replace("+00:00", "Z"), normalized + + +def _validate_json_value(value: object, path: str = "$") -> object: + if value is None or type(value) in {bool, str}: + if type(value) is str: + _text(value, path) + return value + if type(value) is int: + integer = value + if abs(integer) > _MAX_SAFE_INTEGER: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + path, + "integer exceeds the canonical safe range", + ) + return integer + if type(value) is float: + return _finite_number(value, path) + if isinstance(value, Mapping): + normalized: dict[str, object] = {} + for key, item in value.items(): + if type(key) is not str: + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + f"{path}.keys", + "object keys must be strings", + ) + normalized[key] = _validate_json_value(item, f"{path}.{key}") + return dict(sorted(normalized.items())) + if isinstance(value, list | tuple): + return [ + _validate_json_value(item, f"{path}[{index}]") + for index, item in enumerate(value) + ] + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + path, + f"unsupported canonical value type: {type(value).__name__}", + ) + + +def _canonical_json(value: object) -> str: + return json.dumps( + _validate_json_value(value), + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + +def _payload_digest(value: object) -> str: + return f"sha256:{hashlib.sha256(_canonical_json(value).encode('utf-8')).hexdigest()}" + + +def _document_sha256(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def _immutable_float_mapping( + values: Mapping[str, float], + path: str, +) -> Mapping[str, float]: + if not isinstance(values, Mapping): + _fail(PortfolioRiskContractErrorCode.TYPE_ERROR, path, "must be an object") + normalized: dict[str, float] = {} + for raw_key, raw_value in values.items(): + key = _text(raw_key, f"{path}.keys") + if key in normalized: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + path, + "asset labels must be unique", + ) + normalized[key] = _finite_number(raw_value, f"{path}.{key}") + return MappingProxyType(dict(sorted(normalized.items()))) + + +def _mapping_dict(values: Mapping[str, float]) -> dict[str, float]: + return {key: values[key] for key in sorted(values)} + + +@dataclass(frozen=True, slots=True, init=False) +class FreshnessPolicy: + """Versioned maximum ages for manifest and covariance evidence.""" + + schema_version: str + max_manifest_age_seconds: int + max_covariance_age_days: int + + def __init__( + self, + *, + max_manifest_age_seconds: int, + max_covariance_age_days: int, + schema_version: str = FRESHNESS_POLICY_SCHEMA_VERSION, + ) -> None: + if schema_version != FRESHNESS_POLICY_SCHEMA_VERSION: + _fail( + PortfolioRiskContractErrorCode.INVALID_FRESHNESS_POLICY, + "$.schema_version", + "unsupported freshness policy schema version", + ) + + def strict_threshold(value: object, path: str) -> int: + if type(value) is not int: + _fail( + PortfolioRiskContractErrorCode.INVALID_FRESHNESS_POLICY, + path, + "must be a non-negative integer", + ) + threshold = value + if threshold < 0 or threshold > _MAX_SAFE_INTEGER: + _fail( + PortfolioRiskContractErrorCode.INVALID_FRESHNESS_POLICY, + path, + "must be a non-negative canonical safe integer", + ) + return threshold + + object.__setattr__(self, "schema_version", schema_version) + object.__setattr__( + self, + "max_manifest_age_seconds", + strict_threshold(max_manifest_age_seconds, "$.max_manifest_age_seconds"), + ) + object.__setattr__( + self, + "max_covariance_age_days", + strict_threshold(max_covariance_age_days, "$.max_covariance_age_days"), + ) + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": self.schema_version, + "max_manifest_age_seconds": self.max_manifest_age_seconds, + "max_covariance_age_days": self.max_covariance_age_days, + } + + def to_json(self) -> str: + return _canonical_json(self.to_dict()) + + @classmethod + def from_dict(cls, value: object) -> Self: + fields = ( + "schema_version", + "max_manifest_age_seconds", + "max_covariance_age_days", + ) + item = _strict_object( + value, + "$", + fields, + invalid_code=PortfolioRiskContractErrorCode.INVALID_FRESHNESS_POLICY, + ) + return cls( + schema_version=item["schema_version"], + max_manifest_age_seconds=item["max_manifest_age_seconds"], + max_covariance_age_days=item["max_covariance_age_days"], + ) + + +@dataclass(frozen=True, slots=True, init=False) +class ConstraintSetV1: + """Supported constraints that can be recomputed from target and prior weights.""" + + schema_version: str + gross_exposure_max: float | None + net_exposure_min: float | None + net_exposure_max: float | None + single_asset_min: float | None + single_asset_max: float | None + position_count_max: int | None + turnover_max: float | None + + def __init__( + self, + *, + gross_exposure_max: float | None = None, + net_exposure_min: float | None = None, + net_exposure_max: float | None = None, + single_asset_min: float | None = None, + single_asset_max: float | None = None, + position_count_max: int | None = None, + turnover_max: float | None = None, + schema_version: str = CONSTRAINT_SET_SCHEMA_VERSION, + ) -> None: + if schema_version != CONSTRAINT_SET_SCHEMA_VERSION: + _fail( + PortfolioRiskContractErrorCode.UNSUPPORTED_SCHEMA, + "$.schema_version", + "unsupported constraint-set schema version", + ) + + def optional_number( + value: object, + path: str, + *, + non_negative: bool = False, + ) -> float | None: + return ( + None + if value is None + else _finite_number(value, path, non_negative=non_negative) + ) + + gross = optional_number( + gross_exposure_max, + "$.gross_exposure_max", + non_negative=True, + ) + net_min = optional_number(net_exposure_min, "$.net_exposure_min") + net_max = optional_number(net_exposure_max, "$.net_exposure_max") + asset_min = optional_number(single_asset_min, "$.single_asset_min") + asset_max = optional_number(single_asset_max, "$.single_asset_max") + turnover = optional_number(turnover_max, "$.turnover_max", non_negative=True) + positions = ( + None + if position_count_max is None + else _safe_integer(position_count_max, "$.position_count_max") + ) + if net_min is not None and net_max is not None and net_min > net_max: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.net_exposure_min", + "must not exceed net_exposure_max", + ) + if asset_min is not None and asset_max is not None and asset_min > asset_max: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.single_asset_min", + "must not exceed single_asset_max", + ) + object.__setattr__(self, "schema_version", schema_version) + object.__setattr__(self, "gross_exposure_max", gross) + object.__setattr__(self, "net_exposure_min", net_min) + object.__setattr__(self, "net_exposure_max", net_max) + object.__setattr__(self, "single_asset_min", asset_min) + object.__setattr__(self, "single_asset_max", asset_max) + object.__setattr__(self, "position_count_max", positions) + object.__setattr__(self, "turnover_max", turnover) + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": self.schema_version, + "gross_exposure_max": self.gross_exposure_max, + "net_exposure_min": self.net_exposure_min, + "net_exposure_max": self.net_exposure_max, + "single_asset_min": self.single_asset_min, + "single_asset_max": self.single_asset_max, + "position_count_max": self.position_count_max, + "turnover_max": self.turnover_max, + } + + def to_json(self) -> str: + return _canonical_json(self.to_dict()) + + @classmethod + def from_dict(cls, value: object) -> Self: + fields = ( + "schema_version", + "gross_exposure_max", + "net_exposure_min", + "net_exposure_max", + "single_asset_min", + "single_asset_max", + "position_count_max", + "turnover_max", + ) + item = _strict_object(value, "$", fields) + return cls(**item) + + +class ReceiptStatus(StrEnum): + COMPLETED = "completed" + CONVERGED = "converged" + FAILED = "failed" + FALLBACK = "fallback" + + +@dataclass(frozen=True, slots=True, init=False) +class ComputationReceipt: + """Versioned producer receipt whose claims are independently recomputed.""" + + schema_version: str + algorithm: str + algorithm_version: str + implementation_digest: str + parameter_digest: str + input_digest: str + constraint_digest: str + output_digest: str + status: ReceiptStatus + solver_required: bool + solver_name: str | None + solver_version: str | None + solver_config_digest: str | None + iterations: int | None + objective_value: float | None + max_constraint_residual: float + tolerance: float + computed_at: str + + def __init__( + self, + *, + algorithm: str, + algorithm_version: str, + implementation_digest: str, + parameter_digest: str, + input_digest: str, + constraint_digest: str, + output_digest: str, + status: ReceiptStatus | str, + solver_required: bool, + solver_name: str | None, + solver_version: str | None, + solver_config_digest: str | None, + iterations: int | None, + objective_value: float | None, + max_constraint_residual: float, + tolerance: float, + computed_at: datetime | str, + schema_version: str = COMPUTATION_RECEIPT_SCHEMA_VERSION, + ) -> None: + if schema_version != COMPUTATION_RECEIPT_SCHEMA_VERSION: + _fail( + PortfolioRiskContractErrorCode.UNSUPPORTED_SCHEMA, + "$.schema_version", + "unsupported computation-receipt schema version", + ) + try: + normalized_status = ReceiptStatus(status) + except (TypeError, ValueError) as error: + raise PortfolioRiskContractError( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.status", + "unsupported receipt status", + ) from error + if type(solver_required) is not bool: + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.solver_required", + "must be a boolean", + ) + normalized_residual = _finite_number( + max_constraint_residual, + "$.max_constraint_residual", + non_negative=True, + ) + normalized_tolerance = _finite_number( + tolerance, + "$.tolerance", + non_negative=True, + ) + normalized_solver_name: str | None = None + normalized_solver_version: str | None = None + normalized_solver_digest: str | None = None + normalized_iterations: int | None = None + normalized_objective: float | None = None + if solver_required: + normalized_solver_name = _text(solver_name, "$.solver_name") + normalized_solver_version = _semver(solver_version, "$.solver_version") + normalized_solver_digest = _digest( + solver_config_digest, + "$.solver_config_digest", + ) + normalized_iterations = _safe_integer(iterations, "$.iterations", minimum=1) + normalized_objective = _finite_number(objective_value, "$.objective_value") + if normalized_status not in { + ReceiptStatus.CONVERGED, + ReceiptStatus.FAILED, + ReceiptStatus.FALLBACK, + }: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.status", + "solver receipt status must be converged, failed, or fallback", + ) + if ( + normalized_status is ReceiptStatus.CONVERGED + and normalized_residual > normalized_tolerance + ): + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.max_constraint_residual", + "converged solver residual exceeds tolerance", + ) + else: + if normalized_status is not ReceiptStatus.COMPLETED: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.status", + "non-solver receipt status must be completed", + ) + if any( + item is not None + for item in ( + solver_name, + solver_version, + solver_config_digest, + iterations, + objective_value, + ) + ): + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.solver_name", + "non-solver receipt cannot contain solver fields", + ) + if normalized_residual != 0: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.max_constraint_residual", + "non-solver receipt residual must be zero", + ) + normalized_computed_at, _ = _instant(computed_at, "$.computed_at") + object.__setattr__(self, "schema_version", schema_version) + object.__setattr__(self, "algorithm", _text(algorithm, "$.algorithm")) + object.__setattr__( + self, + "algorithm_version", + _semver(algorithm_version, "$.algorithm_version"), + ) + object.__setattr__( + self, + "implementation_digest", + _digest(implementation_digest, "$.implementation_digest"), + ) + object.__setattr__( + self, + "parameter_digest", + _digest(parameter_digest, "$.parameter_digest"), + ) + object.__setattr__(self, "input_digest", _digest(input_digest, "$.input_digest")) + object.__setattr__( + self, + "constraint_digest", + _digest(constraint_digest, "$.constraint_digest"), + ) + object.__setattr__(self, "output_digest", _digest(output_digest, "$.output_digest")) + object.__setattr__(self, "status", normalized_status) + object.__setattr__(self, "solver_required", solver_required) + object.__setattr__(self, "solver_name", normalized_solver_name) + object.__setattr__(self, "solver_version", normalized_solver_version) + object.__setattr__(self, "solver_config_digest", normalized_solver_digest) + object.__setattr__(self, "iterations", normalized_iterations) + object.__setattr__(self, "objective_value", normalized_objective) + object.__setattr__(self, "max_constraint_residual", normalized_residual) + object.__setattr__(self, "tolerance", normalized_tolerance) + object.__setattr__(self, "computed_at", normalized_computed_at) + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": self.schema_version, + "algorithm": self.algorithm, + "algorithm_version": self.algorithm_version, + "implementation_digest": self.implementation_digest, + "parameter_digest": self.parameter_digest, + "input_digest": self.input_digest, + "constraint_digest": self.constraint_digest, + "output_digest": self.output_digest, + "status": self.status.value, + "solver_required": self.solver_required, + "solver_name": self.solver_name, + "solver_version": self.solver_version, + "solver_config_digest": self.solver_config_digest, + "iterations": self.iterations, + "objective_value": self.objective_value, + "max_constraint_residual": self.max_constraint_residual, + "tolerance": self.tolerance, + "computed_at": self.computed_at, + } + + def to_json(self) -> str: + return _canonical_json(self.to_dict()) + + @classmethod + def from_dict(cls, value: object) -> Self: + fields = tuple(cls.__dataclass_fields__) + item = _strict_object(value, "$", fields) + rebuilt = cls(**item) + if rebuilt.to_dict() != item: + _fail( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + "$", + "receipt is not in canonical form", + ) + return rebuilt + + +def _target_payload(target: PortfolioTarget) -> dict[str, object]: + if not isinstance(target, PortfolioTarget): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.portfolio_target", + "PortfolioTarget is required", + ) + created_at, _ = _instant(target.created_at, "$.portfolio_target.created_at") + weights = _immutable_float_mapping(target.weights, "$.portfolio_target.weights") + return { + "target_id": _text(target.target_id, "$.portfolio_target.target_id"), + "backtest_run_id": _text( + target.backtest_run_id, + "$.portfolio_target.backtest_run_id", + ), + "dataset_snapshot_id": _text( + target.dataset_snapshot_id, + "$.portfolio_target.dataset_snapshot_id", + ), + "weights": _mapping_dict(weights), + "created_at": created_at, + } + + +def _constraint_metrics( + weights: Mapping[str, float], + prior_weights: Mapping[str, float] | None, +) -> dict[str, float | int | None]: + gross = float(sum(abs(value) for value in weights.values())) + net = float(sum(weights.values())) + position_count = sum(value != 0 for value in weights.values()) + turnover: float | None = None + if prior_weights is not None: + labels = set(weights) | set(prior_weights) + turnover = float( + sum(abs(weights.get(label, 0.0) - prior_weights.get(label, 0.0)) for label in labels) + ) + return { + "gross_exposure": gross, + "net_exposure": net, + "position_count": position_count, + "turnover_l1": turnover, + } + + +def _constraint_residuals( + constraints: ConstraintSetV1, + weights: Mapping[str, float], + metrics: Mapping[str, float | int | None], +) -> dict[str, float]: + residuals: dict[str, float] = {} + gross = cast(float, metrics["gross_exposure"]) + net = cast(float, metrics["net_exposure"]) + positions = cast(int, metrics["position_count"]) + turnover = cast(float | None, metrics["turnover_l1"]) + active_weights = [value for value in weights.values() if value != 0] + if constraints.gross_exposure_max is not None: + residuals["gross_exposure_max"] = max( + gross - constraints.gross_exposure_max, + 0.0, + ) + if constraints.net_exposure_min is not None: + residuals["net_exposure_min"] = max(constraints.net_exposure_min - net, 0.0) + if constraints.net_exposure_max is not None: + residuals["net_exposure_max"] = max(net - constraints.net_exposure_max, 0.0) + if constraints.single_asset_min is not None: + residuals["single_asset_min"] = max( + (constraints.single_asset_min - min(active_weights)) if active_weights else 0.0, + 0.0, + ) + if constraints.single_asset_max is not None: + residuals["single_asset_max"] = max( + (max(active_weights) - constraints.single_asset_max) if active_weights else 0.0, + 0.0, + ) + if constraints.position_count_max is not None: + residuals["position_count_max"] = float( + max(positions - constraints.position_count_max, 0) + ) + if constraints.turnover_max is not None: + if turnover is None: + _fail( + PortfolioRiskContractErrorCode.MISSING_FIELD, + "$.prior_weights", + "prior weights are required for a turnover constraint", + ) + residuals["turnover_max"] = max(turnover - constraints.turnover_max, 0.0) + return dict(sorted(residuals.items())) + + +def _receipt_material( + *, + backtest_run_ref: BacktestRunRef, + manifest: BacktestEvidenceManifest, + target: PortfolioTarget, + objective_name: str, + objective_version: str, + objective_digest: str, + model_name: str, + model_version: str, + model_digest: str, + expected_return_digest: str, + covariance_digest: str, + scenario_digest: str, + constraints: ConstraintSetV1, + freshness_policy: FreshnessPolicy, + prior_weights: Mapping[str, float] | None, +) -> tuple[dict[str, object], Mapping[str, float], dict[str, float | int | None], dict[str, float]]: + if not isinstance(backtest_run_ref, BacktestRunRef): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.backtest_run_ref", + "complete BacktestRunRef is required", + ) + if not isinstance(manifest, BacktestEvidenceManifest): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.manifest", + "complete BacktestEvidenceManifest is required", + ) + if not isinstance(constraints, ConstraintSetV1): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.constraints", + "ConstraintSetV1 is required", + ) + if not isinstance(freshness_policy, FreshnessPolicy): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.freshness_policy", + "FreshnessPolicy is required", + ) + target_payload = _target_payload(target) + weights = _immutable_float_mapping(target.weights, "$.portfolio_target.weights") + normalized_prior = ( + None + if prior_weights is None + else _immutable_float_mapping(prior_weights, "$.prior_weights") + ) + metrics = _constraint_metrics(weights, normalized_prior) + residuals = _constraint_residuals(constraints, weights, metrics) + run_document_sha256 = _document_sha256(backtest_run_ref.to_json()) + manifest_document_sha256 = _document_sha256(manifest.to_json()) + input_payload: dict[str, object] = { + "run_ref_document_sha256": run_document_sha256, + "manifest_document_sha256": manifest_document_sha256, + "portfolio_target": target_payload, + "objective": { + "name": _text(objective_name, "$.objective_name"), + "version": _semver(objective_version, "$.objective_version"), + "digest": _digest(objective_digest, "$.objective_digest"), + }, + "model": { + "name": _text(model_name, "$.model_name"), + "version": _semver(model_version, "$.model_version"), + "digest": _digest(model_digest, "$.model_digest"), + }, + "expected_return_digest": _digest( + expected_return_digest, + "$.expected_return_digest", + ), + "covariance_digest": _digest(covariance_digest, "$.covariance_digest"), + "scenario_digest": _digest(scenario_digest, "$.scenario_digest"), + "freshness_policy_digest": _payload_digest(freshness_policy.to_dict()), + "prior_weights": None if normalized_prior is None else _mapping_dict(normalized_prior), + } + return input_payload, weights, metrics, residuals + + +def compute_portfolio_receipt_digests( + *, + backtest_run_ref: BacktestRunRef, + manifest: BacktestEvidenceManifest, + target: PortfolioTarget, + objective_name: str, + objective_version: str, + objective_digest: str, + model_name: str, + model_version: str, + model_digest: str, + expected_return_digest: str, + covariance_digest: str, + scenario_digest: str, + constraints: ConstraintSetV1, + freshness_policy: FreshnessPolicy, + prior_weights: Mapping[str, float] | None = None, +) -> Mapping[str, str | float]: + """Return verifier-derived digests for a producer receipt.""" + + input_payload, weights, metrics, residuals = _receipt_material( + backtest_run_ref=backtest_run_ref, + manifest=manifest, + target=target, + objective_name=objective_name, + objective_version=objective_version, + objective_digest=objective_digest, + model_name=model_name, + model_version=model_version, + model_digest=model_digest, + expected_return_digest=expected_return_digest, + covariance_digest=covariance_digest, + scenario_digest=scenario_digest, + constraints=constraints, + freshness_policy=freshness_policy, + prior_weights=prior_weights, + ) + max_residual = max(residuals.values(), default=0.0) + values: dict[str, str | float] = { + "input_digest": _payload_digest(input_payload), + "constraint_digest": _payload_digest(constraints.to_dict()), + "output_digest": _payload_digest( + { + "weights": _mapping_dict(weights), + "metrics": dict(metrics), + "constraint_residuals": residuals, + } + ), + "max_constraint_residual": max_residual, + } + return MappingProxyType(values) + + +@dataclass(frozen=True, slots=True, init=False) +class PortfolioDecision: + """Immutable content-addressed evidence for one portfolio computation.""" + + contract_name: str + schema_version: str + decision_id: str + run_id: str + manifest_id: str + evidence_digest: str + dataset_snapshot_id: str + run_ref_document_sha256: str + manifest_document_sha256: str + source_universe_digest: str + portfolio_asset_set_digest: str + target_id: str + target_weights: Mapping[str, float] + prior_weights: Mapping[str, float] | None + objective_name: str + objective_version: str + objective_digest: str + model_name: str + model_version: str + model_digest: str + expected_return_digest: str + covariance_digest: str + scenario_digest: str + constraints: ConstraintSetV1 + freshness_policy: FreshnessPolicy + receipt: ComputationReceipt + gross_exposure: float + net_exposure: float + turnover_l1: float | None + position_count: int + constraint_residuals: Mapping[str, float] + output_digest: str + effective_at: str + computed_at: str + + def to_dict(self) -> dict[str, object]: + return { + "contract_name": self.contract_name, + "schema_version": self.schema_version, + "decision_id": self.decision_id, + "run_id": self.run_id, + "manifest_id": self.manifest_id, + "evidence_digest": self.evidence_digest, + "dataset_snapshot_id": self.dataset_snapshot_id, + "run_ref_document_sha256": self.run_ref_document_sha256, + "manifest_document_sha256": self.manifest_document_sha256, + "source_universe_digest": self.source_universe_digest, + "portfolio_asset_set_digest": self.portfolio_asset_set_digest, + "target_id": self.target_id, + "target_weights": _mapping_dict(self.target_weights), + "prior_weights": ( + None if self.prior_weights is None else _mapping_dict(self.prior_weights) + ), + "objective_name": self.objective_name, + "objective_version": self.objective_version, + "objective_digest": self.objective_digest, + "model_name": self.model_name, + "model_version": self.model_version, + "model_digest": self.model_digest, + "expected_return_digest": self.expected_return_digest, + "covariance_digest": self.covariance_digest, + "scenario_digest": self.scenario_digest, + "constraints": self.constraints.to_dict(), + "freshness_policy": self.freshness_policy.to_dict(), + "receipt": self.receipt.to_dict(), + "gross_exposure": self.gross_exposure, + "net_exposure": self.net_exposure, + "turnover_l1": self.turnover_l1, + "position_count": self.position_count, + "constraint_residuals": _mapping_dict(self.constraint_residuals), + "output_digest": self.output_digest, + "effective_at": self.effective_at, + "computed_at": self.computed_at, + } + + def to_json(self) -> str: + return _canonical_json(self.to_dict()) + + +def _close_s3_identity( + backtest_run_ref: BacktestRunRef, + manifest: BacktestEvidenceManifest, + target: PortfolioTarget, +) -> None: + if manifest.qualification is not EvidenceQualification.CONTRACT_QUALIFIED: + _fail( + PortfolioRiskContractErrorCode.QUALIFICATION_REJECTED, + "$.manifest.qualification", + "contract-qualified S3 evidence is required", + ) + if manifest.backtest_run_ref is None or manifest.legacy_backtest_run is not None: + _fail( + PortfolioRiskContractErrorCode.QUALIFICATION_REJECTED, + "$.manifest.run_reference.kind", + "non-legacy BacktestRunRef evidence is required", + ) + if manifest.backtest_run_ref.to_dict() != backtest_run_ref.to_dict(): + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.manifest.run_reference.value", + "embedded run reference differs from the supplied run reference", + ) + if not ( + manifest.run_id == backtest_run_ref.run_id == target.backtest_run_id + ): + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.portfolio_target.backtest_run_id", + "manifest, run reference, and target run identities must match", + ) + if target.dataset_snapshot_id != backtest_run_ref.dataset_snapshot_id: + _fail( + PortfolioRiskContractErrorCode.DATASET_IDENTITY_MISMATCH, + "$.portfolio_target.dataset_snapshot_id", + "target and run-reference dataset identities must match", + ) + + +def build_portfolio_decision( + *, + backtest_run_ref: BacktestRunRef, + manifest: BacktestEvidenceManifest, + target: PortfolioTarget, + objective_name: str, + objective_version: str, + objective_digest: str, + model_name: str, + model_version: str, + model_digest: str, + expected_return_digest: str, + covariance_digest: str, + scenario_digest: str, + constraints: ConstraintSetV1, + freshness_policy: FreshnessPolicy, + receipt: ComputationReceipt, + computed_at: datetime | str, + prior_weights: Mapping[str, float] | None = None, +) -> PortfolioDecision: + """Verify and bind a qualified S3-backed portfolio computation.""" + + if not isinstance(receipt, ComputationReceipt): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.receipt", + "ComputationReceipt is required", + ) + input_payload, weights, metrics, residuals = _receipt_material( + backtest_run_ref=backtest_run_ref, + manifest=manifest, + target=target, + objective_name=objective_name, + objective_version=objective_version, + objective_digest=objective_digest, + model_name=model_name, + model_version=model_version, + model_digest=model_digest, + expected_return_digest=expected_return_digest, + covariance_digest=covariance_digest, + scenario_digest=scenario_digest, + constraints=constraints, + freshness_policy=freshness_policy, + prior_weights=prior_weights, + ) + _close_s3_identity(backtest_run_ref, manifest, target) + computed_text, computed_time = _instant(computed_at, "$.computed_at") + effective_text, effective_time = _instant( + target.created_at, + "$.portfolio_target.created_at", + ) + _, evaluation_time = _instant( + backtest_run_ref.evaluation_at, + "$.backtest_run_ref.evaluation_at", + ) + _, run_computed_time = _instant( + backtest_run_ref.computed_at, + "$.backtest_run_ref.computed_at", + ) + _, artifact_time = _instant( + manifest.artifact_available_at, + "$.manifest.artifact_available_at", + ) + if evaluation_time > effective_time or effective_time > computed_time: + _fail( + PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION, + "$.portfolio_target.created_at", + "run evaluation, target creation, and decision computation are out of order", + ) + if run_computed_time > artifact_time or artifact_time > computed_time: + _fail( + PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION, + "$.manifest.artifact_available_at", + "run computation, artifact availability, and decision computation are out of order", + ) + if receipt.computed_at != computed_text: + _fail( + PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION, + "$.receipt.computed_at", + "receipt and decision computation times must match", + ) + manifest_age = (computed_time - artifact_time).total_seconds() + if manifest_age < 0: + _fail( + PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION, + "$.manifest.artifact_available_at", + "manifest availability cannot be in the future", + ) + if manifest_age > freshness_policy.max_manifest_age_seconds: + _fail( + PortfolioRiskContractErrorCode.MANIFEST_STALE, + "$.manifest.artifact_available_at", + "manifest age exceeds the freshness policy", + ) + if receipt.status in {ReceiptStatus.FAILED, ReceiptStatus.FALLBACK}: + _fail( + PortfolioRiskContractErrorCode.QUALIFICATION_REJECTED, + "$.receipt.status", + "failed or fallback computation cannot form a decision", + ) + expected_input_digest = _payload_digest(input_payload) + expected_constraint_digest = _payload_digest(constraints.to_dict()) + expected_output_digest = _payload_digest( + { + "weights": _mapping_dict(weights), + "metrics": dict(metrics), + "constraint_residuals": residuals, + } + ) + expected_residual = max(residuals.values(), default=0.0) + digest_checks = ( + (receipt.input_digest, expected_input_digest, "$.receipt.input_digest"), + ( + receipt.constraint_digest, + expected_constraint_digest, + "$.receipt.constraint_digest", + ), + (receipt.output_digest, expected_output_digest, "$.receipt.output_digest"), + ) + for actual, expected, path in digest_checks: + if actual != expected: + _fail( + PortfolioRiskContractErrorCode.RECEIPT_MISMATCH, + path, + "receipt digest differs from independently recomputed evidence", + ) + if receipt.max_constraint_residual != expected_residual: + _fail( + PortfolioRiskContractErrorCode.RECEIPT_MISMATCH, + "$.receipt.max_constraint_residual", + "receipt residual differs from independently recomputed constraints", + ) + if expected_residual > 0.0: + _fail( + PortfolioRiskContractErrorCode.CONSTRAINT_VIOLATION, + "$.constraints", + "target violates one or more supported constraints", + ) + normalized_prior = ( + None + if prior_weights is None + else _immutable_float_mapping(prior_weights, "$.prior_weights") + ) + run_document_sha256 = _document_sha256(backtest_run_ref.to_json()) + manifest_document_sha256 = _document_sha256(manifest.to_json()) + asset_set_digest = _payload_digest(sorted(weights)) + payload: dict[str, object] = { + "contract_name": "researchhub.portfolio-decision", + "schema_version": PORTFOLIO_DECISION_SCHEMA_VERSION, + "run_id": backtest_run_ref.run_id, + "manifest_id": manifest.manifest_id, + "evidence_digest": manifest.evidence_digest, + "dataset_snapshot_id": backtest_run_ref.dataset_snapshot_id, + "run_ref_document_sha256": run_document_sha256, + "manifest_document_sha256": manifest_document_sha256, + "source_universe_digest": backtest_run_ref.universe_digest, + "portfolio_asset_set_digest": asset_set_digest, + "target_id": target.target_id, + "target_weights": _mapping_dict(weights), + "prior_weights": None if normalized_prior is None else _mapping_dict(normalized_prior), + "objective_name": objective_name, + "objective_version": objective_version, + "objective_digest": objective_digest, + "model_name": model_name, + "model_version": model_version, + "model_digest": model_digest, + "expected_return_digest": expected_return_digest, + "covariance_digest": covariance_digest, + "scenario_digest": scenario_digest, + "constraints": constraints.to_dict(), + "freshness_policy": freshness_policy.to_dict(), + "receipt": receipt.to_dict(), + "gross_exposure": metrics["gross_exposure"], + "net_exposure": metrics["net_exposure"], + "turnover_l1": metrics["turnover_l1"], + "position_count": metrics["position_count"], + "constraint_residuals": residuals, + "output_digest": expected_output_digest, + "effective_at": effective_text, + "computed_at": computed_text, + } + decision_id = ( + "rhportfoliodecisionv1:sha256:" + f"{hashlib.sha256(_canonical_json(payload).encode('utf-8')).hexdigest()}" + ) + instance = object.__new__(PortfolioDecision) + values: dict[str, object] = { + **payload, + "decision_id": decision_id, + "target_weights": weights, + "prior_weights": normalized_prior, + "constraints": constraints, + "freshness_policy": freshness_policy, + "receipt": receipt, + "gross_exposure": cast(float, metrics["gross_exposure"]), + "net_exposure": cast(float, metrics["net_exposure"]), + "turnover_l1": cast(float | None, metrics["turnover_l1"]), + "position_count": cast(int, metrics["position_count"]), + "constraint_residuals": MappingProxyType(residuals), + } + for name, value in values.items(): + object.__setattr__(instance, name, value) + return instance + + +class RiskFindingCode(StrEnum): + COVARIANCE_NOT_PSD = "COVARIANCE_NOT_PSD" + PORTFOLIO_VARIANCE_NON_POSITIVE = "PORTFOLIO_VARIANCE_NON_POSITIVE" + RISK_CONTRIBUTION_NOT_CLOSED = "RISK_CONTRIBUTION_NOT_CLOSED" + RISK_BUDGET_BREACH = "RISK_BUDGET_BREACH" + + +class RiskAssessmentStatus(StrEnum): + READY = "ready" + UNAVAILABLE = "unavailable" + + +@dataclass(frozen=True, slots=True, init=False) +class RiskAssessment: + """Immutable risk result bound to one exact portfolio decision.""" + + contract_name: str + schema_version: str + assessment_id: str + decision_id: str + run_id: str + manifest_id: str + dataset_snapshot_id: str + covariance_data_snapshot_id: str + covariance_snapshot_id: str + covariance_as_of_date: str + covariance_input_digest: str + return_frequency: str + periods_per_year: int + risk_model_name: str + risk_model_version: str + risk_model_digest: str + freshness_policy_digest: str + scenario_digest: str + portfolio_volatility_limit: float | None + risk_budget: Mapping[str, float] + marginal_risk: Mapping[str, float] + component_risk: Mapping[str, float] + percentage_risk: Mapping[str, float] + portfolio_volatility: float | None + group_exposure: Mapping[str, float] + findings: tuple[RiskFindingCode, ...] + status: RiskAssessmentStatus + qualified: bool + + def to_dict(self) -> dict[str, object]: + return { + "contract_name": self.contract_name, + "schema_version": self.schema_version, + "assessment_id": self.assessment_id, + "decision_id": self.decision_id, + "run_id": self.run_id, + "manifest_id": self.manifest_id, + "dataset_snapshot_id": self.dataset_snapshot_id, + "covariance_data_snapshot_id": self.covariance_data_snapshot_id, + "covariance_snapshot_id": self.covariance_snapshot_id, + "covariance_as_of_date": self.covariance_as_of_date, + "covariance_input_digest": self.covariance_input_digest, + "return_frequency": self.return_frequency, + "periods_per_year": self.periods_per_year, + "risk_model_name": self.risk_model_name, + "risk_model_version": self.risk_model_version, + "risk_model_digest": self.risk_model_digest, + "freshness_policy_digest": self.freshness_policy_digest, + "scenario_digest": self.scenario_digest, + "portfolio_volatility_limit": self.portfolio_volatility_limit, + "risk_budget": _mapping_dict(self.risk_budget), + "marginal_risk": _mapping_dict(self.marginal_risk), + "component_risk": _mapping_dict(self.component_risk), + "percentage_risk": _mapping_dict(self.percentage_risk), + "portfolio_volatility": self.portfolio_volatility, + "group_exposure": _mapping_dict(self.group_exposure), + "findings": [finding.value for finding in self.findings], + "status": self.status.value, + "qualified": self.qualified, + } + + def to_json(self) -> str: + return _canonical_json(self.to_dict()) + + +def _validate_covariance_structure( + weights: Mapping[str, float], + covariance: CovarianceSnapshot, +) -> pd.DataFrame: + frame = covariance.covariance + if not isinstance(frame, pd.DataFrame): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.covariance.matrix", + "must be a DataFrame", + ) + if not frame.index.is_unique or not frame.columns.is_unique: + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.covariance.labels", + "row and column labels must be unique", + ) + if any(type(label) is not str or not label.strip() for label in frame.index): + _fail( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + "$.covariance.index", + "asset labels must be non-empty strings", + ) + if any(type(label) is not str or not label.strip() for label in frame.columns): + _fail( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + "$.covariance.columns", + "asset labels must be non-empty strings", + ) + expected = set(weights) + if set(cast(Sequence[str], frame.index.tolist())) != expected: + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.covariance.index", + "covariance and target weights must use the same asset labels", + ) + if set(cast(Sequence[str], frame.columns.tolist())) != expected: + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.covariance.columns", + "covariance and target weights must use the same asset labels", + ) + aligned = frame.reindex(index=sorted(weights), columns=sorted(weights)).astype(float).copy() + values = aligned.to_numpy(dtype=float, copy=True) + if not np.isfinite(values).all(): + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.covariance.matrix", + "covariance values must be finite", + ) + if not np.allclose(values, values.T, rtol=_CLOSURE_RTOL, atol=_CLOSURE_ATOL): + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.covariance.matrix", + "covariance must be symmetric", + ) + return aligned + + +def _series_mapping(series: pd.Series[Any]) -> Mapping[str, float]: + values = { + cast(str, label): _finite_number(value, "$.risk_output") + for label, value in series.items() + } + return MappingProxyType(dict(sorted(values.items()))) + + +def _make_assessment( + *, + decision: PortfolioDecision, + covariance: CovarianceSnapshot, + risk_model_name: str, + risk_model_version: str, + risk_model_digest: str, + portfolio_volatility_limit: float | None, + risk_budget: Mapping[str, float], + marginal_risk: Mapping[str, float], + component_risk: Mapping[str, float], + percentage_risk: Mapping[str, float], + portfolio_volatility: float | None, + group_exposure: Mapping[str, float], + findings: tuple[RiskFindingCode, ...], + status: RiskAssessmentStatus, + qualified: bool, +) -> RiskAssessment: + payload: dict[str, object] = { + "contract_name": "researchhub.risk-assessment", + "schema_version": RISK_ASSESSMENT_SCHEMA_VERSION, + "decision_id": decision.decision_id, + "run_id": decision.run_id, + "manifest_id": decision.manifest_id, + "dataset_snapshot_id": decision.dataset_snapshot_id, + "covariance_data_snapshot_id": covariance.data_snapshot_id, + "covariance_snapshot_id": covariance.snapshot_id, + "covariance_as_of_date": covariance.as_of_date.isoformat(), + "covariance_input_digest": f"sha256:{covariance.input_sha256}", + "return_frequency": covariance.return_frequency, + "periods_per_year": covariance.periods_per_year, + "risk_model_name": risk_model_name, + "risk_model_version": risk_model_version, + "risk_model_digest": risk_model_digest, + "freshness_policy_digest": _payload_digest(decision.freshness_policy.to_dict()), + "scenario_digest": decision.scenario_digest, + "portfolio_volatility_limit": portfolio_volatility_limit, + "risk_budget": _mapping_dict(risk_budget), + "marginal_risk": _mapping_dict(marginal_risk), + "component_risk": _mapping_dict(component_risk), + "percentage_risk": _mapping_dict(percentage_risk), + "portfolio_volatility": portfolio_volatility, + "group_exposure": _mapping_dict(group_exposure), + "findings": [finding.value for finding in findings], + "status": status.value, + "qualified": qualified, + } + assessment_id = ( + "rhriskassessmentv1:sha256:" + f"{hashlib.sha256(_canonical_json(payload).encode('utf-8')).hexdigest()}" + ) + instance = object.__new__(RiskAssessment) + values: dict[str, object] = { + **payload, + "assessment_id": assessment_id, + "risk_budget": risk_budget, + "marginal_risk": marginal_risk, + "component_risk": component_risk, + "percentage_risk": percentage_risk, + "group_exposure": group_exposure, + "findings": findings, + "status": status, + } + for name, value in values.items(): + object.__setattr__(instance, name, value) + return instance + + +def assess_portfolio_risk( + *, + portfolio_decision: PortfolioDecision, + backtest_run_ref: BacktestRunRef, + manifest: BacktestEvidenceManifest, + covariance: CovarianceSnapshot, + risk_model_name: str, + risk_model_version: str, + risk_model_digest: str, + risk_budget: Mapping[str, float] | None = None, + portfolio_volatility_limit: float | None = None, + groups: Mapping[str, str] | None = None, +) -> RiskAssessment: + """Delegate one label-safe decomposition and qualify the resulting evidence.""" + + if not isinstance(portfolio_decision, PortfolioDecision): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.portfolio_decision", + "PortfolioDecision is required", + ) + if not isinstance(backtest_run_ref, BacktestRunRef): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.backtest_run_ref", + "BacktestRunRef is required", + ) + if not isinstance(manifest, BacktestEvidenceManifest): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.manifest", + "BacktestEvidenceManifest is required", + ) + if not isinstance(covariance, CovarianceSnapshot): + _fail( + PortfolioRiskContractErrorCode.TYPE_ERROR, + "$.covariance", + "CovarianceSnapshot is required", + ) + _close_s3_identity( + backtest_run_ref, + manifest, + _decision_target(portfolio_decision), + ) + if _document_sha256(backtest_run_ref.to_json()) != portfolio_decision.run_ref_document_sha256: + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.backtest_run_ref", + "run-reference document differs from the portfolio decision", + ) + if _document_sha256(manifest.to_json()) != portfolio_decision.manifest_document_sha256: + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.manifest", + "manifest document differs from the portfolio decision", + ) + if not ( + covariance.data_snapshot_id + == portfolio_decision.dataset_snapshot_id + == backtest_run_ref.dataset_snapshot_id + ): + _fail( + PortfolioRiskContractErrorCode.DATASET_IDENTITY_MISMATCH, + "$.covariance.data_snapshot_id", + "covariance, target decision, and run-reference datasets must match", + ) + if covariance.window_end_date is None: + _fail( + PortfolioRiskContractErrorCode.UNSUPPORTED_SCHEMA, + "$.covariance.window_end_date", + "a bounded covariance window is required", + ) + if covariance.window_end_date > covariance.as_of_date: + _fail( + PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION, + "$.covariance.window_end_date", + "covariance window cannot end after its as-of date", + ) + _, target_time = _instant(portfolio_decision.effective_at, "$.portfolio_decision.effective_at") + if covariance.as_of_date > target_time.date(): + _fail( + PortfolioRiskContractErrorCode.TIME_ORDER_VIOLATION, + "$.covariance.as_of_date", + "covariance as-of date cannot follow the target date", + ) + covariance_age = (target_time.date() - covariance.as_of_date).days + if covariance_age > portfolio_decision.freshness_policy.max_covariance_age_days: + _fail( + PortfolioRiskContractErrorCode.COVARIANCE_STALE, + "$.covariance.as_of_date", + "covariance age exceeds the freshness policy", + ) + if _SHA256.fullmatch(covariance.input_sha256) is None: + _fail( + PortfolioRiskContractErrorCode.INVALID_FORMAT, + "$.covariance.input_sha256", + "a lowercase SHA-256 input digest is required", + ) + if portfolio_decision.covariance_digest != f"sha256:{covariance.input_sha256}": + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.covariance.input_sha256", + "covariance input digest differs from the portfolio decision", + ) + normalized_name = _text(risk_model_name, "$.risk_model_name") + normalized_version = _semver(risk_model_version, "$.risk_model_version") + normalized_digest = _digest(risk_model_digest, "$.risk_model_digest") + normalized_limit = ( + None + if portfolio_volatility_limit is None + else _finite_number( + portfolio_volatility_limit, + "$.portfolio_volatility_limit", + non_negative=True, + ) + ) + normalized_budget = ( + MappingProxyType({}) + if risk_budget is None + else _immutable_float_mapping(risk_budget, "$.risk_budget") + ) + if any(value < 0 for value in normalized_budget.values()): + _fail( + PortfolioRiskContractErrorCode.INVALID_VALUE, + "$.risk_budget", + "risk budget limits must be non-negative", + ) + if not set(normalized_budget).issubset(portfolio_decision.target_weights): + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.risk_budget", + "risk budget labels must be target asset labels", + ) + group_labels: Mapping[str, str] | None = None + if groups is not None: + if not isinstance(groups, Mapping): + _fail(PortfolioRiskContractErrorCode.TYPE_ERROR, "$.groups", "must be an object") + normalized_groups = { + _text(key, "$.groups.keys"): _text(value, f"$.groups.{key}") + for key, value in groups.items() + } + if set(normalized_groups) != set(portfolio_decision.target_weights): + _fail( + PortfolioRiskContractErrorCode.IDENTITY_MISMATCH, + "$.groups", + "groups must label every target asset exactly once", + ) + group_labels = MappingProxyType(dict(sorted(normalized_groups.items()))) + aligned_covariance = _validate_covariance_structure( + portfolio_decision.target_weights, + covariance, + ) + weights = pd.Series( + _mapping_dict(portfolio_decision.target_weights), + dtype=float, + name="weight", + ) + annualized_covariance = aligned_covariance * covariance.periods_per_year + try: + decomposition = labeled_component_risk(weights, annualized_covariance) + except ValueError as error: + message = str(error) + mapped = { + "covariance must be positive semidefinite": RiskFindingCode.COVARIANCE_NOT_PSD, + "weights and covariance must produce positive portfolio variance": ( + RiskFindingCode.PORTFOLIO_VARIANCE_NON_POSITIVE + ), + }.get(message) + if mapped is None: + raise PortfolioRiskContractError( + PortfolioRiskContractErrorCode.COMPUTATION_FAILURE, + "$.covariance", + "risk computation failed", + ) from error + empty: Mapping[str, float] = MappingProxyType({}) + return _make_assessment( + decision=portfolio_decision, + covariance=covariance, + risk_model_name=normalized_name, + risk_model_version=normalized_version, + risk_model_digest=normalized_digest, + portfolio_volatility_limit=normalized_limit, + risk_budget=normalized_budget, + marginal_risk=empty, + component_risk=empty, + percentage_risk=empty, + portfolio_volatility=None, + group_exposure=empty, + findings=(mapped,), + status=RiskAssessmentStatus.UNAVAILABLE, + qualified=False, + ) + marginal = _series_mapping(decomposition.marginal) + component = _series_mapping(decomposition.component) + percentage = _series_mapping(decomposition.percentage) + volatility = _finite_number( + decomposition.portfolio_volatility, + "$.risk_output.portfolio_volatility", + non_negative=True, + ) + output_labels_closed = ( + set(marginal) + == set(component) + == set(percentage) + == set(portfolio_decision.target_weights) + ) + if not output_labels_closed or not math.isclose( + sum(component.values()), + volatility, + rel_tol=_CLOSURE_RTOL, + abs_tol=_CLOSURE_ATOL, + ) or not math.isclose( + sum(percentage.values()), + 1.0, + rel_tol=_CLOSURE_RTOL, + abs_tol=_CLOSURE_ATOL, + ): + empty = MappingProxyType({}) + return _make_assessment( + decision=portfolio_decision, + covariance=covariance, + risk_model_name=normalized_name, + risk_model_version=normalized_version, + risk_model_digest=normalized_digest, + portfolio_volatility_limit=normalized_limit, + risk_budget=normalized_budget, + marginal_risk=empty, + component_risk=empty, + percentage_risk=empty, + portfolio_volatility=None, + group_exposure=empty, + findings=(RiskFindingCode.RISK_CONTRIBUTION_NOT_CLOSED,), + status=RiskAssessmentStatus.UNAVAILABLE, + qualified=False, + ) + grouped: Mapping[str, float] = MappingProxyType({}) + if group_labels is not None: + group_series = pd.Series(dict(group_labels), dtype="object") + grouped = _series_mapping(decomposition.grouped_component(group_series)) + budget_breach = ( + normalized_limit is not None and volatility > normalized_limit + _CLOSURE_ATOL + ) or any( + percentage[label] > limit + _CLOSURE_ATOL + for label, limit in normalized_budget.items() + ) + findings = ( + (RiskFindingCode.RISK_BUDGET_BREACH,) + if budget_breach + else () + ) + return _make_assessment( + decision=portfolio_decision, + covariance=covariance, + risk_model_name=normalized_name, + risk_model_version=normalized_version, + risk_model_digest=normalized_digest, + portfolio_volatility_limit=normalized_limit, + risk_budget=normalized_budget, + marginal_risk=marginal, + component_risk=component, + percentage_risk=percentage, + portfolio_volatility=volatility, + group_exposure=grouped, + findings=findings, + status=RiskAssessmentStatus.READY, + qualified=not budget_breach, + ) + + +def _decision_target(decision: PortfolioDecision) -> PortfolioTarget: + """Reconstruct only the existing target fields for S3 closure revalidation.""" + + _, created_at = _instant(decision.effective_at, "$.portfolio_decision.effective_at") + return PortfolioTarget( + target_id=decision.target_id, + backtest_run_id=decision.run_id, + dataset_snapshot_id=decision.dataset_snapshot_id, + weights=decision.target_weights, + created_at=created_at, + ) diff --git a/tests/fixtures/portfolio-risk-computation-v1.golden.json b/tests/fixtures/portfolio-risk-computation-v1.golden.json new file mode 100644 index 0000000..d33b1c3 --- /dev/null +++ b/tests/fixtures/portfolio-risk-computation-v1.golden.json @@ -0,0 +1,129 @@ +{ + "portfolio_decision": { + "computed_at": "2026-01-08T03:01:00Z", + "constraint_residuals": { + "gross_exposure_max": 0.0, + "net_exposure_max": 0.0, + "net_exposure_min": 0.0, + "position_count_max": 0.0, + "single_asset_max": 0.0, + "single_asset_min": 0.0, + "turnover_max": 0.0 + }, + "constraints": { + "gross_exposure_max": 1.0, + "net_exposure_max": 1.0, + "net_exposure_min": 1.0, + "position_count_max": 2, + "schema_version": "1.0.0", + "single_asset_max": 0.7, + "single_asset_min": 0.2, + "turnover_max": 0.2 + }, + "contract_name": "researchhub.portfolio-decision", + "covariance_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "dataset_snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57", + "decision_id": "rhportfoliodecisionv1:sha256:0e6e5ce2fa08de8006cc392610695a013327fc80645bfa4d7a908ad3f615bebd", + "effective_at": "2026-01-08T03:00:00Z", + "evidence_digest": "sha256:f3913894d032c389c64eef59058b3cbc694cb9cc14ce9cee2699f0068200b650", + "expected_return_digest": "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd", + "freshness_policy": { + "max_covariance_age_days": 0, + "max_manifest_age_seconds": 3600, + "schema_version": "1.0.0" + }, + "gross_exposure": 1.0, + "manifest_document_sha256": "fbf54218f770528978f9ccd35577e1ab00877143ea397a576e071e75f0afbab0", + "manifest_id": "rhbacktestevidencev1:sha256:681c49cbdfb3e221b273e7bc616602ad80a7ab01206970807ad4294b176ebc75", + "model_digest": "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", + "model_name": "deterministic_weights", + "model_version": "1.0.0", + "net_exposure": 1.0, + "objective_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "objective_name": "long_only_allocation", + "objective_version": "1.0.0", + "output_digest": "sha256:bd3b964c628c8648322d036e57dd6f444ca287017d1578bab3689d07d32b28ce", + "portfolio_asset_set_digest": "sha256:b64e3448a83a5b86466465080361c1a7e1157a27ddccd4b68069cb18caffb74a", + "position_count": 2, + "prior_weights": { + "A": 0.5, + "B": 0.5 + }, + "receipt": { + "algorithm": "bounded_allocation", + "algorithm_version": "1.0.0", + "computed_at": "2026-01-08T03:01:00Z", + "constraint_digest": "sha256:34df0e5c00f503748ff936f9cd415a8947169181f8ca0917bce97dd606e08e94", + "implementation_digest": "sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff", + "input_digest": "sha256:ebd8ff957115e1adfd84eafa5cea49470356194d747b64ee5897858f9dd067b7", + "iterations": null, + "max_constraint_residual": 0.0, + "objective_value": null, + "output_digest": "sha256:bd3b964c628c8648322d036e57dd6f444ca287017d1578bab3689d07d32b28ce", + "parameter_digest": "sha256:0000000000000000000000000000000000000000000000000000000000000000", + "schema_version": "1.0.0", + "solver_config_digest": null, + "solver_name": null, + "solver_required": false, + "solver_version": null, + "status": "completed", + "tolerance": 1e-12 + }, + "run_id": "rhbacktestrunv1:sha256:5036c771c44a2adade9ea590eee0d8cf824ff8f519fadd8ba424e5d914856386", + "run_ref_document_sha256": "6a798adb3e0568aca84ed3e3a285b92d181ec569462fc003952a8c0d8382ae3a", + "scenario_digest": "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", + "schema_version": "1.0.0", + "source_universe_digest": "sha256:5555555555555555555555555555555555555555555555555555555555555555", + "target_id": "portfolio-target:synthetic-v1", + "target_weights": { + "A": 0.6, + "B": 0.4 + }, + "turnover_l1": 0.19999999999999996 + }, + "risk_assessment": { + "assessment_id": "rhriskassessmentv1:sha256:dbc38825cffcf6d95bd0216d22dbba4e0d4a1359ca5924a3ee529b99a7d78b6d", + "component_risk": { + "A": 1.4549226783578566, + "B": 1.4549226783578568 + }, + "contract_name": "researchhub.risk-assessment", + "covariance_as_of_date": "2026-01-08", + "covariance_data_snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57", + "covariance_input_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "covariance_snapshot_id": "covariance:synthetic-v1", + "dataset_snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57", + "decision_id": "rhportfoliodecisionv1:sha256:0e6e5ce2fa08de8006cc392610695a013327fc80645bfa4d7a908ad3f615bebd", + "findings": [], + "freshness_policy_digest": "sha256:833f58f4d1056f0450f7369fd4edd70534cc74e9e55cd60f05b2b3d7a2979763", + "group_exposure": { + "equity": 1.4549226783578566, + "fixed_income": 1.4549226783578568 + }, + "manifest_id": "rhbacktestevidencev1:sha256:681c49cbdfb3e221b273e7bc616602ad80a7ab01206970807ad4294b176ebc75", + "marginal_risk": { + "A": 2.424871130596428, + "B": 3.637306695894642 + }, + "percentage_risk": { + "A": 0.49999999999999983, + "B": 0.49999999999999994 + }, + "periods_per_year": 252, + "portfolio_volatility": 2.909845356715714, + "portfolio_volatility_limit": 10.0, + "qualified": true, + "return_frequency": "1d", + "risk_budget": { + "A": 0.8, + "B": 0.8 + }, + "risk_model_digest": "sha256:2222222222222222222222222222222222222222222222222222222222222222", + "risk_model_name": "euler_volatility", + "risk_model_version": "1.0.0", + "run_id": "rhbacktestrunv1:sha256:5036c771c44a2adade9ea590eee0d8cf824ff8f519fadd8ba424e5d914856386", + "scenario_digest": "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee", + "schema_version": "1.0.0", + "status": "ready" + } +} diff --git a/tests/governance/test_module_spec.py b/tests/governance/test_module_spec.py index b98de24..6d0a088 100644 --- a/tests/governance/test_module_spec.py +++ b/tests/governance/test_module_spec.py @@ -16,7 +16,7 @@ def test_module_spec_declares_pure_research_engine_boundary() -> None: prohibited = " ".join(spec["bounded_context"]["prohibited_responsibilities"]).lower() for term in ("investment advice", "live order", "credentials", "source facts"): assert term in prohibited - assert spec["authority"]["revision"] == 3 + assert spec["authority"]["revision"] == 4 assert { (item["contract_id"], item["version"]) for item in spec["contracts"]["provides"] @@ -25,12 +25,16 @@ def test_module_spec_declares_pure_research_engine_boundary() -> None: ("researchhub.factor-set-ref", "1.0.0"), ("researchhub.backtest-run-ref", "1.0.0"), ("researchhub.backtest-evidence-manifest", "1.0.0"), + ("researchhub.portfolio-decision", "1.0.0"), + ("researchhub.risk-assessment", "1.0.0"), } expected_paths = { "researchhub.factor-definition": "src/quant_engine/factor_contracts.py", "researchhub.factor-set-ref": "src/quant_engine/factor_contracts.py", "researchhub.backtest-run-ref": "src/quant_engine/governed_pipeline.py", "researchhub.backtest-evidence-manifest": "src/quant_engine/artifact.py", + "researchhub.portfolio-decision": "src/quant_engine/portfolio_risk_contracts.py", + "researchhub.risk-assessment": "src/quant_engine/portfolio_risk_contracts.py", } assert all(item["authority"] == "quant_engine" for item in spec["contracts"]["provides"]) assert { @@ -48,6 +52,14 @@ def test_module_spec_declares_pure_research_engine_boundary() -> None: for item in spec["contracts"]["consumes"] ) assert spec["dependencies"] == [] + capabilities = {item["id"]: item for item in spec["capabilities"]} + portfolio_contract = capabilities["portfolio-risk-computation-contracts"] + assert portfolio_contract["status"] == "operational" + summary = portfolio_contract["summary"].lower() + for term in ("receipt", "portfolio decisions", "risk assessments", "without"): + assert term in summary + for term in ("approval", "maker-checker", "publication", "paper", "live"): + assert term in prohibited assert all( command["required"] and not command["network"] for command in spec["verification"]["commands"] diff --git a/tests/test_portfolio_risk_contracts.py b/tests/test_portfolio_risk_contracts.py new file mode 100644 index 0000000..bd35aff --- /dev/null +++ b/tests/test_portfolio_risk_contracts.py @@ -0,0 +1,1004 @@ +"""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() + 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/artifact.py": "e15feec412d3bfff10d8f21ca20813fc65cabc0703940371ed661896147bc379", + "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