claude_dev
96b1924fd5
[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 起服务联调
253 lines
8.5 KiB
Python
253 lines
8.5 KiB
Python
"""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()
|