"""New synthetic S4 evidence; historical valuation is not actual availability.""" from __future__ import annotations import hashlib import json from dataclasses import FrozenInstanceError, replace from typing import Any import pandas as pd import pytest from quant_engine.portfolio_risk_contracts import ( ComputationReceipt, ConstraintSetV1, FreshnessPolicy, PortfolioRiskContractError, RiskAssessmentStatus, RiskFindingCode, ) from quant_engine.artifact import EvidenceQualification, PerformanceEvidenceError from quant_engine.factor_contracts import FactorContractError from quant_engine.governed_pipeline import BacktestContractError from quant_engine.risk import ComponentRiskResult, CovarianceSnapshot, labeled_component_risk import quant_engine.retrospective_portfolio_risk_contracts as contracts from quant_engine.retrospective_artifact_contracts import ( build_retrospective_backtest_evidence_manifest, ) from quant_engine.retrospective_backtest_contracts import RetrospectiveBacktestRunRef from quant_engine.retrospective_portfolio_risk_contracts import ( RetrospectivePortfolioDecision, RetrospectivePortfolioTarget, RetrospectiveRiskAssessment, build_retrospective_portfolio_decision, compute_retrospective_portfolio_receipt_digests, assess_retrospective_portfolio_risk, ) from test_retrospective_artifact_contracts import synthetic_artifact from test_retrospective_backtest_contracts import run_arguments from test_retrospective_data_contracts import digest, replace_at ASSETS = ("rhinstrument:" + "1" * 32, "rhinstrument:" + "2" * 32) CONTRACT_ERRORS = ( FactorContractError, PortfolioRiskContractError, BacktestContractError, PerformanceEvidenceError, ) def portfolio_arguments() -> dict[str, Any]: run = RetrospectiveBacktestRunRef.create(**run_arguments()) artifact = synthetic_artifact(run) manifest = build_retrospective_backtest_evidence_manifest( run, artifact, artifact_available_at="2026-09-08T01:11:00Z" ) target = RetrospectivePortfolioTarget.create( backtest_run_id=run.run_id, dataset_snapshot_id=run.dataset_snapshot_id, weights={ASSETS[0]: 0.6, ASSETS[1]: 0.4}, effective_at="2018-01-05T07:00:00Z", created_at="2026-09-08T01:12:00Z", ) return { "backtest_run_ref": run, "manifest": manifest, "target": target, "objective_name": "synthetic_allocation", "objective_version": "1.0.0", "objective_digest": digest({"synthetic_objective": 1}), "model_name": "bounded_weights", "model_version": "1.0.0", "model_digest": digest({"synthetic_model": 1}), "expected_return_digest": digest({"synthetic_returns": 1}), "covariance_digest": "sha256:" + "a" * 64, "scenario_digest": digest({"synthetic_scenario": 1}), "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=3600, max_covariance_age_days=0 ), "prior_weights": {ASSETS[0]: 0.5, ASSETS[1]: 0.5}, "computed_at": "2026-09-08T01:13:00Z", } def portfolio_receipt(arguments: dict[str, Any], **changes: Any) -> ComputationReceipt: values = compute_retrospective_portfolio_receipt_digests( **{key: value for key, value in arguments.items() if key not in {"computed_at", "receipt"}} ) return ComputationReceipt( **{ "algorithm": "bounded_weights", "algorithm_version": "1.0.0", "implementation_digest": digest({"synthetic_implementation": 1}), "parameter_digest": digest({"synthetic_parameters": 1}), "input_digest": values["input_digest"], "constraint_digest": values["constraint_digest"], "output_digest": values["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": values["max_constraint_residual"], "tolerance": 1e-12, "computed_at": arguments["computed_at"], **changes, } ) def test_target_separates_historical_effective_time_from_actual_creation() -> None: arguments = portfolio_arguments() target = arguments["target"] assert target.effective_at == "2018-01-05T07:00:00Z" assert target.created_at == "2026-09-08T01:12:00Z" assert target.target_id.startswith("rhportfoliotargetv2:sha256:") assert target.to_dict()["usage"] == "retrospective_research" def test_portfolio_decision_preserves_constraints_and_actual_receipt_time() -> None: arguments = portfolio_arguments() decision = build_retrospective_portfolio_decision( **arguments, receipt=portfolio_receipt(arguments) ) assert decision.decision_id.startswith("rhportfoliodecisionv2:sha256:") assert decision.effective_at == "2018-01-05T07:00:00Z" assert decision.created_at == "2026-09-08T01:12:00Z" assert decision.computed_at == "2026-09-08T01:13:00Z" assert decision.gross_exposure == 1.0 assert decision.position_count == 2 assert decision.to_dict()["decision_eligible"] is False def covariance(arguments: dict[str, Any], **changes: Any) -> CovarianceSnapshot: return CovarianceSnapshot( **{ "snapshot_id": "covariance:synthetic-retrospective", "as_of_date": "2018-01-05", "covariance": pd.DataFrame([[0.04, 0.01], [0.01, 0.09]], index=ASSETS, columns=ASSETS), "return_frequency": "1d", "periods_per_year": 252, "method": "provided", "window_start_date": "2018-01-02", "window_end_date": "2018-01-05", "observations": 4, "lookback_sessions": 4, "missing_policy": "complete_case", "data_snapshot_id": arguments["backtest_run_ref"].dataset_snapshot_id, "input_sha256": "a" * 64, **changes, } ) def risk_arguments(arguments: dict[str, Any]) -> dict[str, Any]: decision = build_retrospective_portfolio_decision( **arguments, receipt=portfolio_receipt(arguments) ) return { "portfolio_decision": decision, "backtest_run_ref": arguments["backtest_run_ref"], "manifest": arguments["manifest"], "covariance": covariance(arguments), "risk_model_name": "euler_volatility", "risk_model_version": "1.0.0", "risk_model_digest": digest({"synthetic_risk_model": 1}), "risk_budget": {ASSETS[0]: 0.8, ASSETS[1]: 0.8}, "portfolio_volatility_limit": 10.0, "groups": {ASSETS[0]: "equity", ASSETS[1]: "fixed_income"}, "computed_at": "2026-09-08T01:14:00Z", } def test_risk_uses_historical_business_age_and_actual_computation_time() -> None: arguments = risk_arguments(portfolio_arguments()) result = assess_retrospective_portfolio_risk(**arguments) assert result.assessment_id.startswith("rhriskassessmentv2:sha256:") assert result.qualified is True assert result.effective_at == "2018-01-05T07:00:00Z" assert result.computed_at == "2026-09-08T01:14:00Z" assert result.to_dict()["decision_eligible"] is False assert result.to_dict()["execution_validation"] == "not_validated" assert sum(result.percentage_risk.values()) == pytest.approx(1.0) assert sum(result.component_risk.values()) == pytest.approx(result.portfolio_volatility) def test_new_risk_computation_cannot_reuse_stale_actual_manifest_time() -> None: arguments = risk_arguments(portfolio_arguments()) arguments["computed_at"] = "2026-09-08T02:11:01Z" with pytest.raises(FactorContractError, match="stale"): assess_retrospective_portfolio_risk(**arguments) def target_with(arguments: dict[str, Any], **changes: Any) -> RetrospectivePortfolioTarget: row = arguments["target"].to_dict() return RetrospectivePortfolioTarget.create( **{ key: value for key, value in {**row, **changes}.items() if key in {"backtest_run_id", "dataset_snapshot_id", "weights", "effective_at", "created_at"} } ) def decision_context(arguments: dict[str, Any]) -> dict[str, Any]: return {key: arguments[key] for key in ("backtest_run_ref", "manifest", "target")} def assessment_context(arguments: dict[str, Any]) -> dict[str, Any]: return { key: arguments[key] for key in ("portfolio_decision", "backtest_run_ref", "manifest", "covariance") } def reidentify(row: dict[str, Any], field: str, prefix: str) -> None: row.pop(field, None) encoded = json.dumps( row, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False ) row[field] = prefix + "sha256:" + hashlib.sha256(encoded.encode()).hexdigest() @pytest.mark.parametrize( "parser", [RetrospectivePortfolioTarget, RetrospectivePortfolioDecision, RetrospectiveRiskAssessment], ) def test_json_syntax_failures_use_typed_contract_errors(parser: Any) -> None: with pytest.raises(FactorContractError): parser.from_json(b"{") @pytest.mark.parametrize( "change", [ {"method": "alternate_estimator"}, {"window_start_date": "2018-01-03"}, {"window_end_date": "2018-01-04"}, {"observations": 3}, {"lookback_sessions": 5}, {"missing_policy": "alternate_missing_policy"}, ], ) def test_covariance_estimation_context_is_bound_into_the_result_identity( change: dict[str, Any], ) -> None: base = portfolio_arguments() arguments = risk_arguments(base) original = assess_retrospective_portfolio_risk(**arguments) arguments["covariance"] = covariance(base, **change) changed = assess_retrospective_portfolio_risk(**arguments) assert changed.assessment_id != original.assessment_id def test_canonical_roundtrips_and_immutable_results() -> None: base = portfolio_arguments() target = base["target"] assert RetrospectivePortfolioTarget.from_json(target.to_json().encode()) == target decision = build_retrospective_portfolio_decision(**base, receipt=portfolio_receipt(base)) assert ( RetrospectivePortfolioDecision.from_json(decision.to_json(), **decision_context(base)) == decision ) arguments = risk_arguments(base) result = assess_retrospective_portfolio_risk(**arguments) assert ( RetrospectiveRiskAssessment.from_json( result.to_json().encode(), **assessment_context(arguments) ) == result ) with pytest.raises(TypeError): target.weights[ASSETS[0]] = 0.1 with pytest.raises(FrozenInstanceError): target.created_at = "2018-01-05T07:00:00Z" with pytest.raises(TypeError): decision.target_weights[ASSETS[0]] = 0.1 with pytest.raises(TypeError): result.component_risk[ASSETS[0]] = 0.1 detached = result.to_dict() detached["component_risk"][ASSETS[0]] = 0.1 assert detached != result.to_dict() @pytest.mark.parametrize( "change", [ {"weights": {}}, {"weights": {"SIM0": 1.0}}, {"weights": {ASSETS[0]: float("nan")}}, {"weights": {ASSETS[0]: True}}, {"backtest_run_id": "rhbacktestrunv1:sha256:" + "0" * 64}, {"dataset_snapshot_id": "rhds:sha256:" + "0" * 64}, {"effective_at": "2026-09-09T01:00:00Z"}, {"created_at": "2026-09-08T01:12:00.1234567Z"}, {"effective_at": "2018-01-05T15:00:00+08:00"}, ], ) def test_target_rejects_legacy_ambiguous_and_nonfinite_inputs(change: dict[str, Any]) -> None: with pytest.raises(CONTRACT_ERRORS): target_with(portfolio_arguments(), **change) @pytest.mark.parametrize( "path,value", [ ("usage", "live"), ("historical_availability", "established"), ("schema_version", "1.0.0"), ("extra", True), ("target_id", "rhportfoliotargetv2:sha256:" + "0" * 64), ], ) def test_target_rejects_wire_mutations(path: str, value: Any) -> None: row = portfolio_arguments()["target"].to_dict() row[path] = value with pytest.raises(CONTRACT_ERRORS): RetrospectivePortfolioTarget.from_dict(row) @pytest.mark.parametrize("field", ["input_digest", "constraint_digest", "output_digest"]) def test_receipt_digests_are_recomputed(field: str) -> None: arguments = portfolio_arguments() receipt = portfolio_receipt(arguments, **{field: "sha256:" + "0" * 64}) with pytest.raises(FactorContractError, match="independently recomputed"): build_retrospective_portfolio_decision(**arguments, receipt=receipt) @pytest.mark.parametrize("status", ["failed", "fallback"]) def test_failed_or_fallback_solver_cannot_form_a_decision(status: str) -> None: arguments = portfolio_arguments() receipt = portfolio_receipt( arguments, status=status, solver_required=True, solver_name="synthetic_solver", solver_version="1.0.0", solver_config_digest=digest({"synthetic_solver": 1}), iterations=1, objective_value=0.0, ) with pytest.raises(FactorContractError, match="failed/fallback"): build_retrospective_portfolio_decision(**arguments, receipt=receipt) @pytest.mark.parametrize( "change", [ {"backtest_run_id": "rhbacktestrunv2:sha256:" + "0" * 64}, {"dataset_snapshot_id": "rhdsv2:sha256:" + "0" * 64}, {"weights": {"rhinstrument:" + "f" * 32: 1.0}}, {"created_at": "2026-09-08T01:10:00Z"}, {"created_at": "2026-09-08T01:14:00Z"}, ], ) def test_decision_closes_target_identity_assets_and_actual_time(change: dict[str, Any]) -> None: arguments = portfolio_arguments() receipt = portfolio_receipt(arguments) arguments["target"] = target_with(arguments, **change) with pytest.raises(CONTRACT_ERRORS): build_retrospective_portfolio_decision(**arguments, receipt=receipt) def test_actual_manifest_freshness_boundary_and_receipt_time() -> None: arguments = portfolio_arguments() arguments["computed_at"] = "2026-09-08T02:11:00Z" assert ( build_retrospective_portfolio_decision( **arguments, receipt=portfolio_receipt(arguments) ).computed_at == arguments["computed_at"] ) arguments["computed_at"] = "2026-09-08T02:11:00.000001Z" with pytest.raises(FactorContractError, match="stale"): build_retrospective_portfolio_decision(**arguments, receipt=portfolio_receipt(arguments)) arguments["computed_at"] = "2026-09-08T01:13:00Z" with pytest.raises(FactorContractError, match="receipt actual time"): build_retrospective_portfolio_decision( **arguments, receipt=portfolio_receipt(arguments, computed_at="2026-09-08T01:13:01Z") ) def test_manifest_tables_and_qualification_are_revalidated_at_s4_boundary() -> None: arguments = portfolio_arguments() receipt = portfolio_receipt(arguments) manifest = arguments["manifest"] artifact = manifest._artifact arguments["manifest"] = build_retrospective_backtest_evidence_manifest( arguments["backtest_run_ref"], artifact, artifact_available_at=manifest.artifact_available_at, qualification=EvidenceQualification.EXPLORATORY, ) with pytest.raises(FactorContractError, match="contract-qualified"): build_retrospective_portfolio_decision(**arguments, receipt=receipt) arguments["manifest"] = manifest # Public access is an isolated copy. Simulate corruption of the retained bytes, # beyond that normal interface, to exercise the consumer's independent recheck. artifact._performance.loc[0, "n_days"] += 1 with pytest.raises(CONTRACT_ERRORS): build_retrospective_portfolio_decision(**arguments, receipt=receipt) def test_constraint_residuals_and_prior_assets_cannot_be_bypassed() -> None: arguments = portfolio_arguments() arguments["constraints"] = ConstraintSetV1(gross_exposure_max=0.9) # A solver may report convergence within its tolerance; actual contract constraints still bind. receipt = portfolio_receipt( arguments, status="converged", solver_required=True, solver_name="synthetic_solver", solver_version="1.0.0", solver_config_digest=digest({"synthetic_solver": 1}), iterations=1, objective_value=0.0, tolerance=0.2, ) with pytest.raises(FactorContractError, match="violates supported constraints"): build_retrospective_portfolio_decision(**arguments, receipt=receipt) arguments = portfolio_arguments() arguments["prior_weights"] = {"rhinstrument:" + "f" * 32: 0.5} with pytest.raises(FactorContractError, match="prior assets"): compute_retrospective_portfolio_receipt_digests( **{key: value for key, value in arguments.items() if key != "computed_at"} ) arguments["prior_weights"] = None with pytest.raises(PortfolioRiskContractError, match="prior"): portfolio_receipt(arguments) def test_optional_prior_budget_limit_and_groups_have_explicit_empty_semantics() -> None: base = portfolio_arguments() base["constraints"] = ConstraintSetV1(gross_exposure_max=1.0) base["prior_weights"] = None arguments = risk_arguments(base) arguments.update(risk_budget=None, portfolio_volatility_limit=None, groups=None) result = assess_retrospective_portfolio_risk(**arguments) assert result.qualified is True assert result.risk_budget == {} assert result.group_exposure == {} assert result.groups is None @pytest.mark.parametrize( "path,value", [ ("decision_eligible", True), ("execution_validation", "validated"), ("historical_availability", "established"), ("gross_exposure", True), ("position_count", 2.0), ("target_weights." + ASSETS[0], 0.5), ("schema_version", "1.0.0"), ("observation_cutoff", "2018-01-05T07:00:00Z"), ("extra", True), ], ) def test_decision_rejects_reidentified_forged_wire(path: str, value: Any) -> None: base = portfolio_arguments() row = build_retrospective_portfolio_decision(**base, receipt=portfolio_receipt(base)).to_dict() replace_at(row, path, value) reidentify(row, "decision_id", "rhportfoliodecisionv2:") with pytest.raises(CONTRACT_ERRORS): RetrospectivePortfolioDecision.from_dict(row, **decision_context(base)) @pytest.mark.parametrize( "change", [ {"as_of_date": "2018-01-06"}, {"as_of_date": "2018-01-04", "window_end_date": "2018-01-04"}, {"window_start_date": None, "window_end_date": None}, {"data_snapshot_id": "rhdsv2:sha256:" + "0" * 64}, {"input_sha256": "b" * 64}, ], ) def test_covariance_business_time_bounds_and_source_binding(change: dict[str, Any]) -> None: base = portfolio_arguments() arguments = risk_arguments(base) arguments["covariance"] = covariance(base, **change) with pytest.raises(FactorContractError): assess_retrospective_portfolio_risk(**arguments) @pytest.mark.parametrize( "matrix,index,columns", [ ([[float("nan"), 0.0], [0.0, 0.1]], ASSETS, ASSETS), ([[0.1, 0.1], [0.0, 0.1]], ASSETS, ASSETS), ([[0.1, 0.0], [0.0, 0.1]], (ASSETS[0], ASSETS[0]), ASSETS), ([[0.1, 0.0], [0.0, 0.1]], (ASSETS[0], "unknown"), ASSETS), ([[0.1, 0.0], [0.0, 0.1]], ASSETS, (ASSETS[0], "unknown")), ], ) def test_covariance_structure_is_checked_before_computation( matrix: Any, index: Any, columns: Any ) -> None: base = portfolio_arguments() arguments = risk_arguments(base) arguments["covariance"] = covariance( base, covariance=pd.DataFrame(matrix, index=index, columns=columns) ) with pytest.raises(PortfolioRiskContractError): assess_retrospective_portfolio_risk(**arguments) @pytest.mark.parametrize( "change", [ {"risk_budget": {ASSETS[0]: -0.1}}, {"risk_budget": {"unknown": 0.1}}, {"portfolio_volatility_limit": -0.1}, {"groups": {ASSETS[0]: "equity"}}, {"groups": []}, {"risk_model_version": "latest"}, {"risk_model_name": "/private/model"}, {"computed_at": "2026-09-08T01:12:59Z"}, {"portfolio_decision": object()}, {"covariance": object()}, ], ) def test_risk_rejects_invalid_models_budgets_clocks_and_untyped_inputs( change: dict[str, Any], ) -> None: arguments = risk_arguments(portfolio_arguments()) arguments.update(change) with pytest.raises(CONTRACT_ERRORS): assess_retrospective_portfolio_risk(**arguments) @pytest.mark.parametrize( "matrix,finding", [ ([[1.0, 2.0], [2.0, 1.0]], RiskFindingCode.COVARIANCE_NOT_PSD), ([[0.0, 0.0], [0.0, 0.0]], RiskFindingCode.PORTFOLIO_VARIANCE_NON_POSITIVE), ], ) def test_numerical_unavailability_is_not_qualification( matrix: Any, finding: RiskFindingCode ) -> None: base = portfolio_arguments() arguments = risk_arguments(base) arguments["covariance"] = covariance( base, covariance=pd.DataFrame(matrix, index=ASSETS, columns=ASSETS) ) result = assess_retrospective_portfolio_risk(**arguments) assert result.status is RiskAssessmentStatus.UNAVAILABLE assert result.qualified is False assert result.findings == (finding,) assert result.portfolio_volatility is None def test_risk_uses_the_existing_numeric_implementation_exactly_once( monkeypatch: pytest.MonkeyPatch, ) -> None: arguments = risk_arguments(portfolio_arguments()) calls = [] def recorded(weights: Any, matrix: Any) -> ComponentRiskResult: calls.append((weights, matrix)) return labeled_component_risk(weights, matrix) monkeypatch.setattr(contracts, "labeled_component_risk", recorded) result = assess_retrospective_portfolio_risk(**arguments) assert len(calls) == 1 expected = labeled_component_risk(*calls[0]) assert result.component_risk == expected.component.to_dict() assert result.portfolio_volatility == expected.portfolio_volatility def test_unknown_numeric_failures_are_sanitized(monkeypatch: pytest.MonkeyPatch) -> None: def failed(*args: Any) -> ComponentRiskResult: raise ValueError("synthetic internal detail") monkeypatch.setattr(contracts, "labeled_component_risk", failed) with pytest.raises(PortfolioRiskContractError, match="risk computation failed") as error: assess_retrospective_portfolio_risk(**risk_arguments(portfolio_arguments())) assert "internal detail" not in str(error.value) def test_nonclosed_decomposition_is_unavailable(monkeypatch: pytest.MonkeyPatch) -> None: def nonclosed(weights: Any, matrix: Any) -> ComponentRiskResult: output = labeled_component_risk(weights, matrix) return replace(output, component=output.component * 0.5) monkeypatch.setattr(contracts, "labeled_component_risk", nonclosed) result = assess_retrospective_portfolio_risk(**risk_arguments(portfolio_arguments())) assert result.status is RiskAssessmentStatus.UNAVAILABLE assert result.findings == (RiskFindingCode.RISK_CONTRIBUTION_NOT_CLOSED,) @pytest.mark.parametrize( "change", [{"portfolio_volatility_limit": 0.0}, {"risk_budget": {ASSETS[0]: 0.0}}] ) def test_budget_breach_keeps_ready_but_unqualified_evidence(change: dict[str, Any]) -> None: arguments = risk_arguments(portfolio_arguments()) arguments.update(change) result = assess_retrospective_portfolio_risk(**arguments) assert result.status is RiskAssessmentStatus.READY assert result.qualified is False assert result.findings == (RiskFindingCode.RISK_BUDGET_BREACH,) assert result.decision_eligible is False @pytest.mark.parametrize( "path,value", [ ("decision_eligible", True), ("execution_validation", "validated"), ("historical_availability", "established"), ("qualified", 1), ("portfolio_volatility", 1.0), ("component_risk." + ASSETS[0], 1.0), ("schema_version", "1.0.0"), ("covariance_matrix_digest", "sha256:" + "0" * 64), ("extra", True), ], ) def test_risk_rejects_reidentified_forged_wire(path: str, value: Any) -> None: arguments = risk_arguments(portfolio_arguments()) row = assess_retrospective_portfolio_risk(**arguments).to_dict() replace_at(row, path, value) reidentify(row, "assessment_id", "rhriskassessmentv2:") with pytest.raises(CONTRACT_ERRORS): RetrospectiveRiskAssessment.from_dict(row, **assessment_context(arguments)) @pytest.mark.parametrize( "raw", [ b'{"x":1,"x":2}', b'{ "x":1}', b"[]", b'{"x":NaN}', b'{"x":Infinity}', b'{"x":9007199254740992}', 1, ], ) def test_json_profiles_reject_ambiguous_nonfinite_and_noncanonical_input(raw: Any) -> None: with pytest.raises(CONTRACT_ERRORS): RetrospectivePortfolioTarget.from_json(raw)