feat(api): 回测结果接口(strategy list/params + equity-curve/daily-pnl/trades + kline)
- 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)
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user