claude_dev
3a0e75fdc1
- strategy_registry 枚举 vnpy_ctastrategy 策略（兜底 STRATEGY_NAMES）
- /strategy/list、/strategy/{name}/params
- /task/{id}/equity-curve、/daily-pnl、/trades（BacktestResult JSON 化）
- /kline（read_db_daily 历史 K 线）
- 9 tests passed（4 strategy_registry + 5 routes）
57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
"""Enumerate vnpy_ctastrategy CTA strategies + their parameters.
|
|
|
|
Used by the backtest UI dropdown and dynamic parameter form. Falls back to a
|
|
static name list when vnpy_ctastrategy is not importable (e.g. local dev).
|
|
Task S1.3.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import pkgutil
|
|
|
|
# Fallback strategy names (when vnpy_ctastrategy import fails).
|
|
STRATEGY_NAMES: list[str] = ["DoubleMaStrategy", "BollChannelStrategy", "AtrRsiStrategy"]
|
|
|
|
|
|
def _load_strategy_classes() -> dict[str, type]:
|
|
"""Import all Strategy classes from vnpy_ctastrategy.strategies."""
|
|
classes: dict[str, type] = {}
|
|
try:
|
|
mod = importlib.import_module("vnpy_ctastrategy.strategies")
|
|
for _, name, _ in pkgutil.iter_modules(mod.__path__):
|
|
try:
|
|
m = importlib.import_module(f"vnpy_ctastrategy.strategies.{name}")
|
|
for attr in dir(m):
|
|
obj = getattr(m, attr)
|
|
if isinstance(obj, type) and attr.endswith("Strategy") and hasattr(obj, "parameters"):
|
|
classes[attr] = obj
|
|
except Exception:
|
|
continue
|
|
except Exception:
|
|
pass
|
|
return classes
|
|
|
|
|
|
def list_strategies() -> list[dict]:
|
|
"""Return [{name, class_name}, ...] for the UI dropdown."""
|
|
classes = _load_strategy_classes()
|
|
if classes:
|
|
return [{"name": n, "class_name": n} for n in sorted(classes)]
|
|
return [{"name": n, "class_name": n} for n in STRATEGY_NAMES]
|
|
|
|
|
|
def strategy_params(name: str) -> dict:
|
|
"""Return {parameters: [...], defaults: {...}} for a strategy's dynamic form."""
|
|
classes = _load_strategy_classes()
|
|
cls = classes.get(name)
|
|
if cls is None:
|
|
return {"parameters": [], "defaults": {}}
|
|
params = list(getattr(cls, "parameters", []))
|
|
defaults = {p: getattr(cls, p, None) for p in params}
|
|
return {"parameters": params, "defaults": defaults}
|
|
|
|
|
|
def get_strategy_class(name: str) -> type | None:
|
|
"""Return the strategy class by name (None if unavailable)."""
|
|
return _load_strategy_classes().get(name)
|