chore(deps): align QPK pin to b371322b948e

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,31 @@ 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):
+    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)
+    for position in getattr(snapshot, "positions", ()) or ():
+        symbol = str(getattr(position, "symbol", "")).strip().upper()
+        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)
+    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,25 +189,22 @@ 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)
+    held_risk_symbols = _resolve_held_risk_symbols(
+        ctx,
+        working_state,
+    )
+    sell_reasons, stop_input_blocked = evaluate_held_trend_stops(
+        working_state,
+        held_symbols=held_risk_symbols,
+        prices=prices,
+        indicators_map=indicators_map,
+        selected_candidates=selected_candidates,
+        atr_multiplier=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,
+    )
 
     eligible_buy_symbols, planned_trend_buys = legacy_rotation.plan_trend_buys(
         working_state,
@@ -200,6 +227,8 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision
     ]
     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,
@@ -257,11 +286,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,44 @@ 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_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 = any(
+            position.target_weight is not None
+            and (
+                not math.isfinite(float(position.target_weight))
+                or abs(float(position.target_weight)) > cap
+            )
+            for position in result.decision.positions
+        )
+    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",)))
+    strategy_rejected = strategy_concentration_rejected or strategy_position_count_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),
+        },
     )

src/crypto_strategies/strategies/crypto_live_pool_rotation/rotation.py

@@ -2,7 +2,79 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from datetime import datetime, timezone
+import math
+
+
+_STRATEGY_STOP_POLICY_ID = "crypto_live_pool_rotation.executable_stop"
+_STRATEGY_STOP_POLICY_VERSION = "v1"
+
+
+def _finite_number(value):
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return None
+    number = float(value)
+    return number if math.isfinite(number) else None
+
+
+def evaluate_held_trend_stops(
+    state,
+    *,
+    held_symbols,
+    prices,
+    indicators_map,
+    selected_candidates,
+    atr_multiplier,
+    get_symbol_trade_state_fn,
+    set_symbol_trade_state_fn,
+    translate_fn,
+):
+    """Evaluate every held risk symbol; incomplete inputs block CLEAR."""
+    sell_reasons = {}
+    input_blocked = False
+    for symbol in _normalize_symbol_list(held_symbols):
+        symbol_state = get_symbol_trade_state_fn(state, symbol)
+        indicators = indicators_map.get(symbol)
+        curr_price = _finite_number(prices.get(symbol))
+        atr = _finite_number(indicators.get("atr14")) if isinstance(indicators, Mapping) else None
+        sma60 = _finite_number(indicators.get("sma60")) if isinstance(indicators, Mapping) else None
+        if not symbol_state.get("is_holding") or curr_price is None or atr is None or sma60 is None:
+            input_blocked = True
+            sell_reasons[symbol] = translate_fn("trend_sell_reason_missing_stop_input")
+            continue
+        reason = get_trend_sell_reason(
+            state,
+            symbol,
+            curr_price,
+            indicators,
+            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=translate_fn,
+        )
+        if reason:
+            sell_reasons[symbol] = str(reason)
+    return sell_reasons, input_blocked
+
+
+def build_strategy_stop_evaluation(
+    *,
+    evaluated_at,
+    decision_digest_sha256,
+    outcome,
+    action_result,
+):
+    return {
+        "evaluated": True,
+        "policy_id": _STRATEGY_STOP_POLICY_ID,
+        "policy_version": _STRATEGY_STOP_POLICY_VERSION,
+        "evaluated_at": evaluated_at,
+        "decision_digest_sha256": decision_digest_sha256,
+        "outcome": outcome,
+        "action_result": action_result,
+    }
 
 
 def _normalize_symbol_list(symbols):