diff --git a/sanguo_live/account_monitor.py b/sanguo_live/account_monitor.py
new file mode 100644
index 0000000..b78e5e6
--- /dev/null
+++ b/sanguo_live/account_monitor.py
@@ -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"]
diff --git a/sanguo_live/persistence.py b/sanguo_live/persistence.py
index f896eb9..63895d7 100644
--- a/sanguo_live/persistence.py
+++ b/sanguo_live/persistence.py
@@ -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",
 ]
diff --git a/sanguo_live/runner.py b/sanguo_live/runner.py
index 65d8fdf..596e201 100644
--- a/sanguo_live/runner.py
+++ b/sanguo_live/runner.py
@@ -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] 已退出")
diff --git a/tests/api/test_account_monitor.py b/tests/api/test_account_monitor.py
new file mode 100644
index 0000000..37fc23f
--- /dev/null
+++ b/tests/api/test_account_monitor.py
@@ -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 启动