From 7538d2b5d8c9c98297be8c8a906a9c3821bcccd6 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 5 Aug 2026 16:44:03 +0800 Subject: [PATCH] feat: bind promotion runners to risk provenance Co-Authored-By: Codex --- src/quant_platform_kit/position_sizing.py | 172 +++++++++++++ src/quant_platform_kit/risk/contracts.py | 58 +++++ src/quant_platform_kit/risk/gate.py | 155 +++++++++++- .../backtest_orchestrator.py | 6 + .../strategy_lifecycle/param_optimizer.py | 6 +- tests/test_backtest_orchestrator.py | 27 ++ tests/test_param_optimizer.py | 26 +- tests/test_position_sizing.py | 99 ++++++++ tests/test_risk_gate.py | 230 +++++++++++++++++- ...t_strategy_evidence_package_v2_contract.py | 1 + 10 files changed, 758 insertions(+), 22 deletions(-) diff --git a/src/quant_platform_kit/position_sizing.py b/src/quant_platform_kit/position_sizing.py index 259256e..1ac421c 100644 --- a/src/quant_platform_kit/position_sizing.py +++ b/src/quant_platform_kit/position_sizing.py @@ -4,6 +4,7 @@ from dataclasses import dataclass import math +from typing import Mapping _DEFAULT_MAX_POSITION_PCT = 0.10 @@ -24,6 +25,177 @@ class KellyResult: _BOOTSTRAP_NOMINAL_CAPS = {1: 0.50, 2: 0.25, 3: 0.15} +def _weight_mapping(value: object, *, allow_empty: bool) -> dict[str, float] | None: + if not isinstance(value, Mapping) or (not value and not allow_empty): + return None + normalized: dict[str, float] = {} + for symbol, raw_weight in value.items(): + if ( + not isinstance(symbol, str) + or not symbol + or symbol != symbol.strip() + or isinstance(raw_weight, bool) + or not isinstance(raw_weight, (int, float)) + ): + return None + weight = float(raw_weight) + if not math.isfinite(weight) or weight < 0.0: + return None + normalized[symbol] = weight + return normalized + + +def risk_budgeted_target_weights( + *, + raw_target_weights: Mapping[str, float], + risk_mandate_id: str | None, + risk_fraction: float, + stop_loss_distances: Mapping[str, float], + drawdown_scalar: float, + available_effective_exposure: float, + product_leverage_factors: Mapping[str, int], + inputs_fresh: bool, +) -> dict[str, float]: + """Scale one mandate-bound multi-asset target vector proportionally. + + This is a pure sizing helper, not an allocator or an approval decision. + Invalid, stale, unmandated or over-authority inputs return an empty vector. + """ + raw_weights = _weight_mapping(raw_target_weights, allow_empty=False) + if ( + inputs_fresh is not True + or not isinstance(risk_mandate_id, str) + or not risk_mandate_id + or risk_mandate_id != risk_mandate_id.strip() + or risk_mandate_id == _APPROVED_BOOTSTRAP_MANDATE + or raw_weights is None + or not isinstance(stop_loss_distances, Mapping) + or not isinstance(product_leverage_factors, Mapping) + or set(stop_loss_distances) != set(raw_weights) + or set(product_leverage_factors) != set(raw_weights) + ): + return {} + numeric_inputs = (risk_fraction, drawdown_scalar, available_effective_exposure) + if any( + isinstance(value, bool) or not isinstance(value, (int, float)) + for value in numeric_inputs + ): + return {} + risk_fraction, drawdown_scalar, available_effective_exposure = ( + float(value) for value in numeric_inputs + ) + if ( + not all(math.isfinite(value) for value in numeric_inputs) + or not 0.0 < risk_fraction <= _BOOTSTRAP_LOSS_BUDGET_CAP + or not 0.0 < drawdown_scalar <= 1.0 + or not 0.0 < available_effective_exposure <= _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP + ): + return {} + + stops: dict[str, float] = {} + factors: dict[str, int] = {} + for symbol in raw_weights: + raw_stop = stop_loss_distances[symbol] + factor = product_leverage_factors[symbol] + if ( + isinstance(raw_stop, bool) + or not isinstance(raw_stop, (int, float)) + or not math.isfinite(float(raw_stop)) + or not 0.0 < float(raw_stop) <= 1.0 + or isinstance(factor, bool) + or not isinstance(factor, int) + or factor not in _BOOTSTRAP_NOMINAL_CAPS + ): + return {} + stops[symbol] = float(raw_stop) + factors[symbol] = factor + + active = {symbol: weight for symbol, weight in raw_weights.items() if weight > 0.0} + if not active: + return {} + modeled_loss = sum(active[symbol] * stops[symbol] for symbol in active) + effective_exposure = sum(active[symbol] * factors[symbol] for symbol in active) + if modeled_loss <= 0.0 or effective_exposure <= 0.0: + return {} + + scales = [ + 1.0, + risk_fraction * drawdown_scalar / modeled_loss, + available_effective_exposure / effective_exposure, + ] + scales.extend( + _BOOTSTRAP_NOMINAL_CAPS[factors[symbol]] / weight + for symbol, weight in active.items() + ) + scale = min(scales) + if not math.isfinite(scale) or scale <= 0.0: + return {} + return {symbol: weight * scale for symbol, weight in active.items()} + + +def validate_reduce_only_normalization( + *, + origin_weights: Mapping[str, float], + target_weights: Mapping[str, float], + product_leverage_factors: Mapping[str, int], + effective_exposure_cap: float, + observed_effective_exposure: float, +) -> bool: + """Validate one explicit transition from an over-cap origin toward cash.""" + origin = _weight_mapping(origin_weights, allow_empty=False) + target = _weight_mapping(target_weights, allow_empty=True) + if ( + origin is None + or target is None + or not isinstance(product_leverage_factors, Mapping) + or not (set(origin) | set(target)).issubset(product_leverage_factors) + or isinstance(effective_exposure_cap, bool) + or not isinstance(effective_exposure_cap, (int, float)) + or isinstance(observed_effective_exposure, bool) + or not isinstance(observed_effective_exposure, (int, float)) + ): + return False + cap = float(effective_exposure_cap) + observed = float(observed_effective_exposure) + if ( + not math.isfinite(cap) + or not 0.0 <= cap <= _BOOTSTRAP_EFFECTIVE_EXPOSURE_CAP + or not math.isfinite(observed) + or observed < 0.0 + ): + return False + + factors: dict[str, int] = {} + for symbol, factor in product_leverage_factors.items(): + if ( + isinstance(factor, bool) + or not isinstance(factor, int) + or factor not in _BOOTSTRAP_NOMINAL_CAPS + ): + return False + factors[symbol] = factor + origin_active = {symbol for symbol, weight in origin.items() if weight > 0.0} + target_active = {symbol for symbol, weight in target.items() if weight > 0.0} + if not origin_active or not target_active.issubset(origin_active): + return False + if any(target.get(symbol, 0.0) > origin[symbol] + 1e-9 for symbol in origin): + return False + if any( + weight > _BOOTSTRAP_NOMINAL_CAPS[factors[symbol]] + 1e-9 + for symbol, weight in target.items() + ): + return False + + origin_effective = sum(weight * factors[symbol] for symbol, weight in origin.items()) + target_effective = sum(weight * factors[symbol] for symbol, weight in target.items()) + return ( + abs(origin_effective - observed) <= 1e-9 + and origin_effective > cap + 1e-9 + and target_effective < origin_effective - 1e-9 + and target_effective <= cap + 1e-9 + ) + + def risk_budgeted_target_weight( *, risk_mandate_id: str | None = None, diff --git a/src/quant_platform_kit/risk/contracts.py b/src/quant_platform_kit/risk/contracts.py index fbb2b35..70f7641 100644 --- a/src/quant_platform_kit/risk/contracts.py +++ b/src/quant_platform_kit/risk/contracts.py @@ -26,6 +26,14 @@ _REGIME_RISK_ORDER = (REGIME_NORMAL, REGIME_ELEVATED, REGIME_STRESS) +def _is_lower_hex(value: object, length: int) -> bool: + return ( + isinstance(value, str) + and len(value) == length + and all(character in "0123456789abcdef" for character in value) + ) + + def normalise_regime(raw: str | None) -> str: """Normalise a free-form regime string to one of the canonical constants.""" value = str(raw or "").strip().lower() @@ -145,6 +153,52 @@ class RiskAction: notify: bool = True +@dataclass(frozen=True) +class CandidateRiskIdentity: + """Immutable identity of one mandate-bound promotion candidate.""" + + strategy_profile: str + account_mode: str + strategy_revision: str + runner_revision: str + config_sha256: str + input_manifest_sha256: str + authority_receipt_sha256: str + candidate_sha256: str = field(init=False) + + def __post_init__(self) -> None: + for name in ("strategy_profile", "account_mode"): + value = getattr(self, name) + if not isinstance(value, str) or not value or value != value.strip(): + raise ValueError(f"{name} must be a non-empty canonical string") + for name in ("strategy_revision", "runner_revision"): + if not _is_lower_hex(getattr(self, name), 40): + raise ValueError(f"{name} must be a lowercase 40-character Git revision") + for name in ( + "config_sha256", + "input_manifest_sha256", + "authority_receipt_sha256", + ): + if not _is_lower_hex(getattr(self, name), 64): + raise ValueError(f"{name} must be a lowercase SHA-256 digest") + payload = { + "strategy_profile": self.strategy_profile, + "account_mode": self.account_mode, + "strategy_revision": self.strategy_revision, + "runner_revision": self.runner_revision, + "config_sha256": self.config_sha256, + "input_manifest_sha256": self.input_manifest_sha256, + "authority_receipt_sha256": self.authority_receipt_sha256, + } + encoded = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + object.__setattr__(self, "candidate_sha256", hashlib.sha256(encoded).hexdigest()) + + @dataclass(frozen=True) class RiskGateAssessment: """Immutable redacted evidence from a scoped risk-gate evaluation.""" @@ -159,8 +213,10 @@ class RiskGateAssessment: mandate_version: str | None mandate_authority_receipt_sha256: str | None mandate_scope: str | None + candidate_identity_sha256: str | None decision_digest_sha256: str portfolio_snapshot_digest_sha256: str + normalization_origin_digest_sha256: str | None effective_exposure_cap: float | None observed_effective_exposure: float | None proposed_effective_exposure: float | None @@ -180,8 +236,10 @@ def __post_init__(self) -> None: "mandate_version": self.mandate_version, "mandate_authority_receipt_sha256": self.mandate_authority_receipt_sha256, "mandate_scope": self.mandate_scope, + "candidate_identity_sha256": self.candidate_identity_sha256, "decision_digest_sha256": self.decision_digest_sha256, "portfolio_snapshot_digest_sha256": self.portfolio_snapshot_digest_sha256, + "normalization_origin_digest_sha256": self.normalization_origin_digest_sha256, "effective_exposure_cap": self.effective_exposure_cap, "observed_effective_exposure": self.observed_effective_exposure, "proposed_effective_exposure": self.proposed_effective_exposure, diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index 6973e98..a362865 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -14,7 +14,12 @@ from typing import Any, Mapping from quant_platform_kit.common.models import PortfolioSnapshot -from quant_platform_kit.risk.contracts import RiskGateAssessment, RiskGateResult +from quant_platform_kit.position_sizing import validate_reduce_only_normalization +from quant_platform_kit.risk.contracts import ( + CandidateRiskIdentity, + RiskGateAssessment, + RiskGateResult, +) from quant_platform_kit.risk.engine import build_risk_engine from quant_platform_kit.strategy_contracts import StrategyDecision @@ -90,6 +95,12 @@ def _sha256(value: Any) -> str | None: return value if all(character in "0123456789abcdef" for character in value) else None +def _git_revision(value: Any) -> str | None: + if not isinstance(value, str) or len(value) != 40: + return None + return value if all(character in "0123456789abcdef" for character in value) else None + + def _decision_metrics( decision: StrategyDecision, *, @@ -213,6 +224,11 @@ def _mandate_fields( "authority_scope", "strategy_profile", "account_mode", + "strategy_revision", + "runner_revision", + "config_sha256", + "input_manifest_sha256", + "candidate_identity_sha256", "effective_at", "expires_at", "max_snapshot_age_seconds", @@ -233,6 +249,13 @@ def _mandate_fields( return {}, {"invalid_mandate"} authority_scope = mandate_provenance["authority_scope"] receipt_sha256 = _sha256(mandate_provenance["authority_receipt_sha256"]) + strategy_revision = _git_revision(mandate_provenance["strategy_revision"]) + runner_revision = _git_revision(mandate_provenance["runner_revision"]) + config_sha256 = _sha256(mandate_provenance["config_sha256"]) + input_manifest_sha256 = _sha256(mandate_provenance["input_manifest_sha256"]) + candidate_identity_sha256 = _sha256( + mandate_provenance["candidate_identity_sha256"] + ) effective_at = _parse_utc_timestamp(mandate_provenance["effective_at"]) expires_at = _parse_utc_timestamp(mandate_provenance["expires_at"]) max_snapshot_age_seconds = _finite_number(mandate_provenance["max_snapshot_age_seconds"]) @@ -241,9 +264,20 @@ def _mandate_fields( if ( not isinstance(mandate_provenance["mandate_id"], str) or not isinstance(mandate_provenance["mandate_version"], str) - or not isinstance(mandate_provenance["source_revision"], str) + or _git_revision(mandate_provenance["source_revision"]) is None or not isinstance(mandate_provenance["strategy_profile"], str) + or not mandate_provenance["strategy_profile"] + or mandate_provenance["strategy_profile"] + != mandate_provenance["strategy_profile"].strip() or not isinstance(mandate_provenance["account_mode"], str) + or not mandate_provenance["account_mode"] + or mandate_provenance["account_mode"] + != mandate_provenance["account_mode"].strip() + or strategy_revision is None + or runner_revision is None + or config_sha256 is None + or input_manifest_sha256 is None + or candidate_identity_sha256 is None or authority_scope not in _ALLOWED_MANDATE_SCOPES or receipt_sha256 is None or effective_at is None @@ -279,6 +313,13 @@ def _mandate_fields( "authority_receipt_sha256": receipt_sha256, "authority_scope": authority_scope, "source_revision": mandate_provenance["source_revision"], + "strategy_profile": mandate_provenance["strategy_profile"], + "account_mode": mandate_provenance["account_mode"], + "strategy_revision": strategy_revision, + "runner_revision": runner_revision, + "config_sha256": config_sha256, + "input_manifest_sha256": input_manifest_sha256, + "candidate_identity_sha256": candidate_identity_sha256, "effective_exposure_cap": cap, "max_snapshot_age_seconds": max_snapshot_age_seconds, "loss_budget": loss_budget, @@ -289,6 +330,68 @@ def _mandate_fields( }, set() +def _candidate_binding_errors( + mandate_provenance: Mapping[str, Any] | None, + mandate: Mapping[str, Any], + candidate_identity: CandidateRiskIdentity | None, +) -> set[str]: + if mandate_provenance is None: + return {"candidate_without_mandate"} if candidate_identity is not None else set() + if candidate_identity is None: + return {"missing_candidate_identity"} + if not isinstance(candidate_identity, CandidateRiskIdentity): + return {"invalid_candidate_identity"} + if not mandate: + return set() + comparisons = ( + ( + candidate_identity.strategy_profile, + mandate.get("strategy_profile"), + "candidate_strategy_profile_mismatch", + ), + ( + candidate_identity.account_mode, + mandate.get("account_mode"), + "candidate_account_mode_mismatch", + ), + ( + candidate_identity.strategy_revision, + mandate.get("strategy_revision"), + "candidate_strategy_revision_mismatch", + ), + ( + candidate_identity.runner_revision, + mandate.get("runner_revision"), + "candidate_runner_revision_mismatch", + ), + ( + candidate_identity.config_sha256, + mandate.get("config_sha256"), + "candidate_config_digest_mismatch", + ), + ( + candidate_identity.input_manifest_sha256, + mandate.get("input_manifest_sha256"), + "candidate_input_manifest_digest_mismatch", + ), + ( + candidate_identity.authority_receipt_sha256, + mandate.get("authority_receipt_sha256"), + "candidate_authority_digest_mismatch", + ), + ( + candidate_identity.candidate_sha256, + mandate.get("candidate_identity_sha256"), + "candidate_identity_digest_mismatch", + ), + ) + return { + reason_code + for actual, expected, reason_code in comparisons + if actual != expected + } + + def _position_cap(value: Any, symbol: str, leverage_factor: float) -> float | None: if isinstance(value, Mapping): leverage_class = str(int(leverage_factor)) @@ -329,6 +432,8 @@ def assess_with_evidence( scope: str, mandate_provenance: Mapping[str, Any] | None, market_data: Mapping[str, Any], + candidate_identity: CandidateRiskIdentity | None = None, + normalization_origin_weights: Mapping[str, float] | None = None, ) -> RiskGateResult: """Assess exactly once and fail closed with a redacted canonical receipt.""" now = _utc_now() @@ -336,6 +441,13 @@ def assess_with_evidence( assessment_scope = scope if scope in _ALLOWED_SCOPES else "MEMBER" mandate, mandate_errors = _mandate_fields(mandate_provenance, now=now) reason_codes = set(mandate_errors) + reason_codes.update( + _candidate_binding_errors( + mandate_provenance, + mandate, + candidate_identity, + ) + ) cap = mandate.get("effective_exposure_cap") snapshot_payload, observed, total_equity, snapshot_errors = _snapshot_metrics( portfolio_snapshot, @@ -355,6 +467,7 @@ def assess_with_evidence( reason_codes.update(_budget_authority_errors(decision, mandate)) proposed: float | None = None + normalization_origin_digest_sha256: str | None = None if can_evaluate_policy: factors = mandate["product_leverage_factors"] allowed_assets = mandate["allowed_nonzero_assets"] @@ -377,8 +490,36 @@ def assess_with_evidence( if weight > min(product_cap, nominal_cap): reason_codes.add("product_exposure_cap") weighted_exposure += weight * factor - proposed = max(observed or 0.0, weighted_exposure) - if cap is None or observed is None or observed > cap + 1e-9: + target_weights: dict[str, float] = {} + for symbol, weight in active_positions: + target_weights[symbol] = target_weights.get(symbol, 0.0) + weight + valid_normalization = False + if normalization_origin_weights is not None: + valid_normalization = validate_reduce_only_normalization( + origin_weights=normalization_origin_weights, + target_weights=target_weights, + product_leverage_factors=factors, + effective_exposure_cap=cap, + observed_effective_exposure=observed, + ) + if not valid_normalization: + reason_codes.add("invalid_reduce_only_normalization") + else: + normalized_origin = { + symbol: float(weight) + for symbol, weight in sorted(normalization_origin_weights.items()) + } + normalization_origin_digest_sha256 = _canonical_digest( + {"weights": normalized_origin} + ) + proposed = ( + weighted_exposure + if valid_normalization + else max(observed or 0.0, weighted_exposure) + ) + if cap is None or observed is None or ( + observed > cap + 1e-9 and not valid_normalization + ): reason_codes.add("observed_effective_exposure") if cap is None or proposed > cap + 1e-9: reason_codes.add("effective_exposure_cap") @@ -409,8 +550,14 @@ def assess_with_evidence( mandate_version=mandate.get("mandate_version"), mandate_authority_receipt_sha256=mandate.get("authority_receipt_sha256"), mandate_scope=mandate.get("authority_scope"), + candidate_identity_sha256=( + candidate_identity.candidate_sha256 + if isinstance(candidate_identity, CandidateRiskIdentity) + else None + ), decision_digest_sha256=_canonical_digest(decision_payload), portfolio_snapshot_digest_sha256=_canonical_digest(snapshot_payload), + normalization_origin_digest_sha256=normalization_origin_digest_sha256, effective_exposure_cap=cap, observed_effective_exposure=observed, proposed_effective_exposure=proposed, diff --git a/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py b/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py index 350ef48..73aec82 100644 --- a/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py +++ b/src/quant_platform_kit/strategy_lifecycle/backtest_orchestrator.py @@ -404,6 +404,12 @@ def run_promotion( raise ValueError( f"No BacktestRunner registered for domain={domain!r}. Available: {sorted(self._runners)}" ) + runner_kind = getattr(runner, "runner_kind", None) + if runner_kind != "real": + raise RuntimeError( + "promotion-grade execution requires explicit runner_kind='real'; " + f"received {runner_kind!r}" + ) if not isinstance(runner, PromotionBacktestRunner): raise TypeError( "promotion-grade execution requires explicit run_purged_fold and run_locked_oos runner methods" diff --git a/src/quant_platform_kit/strategy_lifecycle/param_optimizer.py b/src/quant_platform_kit/strategy_lifecycle/param_optimizer.py index 5a8e6f2..26867e2 100644 --- a/src/quant_platform_kit/strategy_lifecycle/param_optimizer.py +++ b/src/quant_platform_kit/strategy_lifecycle/param_optimizer.py @@ -423,11 +423,11 @@ def _auto_register_runner(orchestrator: BacktestOrchestrator, domain: str) -> No continue runner = runner_factory() - runner_kind = str(getattr(runner, "runner_kind", "real") or "real").strip().lower() + runner_kind = getattr(runner, "runner_kind", None) if runner_kind != "real": raise RuntimeError( - f"BacktestRunner for domain={domain!r} is marked runner_kind={runner_kind!r}; " - "placeholder runners are blocked from lifecycle optimization." + f"BacktestRunner for domain={domain!r} requires explicit runner_kind='real'; " + f"received {runner_kind!r}." ) orchestrator.register_runner(domain, runner) diff --git a/tests/test_backtest_orchestrator.py b/tests/test_backtest_orchestrator.py index 9fbbbf0..f761ed0 100644 --- a/tests/test_backtest_orchestrator.py +++ b/tests/test_backtest_orchestrator.py @@ -60,6 +60,7 @@ class _PromotionRecordingRunner(_RecordingRunner): def __init__(self, **result_overrides: Any) -> None: super().__init__() + self.runner_kind = "real" self.result_overrides = result_overrides def _promotion_result( @@ -375,6 +376,7 @@ def test_ordinary_walk_forward_accepts_raw_windows_but_is_non_promotion( ) def test_promotion_run_requires_explicit_runner_capability(self) -> None: + self.runner.runner_kind = "real" with self.assertRaises(TypeError): self.orchestrator.run_promotion( "test_strat", @@ -389,6 +391,31 @@ def test_promotion_run_requires_explicit_runner_capability(self) -> None: cost_model=self._cost_model(), ) + def test_promotion_run_requires_exact_real_runner_marker(self) -> None: + missing = object() + for marker in (missing, None, "", " ", "placeholder", "REAL", " real "): + with self.subTest(marker=marker): + runner = _PromotionRecordingRunner() + if marker is missing: + del runner.runner_kind + else: + runner.runner_kind = marker + + with self.assertRaisesRegex(RuntimeError, "explicit runner_kind='real'"): + self._run_promotion(runner=runner) + + def test_unmarked_runner_remains_valid_for_ordinary_non_promotion_run(self) -> None: + result = self.orchestrator.run( + "test_strat", + domain="us_equity", + params={"lookback": 20}, + start_date=date(2020, 1, 1), + end_date=date(2020, 12, 31), + ) + + self.assertEqual(result.strategy_profile, "test_strat") + self.assertIsNone(getattr(self.runner, "runner_kind", None)) + def test_promotion_run_enforces_and_persists_purged_wfa_identity(self) -> None: run = self._run_promotion() diff --git a/tests/test_param_optimizer.py b/tests/test_param_optimizer.py index 1a25f8e..03a01c3 100644 --- a/tests/test_param_optimizer.py +++ b/tests/test_param_optimizer.py @@ -9,17 +9,21 @@ class ParamOptimizerRunnerRegistrationTests(unittest.TestCase): - def test_auto_register_runner_rejects_placeholder_runner(self) -> None: - orchestrator = BacktestOrchestrator() - - class PlaceholderRunner: - runner_kind = "placeholder" - - fake_module = SimpleNamespace(build_backtest_runner=lambda: PlaceholderRunner()) - - with patch("importlib.import_module", return_value=fake_module): - with self.assertRaisesRegex(RuntimeError, "placeholder runners are blocked"): - _auto_register_runner(orchestrator, "us_equity") + def test_auto_register_runner_requires_exact_real_marker(self) -> None: + missing = object() + for marker in (missing, None, "", " ", "placeholder", "REAL", " real "): + with self.subTest(marker=marker): + orchestrator = BacktestOrchestrator() + runner = SimpleNamespace() + if marker is not missing: + runner.runner_kind = marker + fake_module = SimpleNamespace(build_backtest_runner=lambda: runner) + + with ( + patch("importlib.import_module", return_value=fake_module), + self.assertRaisesRegex(RuntimeError, "explicit runner_kind='real'"), + ): + _auto_register_runner(orchestrator, "us_equity") def test_auto_register_runner_raises_when_all_candidates_fail(self) -> None: orchestrator = BacktestOrchestrator() diff --git a/tests/test_position_sizing.py b/tests/test_position_sizing.py index 8e0a0a5..98f6bda 100644 --- a/tests/test_position_sizing.py +++ b/tests/test_position_sizing.py @@ -8,6 +8,8 @@ KellyResult, estimate_kelly, risk_budgeted_target_weight, + risk_budgeted_target_weights, + validate_reduce_only_normalization, ) @@ -175,5 +177,102 @@ def test_invalid_stale_or_over_budget_inputs_fail_closed(self) -> None: ) +class RiskBudgetedTargetWeightsTests(unittest.TestCase): + def _approved_inputs(self, **overrides: object) -> dict[str, object]: + return { + "raw_target_weights": {"SOXL": 0.70, "SOXX": 0.30}, + "risk_mandate_id": "soxl_p3_research_v1", + "risk_fraction": 0.01, + "stop_loss_distances": {"SOXL": 0.05, "SOXX": 0.05}, + "drawdown_scalar": 1.0, + "available_effective_exposure": 0.50, + "product_leverage_factors": {"SOXL": 3, "SOXX": 1}, + "inputs_fresh": True, + **overrides, + } + + def test_sizes_multi_asset_vector_proportionally_under_all_caps(self) -> None: + result = risk_budgeted_target_weights(**self._approved_inputs()) + + self.assertEqual(set(result), {"SOXL", "SOXX"}) + self.assertAlmostEqual(result["SOXL"], 0.14) + self.assertAlmostEqual(result["SOXX"], 0.06) + self.assertAlmostEqual(result["SOXL"] / result["SOXX"], 7 / 3) + self.assertLessEqual(result["SOXL"], 0.15) + self.assertLessEqual(result["SOXL"] * 3 + result["SOXX"], 0.50) + self.assertLessEqual( + result["SOXL"] * 0.05 + result["SOXX"] * 0.05, + 0.01, + ) + + def test_drawdown_scalar_reduces_aggregate_loss_budget(self) -> None: + result = risk_budgeted_target_weights( + **self._approved_inputs(drawdown_scalar=0.50), + ) + + self.assertAlmostEqual(result["SOXL"], 0.07) + self.assertAlmostEqual(result["SOXX"], 0.03) + + def test_invalid_or_stale_multi_asset_inputs_fail_closed(self) -> None: + invalid_cases = ( + {"risk_mandate_id": None}, + {"risk_mandate_id": "bootstrap_small_account_v2"}, + {"inputs_fresh": False}, + {"risk_fraction": 0.0100001}, + {"drawdown_scalar": float("nan")}, + {"available_effective_exposure": 0.500001}, + {"product_leverage_factors": {"SOXL": 4, "SOXX": 1}}, + {"product_leverage_factors": {"SOXL": 3}}, + {"stop_loss_distances": {"SOXL": 0.05}}, + {"raw_target_weights": {"SOXL": float("inf"), "SOXX": 0.30}}, + {"raw_target_weights": {"SOXL": -0.10, "SOXX": 0.30}}, + ) + for overrides in invalid_cases: + with self.subTest(overrides=overrides): + self.assertEqual( + risk_budgeted_target_weights( + **self._approved_inputs(**overrides), + ), + {}, + ) + + +class ReduceOnlyNormalizationTests(unittest.TestCase): + def test_one_hundred_percent_boxx_can_normalize_to_compliant_boxx_cash(self) -> None: + self.assertTrue( + validate_reduce_only_normalization( + origin_weights={"BOXX": 1.0}, + target_weights={"BOXX": 0.50}, + product_leverage_factors={"BOXX": 1}, + effective_exposure_cap=0.50, + observed_effective_exposure=1.0, + ) + ) + + def test_normalization_rejects_new_exposure_non_reduction_or_bad_origin(self) -> None: + invalid_cases = ( + ({"BOXX": 0.40, "SOXX": 0.10}, {"BOXX": 1, "SOXX": 1}, 1.0), + ({"BOXX": 1.0}, {"BOXX": 1}, 1.0), + ({"BOXX": 0.60}, {"BOXX": 1}, 1.0), + ({"BOXX": 0.50}, {"BOXX": 1}, 0.90), + ({"BOXX": float("nan")}, {"BOXX": 1}, 1.0), + ) + for target_weights, factors, observed in invalid_cases: + with self.subTest( + target_weights=target_weights, + factors=factors, + observed=observed, + ): + self.assertFalse( + validate_reduce_only_normalization( + origin_weights={"BOXX": 1.0}, + target_weights=target_weights, + product_leverage_factors=factors, + effective_exposure_cap=0.50, + observed_effective_exposure=observed, + ) + ) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index 65563b4..0434f5b 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -5,7 +5,12 @@ from unittest.mock import Mock, patch from quant_platform_kit.common.models import PortfolioSnapshot -from quant_platform_kit.risk.contracts import ROUTE_BLOCKED, RiskAction, RiskSignal +from quant_platform_kit.risk.contracts import ( + ROUTE_BLOCKED, + CandidateRiskIdentity, + RiskAction, + RiskSignal, +) from quant_platform_kit.risk.engine import RiskEngine from quant_platform_kit.risk.gate import ( assess_with_evidence, @@ -363,14 +368,38 @@ class AssessWithEvidenceTests(unittest.TestCase): _NOW = datetime(2026, 8, 4, 4, 28, tzinfo=timezone.utc) @staticmethod - def _mandate(**overrides: object) -> dict[str, object]: + def _candidate(**overrides: object) -> CandidateRiskIdentity: + values: dict[str, object] = { + "strategy_profile": "crypto_live_pool_rotation", + "account_mode": "single_strategy_account_v1", + "strategy_revision": "b" * 40, + "runner_revision": "c" * 40, + "config_sha256": "d" * 64, + "input_manifest_sha256": "e" * 64, + "authority_receipt_sha256": "a" * 64, + } + values.update(overrides) + return CandidateRiskIdentity(**values) + + @classmethod + def _mandate( + cls, + candidate: CandidateRiskIdentity | None = None, + **overrides: object, + ) -> dict[str, object]: + candidate = candidate or cls._candidate() mandate: dict[str, object] = { "mandate_id": "binance_crypto_research_only_v1", "mandate_version": "2026-08-04.1", "authority_receipt_sha256": "a" * 64, "authority_scope": "RESEARCH_ONLY", - "strategy_profile": "crypto_live_pool_rotation", - "account_mode": "single_strategy_account_v1", + "strategy_profile": candidate.strategy_profile, + "account_mode": candidate.account_mode, + "strategy_revision": candidate.strategy_revision, + "runner_revision": candidate.runner_revision, + "config_sha256": candidate.config_sha256, + "input_manifest_sha256": candidate.input_manifest_sha256, + "candidate_identity_sha256": candidate.candidate_sha256, "effective_at": "2026-08-04T04:27:55Z", "expires_at": "2026-09-03T15:59:59Z", "max_snapshot_age_seconds": 300, @@ -408,6 +437,7 @@ def test_approved_receipt_is_immutable_and_redacts_digest_inputs(self) -> None: scope="MEMBER", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) redacted_equivalent = assess_with_evidence( decision, @@ -418,15 +448,196 @@ def test_approved_receipt_is_immutable_and_redacts_digest_inputs(self) -> None: scope="MEMBER", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(first.assessment.outcome, "APPROVE") self.assertEqual(first.assessment.effective_exposure_cap, 0.50) self.assertEqual(first.assessment.observed_effective_exposure, 0.10) self.assertEqual(first.assessment.proposed_effective_exposure, 0.20) + self.assertEqual( + first.assessment.candidate_identity_sha256, + self._candidate().candidate_sha256, + ) self.assertEqual(first.assessment.assessment_sha256, redacted_equivalent.assessment.assessment_sha256) self.assertEqual(len(first.decision.positions), 1) + def test_mandate_requires_typed_candidate_and_still_assesses_once(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),), + ) + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = assess_with_evidence( + decision, + self._snapshot(), + scope="MEMBER", + mandate_provenance=self._mandate(), + market_data={}, + candidate_identity=None, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("missing_candidate_identity", result.assessment.reason_codes) + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + engine.assess.assert_called_once_with(decision, self._snapshot(), market_data={}) + + def test_mandate_is_bound_to_exact_candidate_fields_and_digest(self) -> None: + decision = _decision( + positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),), + ) + base_candidate = self._candidate() + cases = ( + ( + "strategy", + self._candidate(strategy_profile="soxl_soxx_trend_income"), + self._mandate(), + "candidate_strategy_profile_mismatch", + ), + ( + "account", + self._candidate(account_mode="smart_portfolio_v1"), + self._mandate(), + "candidate_account_mode_mismatch", + ), + ( + "candidate_digest", + base_candidate, + self._mandate(candidate_identity_sha256="f" * 64), + "candidate_identity_digest_mismatch", + ), + ( + "input_manifest", + self._candidate(input_manifest_sha256="f" * 64), + self._mandate(), + "candidate_input_manifest_digest_mismatch", + ), + ) + for name, candidate, mandate, reason_code in cases: + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + with ( + self.subTest(name=name), + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + patch( + "quant_platform_kit.risk.gate.build_risk_engine", + return_value=engine, + ), + ): + result = assess_with_evidence( + decision, + self._snapshot(), + scope="MEMBER", + mandate_provenance=mandate, + market_data={}, + candidate_identity=candidate, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn(reason_code, result.assessment.reason_codes) + self.assertEqual( + result.assessment.candidate_identity_sha256, + candidate.candidate_sha256, + ) + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + engine.assess.assert_called_once_with( + decision, + self._snapshot(), + market_data={}, + ) + + def test_reduce_only_normalization_can_exit_over_cap_origin_once(self) -> None: + candidate = self._candidate(strategy_profile="soxl_soxx_trend_income") + mandate = self._mandate( + candidate, + product_leverage_factors={"BOXX": 1, "SOXX": 1}, + allowed_nonzero_assets=["BOXX", "SOXX"], + product_caps={"BOXX": 0.50, "SOXX": 0.50}, + nominal_caps={"BOXX": 0.50, "SOXX": 0.50}, + loss_budget=0.01, + ) + decision = _decision( + positions=(PositionTarget(symbol="BOXX", target_weight=0.50),), + ) + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = assess_with_evidence( + decision, + self._snapshot(observed_effective_exposure=1.0), + scope="MEMBER", + mandate_provenance=mandate, + market_data={}, + candidate_identity=candidate, + normalization_origin_weights={"BOXX": 1.0}, + ) + + self.assertEqual(result.assessment.outcome, "APPROVE") + self.assertEqual(result.assessment.proposed_effective_exposure, 0.50) + self.assertIsNotNone(result.assessment.normalization_origin_digest_sha256) + engine.assess.assert_called_once_with( + decision, + self._snapshot(observed_effective_exposure=1.0), + market_data={}, + ) + + def test_invalid_reduce_only_normalization_rejects_and_assesses_once(self) -> None: + candidate = self._candidate(strategy_profile="soxl_soxx_trend_income") + mandate = self._mandate( + candidate, + product_leverage_factors={"BOXX": 1, "SOXX": 1}, + allowed_nonzero_assets=["BOXX", "SOXX"], + product_caps={"BOXX": 0.50, "SOXX": 0.50}, + nominal_caps={"BOXX": 0.50, "SOXX": 0.50}, + loss_budget=0.01, + ) + decision = _decision( + positions=( + PositionTarget(symbol="BOXX", target_weight=0.40), + PositionTarget(symbol="SOXX", target_weight=0.10), + ), + ) + engine = Mock() + engine.assess.return_value = RiskAction(action="approve", reason="passed") + + with ( + patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW), + patch("quant_platform_kit.risk.gate.build_risk_engine", return_value=engine), + ): + result = assess_with_evidence( + decision, + self._snapshot(observed_effective_exposure=1.0), + scope="MEMBER", + mandate_provenance=mandate, + market_data={}, + candidate_identity=candidate, + normalization_origin_weights={"BOXX": 1.0}, + ) + + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn( + "invalid_reduce_only_normalization", + result.assessment.reason_codes, + ) + self.assertEqual(result.decision.positions, ()) + self.assertEqual(result.decision.budgets, ()) + engine.assess.assert_called_once_with( + decision, + self._snapshot(observed_effective_exposure=1.0), + market_data={}, + ) + def test_zero_cap_research_mandate_never_produces_order_authority(self) -> None: decision = StrategyDecision( positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),), @@ -447,6 +658,7 @@ def test_zero_cap_research_mandate_never_produces_order_authority(self) -> None: allowed_nonzero_assets=[], ), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(result.assessment.outcome, "REJECT") @@ -468,6 +680,7 @@ def test_invalid_scope_rejects_fail_closed(self) -> None: scope="STRATEGY", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(result.assessment.outcome, "REJECT") @@ -494,6 +707,7 @@ def test_invalid_snapshot_still_assesses_once_and_keeps_static_reason(self) -> N scope="MEMBER", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(result.assessment.outcome, "REJECT") @@ -525,6 +739,7 @@ def test_unmapped_or_empty_product_caps_reject_fail_closed(self) -> None: scope="MEMBER", mandate_provenance=self._mandate(**cap_overrides), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(result.assessment.outcome, "REJECT") @@ -559,6 +774,7 @@ def test_decision_digest_binds_position_and_budget_execution_fields(self) -> Non scope="MEMBER", mandate_provenance=mandate, market_data={}, + candidate_identity=self._candidate(), ) for decision in decisions ) @@ -607,6 +823,7 @@ def evaluate(self, market_data): scope="MEMBER", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(result.assessment.outcome, "REJECT") @@ -629,6 +846,7 @@ def test_canonical_portfolio_snapshot_matches_mapping_normalization(self) -> Non scope="MEMBER", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) mapping_result = assess_with_evidence( decision, @@ -636,6 +854,7 @@ def test_canonical_portfolio_snapshot_matches_mapping_normalization(self) -> Non scope="MEMBER", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(canonical_result.assessment.outcome, "APPROVE") @@ -655,6 +874,7 @@ def test_value_target_uses_positive_finite_snapshot_equity(self) -> None: scope="MEMBER", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(approved.assessment.outcome, "APPROVE") @@ -675,6 +895,7 @@ def test_value_target_uses_positive_finite_snapshot_equity(self) -> None: scope="MEMBER", mandate_provenance=self._mandate(), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(rejected.assessment.outcome, "REJECT") self.assertEqual(rejected.decision.positions, ()) @@ -700,6 +921,7 @@ def test_mandate_rejects_budget_only_decision_above_authority(self) -> None: allowed_nonzero_assets=[], ), market_data={}, + candidate_identity=self._candidate(), ) self.assertEqual(result.assessment.outcome, "REJECT") diff --git a/tests/test_strategy_evidence_package_v2_contract.py b/tests/test_strategy_evidence_package_v2_contract.py index 041823d..aceb876 100644 --- a/tests/test_strategy_evidence_package_v2_contract.py +++ b/tests/test_strategy_evidence_package_v2_contract.py @@ -385,6 +385,7 @@ def test_accepts_exact_backtest_orchestrator_promotion_output(tmp_path: Path) -> from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore class Runner: + runner_kind = "real" @staticmethod def _result(start_date: date, end_date: date) -> BacktestResult: return BacktestResult(