diff --git a/sanguo_api/__init__.py b/sanguo_api/__init__.py index 6673b9f..5c1027c 100644 --- a/sanguo_api/__init__.py +++ b/sanguo_api/__init__.py @@ -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__"]) diff --git a/sanguo_api/strategy_registry.py b/sanguo_api/strategy_registry.py index 6baf1c6..9733b85 100644 --- a/sanguo_api/strategy_registry.py +++ b/sanguo_api/strategy_registry.py @@ -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) diff --git a/tests/api/test_strategy_registry.py b/tests/api/test_strategy_registry.py index aa1fff4..90f0aae 100644 --- a/tests/api/test_strategy_registry.py +++ b/tests/api/test_strategy_registry.py @@ -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"]