feat: 实盘模拟(live) + 组合回测MVP(portfolio)

[live] 实盘模拟 vnpy+miniQMT 直连(supervisor 轮询, 前后端): - sanguo_live: LiveTradingEngine + AShareCtaTemplate(定寸/禁做空) + runner_supervisor(DB驱动) + persistence(4表WAL) - sanguo_api/routes_live: 9路由(create/start/stop/positions/trades/account/status) - frontend live: New/List/Monitor + api/live.ts; config/live.yaml [portfolio] 组合回测 MVP(BulletTrade, 链路代码完成待验证): - runner_backtest 加 JSON 入口(--json, BacktestEngine 顶层 import) - sanguo_api/routes_portfolio: POST /portfolio/backtest SSH 触发 VPS 跑 - frontend PortfolioBacktest.vue + api/portfolio.ts: 表单+结果+净值曲线 - 路由/菜单注册(/backtest/portfolio 组合回测) - 已知: MVP 链路未端到端验证, agent 改至中途被停; 待 Mac 起服务联调
2026-07-18 20:04:16 +08:00
parent a68cf4905e
commit 96b1924fd5
24 changed files with 3550 additions and 9 deletions
@@ -0,0 +1,252 @@
+"""sanguo_live 单元测试。
+
+分两层:
+  (1) 纯 Python 逻辑层 —— Mac dev 机也跑(config 解析、注册表、默认参数);
+  (2) 依赖 vnpy_ctastrategy 层 —— Mac 未装时单测级 skip,VPS 装齐则跑通。
+
+Mac 跑:``pytest tests/test_live_engine.py -v``(层 1 全 pass + 层 2 skipped,exit 0)。
+VPS 跑:全部 pass(含定寸/禁做空/引擎装配)。
+"""
+from __future__ import annotations
+
+import importlib
+
+import pytest
+
+
+def _has_vnpy_cta() -> bool:
+    """Mac dev 机没装 vnpy_ctastrategy(只装在 VPS)。"""
+    try:
+        importlib.import_module("vnpy_ctastrategy")
+        return True
+    except ImportError:
+        return False
+
+
+# 单测级 skip marker(模块级 importorskip 会跳过整个文件,误伤层 1)
+needs_vnpy_cta = pytest.mark.skipif(
+    not _has_vnpy_cta(),
+    reason="本机未装 vnpy_ctastrategy(仅 VPS 有)— 跳过依赖它的单测",
+)
+
+
+# =============================================================================
+# 层 1:纯 Python 逻辑(Mac dev 机也跑)
+# =============================================================================
+
+
+def test_module_import_tolerant():
+    """``sanguo_live`` 包 import 不应崩(即便本机没 vnpy_ctastrategy)。"""
+    importlib.import_module("sanguo_live")
+    importlib.import_module("sanguo_live.base_template")
+    importlib.import_module("sanguo_live.runner")
+    # engine / strategies import 了 vnpy_qmt/vnpy_ctastrategy 的类绑定,
+    # 但都用 try/except 容错,模块本身能 import。
+    importlib.import_module("sanguo_live.engine")
+    importlib.import_module("sanguo_live.strategies")
+
+
+def test_default_config_fields():
+    from sanguo_live.runner import DEFAULT_CONFIG
+    assert DEFAULT_CONFIG["strategy_class"] == "AShareDoubleMaStrategy"
+    assert DEFAULT_CONFIG["vt_symbol"] == "600000.SSE"
+    s = DEFAULT_CONFIG["setting"]
+    assert s["window"] == 15
+    assert s["size"] == 100
+    assert s["forbid_short"] is True
+    assert s["fast_window"] == 10
+    assert s["slow_window"] == 20
+
+
+def test_load_config_env_override(monkeypatch):
+    """env SANGUO_QMT_ACCOUNT / SANGUO_QMT_PATH 优先于 yaml / 默认。"""
+    monkeypatch.setenv("SANGUO_QMT_ACCOUNT", "12345678")
+    monkeypatch.setenv("SANGUO_QMT_PATH", "/tmp/fake_mini")
+    from sanguo_live.runner import load_config
+    cfg = load_config("/nonexistent/path.yaml")  # 文件不存在 → 走默认
+    assert cfg["account"] == "12345678"
+    assert cfg["mini_path"] == "/tmp/fake_mini"
+
+
+def test_load_config_yaml_merge(tmp_path):
+    """yaml 能覆盖默认 fast_window 等。"""
+    yaml_file = tmp_path / "live.yaml"
+    yaml_file.write_text(
+        "account: '99999999'\n"
+        "vt_symbol: '000001.SZSE'\n"
+        "setting:\n"
+        "  fast_window: 5\n"
+        "  slow_window: 30\n",
+        encoding="utf-8",
+    )
+    from sanguo_live.runner import load_config
+    cfg = load_config(str(yaml_file))
+    assert cfg["account"] == "99999999"
+    assert cfg["vt_symbol"] == "000001.SZSE"
+    assert cfg["setting"]["fast_window"] == 5
+    assert cfg["setting"]["slow_window"] == 30
+    # 未覆盖的字段保留默认
+    assert cfg["setting"]["window"] == 15
+    assert cfg["strategy_class"] == "AShareDoubleMaStrategy"
+
+
+def test_build_strategy_class_known():
+    from sanguo_live.runner import build_strategy_class
+    cls = build_strategy_class("AShareDoubleMaStrategy")
+    assert cls.__name__ == "AShareDoubleMaStrategy"
+
+
+def test_build_strategy_class_unknown_raises():
+    from sanguo_live.runner import build_strategy_class
+    with pytest.raises(ValueError, match="未知策略类"):
+        build_strategy_class("NoSuchStrategy_xyz")
+
+
+def test_strategy_class_has_parameters():
+    """AShareDoubleMaStrategy.parameters 必须暴露 size/forbid_short/window
+    + fast/slow_window(缺一个都会让 update_setting 漏字段)。"""
+    from sanguo_live.strategies import AShareDoubleMaStrategy
+    params = AShareDoubleMaStrategy.parameters
+    for required in ("fast_window", "slow_window", "window",
+                     "size", "forbid_short"):
+        assert required in params, f"缺少 parameter: {required}"
+
+
+# =============================================================================
+# 层 2:依赖 vnpy_ctastrategy(Mac skip,VPS 跑)
+# =============================================================================
+
+
+class _FakeCtaEngine:
+    """记录 send_order 调用,模拟 CtaTemplate 依赖的 cta_engine。"""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple] = []
+
+    def send_order(self, strategy, direction, offset, price, volume,
+                   stop=False, lock=False, net=False):
+        self.calls.append((direction, offset, price, volume, stop, lock, net))
+        return []
+
+    def cancel_all(self, strategy):
+        return None
+
+
+def _make_strategy(cls, setting=None):
+    """构造一个策略实例(trading=True,可发单)。
+
+    ``cls`` 必须是具体类(CtaTemplate 是 ABC,带抽象 on_init,不能直接实例化)。
+    用 ``_ConcreteAShare`` 包装 AShareCtaTemplate 来测基类定寸/禁做空逻辑。
+    """
+    strat = cls(_FakeCtaEngine(), "test_strat", "600000.SSE", setting or {})
+    strat.trading = True
+    return strat
+
+
+def _concrete_asare():
+    """返回 AShareCtaTemplate 的一个具体子类(stub on_init/on_tick/on_bar)。"""
+    from sanguo_live.base_template import AShareCtaTemplate
+
+    class _Concrete(AShareCtaTemplate):
+        author = "test"
+
+        def on_init(self) -> None:  # type: ignore[override]
+            return
+
+        def on_tick(self, tick) -> None:  # type: ignore[override]
+            return
+
+        def on_bar(self, bar) -> None:  # type: ignore[override]
+            return
+
+    return _Concrete
+
+
+@needs_vnpy_cta
+def test_buy_volume_multiplied_by_size():
+    """buy(1) 实际下单 volume=100(1 手 × size)。"""
+    strat = _make_strategy(_concrete_asare(), {"size": 100})
+    strat.buy(10.0, 1)
+    assert len(strat.cta_engine.calls) == 1
+    _, _, price, volume, *_ = strat.cta_engine.calls[0]
+    assert price == 10.0
+    assert volume == 100
+
+
+@needs_vnpy_cta
+def test_buy_custom_size_multiplier():
+    """size=200 → buy(2) 下 400。"""
+    strat = _make_strategy(_concrete_asare(), {"size": 200})
+    strat.buy(8.8, 2)
+    assert strat.cta_engine.calls[0][3] == 400
+
+
+@needs_vnpy_cta
+def test_sell_volume_multiplied_by_size():
+    """sell(平多)同样定寸。"""
+    strat = _make_strategy(_concrete_asare(), {"size": 100})
+    strat.sell(11.0, 1)
+    assert strat.cta_engine.calls[0][3] == 100
+
+
+@needs_vnpy_cta
+def test_cover_volume_multiplied_by_size():
+    """cover 也定寸(策略逻辑误调时不至于下零股)。"""
+    strat = _make_strategy(_concrete_asare(), {"size": 100})
+    strat.cover(11.0, 1)
+    assert strat.cta_engine.calls[0][3] == 100
+
+
+@needs_vnpy_cta
+def test_short_blocked_by_default():
+    """forbid_short=True(默认) → short 返回 [],不触达 send_order。"""
+    class _ExplodingEngine:
+        def send_order(self, *a, **kw):
+            raise AssertionError("short 不应到达 send_order")
+
+        def write_log(self, msg, strategy=None):
+            return
+
+    strat = _concrete_asare()(_ExplodingEngine(), "t", "600000.SSE", {})
+    strat.trading = True
+    result = strat.short(10.0, 1)
+    assert result == []
+
+
+@needs_vnpy_cta
+def test_short_passes_when_forbid_disabled():
+    """forbid_short=False → short 定寸后透传到基类(A 股不推荐,仅测试逻辑)。"""
+    strat = _make_strategy(_concrete_asare(),
+                           {"size": 100, "forbid_short": False})
+    strat.short(10.0, 1)
+    assert len(strat.cta_engine.calls) == 1
+    assert strat.cta_engine.calls[0][3] == 100  # 1 手 × 100
+
+
+@needs_vnpy_cta
+def test_double_ma_strategy_uses_15min_window():
+    """AShareDoubleMaStrategy 默认 window=15(不是 1min)。"""
+    from sanguo_live.strategies import AShareDoubleMaStrategy
+    strat = _make_strategy(AShareDoubleMaStrategy, {})
+    assert strat.window == 15
+    assert strat.size == 100
+    assert strat.forbid_short is True
+
+
+@needs_vnpy_cta
+def test_engine_assembly_and_close():
+    """LiveTradingEngine 初始化 → MainEngine 装入 QMT gateway + CTA app,close 干净退出。
+
+    会真启动 EventEngine 线程,测试结束必须 close。
+    """
+    from sanguo_live.engine import LiveTradingEngine
+    eng = LiveTradingEngine()
+    try:
+        assert eng.cta_engine is not None
+        assert "QMT" in eng.main_engine.gateways
+        # 查询方法不抛(连接前可能返回空)
+        assert isinstance(eng.get_all_accounts(), list)
+        assert isinstance(eng.get_positions(), list)
+        assert isinstance(eng.get_orders(), list)
+    finally:
+        eng.close()