feat(strategy): registry 扩展自研目录扫描(AST)+文件读写+py_compile; __init__ 惰性化(纯逻辑模块脱离 fastapi 可单测) [nas]
This commit is contained in:
+25
-13
@@ -1,16 +1,28 @@
|
||||
"""
|
||||
Sanguo Quant API Module
|
||||
FastAPI-based REST API for backtesting services
|
||||
"""
|
||||
|
||||
from .app import create_app
|
||||
from .schemas import CtaBacktestRequest, OptimizeRequest, FactorAnalysisRequest
|
||||
"""Sanguo Quant API Module (惰性 re-export)。
|
||||
|
||||
子模块（strategy_registry / instance_store 等纯逻辑）可直接 import，不触发
|
||||
fastapi/pydantic —— 这样数据层 venv 也能单测纯逻辑模块。create_app / schemas 走
|
||||
PEP 562 `__getattr__`，仅当 `from sanguo_api import create_app` 访问时才加载
|
||||
（调用方应优先 `from sanguo_api.app import create_app`）。
|
||||
"""
|
||||
__version__ = "0.1.0"
|
||||
|
||||
__all__ = [
|
||||
"create_app",
|
||||
"CtaBacktestRequest",
|
||||
"OptimizeRequest",
|
||||
"FactorAnalysisRequest"
|
||||
]
|
||||
_LAZY = {
|
||||
"create_app": "sanguo_api.app",
|
||||
"CtaBacktestRequest": "sanguo_api.schemas",
|
||||
"OptimizeRequest": "sanguo_api.schemas",
|
||||
"FactorAnalysisRequest": "sanguo_api.schemas",
|
||||
}
|
||||
|
||||
|
||||
def __getattr__(name: str):
|
||||
"""惰性加载 re-export 项（首次访问才 import，避免拖入 fastapi/pydantic）。"""
|
||||
if name in _LAZY:
|
||||
import importlib
|
||||
mod = importlib.import_module(_LAZY[name])
|
||||
return getattr(mod, name)
|
||||
raise AttributeError(f"module 'sanguo_api' has no attribute {name!r}")
|
||||
|
||||
|
||||
def __dir__() -> list[str]:
|
||||
return sorted(list(_LAZY) + ["__version__"])
|
||||
|
||||
@@ -6,8 +6,11 @@ Task S1.3.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import glob
|
||||
import importlib
|
||||
import logging
|
||||
import os
|
||||
import pkgutil
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -37,10 +40,20 @@ def _load_strategy_classes() -> dict[str, type]:
|
||||
|
||||
|
||||
def list_strategies() -> list[dict]:
|
||||
"""Return [{name, class_name}, ...] for the UI dropdown."""
|
||||
"""Return [{name, class_name}, ...] for the UI dropdown.
|
||||
|
||||
vnpy_ctastrategy 导入的 CTA 类 + 自研目录 AST 扫出的 CTA 类名（不 import，
|
||||
避免重依赖副作用）。全空时降级 STRATEGY_NAMES。
|
||||
"""
|
||||
classes = _load_strategy_classes()
|
||||
if classes:
|
||||
return [{"name": n, "class_name": n} for n in sorted(classes)]
|
||||
items = [{"name": n, "class_name": n} for n in sorted(classes)]
|
||||
known = {i["name"] for i in items}
|
||||
for f in _scan_self_owned():
|
||||
if f["type"] == "cta" and f["class_name"] not in known:
|
||||
items.append({"name": f["class_name"], "class_name": f["class_name"]})
|
||||
known.add(f["class_name"])
|
||||
if items:
|
||||
return items
|
||||
return [{"name": n, "class_name": n} for n in STRATEGY_NAMES]
|
||||
|
||||
|
||||
@@ -58,3 +71,120 @@ def strategy_params(name: str) -> dict:
|
||||
def get_strategy_class(name: str) -> type | None:
|
||||
"""Return the strategy class by name (None if unavailable)."""
|
||||
return _load_strategy_classes().get(name)
|
||||
|
||||
|
||||
# ===== 自研策略目录扫描 + 文件读写（在线编辑，spec §12）=====
|
||||
|
||||
# 自研策略目录（相对仓库根 / 容器 /app）。sanguo_live/strategies 不存在，勿加。
|
||||
SELF_OWNED_DIRS: list[str] = [
|
||||
"sanguo_trader/strategy",
|
||||
"sanguo_portfolio/strategies",
|
||||
]
|
||||
|
||||
|
||||
def _repo_root() -> str:
|
||||
"""strategy_registry.py 所在包的上一级 = 仓库根（/app 或开发目录）。"""
|
||||
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def _classify_strategy_type(tree: ast.Module) -> str:
|
||||
"""AST 判定 portfolio vs cta：import 了 vnpy StrategyTemplate → 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.Import):
|
||||
for alias in node.names:
|
||||
if "StrategyTemplate" in alias.name:
|
||||
return "cta"
|
||||
return "portfolio"
|
||||
|
||||
|
||||
def _extract_class_names(tree: ast.Module) -> list[str]:
|
||||
"""AST 提取所有 `class XxxStrategy` 类名。"""
|
||||
return [
|
||||
n.name for n in ast.walk(tree)
|
||||
if isinstance(n, ast.ClassDef) and n.name.endswith("Strategy")
|
||||
]
|
||||
|
||||
|
||||
def _scan_self_owned() -> list[dict]:
|
||||
"""AST 扫描自研目录（不 import，避免重依赖副作用）。"""
|
||||
root = _repo_root()
|
||||
out: list[dict] = []
|
||||
for rel in SELF_OWNED_DIRS:
|
||||
abs_dir = os.path.join(root, rel)
|
||||
for py in sorted(glob.glob(os.path.join(abs_dir, "*.py"))):
|
||||
name = os.path.basename(py)
|
||||
if name.startswith("_"):
|
||||
continue
|
||||
try:
|
||||
src = open(py, encoding="utf-8").read()
|
||||
tree = ast.parse(src, filename=name)
|
||||
except (SyntaxError, OSError):
|
||||
continue
|
||||
classes = _extract_class_names(tree)
|
||||
if not classes:
|
||||
continue
|
||||
out.append({
|
||||
"name": name,
|
||||
"dir": rel + "/",
|
||||
"class_name": classes[0],
|
||||
"type": _classify_strategy_type(tree),
|
||||
"lines": len(src.splitlines()),
|
||||
"modified": "",
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def list_strategy_files() -> dict:
|
||||
"""GET /strategy/files → {files: [...]}（不含 code，前端列表用）。"""
|
||||
return {"files": _scan_self_owned()}
|
||||
|
||||
|
||||
def read_strategy_file(name: str) -> dict:
|
||||
"""按文件名读单个策略文件全量（含 code）。找不到 → ValueError。"""
|
||||
root = _repo_root()
|
||||
for rel in SELF_OWNED_DIRS:
|
||||
path = os.path.join(root, rel, name)
|
||||
if os.path.exists(path):
|
||||
src = open(path, encoding="utf-8").read()
|
||||
tree = ast.parse(src, filename=name)
|
||||
classes = _extract_class_names(tree)
|
||||
return {
|
||||
"name": name,
|
||||
"dir": rel + "/",
|
||||
"class_name": classes[0] if classes else "",
|
||||
"type": _classify_strategy_type(tree),
|
||||
"lines": len(src.splitlines()),
|
||||
"modified": "",
|
||||
"code": src,
|
||||
}
|
||||
raise ValueError(f"策略文件不存在: {name}")
|
||||
|
||||
|
||||
def write_strategy_file(name: str, code: str) -> None:
|
||||
"""写策略文件。py_compile 语法校验失败 → ValueError（拒绝坏代码落盘）。"""
|
||||
import py_compile
|
||||
import tempfile
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as tf:
|
||||
tf.write(code)
|
||||
tmp = tf.name
|
||||
try:
|
||||
py_compile.compile(tmp, doraise=True)
|
||||
except py_compile.PyCompileError as e:
|
||||
raise ValueError(f"语法错误，未保存: {e}") from e
|
||||
finally:
|
||||
os.unlink(tmp)
|
||||
|
||||
root = _repo_root()
|
||||
for rel in SELF_OWNED_DIRS:
|
||||
path = os.path.join(root, rel, name)
|
||||
if os.path.exists(path):
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
f.write(code)
|
||||
return
|
||||
# 新文件落 cta 目录（MVP 统一落 sanguo_trader/strategy）
|
||||
target = os.path.join(root, SELF_OWNED_DIRS[0], name)
|
||||
os.makedirs(os.path.dirname(target), exist_ok=True)
|
||||
with open(target, "w", encoding="utf-8") as f:
|
||||
f.write(code)
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
"""Tests for sanguo_api.strategy_registry (Task S1.3)."""
|
||||
from sanguo_api.strategy_registry import list_strategies, strategy_params, STRATEGY_NAMES
|
||||
from sanguo_api.strategy_registry import (
|
||||
list_strategies, strategy_params, STRATEGY_NAMES,
|
||||
list_strategy_files, read_strategy_file,
|
||||
)
|
||||
|
||||
|
||||
def test_list_strategies_shape():
|
||||
@@ -27,3 +30,30 @@ def test_strategy_params_keys():
|
||||
def test_strategy_params_unknown_returns_empty():
|
||||
p = strategy_params("NoSuchStrategy_xyz")
|
||||
assert p == {"parameters": [], "defaults": {}}
|
||||
|
||||
|
||||
def test_list_strategy_files_returns_dirs():
|
||||
data = list_strategy_files()
|
||||
assert "files" in data and isinstance(data["files"], list)
|
||||
dirs = {f["dir"] for f in data["files"]}
|
||||
# 自研目录至少出现一个（sanguo_portfolio/strategies 有真实策略文件）
|
||||
assert any("sanguo_trader/strategy" in d or "sanguo_portfolio/strategies" in d for d in dirs)
|
||||
|
||||
|
||||
def test_list_strategy_files_item_shape():
|
||||
data = list_strategy_files()
|
||||
if not data["files"]:
|
||||
return
|
||||
f = data["files"][0]
|
||||
for k in ("name", "dir", "class_name", "type"):
|
||||
assert k in f
|
||||
|
||||
|
||||
def test_read_strategy_file_returns_code():
|
||||
data = list_strategy_files()
|
||||
if not data["files"]:
|
||||
return
|
||||
f = data["files"][0]
|
||||
content = read_strategy_file(f["name"])
|
||||
assert "code" in content and isinstance(content["code"], str)
|
||||
assert content["class_name"] == f["class_name"]
|
||||
|
||||
Reference in New Issue
Block a user