feat(live): B1全局账户监视器+qmt_account_snapshot快照表——预算制地基(多策略共享账户spec§B1)——①supervisor内daemon线程AccountMonitor:专属probe连接(session id=880811 int,远离bullet_trade的time*1000量级)60s查QMT资金+持仓,upsert单行全局快照(account主键,不挂实例,positions JSON)②账户来源三并集=sticky快照行(实例删光仍记得账号+mini_path,重建期校验不断供)<live_accounts行<config watch_accounts/env③断连自愈:asset None→关旧连接下轮重建,失败日志节流(首败WARN+每30败心跳)④零实盘实例也运行;xtquant缺失(NAS/Mac)warning后空转不炸supervisor⑤get_fresh_account_snapshot(10min过期→None)供B3预算校验fail-closed;+12测试(假xtquant注入,union/断连重建/sticky路径/过期判定);策略session在途A1修改(strategies/*+test_instance_view_isolation)不属本commit [vps]
This commit is contained in:
@@ -0,0 +1,235 @@
|
||||
"""B1 全局账户监视器(spec §multi-strategy-instance-budget §B1)。
|
||||
|
||||
supervisor 内 daemon 线程:独立 probe 连接(专属 int session id)每 60s 查一次
|
||||
QMT 账户资金+持仓,upsert 单行全局快照 ``qmt_account_snapshot``(不挂实例)。
|
||||
|
||||
设计要点:
|
||||
- 与实盘实例解耦:零实盘实例时照常运行(删光重建期预算校验数据不断供)。
|
||||
- 账户来源三并集(后者覆盖前者路径):
|
||||
1. 已有快照行(sticky:实例删光后监视器仍记得账号+mini_path)
|
||||
2. live_accounts 行(account + mini_path)
|
||||
3. config ``live_trading.watch_accounts`` / env ``SANGUO_QMT_ACCOUNT``
|
||||
- xtquant 导入失败或 QMT 客户端不在(NAS/Mac/夜间)→ 告警后空转,不炸 supervisor。
|
||||
- 失败日志节流:同一 mini_path 首败 WARN,此后每 30 败心跳一次。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 专属 probe 会话 id:int,量级刻意远离 bullet_trade 默认的 int(time*1000)(~1.7e12),
|
||||
# 不与引擎连接撞 session。
|
||||
PROBE_SESSION_ID = 880811
|
||||
|
||||
_DEFAULT_INTERVAL_SEC = 60.0
|
||||
_HEARTBEAT_EVERY_N_FAILURES = 30
|
||||
|
||||
|
||||
def _import_qmt() -> tuple[type, type]:
|
||||
"""懒加载 xtquant(测试用 monkeypatch 本函数注入假实现)。"""
|
||||
from xtquant.xttrader import XtQuantTrader # type: ignore
|
||||
from xtquant.xttype import StockAccount # type: ignore
|
||||
return XtQuantTrader, StockAccount
|
||||
|
||||
|
||||
def _extra_from_config() -> dict[str, str]:
|
||||
"""config/data_platform.yaml live_trading.watch_accounts + env 兜底。
|
||||
|
||||
返回 {account: mini_path};读不到/没配 → 空 dict(不炸)。
|
||||
"""
|
||||
import yaml
|
||||
|
||||
extra: dict[str, str] = {}
|
||||
cfg_path = os.path.join(os.path.dirname(os.path.dirname(
|
||||
os.path.abspath(__file__))), "config", "data_platform.yaml")
|
||||
try:
|
||||
with open(cfg_path, encoding="utf-8") as f:
|
||||
lt = (yaml.safe_load(f) or {}).get("live_trading") or {}
|
||||
path = str(lt.get("watch_mini_path") or "")
|
||||
for acc in lt.get("watch_accounts") or []:
|
||||
if str(acc).strip():
|
||||
extra[str(acc).strip()] = path
|
||||
except (OSError, ValueError) as e:
|
||||
logger.debug("[account-monitor] 读 watch 配置失败(忽略): %s", e)
|
||||
if os.environ.get("SANGUO_QMT_ACCOUNT"):
|
||||
extra.setdefault(os.environ["SANGUO_QMT_ACCOUNT"].strip(),
|
||||
os.environ.get("SANGUO_QMT_PATH", ""))
|
||||
return extra
|
||||
|
||||
|
||||
class AccountMonitor(threading.Thread):
|
||||
"""全局账户快照监视线程。用法:monitor = AccountMonitor(db); monitor.start()。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
db_path: str,
|
||||
interval_sec: float = _DEFAULT_INTERVAL_SEC,
|
||||
session_id: int = PROBE_SESSION_ID,
|
||||
extra_accounts: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
super().__init__(daemon=True, name="account-monitor")
|
||||
self.db_path = db_path
|
||||
self.interval_sec = interval_sec
|
||||
self.session_id = session_id
|
||||
# None=首 poll 时读 config;{}=禁用(测试用)
|
||||
self.extra_accounts = extra_accounts
|
||||
self._stop_event = threading.Event()
|
||||
self._traders: dict[str, Any] = {} # mini_path → XtQuantTrader
|
||||
self._fail_counts: dict[str, int] = {} # mini_path → 连续失败计数
|
||||
|
||||
# ---------------- 生命周期 ----------------
|
||||
|
||||
def stop(self) -> None:
|
||||
self._stop_event.set()
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
_import_qmt()
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning(
|
||||
"[account-monitor] xtquant 不可用(%s),账户快照不采集", e)
|
||||
return
|
||||
logger.info("[account-monitor] 启动 (interval=%.0fs db=%s)",
|
||||
self.interval_sec, self.db_path)
|
||||
while True:
|
||||
try:
|
||||
self.poll_once()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.warning("[account-monitor] poll 异常", exc_info=True)
|
||||
if self._stop_event.wait(self.interval_sec):
|
||||
break
|
||||
self._close_all()
|
||||
|
||||
def _close_all(self) -> None:
|
||||
for path, trader in list(self._traders.items()):
|
||||
try:
|
||||
trader.stop()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
self._traders.clear()
|
||||
|
||||
# ---------------- 采集 ----------------
|
||||
|
||||
def poll_once(self) -> int:
|
||||
"""扫全部 watch 目标,写快照。返回成功写入的行数(测试用)。"""
|
||||
from sanguo_live.persistence import (
|
||||
list_accounts, list_snapshot_accounts, upsert_account_snapshot,
|
||||
)
|
||||
|
||||
if self.extra_accounts is None:
|
||||
self.extra_accounts = _extra_from_config()
|
||||
|
||||
# 三并集:sticky 快照行 < live_accounts < config/env(后者路径覆盖)
|
||||
targets: dict[str, str] = {}
|
||||
for row in list_snapshot_accounts(self.db_path):
|
||||
acc = (row.get("account") or "").strip()
|
||||
if acc:
|
||||
targets[acc] = (row.get("mini_path") or "").strip()
|
||||
for row in list_accounts(self.db_path):
|
||||
acc = (row.get("account") or "").strip()
|
||||
if acc:
|
||||
targets[acc] = (row.get("mini_path") or "").strip()
|
||||
targets.update(self.extra_accounts)
|
||||
|
||||
written = 0
|
||||
for account, mini_path in sorted(targets.items()):
|
||||
try:
|
||||
if self._poll_account(account, mini_path,
|
||||
upsert_account_snapshot):
|
||||
written += 1
|
||||
except Exception: # noqa: BLE001
|
||||
logger.warning("[account-monitor] 账户 %s 采集异常",
|
||||
account, exc_info=True)
|
||||
return written
|
||||
|
||||
def _poll_account(self, account: str, mini_path: str, upsert: Any) -> bool:
|
||||
"""单账户采集。返回是否写入。mini_path 为空 → 告警跳过(连不上 QMT)。"""
|
||||
if not mini_path:
|
||||
logger.warning(
|
||||
"[account-monitor] 账户 %s 无 mini_path,跳过(需 live_accounts "
|
||||
"行携带或快照 sticky 记录)", account)
|
||||
return False
|
||||
trader = self._ensure_trader(mini_path)
|
||||
if trader is None:
|
||||
return False
|
||||
XtQuantTrader, StockAccount = _import_qmt()
|
||||
acc_obj = StockAccount(account)
|
||||
asset = trader.query_stock_asset(acc_obj)
|
||||
if asset is None:
|
||||
self._note_failure(mini_path, "query_stock_asset None")
|
||||
self._reset_trader(mini_path) # 可能断连,下轮重建
|
||||
return False
|
||||
positions = trader.query_stock_positions(acc_obj) or []
|
||||
rows = []
|
||||
for p in positions:
|
||||
vol = float(getattr(p, "volume", 0) or 0)
|
||||
if vol <= 0:
|
||||
continue
|
||||
rows.append({
|
||||
"symbol": str(getattr(p, "stock_code", "") or ""),
|
||||
"volume": vol,
|
||||
"can_use": float(getattr(p, "can_use_volume", 0) or 0),
|
||||
"avg_price": float(getattr(p, "avg_price", 0) or 0),
|
||||
"mv": float(getattr(p, "market_value", 0) or 0),
|
||||
})
|
||||
upsert(
|
||||
self.db_path, account,
|
||||
cash=float(getattr(asset, "cash", 0) or 0),
|
||||
market_value=float(getattr(asset, "market_value", 0) or 0),
|
||||
total=float(getattr(asset, "total_asset", 0) or 0),
|
||||
positions=rows,
|
||||
mini_path=mini_path,
|
||||
)
|
||||
self._fail_counts.pop(mini_path, None)
|
||||
logger.info(
|
||||
"[account-monitor] 快照 %s: cash=%.0f mv=%.0f total=%.0f 持仓%d只",
|
||||
account, float(getattr(asset, "cash", 0) or 0),
|
||||
float(getattr(asset, "market_value", 0) or 0),
|
||||
float(getattr(asset, "total_asset", 0) or 0), len(rows))
|
||||
return True
|
||||
|
||||
# ---------------- 连接管理 ----------------
|
||||
|
||||
def _ensure_trader(self, mini_path: str) -> Any:
|
||||
"""按 mini_path 复用/新建 probe 连接;失败返回 None(带节流告警)。"""
|
||||
trader = self._traders.get(mini_path)
|
||||
if trader is not None:
|
||||
return trader
|
||||
XtQuantTrader, _StockAccount = _import_qmt()
|
||||
try:
|
||||
t = XtQuantTrader(mini_path, self.session_id)
|
||||
t.start()
|
||||
if t.connect() not in (0, None):
|
||||
raise RuntimeError(f"connect 返回 {t.connect()}")
|
||||
self._traders[mini_path] = t
|
||||
return t
|
||||
except Exception as e: # noqa: BLE001
|
||||
self._note_failure(mini_path, repr(e))
|
||||
try:
|
||||
t.stop()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
return None
|
||||
|
||||
def _reset_trader(self, mini_path: str) -> None:
|
||||
trader = self._traders.pop(mini_path, None)
|
||||
if trader is not None:
|
||||
try:
|
||||
trader.stop()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
def _note_failure(self, mini_path: str, reason: str) -> None:
|
||||
n = self._fail_counts.get(mini_path, 0) + 1
|
||||
self._fail_counts[mini_path] = n
|
||||
if n == 1 or n % _HEARTBEAT_EVERY_N_FAILURES == 0:
|
||||
logger.warning(
|
||||
"[account-monitor] 连接失败 %s (第%d次): %s", mini_path, n, reason)
|
||||
|
||||
|
||||
__all__ = ["AccountMonitor", "PROBE_SESSION_ID", "_import_qmt",
|
||||
"_extra_from_config"]
|
||||
@@ -69,6 +69,15 @@ CREATE TABLE IF NOT EXISTS live_balance (
|
||||
market_value REAL,
|
||||
total REAL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS qmt_account_snapshot (
|
||||
account TEXT PRIMARY KEY,
|
||||
mini_path TEXT,
|
||||
cash REAL,
|
||||
market_value REAL,
|
||||
total REAL,
|
||||
positions TEXT,
|
||||
updated_at TEXT
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_live_trades_account ON live_trades(account_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_live_balance_account ON live_balance(account_id, date);
|
||||
"""
|
||||
@@ -295,10 +304,94 @@ def get_first_balance(db_path: str, account_id: int) -> dict | None:
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
# ----------------- qmt_account_snapshot (B1 全局账户快照,spec §B1) -----------------
|
||||
|
||||
def upsert_account_snapshot(
|
||||
db_path: str, account: str, cash: float, market_value: float,
|
||||
total: float, positions: list[dict], mini_path: str = "",
|
||||
) -> None:
|
||||
"""upsert 单行全局快照(按 QMT 账号一行,不挂实例)。
|
||||
|
||||
positions = [{symbol,volume,can_use,avg_price,mv}] 存 JSON。
|
||||
mini_path 为空时不覆盖已有值(sticky:删光重建期监视器靠它记住路径)。
|
||||
"""
|
||||
now = _now()
|
||||
with sqlite3.connect(db_path) as conn:
|
||||
conn.execute(
|
||||
"""INSERT INTO qmt_account_snapshot
|
||||
(account, mini_path, cash, market_value, total, positions, updated_at)
|
||||
VALUES (?,?,?,?,?,?,?)
|
||||
ON CONFLICT(account) DO UPDATE SET
|
||||
mini_path=CASE WHEN ?!='' THEN ? ELSE qmt_account_snapshot.mini_path END,
|
||||
cash=excluded.cash, market_value=excluded.market_value,
|
||||
total=excluded.total, positions=excluded.positions,
|
||||
updated_at=excluded.updated_at""",
|
||||
(account, mini_path, cash, market_value, total,
|
||||
json.dumps(positions, ensure_ascii=False), now,
|
||||
mini_path, mini_path),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def get_account_snapshot(db_path: str, account: str) -> dict | None:
|
||||
"""读快照(positions JSON 解析回 list);无行返回 None。"""
|
||||
with sqlite3.connect(db_path) as conn:
|
||||
conn.row_factory = sqlite3.Row
|
||||
cur = conn.execute(
|
||||
"SELECT * FROM qmt_account_snapshot WHERE account=?", (account,)
|
||||
)
|
||||
row = cur.fetchone()
|
||||
if not row:
|
||||
return None
|
||||
d = dict(row)
|
||||
try:
|
||||
d["positions"] = json.loads(d.get("positions") or "[]")
|
||||
except (ValueError, TypeError):
|
||||
d["positions"] = []
|
||||
return d
|
||||
|
||||
|
||||
def list_snapshot_accounts(db_path: str) -> list[dict]:
|
||||
"""全部快照行(account+mini_path)——监视器 sticky 账户来源。"""
|
||||
with sqlite3.connect(db_path) as conn:
|
||||
conn.row_factory = sqlite3.Row
|
||||
cur = conn.execute(
|
||||
"SELECT account, mini_path FROM qmt_account_snapshot"
|
||||
)
|
||||
return [dict(r) for r in cur.fetchall()]
|
||||
|
||||
|
||||
def _snapshot_age_sec(updated_at: str | None) -> float:
|
||||
"""updated_at(ISO,UTC)距今秒数;解析失败按无穷旧处理。"""
|
||||
if not updated_at:
|
||||
return float("inf")
|
||||
try:
|
||||
ts = datetime.fromisoformat(updated_at)
|
||||
if ts.tzinfo is None:
|
||||
ts = ts.replace(tzinfo=timezone.utc)
|
||||
return (datetime.now(timezone.utc) - ts).total_seconds()
|
||||
except ValueError:
|
||||
return float("inf")
|
||||
|
||||
|
||||
def get_fresh_account_snapshot(
|
||||
db_path: str, account: str, max_age_sec: float = 600.0
|
||||
) -> dict | None:
|
||||
"""新鲜快照(默认 10 分钟内);缺失/过期返回 None——B3 预算校验 fail-closed 用。"""
|
||||
snap = get_account_snapshot(db_path, account)
|
||||
if snap is None:
|
||||
return None
|
||||
if _snapshot_age_sec(snap.get("updated_at")) > max_age_sec:
|
||||
return None
|
||||
return snap
|
||||
|
||||
|
||||
__all__ = [
|
||||
"init_db", "save_account", "list_accounts", "get_account",
|
||||
"update_account_status", "list_running_accounts",
|
||||
"save_trade", "list_trades",
|
||||
"save_positions", "load_positions",
|
||||
"save_balance", "list_balance", "get_last_balance", "get_first_balance",
|
||||
"upsert_account_snapshot", "get_account_snapshot",
|
||||
"list_snapshot_accounts", "get_fresh_account_snapshot",
|
||||
]
|
||||
|
||||
@@ -441,6 +441,11 @@ def run_supervisor(
|
||||
)
|
||||
db = db_path or default_supervisor_db_path()
|
||||
init_db(db)
|
||||
# B1 全局账户监视器:独立线程 60s 刷 qmt_account_snapshot(零实盘实例也跑,
|
||||
# 删光重建期预算校验数据不断供)。xtquant 缺失时自行空转,不影响 supervisor。
|
||||
from sanguo_live.account_monitor import AccountMonitor
|
||||
monitor = AccountMonitor(db)
|
||||
monitor.start()
|
||||
logger.info("[supervisor] 启动 (db=%s poll=%.1fs snapshot=%.1fs)",
|
||||
db, poll_interval_sec, snapshot_interval_sec)
|
||||
|
||||
@@ -516,4 +521,5 @@ def run_supervisor(
|
||||
logger.info("[supervisor] 退出清理组合实盘 (account=%s)", aid)
|
||||
_stop_portfolio_subprocess(proc)
|
||||
portfolio_procs.clear()
|
||||
monitor.stop()
|
||||
logger.info("[supervisor] 已退出")
|
||||
|
||||
@@ -0,0 +1,248 @@
|
||||
"""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 启动
|
||||
Reference in New Issue
Block a user