claude_dev
8d55e414fa
审计发现包装层系统性失真(2 CRITICAL+7 HIGH)，vnpy底座可信但A股场景未适配： - C1 定寸: engine.size=N(满仓手数)，策略volume=1手=N股，开平对称(pos归零) - C2 做空拦截: SHORT+OPEN拒单，long-only，SHORT+CLOSE平多允许 - H3 A股费用: AShareDailyResult重算(佣金保底5元/印花税卖方/过户费沪市) - H4 收益口径: simple return从balance算(不再用vnpy log return喂empyrical) - H5+口径: benchmark ffill对齐不缩样本; sizing_shares_per_lot暴露 - H7 退化检测: 零成交/空数据标degenerate不静默done - H8 task_id: optimize/factor用uuid4(原id()内存地址) - 静默except改warning 验证: 容器内真实vnpy DoubleMa 600000 2022-2024, total_return 1e-6→42.3%, end_balance 100万→142万, SHORT+OPEN成交0笔, N=7800股/手. 22 backtest测试全绿(含集成测试), API健康200.
61 lines
2.3 KiB
Python
61 lines
2.3 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 logging
|
|
import pkgutil
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 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 as e:
|
|
logger.warning("导入策略模块 %s 失败: %s", name, e)
|
|
continue
|
|
except Exception as e:
|
|
logger.warning("加载 vnpy_ctastrategy 策略列表失败（降级为静态列表）: %s", e)
|
|
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)
|