diff --git a/src/hk_equity_strategies/entrypoints/__init__.py b/src/hk_equity_strategies/entrypoints/__init__.py index 14f01b0..43167ea 100644 --- a/src/hk_equity_strategies/entrypoints/__init__.py +++ b/src/hk_equity_strategies/entrypoints/__init__.py @@ -9,7 +9,7 @@ from hk_equity_strategies.strategies import hk_global_etf_tactical_rotation as global_etf_strategy from hk_equity_strategies.strategies import hk_low_vol_dividend_quality_snapshot as low_vol_dividend_strategy -from ._common import get_current_holdings, merge_runtime_config, require_market_data, weights_to_positions +from ._common import apply_risk_gate, get_current_holdings, merge_runtime_config, require_market_data, weights_to_positions def evaluate_hk_global_etf_tactical_rotation(ctx: StrategyContext) -> StrategyDecision: @@ -31,11 +31,12 @@ def evaluate_hk_global_etf_tactical_rotation(ctx: StrategyContext) -> StrategyDe risk_flags: tuple[str, ...] = () if has_cash_residual: risk_flags += ("cash_residual",) - return StrategyDecision( + decision = StrategyDecision( positions=weights_to_positions(weights), risk_flags=risk_flags, diagnostics=diagnostics, ) + return apply_risk_gate(decision) hk_global_etf_tactical_rotation_entrypoint = CallableStrategyEntrypoint( @@ -63,11 +64,12 @@ def evaluate_hk_low_vol_dividend_quality_snapshot(ctx: StrategyContext) -> Strat risk_flags: tuple[str, ...] = () if has_cash_residual: risk_flags += ("cash_residual",) - return StrategyDecision( + decision = StrategyDecision( positions=weights_to_positions(weights), risk_flags=risk_flags, diagnostics=diagnostics, ) + return apply_risk_gate(decision) hk_low_vol_dividend_quality_snapshot_entrypoint = CallableStrategyEntrypoint( @@ -83,7 +85,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 - return _eval(ctx) + return apply_risk_gate(_eval(ctx)) from hk_equity_strategies.combo_manifests import hk_equity_combo_manifest # noqa: E402 — intentional late import diff --git a/src/hk_equity_strategies/entrypoints/_common.py b/src/hk_equity_strategies/entrypoints/_common.py index d5e88da..47b84e4 100644 --- a/src/hk_equity_strategies/entrypoints/_common.py +++ b/src/hk_equity_strategies/entrypoints/_common.py @@ -1,8 +1,105 @@ from __future__ import annotations +import logging from typing import Any -from quant_platform_kit.strategy_contracts import PositionTarget, StrategyContext +from quant_platform_kit.strategy_contracts import PositionTarget, StrategyContext, StrategyDecision + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# 风控硬门 — 每个 entrypoint 返回 StrategyDecision 前必须调用 +# --------------------------------------------------------------------------- + + +def apply_risk_gate( + decision: StrategyDecision, + *, + max_single_weight: float = 1.0, + max_positions: int = 20, + max_total_exposure: float = 1.0, +) -> 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%}", + }, + ) + + # 通过 + risk_flags.append("risk_gate:passed") + return StrategyDecision( + positions=decision.positions, + risk_flags=tuple(risk_flags), + diagnostics={**(decision.diagnostics or {}), "risk_gate": "APPROVE"}, + ) + def merge_runtime_config(default_config: dict[str, object], ctx: StrategyContext) -> dict[str, object]: