alphax/app/core/strategy_registry.py
2026-05-28 20:30:38 +08:00

126 lines
4.1 KiB
Python

"""Central registry for strategy identity and display metadata."""
from __future__ import annotations
from dataclasses import dataclass, field
MAIN_COMPOSITE_STRATEGY = "main_composite_v1"
BOX_RETEST_1H_STRATEGY = "box_retest_1h_v1"
BOX_RETEST_4H_STRATEGY = "box_retest_4h_v1"
@dataclass(frozen=True)
class StrategyDefinition:
strategy_code: str
strategy_name: str
description: str = ""
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] = {
MAIN_COMPOSITE_STRATEGY: StrategyDefinition(
strategy_code=MAIN_COMPOSITE_STRATEGY,
strategy_name="综合确认主链路",
description="迁移期兼容主链路,承载现有综合筛选与确认逻辑。",
mode="paper_enabled",
),
BOX_RETEST_1H_STRATEGY: StrategyDefinition(
strategy_code=BOX_RETEST_1H_STRATEGY,
strategy_name="1H箱体突破回踩",
description="小时级底部箱体突破后回踩箱体上沿或EMA承接的早期结构策略。",
mode="paper_only",
entry_gate_config={
"min_entry_score_buy_now": 1,
"min_entry_score_wait_pullback": 0,
"min_rr_buy_now": 1.2,
"max_wait_pullback_deviation_pct": 20,
"breakout_distance_wait_pct": 25,
"gain_24h_wait_pct": 12,
},
paper_config={
"entry_min_rr": 1.5,
"order_min_rr": 1.5,
"order_min_distance_to_entry_pct": 0,
"order_require_current_trigger": False,
"dynamic_leverage_enabled": True,
"dynamic_leverage_min": 3,
},
),
BOX_RETEST_4H_STRATEGY: StrategyDefinition(
strategy_code=BOX_RETEST_4H_STRATEGY,
strategy_name="4H箱体突破回踩",
description="底部箱体突破后回踩箱体上沿或EMA承接的结构策略。",
mode="paper_only",
entry_gate_config={
"min_entry_score_buy_now": 1,
"min_entry_score_wait_pullback": 0,
"min_rr_buy_now": 1.2,
"max_wait_pullback_deviation_pct": 20,
"breakout_distance_wait_pct": 25,
"gain_24h_wait_pct": 12,
},
paper_config={
"entry_min_rr": 1.5,
"order_min_rr": 1.5,
"order_min_distance_to_entry_pct": 0,
"order_require_current_trigger": False,
"dynamic_leverage_enabled": True,
"dynamic_leverage_min": 3,
},
),
}
def normalize_strategy_code(strategy_code: str | None) -> str:
code = str(strategy_code or "").strip()
return code or MAIN_COMPOSITE_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_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