"""Central registry for strategy identity and display metadata.""" from __future__ import annotations from dataclasses import dataclass, field LONG_INTRADAY_MOMENTUM_STRATEGY = "long_intraday_momentum_15m_1h_v1" LONG_MOMENTUM_BREAKOUT_STRATEGY = LONG_INTRADAY_MOMENTUM_STRATEGY LONG_SECOND_WAVE_PULLBACK_STRATEGY = "long_second_wave_pullback_1h_v1" LONG_COMPRESSION_BREAKOUT_STRATEGY = "long_compression_breakout_1h_4h_v1" LONG_BOX_RETEST_4H_STRATEGY = "long_box_retest_4h_v1" SHORT_BREAKDOWN_RETEST_STRATEGY = "short_breakdown_retest_1h_v1" SHORT_WEAK_BOUNCE_FAILURE_STRATEGY = "short_weak_bounce_failure_15m_1h_v1" # Compatibility aliases for old imports. These aliases intentionally map old # names to the new active strategy pool so new data never emits retired codes. MAIN_COMPOSITE_STRATEGY = LONG_MOMENTUM_BREAKOUT_STRATEGY BOX_RETEST_1H_STRATEGY = LONG_SECOND_WAVE_PULLBACK_STRATEGY BOX_RETEST_4H_STRATEGY = LONG_BOX_RETEST_4H_STRATEGY VOLUME_IGNITION_1H_STRATEGY = LONG_INTRADAY_MOMENTUM_STRATEGY COMPRESSION_BREAKOUT_4H_STRATEGY = LONG_COMPRESSION_BREAKOUT_STRATEGY INTRADAY_MOMENTUM_15M_STRATEGY = LONG_INTRADAY_MOMENTUM_STRATEGY BREAKDOWN_RETEST_SHORT_1H_STRATEGY = SHORT_BREAKDOWN_RETEST_STRATEGY @dataclass(frozen=True) class StrategyDefinition: strategy_code: str strategy_name: str description: str = "" direction: str = "long" frequency_profile: str = "intraday" mode: str = "paper_only" status: str = "active" entry_gate_config: dict = field(default_factory=dict) paper_config: dict = field(default_factory=dict) STRATEGY_DEFINITIONS: dict[str, StrategyDefinition] = { LONG_INTRADAY_MOMENTUM_STRATEGY: StrategyDefinition( strategy_code=LONG_INTRADAY_MOMENTUM_STRATEGY, strategy_name="多头日内动量启动", description="15m当前突破叠加1H成交量/波动增强，捕捉山寨币日内到1-3天启动段。", direction="long", frequency_profile="intraday", mode="paper_enabled", entry_gate_config={ "min_entry_score_buy_now": 3, "min_entry_score_wait_pullback": 2, "min_rr_buy_now": 1.25, "breakout_distance_wait_pct": 8, "gain_24h_wait_pct": 10, }, paper_config={ "frequency_profile": "intraday_trading", "target_trades_per_day_min": 3, "target_trades_per_day_max": 5, "entry_min_rec_score": 25, "order_min_rec_score": 25, "entry_min_rr": 1.25, "order_min_rr": 1.25, "order_min_distance_to_entry_pct": 0, "order_require_current_trigger": True, "order_expire_hours": 8, "trailing_activate_pnl_pct": 2.0, "trailing_volatility_min_activation_pct": 1.8, "dynamic_leverage_enabled": True, "dynamic_leverage_min": 1, }, ), LONG_SECOND_WAVE_PULLBACK_STRATEGY: StrategyDefinition( strategy_code=LONG_SECOND_WAVE_PULLBACK_STRATEGY, strategy_name="多头二波回踩", description="强势榜或放量币第一波后回踩EMA、箱体上沿或前高转支撑，再次承接的短线策略。", direction="long", frequency_profile="intraday", mode="paper_enabled", entry_gate_config={ "min_entry_score_buy_now": 2, "min_entry_score_wait_pullback": 0, "min_rr_buy_now": 1.25, "max_wait_pullback_deviation_pct": 10, "breakout_distance_wait_pct": 12, "gain_24h_wait_pct": 18, }, paper_config={ "frequency_profile": "intraday_trading", "target_trades_per_day_min": 3, "target_trades_per_day_max": 5, "entry_min_rec_score": 22, "order_min_rec_score": 22, "entry_min_rr": 1.25, "order_min_rr": 1.25, "order_min_distance_to_entry_pct": 0, "order_require_current_trigger": False, "order_expire_hours": 10, "trailing_activate_pnl_pct": 2.0, "trailing_volatility_min_activation_pct": 1.8, "dynamic_leverage_enabled": True, "dynamic_leverage_min": 1, }, ), LONG_COMPRESSION_BREAKOUT_STRATEGY: StrategyDefinition( strategy_code=LONG_COMPRESSION_BREAKOUT_STRATEGY, strategy_name="多头压缩突破", description="1H/4H低波动压缩后突然放量突破，捕捉启动前后第一段。", direction="long", frequency_profile="intraday", mode="paper_enabled", entry_gate_config={ "min_entry_score_buy_now": 2, "min_entry_score_wait_pullback": 1, "min_rr_buy_now": 1.25, "breakout_distance_wait_pct": 10, "gain_24h_wait_pct": 12, }, paper_config={ "frequency_profile": "intraday_trading", "target_trades_per_day_min": 2, "target_trades_per_day_max": 4, "entry_min_rec_score": 24, "order_min_rec_score": 24, "entry_min_rr": 1.25, "order_min_rr": 1.25, "order_min_distance_to_entry_pct": 0, "order_require_current_trigger": True, "order_expire_hours": 8, "trailing_activate_pnl_pct": 2.0, "trailing_volatility_min_activation_pct": 1.8, "dynamic_leverage_enabled": True, "dynamic_leverage_min": 1, }, ), LONG_BOX_RETEST_4H_STRATEGY: StrategyDefinition( strategy_code=LONG_BOX_RETEST_4H_STRATEGY, strategy_name="多头4H箱体回踩", description="4H箱体突破后第一次或第二次回踩箱体上沿/均线承接。", direction="long", frequency_profile="intraday", mode="paper_enabled", entry_gate_config={ "min_entry_score_buy_now": 2, "min_entry_score_wait_pullback": 0, "min_rr_buy_now": 1.3, "max_wait_pullback_deviation_pct": 12, "breakout_distance_wait_pct": 14, "gain_24h_wait_pct": 18, }, paper_config={ "frequency_profile": "intraday_trading", "target_trades_per_day_min": 1, "target_trades_per_day_max": 3, "entry_min_rec_score": 24, "order_min_rec_score": 24, "entry_min_rr": 1.3, "order_min_rr": 1.3, "order_min_distance_to_entry_pct": 0, "order_require_current_trigger": False, "order_expire_hours": 12, "trailing_activate_pnl_pct": 2.2, "trailing_volatility_min_activation_pct": 2.0, "dynamic_leverage_enabled": True, "dynamic_leverage_min": 1, }, ), SHORT_BREAKDOWN_RETEST_STRATEGY: StrategyDefinition( strategy_code=SHORT_BREAKDOWN_RETEST_STRATEGY, strategy_name="空头破位反抽", description="1H支撑或箱体下沿破位后反抽失败，叠加15m弱确认和相对弱势的空头短线策略。", direction="short", frequency_profile="intraday", mode="paper_enabled", entry_gate_config={ "direction": "short", "min_entry_score_buy_now": 2, "min_entry_score_wait_pullback": 1, "min_rr_buy_now": 1.3, "breakdown_distance_wait_pct": 10, "max_retest_deviation_pct": 8, }, paper_config={ "frequency_profile": "intraday_trading", "target_trades_per_day_min": 3, "target_trades_per_day_max": 5, "entry_min_rec_score": 18, "order_min_rec_score": 18, "entry_min_rr": 1.3, "order_min_rr": 1.3, "order_min_distance_to_entry_pct": 0, "order_require_current_trigger": False, "order_expire_hours": 8, "trailing_activate_pnl_pct": 2.0, "trailing_volatility_min_activation_pct": 1.8, "dynamic_leverage_enabled": True, "dynamic_leverage_min": 1, }, ), SHORT_WEAK_BOUNCE_FAILURE_STRATEGY: StrategyDefinition( strategy_code=SHORT_WEAK_BOUNCE_FAILURE_STRATEGY, strategy_name="空头弱反弹失败", description="弱势环境下15m/1H反弹无量，反抽均线或前支撑后再次转弱。", direction="short", frequency_profile="intraday", mode="paper_enabled", entry_gate_config={ "direction": "short", "min_entry_score_buy_now": 2, "min_entry_score_wait_pullback": 1, "min_rr_buy_now": 1.3, "max_retest_deviation_pct": 8, }, paper_config={ "frequency_profile": "intraday_trading", "target_trades_per_day_min": 1, "target_trades_per_day_max": 3, "entry_min_rec_score": 18, "order_min_rec_score": 18, "entry_min_rr": 1.3, "order_min_rr": 1.3, "order_min_distance_to_entry_pct": 0, "order_require_current_trigger": True, "order_expire_hours": 6, "trailing_activate_pnl_pct": 2.0, "trailing_volatility_min_activation_pct": 1.8, "dynamic_leverage_enabled": True, "dynamic_leverage_min": 1, }, ), } def normalize_strategy_code(strategy_code: str | None) -> str: code = str(strategy_code or "").strip() legacy_map = { "main_composite_v1": LONG_INTRADAY_MOMENTUM_STRATEGY, "long_momentum_breakout_15m_1h_v1": LONG_INTRADAY_MOMENTUM_STRATEGY, "volume_ignition_1h_v1": LONG_INTRADAY_MOMENTUM_STRATEGY, "intraday_momentum_15m_v1": LONG_INTRADAY_MOMENTUM_STRATEGY, "box_retest_1h_v1": LONG_SECOND_WAVE_PULLBACK_STRATEGY, "box_retest_4h_v1": LONG_BOX_RETEST_4H_STRATEGY, "compression_breakout_4h_v1": LONG_COMPRESSION_BREAKOUT_STRATEGY, "breakdown_retest_short_1h_v1": SHORT_BREAKDOWN_RETEST_STRATEGY, } if code in legacy_map: return legacy_map[code] return code or LONG_INTRADAY_MOMENTUM_STRATEGY def strategy_definition(strategy_code: str | None) -> StrategyDefinition: code = normalize_strategy_code(strategy_code) return STRATEGY_DEFINITIONS.get( code, StrategyDefinition( strategy_code=code, strategy_name=code, description="未注册策略，请补充 strategy_registry。", status="unknown", ), ) def strategy_label(strategy_code: str | None) -> str: return strategy_definition(strategy_code).strategy_name def strategy_direction(strategy_code: str | None) -> str: direction = str(strategy_definition(strategy_code).direction or "long").strip().lower() return direction if direction in {"long", "short", "both"} else "long" def is_strategy_allowed_for_side(strategy_code: str | None, side: str | None) -> bool: direction = strategy_direction(strategy_code) normalized_side = "short" if str(side or "").strip().lower() == "short" else "long" return direction == "both" or direction == normalized_side def strategy_entry_gate_config(strategy_code: str | None) -> dict: return dict(strategy_definition(strategy_code).entry_gate_config or {}) def strategy_paper_config(strategy_code: str | None) -> dict: return dict(strategy_definition(strategy_code).paper_config or {}) def registered_strategy_codes() -> list[str]: return list(STRATEGY_DEFINITIONS.keys()) def strategy_catalog_seed_rows(strategy_version: str = "") -> list[dict]: rows = [] for item in STRATEGY_DEFINITIONS.values(): rows.append( { "strategy_code": item.strategy_code, "strategy_name": item.strategy_name, "strategy_version": strategy_version or "", "status": item.status, "mode": item.mode, "description": item.description, } ) return rows