fix(risk): restore hard exposure and trailing-state gates

pyproject.toml

@@ -9,7 +9,7 @@ description = "Shared crypto strategy catalog and implementations"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@776fe71e57e2924fcd1c73126f41d244242240bb",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@b371322b948e4298920a7d8613b155245dcd5f8d",
 ]
 
 [tool.setuptools]

qsl.toml

@@ -4,5 +4,5 @@ upgrade_ring = "ring_b"
 [compat]
 bundle = "2026.07.4"
 requires = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@776fe71e57e2924fcd1c73126f41d244242240bb",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@b371322b948e4298920a7d8613b155245dcd5f8d",
 ]

src/crypto_strategies/entrypoints/__init__.py

@@ -2,6 +2,7 @@
 
 from collections.abc import Callable, Mapping
 from copy import deepcopy
+import math
 
 from quant_platform_kit.strategy_contracts import (
     BudgetIntent,
@@ -19,6 +20,10 @@
 )
 
 from ._common import apply_risk_gate, record_strategy_decision
+from crypto_strategies.strategies.crypto_live_pool_rotation.rotation import (
+    build_strategy_stop_evaluation,
+    evaluate_held_trend_stops,
+)
 
 
 """Unified crypto strategy entrypoints built on top of legacy core/rotation modules."""
@@ -116,6 +121,48 @@ def _set_symbol_trade_state(state, symbol, symbol_state):
     return _get_symbol_trade_state, _set_symbol_trade_state
 
 
+def _resolve_held_risk_symbols(
+    ctx: StrategyContext,
+    state,
+    *,
+    trend_universe_symbols,
+    get_symbol_trade_state_fn,
+):
+    held = {
+        str(symbol).strip().upper()
+        for symbol, payload in state.items()
+        if isinstance(payload, Mapping)
+        and payload.get("is_holding")
+        and str(symbol).strip().upper() != "BTCUSDT"
+    }
+    snapshot = _resolve_portfolio_snapshot(ctx)
+    candidate_symbols = {
+        str(symbol).strip().upper()
+        for symbol in trend_universe_symbols
+        if str(symbol).strip().upper() != "BTCUSDT"
+    }
+    for position in getattr(snapshot, "positions", ()) or ():
+        symbol = str(getattr(position, "symbol", "")).strip().upper()
+        if symbol and symbol != "BTCUSDT":
+            candidate_symbols.add(symbol)
+        quantity = getattr(position, "quantity", 0.0)
+        market_value = getattr(position, "market_value", 0.0)
+        values = (quantity, market_value)
+        if symbol and symbol != "BTCUSDT" and any(
+            isinstance(value, (int, float))
+            and not isinstance(value, bool)
+            and math.isfinite(float(value))
+            and float(value) != 0.0
+            for value in values
+        ):
+            held.add(symbol)
+    for symbol in candidate_symbols:
+        symbol_state = get_symbol_trade_state_fn(state, symbol)
+        if isinstance(symbol_state, Mapping) and symbol_state.get("is_holding"):
+            held.add(symbol)
+    return tuple(sorted(held))
+
+
 def _load_legacy_modules():
     from crypto_strategies.strategies.crypto_live_pool_rotation import core as legacy_core
     from crypto_strategies.strategies.crypto_live_pool_rotation import rotation as legacy_rotation
@@ -159,68 +206,75 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision
         weight_mode=str(config.get("weight_mode", "inverse_vol")),
     )
 
