318 lines
13 KiB
Python
318 lines
13 KiB
Python
"""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
|
|
|
|
import yaml
|
|
|
|
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
|
|
|
|
_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 同哲学)
|
|
|
|
|
|
def _cfg_fail_note(reason: str) -> None:
|
|
"""配置读取失败节流告警:首败 WARN+此后每 30 败心跳一次(A-1)."""
|
|
global _CFG_FAILURES
|
|
_CFG_FAILURES += 1
|
|
if _CFG_FAILURES == 1 or _CFG_FAILURES % _HEARTBEAT_EVERY_N_FAILURES == 0:
|
|
logger.warning("[account-monitor] watch 配置读取失败(第%d次): %s",
|
|
_CFG_FAILURES, reason)
|
|
|
|
|
|
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(cfg_path: str | None = None) -> dict[str, str]:
|
|
"""config/data_platform.yaml live_trading.watch_accounts + env 兜底。
|
|
|
|
返回 {account: mini_path};读不到/没配/解析失败 → 空 dict(不炸)。
|
|
A-1(10-05 审计): 宽捕 (OSError, ValueError, yaml.YAMLError)——坏 YAML 旧码
|
|
未捕=poll 线程逐轮裸抛再被 run() 兜成刷屏 warning;现经 _cfg_fail_note 节流。
|
|
ImportError 不入捕:yaml 已上移模块级 import, 缺依赖=进程启动即 fail loud
|
|
(runner.py 同款; 懒加载版在 except 里引用 yaml.YAMLError 有 NameError 雷)。
|
|
"""
|
|
extra: dict[str, str] = {}
|
|
if cfg_path is None:
|
|
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:
|
|
loaded = yaml.safe_load(f) or {}
|
|
if not isinstance(loaded, dict):
|
|
raise ValueError(f"配置根节点非映射: {type(loaded).__name__}")
|
|
lt = loaded.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, yaml.YAMLError) as e:
|
|
_cfg_fail_note(repr(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 | 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
|
|
# 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 跳过计数(长尾②)
|
|
|
|
# ---------------- 生命周期 ----------------
|
|
|
|
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:
|
|
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, "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
|
|
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 = _num(p, "volume")
|
|
if vol <= 0:
|
|
continue
|
|
rows.append({
|
|
"symbol": str(getattr(p, "stock_code", "") or ""),
|
|
"volume": vol,
|
|
"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=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, cash, mv, total, 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()
|
|
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
|
|
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"]
|