304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""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
|