-    sell_reasons: dict[str, str] = {}
-    atr_multiplier = float(config.get("atr_multiplier", 2.5))
-    for symbol in trend_universe_symbols:
-        curr_price = prices.get(symbol)
-        if curr_price is None:
-            continue
-        reason = legacy_rotation.get_trend_sell_reason(
-            working_state,
-            symbol,
-            curr_price,
-            indicators_map.get(symbol),
-            selected_candidates,
-            atr_multiplier,
-            get_symbol_trade_state_fn=get_symbol_trade_state_fn,
-            set_symbol_trade_state_fn=set_symbol_trade_state_fn,
-            translate_fn=translator,
-        )
-        if reason:
-            sell_reasons[symbol] = str(reason)
-
-    eligible_buy_symbols, planned_trend_buys = legacy_rotation.plan_trend_buys(
+    atr_multiplier = config.get("atr_multiplier", 2.5)
+    held_risk_symbols = _resolve_held_risk_symbols(
+        ctx,
         working_state,
-        runtime_trend_universe={symbol: {"base_asset": symbol[:-4]} for symbol in trend_universe_symbols},
-        selected_candidates=selected_candidates,
-        trend_indicators=indicators_map,
+        trend_universe_symbols=trend_universe_symbols,
+        get_symbol_trade_state_fn=get_symbol_trade_state_fn,
+    )
+    sell_reasons, stop_input_blocked = evaluate_held_trend_stops(
+        working_state,
+        held_symbols=held_risk_symbols,
         prices=prices,
-        available_trend_buy_budget=float(budgets["trend_usdt_pool"]),
-        allow_new_trend_entries=bool(config.get("allow_new_trend_entries", True)),
+        indicators_map=indicators_map,
+        selected_candidates=selected_candidates,
+        atr_multiplier=atr_multiplier,
         get_symbol_trade_state_fn=get_symbol_trade_state_fn,
-        allocate_trend_buy_budget_fn=legacy_core.allocate_trend_buy_budget,
+        set_symbol_trade_state_fn=set_symbol_trade_state_fn,
+        translate_fn=translator,
     )
 
-    positions = [
-        PositionTarget(
-            symbol="BTCUSDT",
-            target_weight=float(budgets["btc_target_ratio"]),
-            role="core",
+    if stop_input_blocked:
+        eligible_buy_symbols, planned_trend_buys = (), {}
+        positions = []
+        budget_intents = ()
+    else:
+        eligible_buy_symbols, planned_trend_buys = legacy_rotation.plan_trend_buys(
+            working_state,
+            runtime_trend_universe={
+                symbol: {"base_asset": symbol[:-4]} for symbol in trend_universe_symbols
+            },
+            selected_candidates=selected_candidates,
+            trend_indicators=indicators_map,
+            prices=prices,
+            available_trend_buy_budget=float(budgets["trend_usdt_pool"]),
+            allow_new_trend_entries=bool(config.get("allow_new_trend_entries", True)),
+            get_symbol_trade_state_fn=get_symbol_trade_state_fn,
+            allocate_trend_buy_budget_fn=legacy_core.allocate_trend_buy_budget,
         )
-    ]
-    trend_target_ratio = float(budgets["trend_target_ratio"])
-    for symbol, payload in sorted(selected_candidates.items()):
-        positions.append(
+        positions = [
             PositionTarget(
-                symbol=symbol,
-                target_weight=trend_target_ratio * float(payload["weight"]),
-                role="trend_rotation",
+                symbol="BTCUSDT",
+                target_weight=float(budgets["btc_target_ratio"]),
+                role="core",
+            )
+        ]
+        trend_target_ratio = float(budgets["trend_target_ratio"])
+        for symbol, payload in sorted(selected_candidates.items()):
+            if symbol in sell_reasons:
+                continue
+            positions.append(
+                PositionTarget(
+                    symbol=symbol,
+                    target_weight=trend_target_ratio * float(payload["weight"]),
+                    role="trend_rotation",
+                )
             )
-        )
 
-    budget_intents = (
-        BudgetIntent(
-            name="btc_core_dca_pool",
-            symbol="BTCUSDT",
-            amount=float(budgets["dca_usdt_pool"]),
-            purpose="btc_core_accumulation",
-        ),
-        BudgetIntent(
-            name="trend_rotation_pool",
-            amount=float(budgets["trend_usdt_pool"]),
-            purpose="trend_rotation",
-        ),
-    )
+        budget_intents = (
+            BudgetIntent(
+                name="btc_core_dca_pool",
+                symbol="BTCUSDT",
+                amount=float(budgets["dca_usdt_pool"]),
+                purpose="btc_core_accumulation",
+            ),
+            BudgetIntent(
+                name="trend_rotation_pool",
+                amount=float(budgets["trend_usdt_pool"]),
+                purpose="trend_rotation",
+            ),
+        )
 
     risk_flags: tuple[str, ...] = ()
     if not btc_snapshot.get("regime_on"):
@@ -257,11 +311,37 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision
     }
     decision = StrategyDecision(
         positions=tuple(positions),
-        budgets=budget_intents,
-        risk_flags=risk_flags,
+        budgets=() if sell_reasons else budget_intents,
+        risk_flags=risk_flags + (("rejected:strategy_stop_input",) if stop_input_blocked else ()),
         diagnostics=diagnostics,
     )
+    if sell_reasons:
+        decision = StrategyDecision(
+            positions=(),
+            budgets=(),
+            risk_flags=decision.risk_flags,
+            diagnostics=decision.diagnostics,
+        )
     decision = apply_risk_gate(decision, ctx=ctx)
+    member_assessment = decision.diagnostics["member_risk_assessment"]
+    stop_outcome = "TRIGGERED" if sell_reasons else "CLEAR"
+    stop_action_result = "NOT_REQUIRED"
+    if stop_outcome == "TRIGGERED":
+        stop_action_result = "BLOCKED"
+    decision = StrategyDecision(
+        positions=decision.positions,
+        budgets=decision.budgets,
+        risk_flags=decision.risk_flags,
+        diagnostics={
+            **dict(decision.diagnostics),
+            "strategy_stop_evaluation": build_strategy_stop_evaluation(
+                evaluated_at=member_assessment["evaluated_at"],
+                decision_digest_sha256=member_assessment["decision_digest_sha256"],
+                outcome=stop_outcome,
+                action_result=stop_action_result,
+            ),
+        },
+    )
     record_strategy_decision(
         ctx,
         decision,

src/crypto_strategies/entrypoints/_common.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
 
 import logging
+import math
 from collections.abc import Mapping
+from dataclasses import asdict
 from typing import Any
 
-from quant_platform_kit.risk.gate import apply_risk_gate as _qpk_apply_risk_gate
+from quant_platform_kit.risk.gate import assess_with_evidence as _qpk_assess_with_evidence
 from quant_platform_kit.risk.gate import enrich_decision_risk_diagnostics
 from quant_platform_kit.risk.portfolio_diagnostics import extract_portfolio_risk_diagnostics
-from quant_platform_kit.strategy_contracts import PositionTarget, StrategyContext, StrategyDecision
+from quant_platform_kit.strategy_contracts import StrategyContext, StrategyDecision
 from quant_platform_kit.strategy_lifecycle.performance_monitor import PerformanceMonitor
 
 logger = logging.getLogger(__name__)
@@ -49,16 +51,16 @@ def apply_risk_gate(
     decision: StrategyDecision,
     *,
     ctx: StrategyContext | None = None,
-    max_single_weight: float = 1.0,
-    max_positions: int = 20,
-    max_total_exposure: float = 1.0,
+    max_single_weight: float | None = None,
     portfolio_snapshot: Any | None = None,
     market_data: Mapping[str, Any] | None = None,
 ) -> StrategyDecision:
-    """QPK unified risk gate: stop-loss, circuit breaker, concentration (task 8)."""
-    snapshot = portfolio_snapshot if portfolio_snapshot is not None else (
-        ctx.portfolio if ctx is not None else None
-    )
+    """Run the QPK MEMBER gate and propagate only its redacted assessment."""
+    snapshot = portfolio_snapshot
+    if snapshot is None and ctx is not None:
+        snapshot = ctx.portfolio
+        if snapshot is None:
+            snapshot = ctx.market_data.get("portfolio_snapshot")
     if snapshot is not None:
         portfolio_diag = extract_portfolio_risk_diagnostics(snapshot)
         decision = enrich_decision_risk_diagnostics(
@@ -68,11 +70,63 @@ def apply_risk_gate(
         )
     if market_data is None and ctx is not None:
         market_data = dict(ctx.market_data or {})
-    return _qpk_apply_risk_gate(
+    mandate_provenance = None if ctx is None else ctx.artifacts.get("mandate_provenance")
+    if not isinstance(mandate_provenance, Mapping):
+        mandate_provenance = {}
+    result = _qpk_assess_with_evidence(
         decision,
-        max_single_weight=max_single_weight,
-        max_positions=max_positions,
-        max_total_exposure=max_total_exposure,
-        portfolio_snapshot=snapshot,
-        market_data=market_data,
+        snapshot,
+        scope="MEMBER",
+        mandate_provenance=mandate_provenance,
+        market_data=market_data or {},
+    )
+    risk_flags = tuple(
+        dict.fromkeys(tuple(decision.risk_flags or ()) + tuple(result.decision.risk_flags or ()))
+    )
+    strategy_weights = []
+    strategy_weight_invalid = False
+    for position in result.decision.positions:
+        try:
+            weight = float(position.target_weight)
+        except (TypeError, ValueError):
+            strategy_weight_invalid = True
+            break
+        if not math.isfinite(weight):
+            strategy_weight_invalid = True
+            break
+        strategy_weights.append(abs(weight))
+    strategy_concentration_rejected = False
+    if max_single_weight is not None:
+        cap = float(max_single_weight)
+        if not math.isfinite(cap) or not 0.0 <= cap <= 1.0:
+            raise ValueError("max_single_weight must be finite and between 0 and 1")
+        strategy_concentration_rejected = strategy_weight_invalid or any(
+            weight > cap for weight in strategy_weights
+        )
+    if strategy_concentration_rejected:
+        risk_flags = tuple(dict.fromkeys(risk_flags + ("rejected:strategy_concentration",)))
+    strategy_position_count_rejected = len(result.decision.positions) > 20
+    if strategy_position_count_rejected:
+        risk_flags = tuple(dict.fromkeys(risk_flags + ("rejected:too_many_positions",)))
+    total_exposure = sum(strategy_weights)
+    strategy_total_exposure_rejected = (
+        strategy_weight_invalid
+        or not math.isfinite(total_exposure)
+        or total_exposure > 1.0 + 1e-9
+    )
+    if strategy_total_exposure_rejected:
+        risk_flags = tuple(dict.fromkeys(risk_flags + ("rejected:overexposed",)))
+    strategy_rejected = (
+        strategy_concentration_rejected
+        or strategy_position_count_rejected
+        or strategy_total_exposure_rejected
+    )
+    return StrategyDecision(
+        positions=() if strategy_rejected else result.decision.positions,
+        budgets=() if strategy_rejected else result.decision.budgets,
+        risk_flags=risk_flags,
+        diagnostics={
+            **dict(result.decision.diagnostics or {}),
+            "member_risk_assessment": asdict(result.assessment),
+        },
     )