feat: bind promotion runners to risk provenance

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,

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,