From 4cfd8a1a181dce9ea1feb8f023d3f9be63d9371a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 9 Jul 2026 06:20:15 +0800 Subject: [PATCH] feat(risk): wire entrypoints to QPK stop-loss diagnostics (task 8) Delegate apply_risk_gate to QuantPlatformKit and enrich decisions from ctx.portfolio unrealized_pnl_pct / consecutive_losses before gating. Co-Authored-By: Claude Co-authored-by: Cursor --- .../entrypoints/__init__.py | 6 +- .../entrypoints/_common.py | 104 +++++------------- tests/test_entrypoint_risk_gate.py | 24 ++++ 3 files changed, 53 insertions(+), 81 deletions(-) create mode 100644 tests/test_entrypoint_risk_gate.py diff --git a/src/hk_equity_strategies/entrypoints/__init__.py b/src/hk_equity_strategies/entrypoints/__init__.py index 8aab40b..9f29d01 100644 --- a/src/hk_equity_strategies/entrypoints/__init__.py +++ b/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, diff --git a/src/hk_equity_strategies/entrypoints/_common.py b/src/hk_equity_strategies/entrypoints/_common.py index a8e20b2..ec656cc 100644 --- a/src/hk_equity_strategies/entrypoints/_common.py +++ b/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, ) diff --git a/tests/test_entrypoint_risk_gate.py b/tests/test_entrypoint_risk_gate.py new file mode 100644 index 0000000..8bf9f1c --- /dev/null +++ b/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