diff --git a/src/crypto_strategies/entrypoints/__init__.py b/src/crypto_strategies/entrypoints/__init__.py index f2e9af8..91b2d9d 100644 --- a/src/crypto_strategies/entrypoints/__init__.py +++ b/src/crypto_strategies/entrypoints/__init__.py @@ -18,6 +18,8 @@ crypto_trend_rotation_manifest, ) +from ._common import apply_risk_gate + """Unified crypto strategy entrypoints built on top of legacy core/rotation modules.""" @@ -253,12 +255,13 @@ def evaluate_crypto_live_pool_rotation(ctx: StrategyContext) -> StrategyDecision **dict(ctx.artifacts.get("trend_pool_contract", {})), }, } - return StrategyDecision( + decision = StrategyDecision( positions=tuple(positions), budgets=budget_intents, risk_flags=risk_flags, diagnostics=diagnostics, ) + return apply_risk_gate(decision) crypto_live_pool_rotation_entrypoint = CallableStrategyEntrypoint( @@ -334,12 +337,13 @@ def evaluate_crypto_btc_dca(ctx: StrategyContext) -> StrategyDecision: "mayer_multiple": metadata.get("mayer_multiple", float("nan")), } - return StrategyDecision( + decision = StrategyDecision( positions=tuple(positions), budgets=budget_intents, risk_flags=risk_flags, diagnostics=diagnostics, ) + return apply_risk_gate(decision, max_single_weight=0.50) crypto_btc_dca_entrypoint = CallableStrategyEntrypoint( @@ -416,12 +420,13 @@ def evaluate_crypto_trend_rotation(ctx: StrategyContext) -> StrategyDecision: "profile": metadata.get("profile"), } - return StrategyDecision( + decision = StrategyDecision( positions=tuple(positions), budgets=budget_intents, risk_flags=risk_flags, diagnostics=diagnostics, ) + return apply_risk_gate(decision, max_single_weight=0.50) crypto_trend_rotation_entrypoint = CallableStrategyEntrypoint( @@ -604,12 +609,13 @@ def evaluate_crypto_equity_combo(ctx: StrategyContext) -> StrategyDecision: }, } - return StrategyDecision( + decision = StrategyDecision( positions=tuple(positions), budgets=budget_intents, risk_flags=risk_flags, diagnostics=diagnostics, ) + return apply_risk_gate(decision) crypto_equity_combo_entrypoint = CallableStrategyEntrypoint( diff --git a/src/crypto_strategies/entrypoints/_common.py b/src/crypto_strategies/entrypoints/_common.py new file mode 100644 index 0000000..4101a85 --- /dev/null +++ b/src/crypto_strategies/entrypoints/_common.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import logging + +from quant_platform_kit.strategy_contracts import PositionTarget, StrategyDecision + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# 风控硬门 — 每个 entrypoint 返回 StrategyDecision 前必须调用 +# --------------------------------------------------------------------------- + + +def _position_weight(position: PositionTarget) -> float | None: + if position.target_weight is not None: + return abs(float(position.target_weight)) + return None + + +def _reject_risk_gate( + decision: StrategyDecision, + *, + risk_flag: str, + reason: str, +) -> StrategyDecision: + return StrategyDecision( + positions=(), + budgets=decision.budgets, + risk_flags=(*decision.risk_flags, risk_flag), + diagnostics={ + **(decision.diagnostics or {}), + "risk_gate": "REJECT", + "reason": reason, + }, + ) + + +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 () + + if not positions: + return StrategyDecision( + positions=decision.positions, + budgets=decision.budgets, + risk_flags=decision.risk_flags, + diagnostics={**(decision.diagnostics or {}), "risk_gate": "APPROVE"}, + ) + + weight_positions = [p for p in positions if _position_weight(p) is not None] + + if max_single_weight < 1.0 and weight_positions: + for p in weight_positions: + weight = _position_weight(p) + assert weight is not None + 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 _reject_risk_gate( + decision, + risk_flag="rejected:concentration", + reason=f"{p.symbol} {weight:.1%} > {max_single_weight:.0%} 上限", + ) + + if len(positions) > max_positions: + logger.warning( + "risk_gate REJECT position_count: %d > %d", len(positions), max_positions, + ) + return _reject_risk_gate( + decision, + risk_flag="rejected:too_many_positions", + reason=f"{len(positions)} 个持仓 > {max_positions} 上限", + ) + + if weight_positions: + total_weight = sum(_position_weight(p) or 0.0 for p in weight_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 _reject_risk_gate( + decision, + risk_flag="rejected:overexposed", + reason=f"总仓位 {total_weight:.1%} > {max_total_exposure:.0%}", + ) + + return StrategyDecision( + positions=decision.positions, + budgets=decision.budgets, + risk_flags=decision.risk_flags, + diagnostics={**(decision.diagnostics or {}), "risk_gate": "APPROVE"}, + )