fix(monitor): account 群长尾六修+两件冻结(终裁§3.2,标尺=应急班车不污染10-08归因)——六件全neutral: connect错误分支单次调用/无mini_path告警节流(首现+30轮心跳)/probe会话id派生per-process(base+pid防重启撞session)/同步查询超时护栏30s(daemon线程join,超时=当轮失败+重置,bs.login挂死同族,fail-closed语义不变)/StockAccount显式account_type/快照字段真缺失首现WARN按0(静默零值与真零可辨,值语义不动); 冻结: traded_at LIKE宽匹配=fills v1.2首实评取数面窗口后再动/wmic时区=确认关单(三日判读绿实证); spec §9.A-19同push; 测试_patch_qmt补类属性复位(裸setattr跨测泄漏) [vps]
CI/CD / nas-deploy (push) Blocked by required conditions
CI/CD / nas-verify (push) Blocked by required conditions
CI/CD / test (push) Successful in 43s

This commit is contained in:
2026-10-07 13:19:18 +08:00
parent c8ba9ea15f
commit e8902c21e8
3 changed files with 210 additions and 25 deletions
@@ -422,6 +422,8 @@ kind 语义(检查分发依据,check_entry data_monitor.py:56-232):
**A-18 QMT 账号硬编码扩散(10-05 用户拍板 ⚖️-2 / P2-18 根治)**:现象=模拟柜台账号 66639661 曾硬编码 32 文件 88 处(灾备腿纯常量/桥腿 env 带默认/wrapper 注入字面量/env 名分裂三个),换号需动 14 生产文件+≥4 生产 wrapper。根因三层=schtask 无 per-task env+机器级 env 未设(wrapper 被迫字面量)/「部署身份」无单一权威源(三代各自发明)/模拟号零安全压力致 copy-paste 传播(run_shadow_bridge.ps1 同文件 redis 口令纪律在线、账号却是字面量=铁证)。修法=config/qmt_identity.json 单一权威源(JSON 非 yaml:灾备腿 stdlib 纯净不能依赖 pyyaml)+gate_common.load_identity 正典(:14-59)+各腿镜像+CI 同步测+码面字面量 grep-pin(tests/trader/test_qmt_identity.py,只 config 三件套可含字面量)。判别复用=换号只改 json+两 yaml(live.yaml/data_platform.yaml 由同步测钉一致)。全案=audit/20261005_qmt_identity_env_plan.md;生产 wrapper 字面量(库外 C:\sanguo_bigqmt)10-08 窗口后按非代码资产流程收尾。
**A-19 account 群长尾六修+两件冻结(10-07 长尾批,终裁 §3.2;标尺=应急班车捎带部署不污染 10-08 窗口归因)**:account_monitor 系被动采集器(快照喂 console/API,不进任何灯色判定)六件全 neutral 当班修(sanguo_live/account_monitor.py,254→317 行):①connect 错误分支单次调用(旧二次调 `t.connect()` 拼消息=副作用+二次值可漂)②无 mini_path 跳过告警节流(首现+每 30 轮心跳;79c6b8b 只节流了配置读取失败)③probe 会话 id 派生 per-process(`_derive_session_id`=base+pid mod,进程内稳定跨进程不撞,防重启与残留旧连接撞 session)④同步查询超时护栏 `_call_with_timeout` 30s(daemon 线程 join;超时=当轮失败 note_failure+重置连接,bs.login 挂死同族防御,fail-closed 语义不变)⑤`StockAccount(account,"STOCK")` 显式账户类型(=现默认零行为差,防上游 API 漂移改默认)⑥快照字段真缺失(attr 缺失≠真零)`_num` 每进程首现 WARN 后按 0(静默零值与真零可辨,值语义不动)。**冻结两件**:`traded_at LIKE` 宽匹配=**10-08 21:30 fills v1.2 首实评取数面(strategy_checks.py:155 `_day_fills`),任何口径变动毁首评归因**→窗口绿后再修(届时有真班基线可对拍);wmic 时区假设=确认关单零代码(连续三日 wmic_shadow=3≡db_running=3 判读绿实证+§4.2-3 CreationDate 保守向留痕在位)。测试=tests/api/test_account_monitor.py 六件契约+`_patch_qmt` 补类属性复位(裸 setattr 跨测试泄漏防护)。
### §9.B 数据面(data,25 条;格式=现象/根因/判别法/修法坐标)
1. **mirror 假黄连三日(0926~28)**:push.log DONE 行只带 [HH:MM:SS] 不带日期(仅 start 行带)→旧 `want_day in line` 过滤永剩 start 行。判别=DONE 行格式先实证。修=start 段归属法(nas_monitor_check.py:166-185)。
+85 -22
View File
@@ -23,13 +23,57 @@ import yaml
logger = logging.getLogger(__name__)
# 专属 probe 会话 id:int,量级刻意远离 bullet_trade 默认的 int(time*1000)(~1.7e12),
# 专属 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
_QUERY_TIMEOUT_SEC = 30.0 # 同步查询超时护栏(长尾④,bs.login 挂死同族)
_MISSING = object() # attr 缺失哨兵(≠值为 None/0, 长尾⑥)
_MISSING_WARNED: set = set() # (类名, 字段) 每进程只首现告警一次
def _derive_session_id() -> int:
"""probe 会话 id 派生:进程内稳定、跨进程不撞——固定 id 在进程重启后与
残留旧连接撞 session(长尾③);base+pid 抖动,量级仍远离引擎 int(time*1000)."""
return PROBE_SESSION_ID + os.getpid() % 100000
def _call_with_timeout(fn, *args):
"""同步查询超时护栏(长尾④):超时=TimeoutError→当轮失败(fail-closed 语义
不变,「无限挂」变「超时重试」);守护线程泄漏有界,不阻塞后续轮次。"""
box: dict[str, Any] = {}
def _run() -> None:
try:
box["ret"] = fn(*args)
except BaseException as e: # noqa: BLE001
box["err"] = e
t = threading.Thread(target=_run, daemon=True)
t.start()
t.join(_QUERY_TIMEOUT_SEC)
if t.is_alive():
raise TimeoutError(f"query timeout {_QUERY_TIMEOUT_SEC}s")
if "err" in box:
raise box["err"]
return box.get("ret")
def _num(obj: Any, field: str) -> float:
"""快照数值读取(长尾⑥):attr 真缺失(≠值为 None/0)→每进程首现 WARN 后按 0
——静默零值与真零可辨;值语义不动(None/缺省照旧折 0)。"""
v = getattr(obj, field, _MISSING)
if v is _MISSING:
key = (type(obj).__name__, field)
if key not in _MISSING_WARNED:
_MISSING_WARNED.add(key)
logger.warning("[account-monitor] 快照字段缺失 %s.%s 按 0 记", *key)
return 0.0
return float(v or 0)
_CFG_FAILURES = 0 # watch 配置读取失败计数(节流告警用, _note_failure 同哲学)
@@ -87,18 +131,21 @@ class AccountMonitor(threading.Thread):
self,
db_path: str,
interval_sec: float = _DEFAULT_INTERVAL_SEC,
session_id: int = PROBE_SESSION_ID,
session_id: int | None = None,
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=进程内派生(跨进程不撞,长尾③);显式传值=测试/运维覆盖
self.session_id = session_id if session_id is not None \
else _derive_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 → 连续失败计数
self._skip_counts: dict[str, int] = {} # account → 无 mini_path 跳过计数(长尾②)
# ---------------- 生命周期 ----------------
@@ -166,49 +213,64 @@ class AccountMonitor(threading.Thread):
return written
def _poll_account(self, account: str, mini_path: str, upsert: Any) -> bool:
"""单账户采集。返回是否写入。mini_path 为空 → 告警跳过(连不上 QMT)。"""
"""单账户采集。返回是否写入。mini_path 为空 → 节流告警跳过(连不上 QMT)。"""
if not mini_path:
logger.warning(
"[account-monitor] 账户 %s 无 mini_path,跳过(需 live_accounts "
"行携带或快照 sticky 记录)", account)
n = self._skip_counts.get(account, 0) + 1 # 长尾②:节流
self._skip_counts[account] = n
if n == 1 or n % _HEARTBEAT_EVERY_N_FAILURES == 0:
logger.warning(
"[account-monitor] 账户 %s 无 mini_path,跳过(需 live_accounts "
"行携带或快照 sticky 记录; 第%d次)", account, n)
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)
acc_obj = StockAccount(account, "STOCK") # 长尾⑤:显式账户类型
try: # 长尾④:超时护栏
asset = _call_with_timeout(trader.query_stock_asset, acc_obj)
except TimeoutError as e:
self._note_failure(mini_path, repr(e))
self._reset_trader(mini_path)
return False
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 []
try:
positions = _call_with_timeout(
trader.query_stock_positions, acc_obj) or []
except TimeoutError as e:
self._note_failure(mini_path, repr(e))
self._reset_trader(mini_path)
return False
rows = []
for p in positions:
vol = float(getattr(p, "volume", 0) or 0)
vol = _num(p, "volume")
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),
"can_use": _num(p, "can_use_volume"),
"avg_price": _num(p, "avg_price"),
"mv": _num(p, "market_value"),
})
cash = _num(asset, "cash")
mv = _num(asset, "market_value")
total = _num(asset, "total_asset")
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),
cash=cash,
market_value=mv,
total=total,
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))
account, cash, mv, total, len(rows))
return True
# ---------------- 连接管理 ----------------
@@ -222,8 +284,9 @@ class AccountMonitor(threading.Thread):
try:
t = XtQuantTrader(mini_path, self.session_id)
t.start()
if t.connect() not in (0, None):
raise RuntimeError(f"connect 返回 {t.connect()}")
rc = t.connect() # 长尾①:单次调用
if rc not in (0, None):
raise RuntimeError(f"connect 返回 {rc}")
self._traders[mini_path] = t
return t
except Exception as e: # noqa: BLE001
+123 -3
View File
@@ -75,6 +75,7 @@ class _FakeTrader:
"""假 XtQuantTrader:类属性记录实例,可编程返回资产/持仓。"""
instances = []
connect_ret = 0
connect_calls = 0
asset = None
positions = []
@@ -88,6 +89,7 @@ class _FakeTrader:
return 0
def connect(self):
type(self).connect_calls += 1
return type(self).connect_ret
def stop(self):
@@ -104,12 +106,19 @@ class _FakeTrader:
def _patch_qmt(monkeypatch, **attrs):
# 类属性默认先复位(裸 setattr 不随 monkeypatch 回滚,跨测试泄漏防护)
_FakeTrader.instances = []
_FakeTrader.connect_calls = 0
_FakeTrader.connect_ret = 0
_FakeTrader.asset = None
_FakeTrader.positions = []
for k, v in attrs.items():
setattr(_FakeTrader, k, v)
_FakeTrader.instances = []
monkeypatch.setattr(
am, "_import_qmt",
lambda: (_FakeTrader, lambda acc: SimpleNamespace(account=acc)))
lambda: (_FakeTrader,
lambda acc, account_type="STOCK": SimpleNamespace(
account=acc)))
def test_watch_targets_union(db, monkeypatch):
@@ -166,7 +175,7 @@ def test_poll_once_upserts_snapshot(db, monkeypatch):
# 连接复用:同路径第二次 poll 不新建 trader
assert m.poll_once() == 1
assert len(_FakeTrader.instances) == 1
assert _FakeTrader.instances[0].session_id == am.PROBE_SESSION_ID
assert _FakeTrader.instances[0].session_id == am._derive_session_id()
def test_poll_query_none_resets_and_no_write(db, monkeypatch, caplog):
@@ -270,3 +279,114 @@ def test_extra_from_config_fail_safe_and_throttled(tmp_path, monkeypatch, caplog
assert am._extra_from_config(
cfg_path=str(tmp_path / "none.yaml")) == {} # 文件缺失
assert len(caplog.records) == 1 # 两次失败只首败告警(节流契约)
# ---------------- 10-07 长尾批(monitoring 终裁 §3.2 account 群六件) ----------------
def test_connect_single_call_in_error_path(db, monkeypatch):
"""长尾①: connect 失败分支不得重调 connect() 拼错误消息(二次调用=副作用+值可漂)."""
_patch_qmt(monkeypatch, connect_ret=-1, asset=None, positions=[])
lp.upsert_account_snapshot(db, "A1", 1, 0, 1, [], "P1")
m = am.AccountMonitor(db, extra_accounts={})
assert m.poll_once() == 0
assert _FakeTrader.connect_calls == 1
def test_empty_mini_path_warn_throttled(db, monkeypatch, caplog):
"""长尾②: 无 mini_path 跳过告警须节流——首现 WARN,后续轮静默
(79c6b8b 只节流了配置读取失败,此枝每轮刷屏)."""
_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"):
for _ in range(3):
assert m.poll_once() == 0
assert len([r for r in caplog.records
if "无 mini_path" in r.message]) == 1
def test_session_id_derived_per_process(db, monkeypatch):
"""长尾③: probe 会话 id 派生=进程内稳定、跨进程不撞(固定 id 在进程重启后
与残留旧连接撞 session);量级仍远离引擎 int(time*1000)."""
_patch_qmt(monkeypatch,
asset=SimpleNamespace(cash=1.0, market_value=0.0,
total_asset=1.0))
lp.upsert_account_snapshot(db, "A1", 1, 0, 1, [], "P1")
m = am.AccountMonitor(db, extra_accounts={})
assert m.poll_once() == 1
sid = _FakeTrader.instances[0].session_id
assert sid == am._derive_session_id() == am._derive_session_id()
assert isinstance(sid, int) and sid >= am.PROBE_SESSION_ID
def test_query_timeout_fail_closed(db, monkeypatch, caplog):
"""长尾④: 同步查询挂死→超时护栏=当轮失败+重置连接,不挂 poll 线程
(bs.login 挂死同族防御;fail-closed 语义不变)."""
import time
class _Hanging(_FakeTrader):
def query_stock_asset(self, acc):
time.sleep(5)
return None
_patch_qmt(monkeypatch, asset=None, positions=[])
monkeypatch.setattr(
am, "_import_qmt",
lambda: (_Hanging,
lambda acc, account_type="STOCK": SimpleNamespace(
account=acc)))
monkeypatch.setattr(am, "_QUERY_TIMEOUT_SEC", 0.05)
lp.upsert_account_snapshot(db, "A1", 1, 0, 1, [], "P1")
m = am.AccountMonitor(db, extra_accounts={})
t0 = time.monotonic()
with caplog.at_level("WARNING"):
assert m.poll_once() == 0
assert time.monotonic() - t0 < 3 # 未被 5s sleep 拖住
assert _FakeTrader.instances[-1].stopped # 超时即重置连接
assert any("timeout" in r.message.lower() for r in caplog.records)
def test_stock_account_type_explicit(db, monkeypatch):
"""长尾⑤: StockAccount 显式携带 account_type('STOCK'=现默认,零行为差)
——防上游 API 漂移改默认时探针静默换了账户类型."""
made = []
def _mk(acc, account_type="UNSET"):
made.append((acc, account_type))
return SimpleNamespace(account=acc)
_patch_qmt(monkeypatch,
asset=SimpleNamespace(cash=1.0, market_value=0.0,
total_asset=1.0))
monkeypatch.setattr(am, "_import_qmt", lambda: (_FakeTrader, _mk))
lp.upsert_account_snapshot(db, "A1", 1, 0, 1, [], "P1")
m = am.AccountMonitor(db, extra_accounts={})
assert m.poll_once() == 1
assert made and made[0][1] == "STOCK"
def test_missing_attr_visible_not_silent_zero(db, monkeypatch, caplog):
"""长尾⑥: 持仓对象真缺字段(attr 缺失≠真零)→快照值仍按 0(语义不动)但
每进程首现 WARN——静默零值与真零可辨."""
_patch_qmt(monkeypatch,
asset=SimpleNamespace(cash=1.0, market_value=0.0,
total_asset=1.0),
positions=[SimpleNamespace(stock_code="600036.SH",
volume=100)]) # 缺 can_use/avg/mv
lp.upsert_account_snapshot(db, "A1", 1, 0, 1, [], "P1")
m = am.AccountMonitor(db, extra_accounts={})
with caplog.at_level("WARNING"):
assert m.poll_once() == 1
n1 = len([r for r in caplog.records if "字段缺失" in r.message])
assert n1 == 3
with caplog.at_level("WARNING"):
assert m.poll_once() == 1
assert len([r for r in caplog.records if "字段缺失" in r.message]) == n1
snap = lp.get_account_snapshot(db, "A1")
assert snap["positions"][0]["volume"] == 100.0
assert snap["positions"][0]["avg_price"] == 0.0 # 值语义不动