165 lines
6.7 KiB
Python
165 lines
6.7 KiB
Python
"""Tests for 组合策略实盘(R3-1): live_accounts 组合列 + create 路由 + supervisor env 映射。
|
|
|
|
runner_live 的引擎装配依赖 miniQMT/VPS,不在单测范围;这里测的是
|
|
「web 建组合实盘 → DB 字段落对 → supervisor 能翻译出正确 env」这条契约链。
|
|
直调路由函数风格与 test_paper_lifecycle.py 一致(免 auth)。
|
|
"""
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from sanguo_api import routes_live as rl
|
|
from sanguo_live import persistence as live_persistence
|
|
from sanguo_live import runner as live_runner
|
|
|
|
|
|
@pytest.fixture()
|
|
def live_db(tmp_path):
|
|
db = str(tmp_path / "live.db")
|
|
rl.set_db_path(db)
|
|
return db
|
|
|
|
|
|
def _create_portfolio(db, **kw):
|
|
req = rl.LiveCreateRequest(
|
|
name=kw.get("name", "组合实盘1"), account=kw.get("account", "66639661"),
|
|
strategy_name=kw.get("strategy_name", "portfolio_all_weather"),
|
|
strategy_type="portfolio", strategy_class=kw.get("strategy_class", "all_weather"),
|
|
pool=kw.get("pool", "hs300_subset"), max_pool=kw.get("max_pool", 30),
|
|
benchmark=kw.get("benchmark", "000300.XSHG"),
|
|
interval=kw.get("interval", ""),
|
|
initial_capital=kw.get("initial_capital", 500000),
|
|
)
|
|
return rl.create_live(req)["account_id"]
|
|
|
|
|
|
def test_save_account_portfolio_fields(live_db):
|
|
aid = live_persistence.save_account(live_db, {
|
|
"name": "portfolio-live", "account": "66639661",
|
|
"strategy_class": "all_weather", "strategy_name": "portfolio_all_weather",
|
|
"strategy_type": "portfolio", "pool": "hs300_subset",
|
|
"max_pool": 30, "benchmark": "000300.XSHG",
|
|
"status": "running", "vt_symbol": "hs300_subset",
|
|
})
|
|
acc = live_persistence.get_account(live_db, aid)
|
|
assert acc["strategy_type"] == "portfolio"
|
|
assert acc["pool"] == "hs300_subset"
|
|
assert acc["max_pool"] == 30
|
|
assert acc["benchmark"] == "000300.XSHG"
|
|
|
|
|
|
def test_init_db_migrates_old_live_accounts(tmp_path):
|
|
"""旧库(无组合列)init_db 后补列且默认 cta。"""
|
|
db = str(tmp_path / "old.db")
|
|
old_schema = """
|
|
CREATE TABLE live_accounts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, account TEXT,
|
|
vt_symbol TEXT, strategy_class TEXT, strategy_name TEXT, setting TEXT,
|
|
status TEXT, interval TEXT, initial_capital REAL,
|
|
connect_wait_sec INTEGER, init_wait_sec INTEGER, mini_path TEXT,
|
|
error_msg TEXT, created_at TEXT, updated_at TEXT
|
|
);
|
|
"""
|
|
with sqlite3.connect(db) as conn:
|
|
conn.executescript(old_schema)
|
|
conn.execute("INSERT INTO live_accounts (name, status) VALUES ('old', 'stopped')")
|
|
live_persistence.init_db(db)
|
|
acc = live_persistence.get_account(db, 1)
|
|
assert acc["strategy_type"] == "cta" # ALTER DEFAULT 生效于旧行
|
|
|
|
|
|
def test_create_portfolio_live(live_db):
|
|
aid = _create_portfolio(live_db)
|
|
acc = live_persistence.get_account(live_db, aid)
|
|
assert acc["strategy_type"] == "portfolio"
|
|
assert acc["vt_symbol"] == "hs300_subset" # 组合行 vt_symbol=池名
|
|
assert acc["interval"] == "d" # 未传周期 → 组合默认日线
|
|
assert acc["status"] == "stopped"
|
|
|
|
|
|
def test_create_portfolio_interval_passthrough(live_db):
|
|
"""组合实盘周期由前端下拉传(miniQMT 档位),后端不写死。"""
|
|
aid = _create_portfolio(live_db, interval="15m")
|
|
acc = live_persistence.get_account(live_db, aid)
|
|
assert acc["interval"] == "15m"
|
|
|
|
|
|
def test_create_portfolio_rejects_empty_strategy(live_db):
|
|
with pytest.raises(Exception):
|
|
_create_portfolio(live_db, strategy_class="")
|
|
|
|
|
|
def test_create_cta_defaults_unchanged(live_db):
|
|
req = rl.LiveCreateRequest(account="66639661", strategy_name="dm1")
|
|
aid = rl.create_live(req)["account_id"]
|
|
acc = live_persistence.get_account(live_db, aid)
|
|
assert acc["strategy_type"] == "cta"
|
|
assert acc["interval"] == "15m"
|
|
assert acc["vt_symbol"] == "600000.SSE"
|
|
|
|
|
|
def test_portfolio_env_mapping():
|
|
"""live_accounts 行 → runner_live env(env 是组合实盘唯一参数契约)。"""
|
|
acc = {
|
|
"id": 7, "account": "66639661", "mini_path": r"C:\qmt\userdata_mini",
|
|
"strategy_class": "value_selection", "max_pool": 20,
|
|
"benchmark": "000905.XSHG", "initial_capital": 2_000_000,
|
|
}
|
|
env = live_runner._portfolio_env_for(acc, "live.db")
|
|
assert env["SANGUO_QMT_ACCOUNT"] == "66639661"
|
|
assert env["SANGUO_QMT_PATH"] == r"C:\qmt\userdata_mini"
|
|
assert env["SANGUO_LIVE_STRATEGY"] == "value_selection"
|
|
assert env["SANGUO_LIVE_MAX_POOL"] == "20"
|
|
assert env["SANGUO_LIVE_BENCHMARK"] == "000905.XSHG"
|
|
assert env["SANGUO_LIVE_CASH"] == "2000000"
|
|
assert env["SANGUO_LIVE_DB"] == "live.db"
|
|
assert env["SANGUO_LIVE_ACCOUNT_ID"] == "7"
|
|
|
|
|
|
def test_runner_live_env_defaults(monkeypatch):
|
|
"""runner_live.live_env 带默认值(手动跑不传参也不炸)。"""
|
|
from sanguo_portfolio import runner_live
|
|
for k in ("SANGUO_LIVE_STRATEGY", "SANGUO_LIVE_MAX_POOL", "SANGUO_LIVE_BENCHMARK",
|
|
"SANGUO_LIVE_CASH", "SANGUO_QMT_ACCOUNT", "SANGUO_QMT_PATH",
|
|
"SANGUO_LIVE_DB", "SANGUO_LIVE_ACCOUNT_ID"):
|
|
monkeypatch.delenv(k, raising=False)
|
|
cfg = runner_live.live_env()
|
|
assert cfg["strategy"] == "all_weather"
|
|
assert cfg["max_pool"] == "30"
|
|
assert cfg["benchmark"] == "000300.XSHG"
|
|
assert cfg["account"] == "" # 空 → run_live 拒绝启动(防误下单)
|
|
|
|
|
|
def test_live_strategy_adapter_builds_all_strategies(monkeypatch):
|
|
"""适配文件的策略工厂:env → StrategyTemplate(4 策略各识别一次)。"""
|
|
from sanguo_portfolio import live_strategy
|
|
|
|
class _FakeProvider:
|
|
pass
|
|
|
|
for name, cls_name in (
|
|
("all_weather", "AllWeatherStrategy"),
|
|
("momentum_timing", "MomentumTimingStrategy"),
|
|
("value_selection", "ValueSelectionStrategy"),
|
|
("small_cap", "SmallCapStrategy"),
|
|
):
|
|
monkeypatch.setenv("SANGUO_LIVE_STRATEGY", name)
|
|
s = live_strategy._build_live_strategy(_FakeProvider())
|
|
assert type(s).__name__ == cls_name
|
|
monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "nope")
|
|
with pytest.raises(ValueError):
|
|
live_strategy._build_live_strategy(_FakeProvider())
|
|
|
|
|
|
def test_normalize_vt_symbol_bare_code_gets_exchange_suffix():
|
|
"""CTA 实盘标的:裸 6 位码自动补交易所后缀(6→SSE,0/3→SZSE)。"""
|
|
from sanguo_api.routes_live import _normalize_vt_symbol
|
|
|
|
assert _normalize_vt_symbol("600000") == "600000.SSE"
|
|
assert _normalize_vt_symbol("000001") == "000001.SZSE"
|
|
assert _normalize_vt_symbol("300750") == "300750.SZSE"
|
|
# 已带后缀/非 6 位(池名等)原样
|
|
assert _normalize_vt_symbol("600000.SSE") == "600000.SSE"
|
|
assert _normalize_vt_symbol("hs300_subset") == "hs300_subset"
|
|
assert _normalize_vt_symbol("") == ""
|