feat(strategy): 策略库灌入vnpy内置8策略模板(DoubleMa/AtrRsi/BollChannel/DualThrust/KingKeltner/MultiSignal/MultiTimeframe/TurtleSignal); registry自研目录优先加载(编辑副本即刻生效,pip兜底); 分类器识别vnpy_ctastrategy import风格 [vps]

sanguo_api/strategy_registry.py

@@ -59,8 +59,7 @@ def list_strategies() -> list[dict]:
 
 def strategy_params(name: str) -> dict:
     """Return {parameters: [...], defaults: {...}} for a strategy's dynamic form."""
-    classes = _load_strategy_classes()
-    cls = classes.get(name)
+    cls = _load_self_owned_module_classes().get(name) or _load_strategy_classes().get(name)
     if cls is None:
         return {"parameters": [], "defaults": {}}
     params = list(getattr(cls, "parameters", []))
@@ -68,8 +67,37 @@ def strategy_params(name: str) -> dict:
     return {"parameters": params, "defaults": defaults}
 
 
+def _load_self_owned_module_classes() -> dict[str, type]:
+    """从自研 CTA 目录(sanguo_trader/strategy)动态 import 策略类。
+
+    让「复制进来的 vnpy 内置模板 + 在线编辑的自研策略」真正可被回测加载,
+    且优先于 pip 包同名类(编辑副本即刻生效)。import 失败(如本机 dev 无
+    vnpy_ctastrategy)静默降级到 pip 包。
+    """
+    classes: dict[str, type] = {}
+    pkg = SELF_OWNED_DIRS[0].replace("/", ".")
+    for f in _scan_self_owned():
+        if f["type"] != "cta" or f["dir"] != SELF_OWNED_DIRS[0] + "/":
+            continue
+        stem = f["name"][:-3]
+        try:
+            mod = importlib.import_module(f"{pkg}.{stem}")
+            cls = getattr(mod, f["class_name"], None)
+            if isinstance(cls, type):
+                classes[f["class_name"]] = cls
+        except Exception as e:
+            logger.debug("自研策略 %s 导入失败(降级 pip 包): %s", f["name"], e)
+    return classes
+
+
 def get_strategy_class(name: str) -> type | None:
-    """Return the strategy class by name (None if unavailable)."""
+    """Return the strategy class by name (None if unavailable).
+
+    自研目录优先(编辑生效),pip 包兜底。
+    """
+    cls = _load_self_owned_module_classes().get(name)
+    if cls is not None:
+        return cls
     return _load_strategy_classes().get(name)
 
 
@@ -88,13 +116,18 @@ def _repo_root() -> str:
 
 
 def _classify_strategy_type(tree: ast.Module) -> str:
-    """AST 判定 portfolio vs cta：import 了 vnpy StrategyTemplate → cta，否则 portfolio。"""
+    """AST 判定 portfolio vs cta：
+    - import 了 bullet_trade 风格 StrategyTemplate → cta
+    - from vnpy_ctastrategy import ...（vnpy 内置/自研 CTA 文件风格）→ cta
+    否则 portfolio。"""
     for node in ast.walk(tree):
-        if isinstance(node, ast.ImportFrom) and (node.module or "").endswith("StrategyTemplate"):
-            return "cta"
+        if isinstance(node, ast.ImportFrom):
+            mod = node.module or ""
+            if mod.endswith("StrategyTemplate") or mod.endswith("vnpy_ctastrategy"):
+                return "cta"
         if isinstance(node, ast.Import):
             for alias in node.names:
-                if "StrategyTemplate" in alias.name:
+                if "StrategyTemplate" in alias.name or "vnpy_ctastrategy" in alias.name:
                     return "cta"
     return "portfolio"