feat(risk): wire QPK stop-loss diagnostics (task 8)

src/hk_equity_strategies/entrypoints/__init__.py

@@ -43,7 +43,7 @@ def evaluate_hk_global_etf_tactical_rotation(ctx: StrategyContext) -> StrategyDe
         risk_flags=risk_flags,
         diagnostics=diagnostics,
     )
-    decision = apply_risk_gate(decision)
+    decision = apply_risk_gate(decision, ctx=ctx)
     record_strategy_decision(
         ctx,
         decision,
@@ -83,7 +83,7 @@ def evaluate_hk_low_vol_dividend_quality_snapshot(ctx: StrategyContext) -> Strat
         risk_flags=risk_flags,
         diagnostics=diagnostics,
     )
-    decision = apply_risk_gate(decision)
+    decision = apply_risk_gate(decision, ctx=ctx)
     record_strategy_decision(
         ctx,
         decision,
@@ -106,7 +106,7 @@ def evaluate_hk_low_vol_dividend_quality_snapshot(ctx: StrategyContext) -> Strat
 
 def evaluate_hk_equity_combo(ctx: StrategyContext) -> StrategyDecision:
     from hk_equity_strategies.combo_entrypoints import evaluate_hk_equity_combo as _eval
-    decision = apply_risk_gate(_eval(ctx))
+    decision = apply_risk_gate(_eval(ctx), ctx=ctx)
     record_strategy_decision(
         ctx,
         decision,

src/hk_equity_strategies/entrypoints/_common.py

@@ -1,8 +1,12 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Mapping
 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 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_lifecycle.performance_monitor import PerformanceMonitor
 
@@ -40,89 +44,33 @@ def record_strategy_decision(
 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,
+    portfolio_snapshot: Any | None = None,
+    market_data: Mapping[str, Any] | None = None,
 ) -> StrategyDecision:
-    """对所有 StrategyDecision 施加硬风控门。
-
-    检查项：
-    1. 单仓位集中度（> max_single_weight → REJECT，默认 100% 即不限制）
-    2. 持仓数量（> max_positions → REJECT）
-    3. 总仓位超限（> max_total_exposure → REJECT）
-
-    各策略类型可根据自身特点调整门限：
-    - ETF 轮动：max_single_weight=1.0（ETF 本身就是分散的篮子）
-    - 个股精选：max_single_weight=0.10
-    - 加密货币：max_single_weight=0.20, max_positions=10
-
-    如果 REJECT，返回空仓决策并标注拒绝原因。
-    这个函数不可绕过 —— AGENTS.md 要求所有 entrypoint 必须调用。
-    """
-    positions = decision.positions or ()
-    risk_flags = list(decision.risk_flags or ())
-
-    # 空仓放行（risk_off 场景）
-    if not positions:
-        return decision
-
-    # 1. 集中度检查（默认不限制，由策略自行设定）
-    if max_single_weight < 1.0:
-        for p in positions:
-            weight = abs(float(p.target_weight))
-            if weight > max_single_weight:
-                logger.warning(
-                    "risk_gate REJECT concentration: symbol=%s weight=%.2f%% limit=%.0f%%",
-                    p.symbol, weight * 100, max_single_weight * 100,
-                )
-                return StrategyDecision(
-                    positions=(),
-                    risk_flags=("rejected:concentration",),
-                    diagnostics={
-                        **(decision.diagnostics or {}),
-                        "risk_gate": "REJECT",
-                        "reason": f"{p.symbol} {weight:.1%} > {max_single_weight:.0%} 上限",
-                    },
-                )
-
-    # 2. 持仓数量检查
-    if len(positions) > max_positions:
-        logger.warning(
-            "risk_gate REJECT position_count: %d > %d", len(positions), max_positions,
-        )
-        return StrategyDecision(
-            positions=(),
-            risk_flags=("rejected:too_many_positions",),
-            diagnostics={
-                **(decision.diagnostics or {}),
-                "risk_gate": "REJECT",
-                "reason": f"{len(positions)} 个持仓 > {max_positions} 上限",
-            },
-        )
-
-    # 3. 总仓位检查
-    total_weight = sum(abs(float(p.target_weight)) for p in positions)
-    if total_weight > max_total_exposure + 1e-9:
-        logger.warning(
-            "risk_gate REJECT total_exposure: %.2f%% > %.0f%%",
-            total_weight * 100, max_total_exposure * 100,
-        )
-        return StrategyDecision(
-            positions=(),
-            risk_flags=("rejected:overexposed",),
-            diagnostics={
-                **(decision.diagnostics or {}),
-                "risk_gate": "REJECT",
-                "reason": f"总仓位 {total_weight:.1%} > {max_total_exposure:.0%}",
-            },
+    """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
+    )
+    if snapshot is not None:
+        portfolio_diag = extract_portfolio_risk_diagnostics(snapshot)
+        decision = enrich_decision_risk_diagnostics(
+            decision,
+            unrealized_pnl_pct=portfolio_diag.get("unrealized_pnl_pct"),
+            consecutive_losses=portfolio_diag.get("consecutive_losses"),
         )
-
-    # 通过
-    risk_flags.append("risk_gate:passed")
-    return StrategyDecision(
-        positions=decision.positions,
-        risk_flags=tuple(risk_flags),
-        diagnostics={**(decision.diagnostics or {}), "risk_gate": "APPROVE"},
+    if market_data is None and ctx is not None:
+        market_data = dict(ctx.market_data or {})
+    return _qpk_apply_risk_gate(
+        decision,
+        max_single_weight=max_single_weight,
+        max_positions=max_positions,
+        max_total_exposure=max_total_exposure,
+        portfolio_snapshot=snapshot,
+        market_data=market_data,
     )
 
 

tests/test_entrypoint_risk_gate.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from quant_platform_kit.common.models import PortfolioSnapshot, Position
+from quant_platform_kit.strategy_contracts import PositionTarget, StrategyContext, StrategyDecision
+
+from hk_equity_strategies.entrypoints._common import apply_risk_gate
+
+
+def test_apply_risk_gate_enriches_stop_loss_diagnostics_from_portfolio() -> None:
+    snapshot = PortfolioSnapshot(
+        as_of=datetime(2026, 7, 9, tzinfo=timezone.utc),
+        total_equity=1000.0,
+        positions=(
+            Position(symbol="2800", quantity=100.0, market_value=700.0, average_cost=10.0),
+        ),
+        metadata={"consecutive_losses": 2},
+    )
+    ctx = StrategyContext(as_of=snapshot.as_of, portfolio=snapshot, market_data={}, state={}, runtime_config={})
+    decision = StrategyDecision(positions=(PositionTarget(symbol="2800", target_weight=0.5),))
+    result = apply_risk_gate(decision, ctx=ctx)
+    assert result.positions == ()
+    assert "rejected:stop_loss" in result.risk_flags