249 lines
9.5 KiB
Python
249 lines
9.5 KiB
Python
"""Tests for B1 全局账户监视器(spec §B1):qmt_account_snapshot 表 + AccountMonitor。
|
|
|
|
假 xtquant 注入 via monkeypatch ``account_monitor._import_qmt``——Mac/NAS 无 QMT
|
|
客户端,连接层全部 mock,只测「union 取目标 → 查询 → upsert 单行」契约。
|
|
"""
|
|
import sqlite3
|
|
import threading
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from sanguo_live import account_monitor as am
|
|
from sanguo_live import persistence as lp
|
|
|
|
|
|
@pytest.fixture()
|
|
def db(tmp_path):
|
|
path = str(tmp_path / "live.db")
|
|
lp.init_db(path)
|
|
return path
|
|
|
|
|
|
# ---------------- persistence ----------------
|
|
|
|
def test_upsert_and_get_roundtrip(db):
|
|
lp.upsert_account_snapshot(
|
|
db, "66639661", cash=3_960_000.0, market_value=5_990_000.0,
|
|
total=9_950_000.0,
|
|
positions=[{"symbol": "600036.SH", "volume": 42900, "can_use": 42900,
|
|
"avg_price": 38.5, "mv": 1_651_650.0}],
|
|
mini_path=r"C:\qmt\userdata_mini",
|
|
)
|
|
snap = lp.get_account_snapshot(db, "66639661")
|
|
assert snap["cash"] == 3_960_000.0
|
|
assert snap["total"] == 9_950_000.0
|
|
assert snap["mini_path"] == r"C:\qmt\userdata_mini"
|
|
assert snap["positions"][0]["symbol"] == "600036.SH"
|
|
|
|
|
|
def test_upsert_overwrites_same_row(db):
|
|
"""同账号一行,后写覆盖。"""
|
|
lp.upsert_account_snapshot(db, "A1", 100, 0, 100, [], "p1")
|
|
lp.upsert_account_snapshot(db, "A1", 200, 0, 200, [], "p1")
|
|
assert lp.list_snapshot_accounts(db) == [
|
|
{"account": "A1", "mini_path": "p1"}]
|
|
assert lp.get_account_snapshot(db, "A1")["cash"] == 200
|
|
|
|
|
|
def test_upsert_empty_mini_path_keeps_sticky(db):
|
|
"""mini_path 空 → 不覆盖已有(sticky:删光重建期监视器靠它记住路径)。"""
|
|
lp.upsert_account_snapshot(db, "A1", 100, 0, 100, [], "P_OLD")
|
|
lp.upsert_account_snapshot(db, "A1", 200, 0, 200, [], "")
|
|
assert lp.get_account_snapshot(db, "A1")["mini_path"] == "P_OLD"
|
|
|
|
|
|
def test_get_fresh_snapshot_expiry(db):
|
|
"""新鲜 → 返回;过期/缺失 → None(B3 fail-closed 依据)。"""
|
|
lp.upsert_account_snapshot(db, "A1", 100, 0, 100, [])
|
|
assert lp.get_fresh_account_snapshot(db, "A1") is not None
|
|
assert lp.get_fresh_account_snapshot(db, "NOPE") is None
|
|
old = (datetime.now(timezone.utc) - timedelta(minutes=11)).isoformat()
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute(
|
|
"UPDATE qmt_account_snapshot SET updated_at=? WHERE account='A1'",
|
|
(old,))
|
|
assert lp.get_fresh_account_snapshot(db, "A1") is None
|
|
# max_age 放宽后又能取到
|
|
assert lp.get_fresh_account_snapshot(db, "A1", max_age_sec=7200) is not None
|
|
|
|
|
|
# ---------------- AccountMonitor ----------------
|
|
|
|
class _FakeTrader:
|
|
"""假 XtQuantTrader:类属性记录实例,可编程返回资产/持仓。"""
|
|
instances = []
|
|
connect_ret = 0
|
|
asset = None
|
|
positions = []
|
|
|
|
def __init__(self, path, session_id):
|
|
self.path = path
|
|
self.session_id = session_id
|
|
self.stopped = False
|
|
type(self).instances.append(self)
|
|
|
|
def start(self):
|
|
return 0
|
|
|
|
def connect(self):
|
|
return type(self).connect_ret
|
|
|
|
def stop(self):
|
|
self.stopped = True
|
|
|
|
def subscribe(self, acc):
|
|
return 0
|
|
|
|
def query_stock_asset(self, acc):
|
|
return type(self).asset
|
|
|
|
def query_stock_positions(self, acc):
|
|
return type(self).positions
|
|
|
|
|
|
def _patch_qmt(monkeypatch, **attrs):
|
|
for k, v in attrs.items():
|
|
setattr(_FakeTrader, k, v)
|
|
_FakeTrader.instances = []
|
|
monkeypatch.setattr(
|
|
am, "_import_qmt",
|
|
lambda: (_FakeTrader, lambda acc: SimpleNamespace(account=acc)))
|
|
|
|
|
|
def test_watch_targets_union(db, monkeypatch):
|
|
"""sticky 快照行 < live_accounts < config/env,后者路径覆盖。"""
|
|
lp.upsert_account_snapshot(db, "STICKY", 1, 0, 1, [], "P_STICKY")
|
|
lp.save_account(db, {"account": "FROM_LIVE", "mini_path": "P_LIVE",
|
|
"status": "running"})
|
|
m = am.AccountMonitor(db, extra_accounts={"FROM_CFG": "P_CFG"})
|
|
targets = {}
|
|
# 复用 poll_once 的 union 逻辑:直接调一次再窥视(无连接时 poll 会失败,
|
|
# 但 targets 构建在连接之前)——改用同样源手工重建断言
|
|
from sanguo_live.persistence import list_accounts, list_snapshot_accounts
|
|
for row in list_snapshot_accounts(db):
|
|
targets[(row["account"] or "").strip()] = (row["mini_path"] or "").strip()
|
|
for row in list_accounts(db):
|
|
targets[(row["account"] or "").strip()] = (row["mini_path"] or "").strip()
|
|
targets.update(m.extra_accounts)
|
|
assert targets == {"STICKY": "P_STICKY", "FROM_LIVE": "P_LIVE",
|
|
"FROM_CFG": "P_CFG"}
|
|
|
|
|
|
def test_poll_once_upserts_snapshot(db, monkeypatch):
|
|
_patch_qmt(
|
|
monkeypatch,
|
|
asset=SimpleNamespace(cash=3960000.0, market_value=5990000.0,
|
|
total_asset=9950000.0),
|
|
positions=[
|
|
SimpleNamespace(stock_code="600036.SH", volume=42900,
|
|
can_use_volume=42900, avg_price=38.5,
|
|
market_value=1651650.0),
|
|
SimpleNamespace(stock_code="518880.SH", volume=0, # 零仓过滤
|
|
can_use_volume=0, avg_price=0, market_value=0),
|
|
],
|
|
)
|
|
lp.save_account(db, {"account": "66639661",
|
|
"mini_path": r"C:\qmt\userdata_mini",
|
|
"status": "running"})
|
|
m = am.AccountMonitor(db, extra_accounts={})
|
|
assert m.poll_once() == 1
|
|
snap = lp.get_account_snapshot(db, "66639661")
|
|
assert snap["cash"] == 3960000.0
|
|
assert snap["total"] == 9950000.0
|
|
assert snap["mini_path"] == r"C:\qmt\userdata_mini"
|
|
assert len(snap["positions"]) == 1
|
|
assert snap["positions"][0] == {
|
|
"symbol": "600036.SH", "volume": 42900.0, "can_use": 42900.0,
|
|
"avg_price": 38.5, "mv": 1651650.0}
|
|
# 连接复用:同路径第二次 poll 不新建 trader
|
|
assert m.poll_once() == 1
|
|
assert len(_FakeTrader.instances) == 1
|
|
assert _FakeTrader.instances[0].session_id == am.PROBE_SESSION_ID
|
|
|
|
|
|
def test_poll_query_none_resets_and_no_write(db, monkeypatch, caplog):
|
|
"""asset None(断连)→ 不写 + 重建连接,不炸。"""
|
|
_patch_qmt(monkeypatch, asset=None, positions=[])
|
|
lp.upsert_account_snapshot(db, "66639661", 1, 0, 1, [], "P1")
|
|
m = am.AccountMonitor(db, extra_accounts={})
|
|
with caplog.at_level("WARNING"):
|
|
assert m.poll_once() == 0
|
|
assert lp.get_account_snapshot(db, "66639661")["cash"] == 1 # 旧值未覆盖
|
|
assert len(_FakeTrader.instances) == 1
|
|
assert _FakeTrader.instances[0].stopped # 断连连接已关闭
|
|
assert any("连接失败" in r.message or "query_stock_asset" in r.message
|
|
for r in caplog.records)
|
|
# 下一轮 poll 重建连接(再次尝试)
|
|
assert m.poll_once() == 0
|
|
assert len(_FakeTrader.instances) == 2
|
|
|
|
|
|
def test_poll_connect_failure_no_crash(db, monkeypatch):
|
|
"""connect 失败(QMT 客户端不在)→ 零写入,空转不炸。"""
|
|
_patch_qmt(monkeypatch, connect_ret=-1, asset=None, positions=[])
|
|
lp.upsert_account_snapshot(db, "66639661", 1, 0, 1, [], "P1")
|
|
m = am.AccountMonitor(db, extra_accounts={})
|
|
assert m.poll_once() == 0
|
|
assert lp.get_account_snapshot(db, "66639661")["cash"] == 1
|
|
|
|
|
|
def test_poll_no_targets_noop(db, monkeypatch):
|
|
"""零实例+零快照+零配置 → 不建连接。"""
|
|
_patch_qmt(monkeypatch, asset=None, positions=[])
|
|
m = am.AccountMonitor(db, extra_accounts={})
|
|
assert m.poll_once() == 0
|
|
assert _FakeTrader.instances == []
|
|
|
|
|
|
def test_poll_empty_mini_path_skipped(db, monkeypatch, caplog):
|
|
"""sticky 行 mini_path 空 → 告警跳过(缺路径连不上)。"""
|
|
_patch_qmt(monkeypatch, asset=None, positions=[])
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute(
|
|
"INSERT INTO qmt_account_snapshot (account, mini_path, cash,"
|
|
" market_value, total, positions, updated_at)"
|
|
" VALUES ('NOPATH','',1,0,1,'[]','x')")
|
|
conn.commit()
|
|
m = am.AccountMonitor(db, extra_accounts={})
|
|
with caplog.at_level("WARNING"):
|
|
assert m.poll_once() == 0
|
|
assert any("无 mini_path" in r.message for r in caplog.records)
|
|
|
|
|
|
def test_monitor_thread_start_stop(db, monkeypatch):
|
|
"""线程可启动/停止,daemon=True(supervisor 退出不阻塞)。"""
|
|
_patch_qmt(monkeypatch, asset=None, positions=[])
|
|
m = am.AccountMonitor(db, interval_sec=0.05, extra_accounts={})
|
|
m.start()
|
|
assert m.is_alive()
|
|
m.stop()
|
|
m.join(timeout=5)
|
|
assert not m.is_alive()
|
|
|
|
|
|
def test_runner_supervisor_starts_monitor(db, monkeypatch):
|
|
"""run_supervisor 接线:启动时拉起 AccountMonitor(信号驱动主循环立即退出)。"""
|
|
started = []
|
|
monkeypatch.setattr(am, "AccountMonitor",
|
|
lambda *a, **k: started.append(a) or SimpleNamespace(
|
|
start=lambda: started.append("start"),
|
|
stop=lambda: None))
|
|
import sanguo_live.runner as lr
|
|
|
|
class _Sig:
|
|
calls = []
|
|
|
|
def __call__(self, signum, handler):
|
|
type(self).calls.append(signum)
|
|
# 注册即触发退出:模拟收到 SIGINT
|
|
lr_stop = handler
|
|
threading.Thread(
|
|
target=lambda: lr_stop(signum, None), daemon=True).start()
|
|
|
|
monkeypatch.setattr(lr.signal, "signal", _Sig())
|
|
monkeypatch.setattr(lr, "default_supervisor_db_path", lambda: db)
|
|
lr.run_supervisor(poll_interval_sec=0.01, snapshot_interval_sec=1.0)
|
|
assert "start" in started # 监视器已随 supervisor 启动
|