400 lines
18 KiB
Python
400 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""策略域体检检查函数库 — 数据面 monitor_checks 的策略镜像(目录内自含).
|
||
|
||
判据全只读: 主库(paper_accounts/live_trades/paper_trades/dbbardata/order_events)
|
||
+ pipeline.db(is_daily/strategy_weekly) + logs/live_*.log+shadow_*.log + wmic 进程快照
|
||
+ data/monitor/health_latest.json(data-monitor 存活外盯,只读不写)。
|
||
返回统一形状 {"key","status","expected","actual","evidence"}; status 取
|
||
green / 注册表项 severity(违例时)。
|
||
|
||
wmic 纪律(照 scripts/ci/vps_count_engines.py): 只在 name='python.exe' 的进程里数
|
||
cmdline 含 'sanguo_trader.shadow --account' 的条目——监控自身/其它 python 进程
|
||
不含该串, 天然无自匹配(无需 +1 修正)。
|
||
"""
|
||
import datetime as dt
|
||
import glob
|
||
import os
|
||
import sqlite3
|
||
import subprocess
|
||
|
||
from sanguo_portfolio import order_event_store # noqa: E402 monitor 先补 sys.path
|
||
|
||
NEEDLE = "sanguo_trader.shadow --account"
|
||
ERROR_TOKENS = ("ERROR", "Traceback")
|
||
LEFTOVER_STATUS = ("open", "ack", "partial") # expire 为终态不算残留
|
||
# 现役引擎日志=shadow_*.log（LIVE=0 时代）；live_* 一并盯，live 桌回归日自动覆盖
|
||
LOG_GLOBS = ("live_*.log", "shadow_*.log")
|
||
|
||
|
||
def run_wmic():
|
||
"""wmic 进程快照 CSV(仅 python.exe); 非 Windows/无 wmic=异常上抛由调用方降黄."""
|
||
out = subprocess.run(
|
||
["wmic", "process", "where", "name='python.exe'",
|
||
"get", "commandline,creationdate", "/format:csv"],
|
||
capture_output=True, timeout=120, check=True)
|
||
return out.stdout.decode("utf-8", errors="replace")
|
||
|
||
|
||
def parse_wmic_csv(text):
|
||
"""CSV→[(cmdline, creation datetime|None)]; 首行表头跳过, 空字段行→("", None).
|
||
|
||
CreationDate=YYYYMMDDHHMMSS.ffffff+zzz 取前 14 位解析; 行尾逗号 rpartition
|
||
(cmdline 可能自带逗号), 首个逗号 partition 剥 NODE 列。
|
||
"""
|
||
rows = []
|
||
lines = [ln for ln in text.replace("\r\n", "\n").split("\n") if ln.strip()]
|
||
for line in lines[1:]:
|
||
head, _, raw_created = line.rpartition(",")
|
||
_, _, cmdline = head.partition(",")
|
||
created = None
|
||
raw_created = raw_created.strip()
|
||
if raw_created[:14].isdigit():
|
||
try:
|
||
created = dt.datetime.strptime(raw_created[:14], "%Y%m%d%H%M%S")
|
||
except ValueError:
|
||
created = None
|
||
rows.append((cmdline, created))
|
||
return rows
|
||
|
||
|
||
def shadow_processes(text):
|
||
"""shadow 引擎进程 [(cmdline, creation)]; 计数=自匹配免疫的口径."""
|
||
return [(cmd, c) for cmd, c in parse_wmic_csv(text) if NEEDLE in cmd]
|
||
|
||
|
||
def trading_days(db_path, today, lookback=50):
|
||
"""确认集=dbbardata 000300 实证日期(近 lookback 日, [:10] 归一为 ISO 日);
|
||
库不可达=[] 调用方降级工作日启发式(monitor_checks.trading_days 同款)."""
|
||
cutoff = (today - dt.timedelta(days=lookback)).isoformat()
|
||
try:
|
||
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=10)
|
||
try:
|
||
rows = conn.execute(
|
||
"SELECT DISTINCT datetime FROM dbbardata "
|
||
"WHERE symbol='000300' AND exchange='SSE' AND interval='d' "
|
||
"AND datetime>=? ORDER BY datetime", (cutoff,)).fetchall()
|
||
finally:
|
||
conn.close()
|
||
return sorted({str(r[0])[:10] for r in rows if r[0]})
|
||
except sqlite3.Error:
|
||
return []
|
||
|
||
|
||
def _ro(path):
|
||
return sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=10)
|
||
|
||
|
||
def _count_running(db_path):
|
||
"""running 账户数, 排除 live 桌(mode='live', live_orchestrator 权威口径)——
|
||
本检查族的 wmic needle 数的是 shadow 进程, live 桌回归日口径不再混桌
|
||
(10-01 二次审计 P1-3 同族; mode 未标 NULL/其它=影子/eod 舰队都计入)."""
|
||
conn = _ro(db_path)
|
||
try:
|
||
return int(conn.execute(
|
||
"SELECT COUNT(*) FROM paper_accounts "
|
||
"WHERE status='running' AND (mode IS NULL OR mode!='live')"
|
||
).fetchone()[0])
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _desk_accounts(db_path, mode):
|
||
"""desk 账户行 [(id, instance_id, strategies_json)]; live 桌=mode='live'."""
|
||
conn = _ro(db_path)
|
||
try:
|
||
return conn.execute(
|
||
"SELECT id, instance_id, strategies FROM paper_accounts "
|
||
"WHERE mode=? AND status='running'", (mode,)).fetchall()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _dual_track_pairs(db_path):
|
||
"""live↔shadow 配对 [(live_id, shadow_id)] + 未配对清单(10-05 审计 P3:
|
||
单边 instance_id 漂移曾静默消失在 pairs 外=reconcile 覆盖面暗洞——
|
||
返回清单供 evidence 点名, 灯色语义不动)(reconcile_report v2 语义镜像:
|
||
instance_id 优先精确配对, 缺 instance 回退 strategies 首名; 10-01 审计
|
||
P1-3——全表总量口径在 live 桌回归日必误红, 逐对账才是双轨本义)."""
|
||
import json as _json
|
||
|
||
def _first_name(s):
|
||
try:
|
||
return (_json.loads(s or "[]") or [{}])[0].get("name", "")
|
||
except (ValueError, IndexError, KeyError):
|
||
return ""
|
||
|
||
lives = [(r[0], r[1], _first_name(r[2]))
|
||
for r in _desk_accounts(db_path, "live")]
|
||
shadows = [(r[0], r[1], _first_name(r[2]))
|
||
for r in _desk_accounts(db_path, "shadow")]
|
||
pairs, used = [], set()
|
||
unpaired_live = []
|
||
for lid, linst, lname in lives:
|
||
cand = next(((s, ) for s, sinst, _ in shadows
|
||
if sinst and sinst == linst and s not in used), None) \
|
||
or next(((s, ) for s, _, sname in shadows
|
||
if not (linst or None) and lname and sname == lname
|
||
and s not in used), None)
|
||
if cand:
|
||
used.add(cand[0])
|
||
pairs.append((lid, cand[0]))
|
||
else:
|
||
unpaired_live.append(lid)
|
||
unpaired_shadow = [sid for sid, _, _ in shadows if sid not in used]
|
||
return pairs, unpaired_live, unpaired_shadow
|
||
|
||
|
||
def _day_fills(db_path, table, account_id, today):
|
||
"""当日某账户成交 (笔数, 金额)——live_trades 按 traded_at 前缀,
|
||
paper_trades 只数 rejected=0 按 bar_date."""
|
||
conn = _ro(db_path)
|
||
try:
|
||
if table == "live_trades":
|
||
row = conn.execute(
|
||
"SELECT COUNT(*), COALESCE(SUM(price*volume),0) FROM live_trades "
|
||
"WHERE account_id=? AND traded_at LIKE ?",
|
||
(account_id, today + "%")).fetchone()
|
||
else:
|
||
row = conn.execute(
|
||
"SELECT COUNT(*), COALESCE(SUM(price*volume),0) FROM paper_trades "
|
||
"WHERE account_id=? AND rejected=0 AND bar_date=?",
|
||
(account_id, today)).fetchone()
|
||
return int(row[0]), float(row[1] or 0)
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _pipeline_one(ctx, sql, args=()):
|
||
conn = _ro(ctx["pipeline_db"])
|
||
try:
|
||
return conn.execute(sql, args).fetchone()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _done(e, status, expected, actual, evidence=None):
|
||
return {"key": e["key"], "status": status, "expected": expected,
|
||
"actual": actual, "evidence": evidence or {}}
|
||
|
||
|
||
# ---------- 盘前班: 引擎面 ----------
|
||
|
||
def engines_expected(ctx, e):
|
||
running = _count_running(ctx["main_db"])
|
||
want = e["expected"]
|
||
status = "green" if running == want else e["severity"]
|
||
return _done(e, status, want, running,
|
||
{"running": running, "expected": want})
|
||
|
||
|
||
def engines_actual(ctx, e):
|
||
shadows = shadow_processes(run_wmic())
|
||
running = _count_running(ctx["main_db"])
|
||
status = "green" if len(shadows) == running else e["severity"]
|
||
return _done(e, status, running, len(shadows),
|
||
{"wmic_shadow": len(shadows), "db_running": running})
|
||
|
||
|
||
def engines_no_restart(ctx, e):
|
||
"""近 N 个**确认**交易日盘窗(09:15~15:30)内出生的 shadow 进程=会话日重启
|
||
(§11.0; 10-01 二次审计 P2-3 重设计)。旧「creation>=当日00:00」双失效:
|
||
盘中重启次日必绿(只盖 00:00-08:05)且 08:05 当日必然未确认→red 恒降黄。
|
||
新语义: 违例日均为已确认交易日→red 可达, 不经未确认日降级; 夜间/盘前
|
||
出生=合法 deploy 窗不计; 连休无新增确认日=自然绿。"""
|
||
shadows = shadow_processes(run_wmic())
|
||
days = trading_days(ctx["main_db"], ctx["now"].date())
|
||
if not days:
|
||
# 空确认日历(库不可达/脏数据)=降黄 fail-safe:旧码 any(空窗)恒 False→
|
||
# 零违例→绿=vacuous 绿, 数据坏灯色假安(S-2; 正常连休期确认集仍有
|
||
# 历史交易日不触此枝; data 域 D-5/6 同哲学)
|
||
return _done(e, "yellow", 0, "no-confirmed-days",
|
||
{"violations": [], "procs": len(shadows), "windows": [],
|
||
"reason": "empty confirmed calendar"})
|
||
windows = []
|
||
for d in sorted(days)[-e["lookback_confirmed"]:]:
|
||
ymd = d.replace("-", "")
|
||
windows.append((ymd + "0915", ymd + "1530"))
|
||
bad = []
|
||
for cmd, c in shadows:
|
||
if c is None:
|
||
# 复核留痕(P3 fail-safe 审计项): CreationDate 解析失败按违例计=
|
||
# 保守向——wmic 口径变化若静默跳过, 盘中重启检测整列失明;
|
||
# 空字段行("HOST,,")不含 needle 已被 shadow_processes 过滤,
|
||
# 此枝只剩真行解析失败, created:null 与真违例(created 有值)可辨
|
||
bad.append({"cmd": cmd, "created": None})
|
||
continue
|
||
ts = c.strftime("%Y%m%d%H%M")
|
||
if any(a <= ts <= b for a, b in windows):
|
||
bad.append({"cmd": cmd, "created": c.isoformat()})
|
||
status = "green" if not bad else e["severity"]
|
||
return _done(e, status, 0, len(bad),
|
||
{"violations": bad[:20], "procs": len(shadows),
|
||
"windows": [f"{a[:8]}" for a, _ in windows]})
|
||
|
||
|
||
# ---------- 盘前班: 日志面 ----------
|
||
|
||
def logs_error_storm(ctx, e):
|
||
"""live_*+shadow_* 26h 窗(两班通用, 旧失败不重炒——数据面 _LOG_WINDOW_H 同值)
|
||
内 ERROR/Traceback 行数 ≤ 阈值. 09-30 修订: 修前只盯 live_*.log, LIVE=0 时代
|
||
全是 stopped 旧文件=窗内恒 0(vacuous 绿), 09-30 08:05 首班 logs_stale 误黄实证."""
|
||
window = ctx["now"] - dt.timedelta(hours=e["window_hours"])
|
||
total = files = unreadable = 0
|
||
for pattern in LOG_GLOBS:
|
||
for p in glob.glob(os.path.join(ctx["logs_dir"], pattern)):
|
||
try:
|
||
if dt.datetime.fromtimestamp(os.path.getmtime(p)) < window:
|
||
continue
|
||
with open(p, encoding="utf-8", errors="replace") as f:
|
||
text = f.read()
|
||
except OSError:
|
||
# 读不动不静默(P3 漏报向收敛): 计数入 evidence 留痕——
|
||
# 文件锁/权限异常时值班可见; 灯色公式不动(10-08 纪律)
|
||
unreadable += 1
|
||
continue
|
||
files += 1
|
||
total += sum(1 for ln in text.splitlines()
|
||
if any(tok in ln for tok in ERROR_TOKENS))
|
||
status = "green" if total <= e["max_errors"] else e["severity"]
|
||
return _done(e, status, f"<={e['max_errors']}", total,
|
||
{"errors": total, "files_in_window": files,
|
||
"unreadable_files": unreadable,
|
||
"window_hours": e["window_hours"]})
|
||
|
||
|
||
def logs_stale(ctx, e):
|
||
"""最新 live_*+shadow_* mtime 距今 ≤ max_age_days(连休容忍; 交易日班评——
|
||
休市表剔假日, 表外工作日照评=复市日 morn 一班诚实黄, P3-26 docstring 随迁)."""
|
||
mtimes = []
|
||
for pattern in LOG_GLOBS:
|
||
for p in glob.glob(os.path.join(ctx["logs_dir"], pattern)):
|
||
try:
|
||
mtimes.append(os.path.getmtime(p))
|
||
except OSError:
|
||
continue
|
||
if not mtimes:
|
||
return _done(e, e["severity"], "any", None, {"files": 0})
|
||
newest = dt.datetime.fromtimestamp(max(mtimes))
|
||
age_h = (ctx["now"] - newest).total_seconds() / 3600
|
||
status = "green" if age_h <= e["max_age_days"] * 24 else e["severity"]
|
||
return _done(e, status, f"<={e['max_age_days']}d", f"{age_h:.0f}h",
|
||
{"newest_age_h": round(age_h, 1), "files": len(mtimes)})
|
||
|
||
|
||
# ---------- 盘后班: 产物+订单链 ----------
|
||
|
||
def is_report_ran(ctx, e):
|
||
row = _pipeline_one(ctx, "SELECT 1 FROM is_daily WHERE day=? LIMIT 1",
|
||
(ctx["today"],))
|
||
status = "green" if row else e["severity"]
|
||
return _done(e, status, "present", bool(row), {"day": ctx["today"]})
|
||
|
||
|
||
def is_identity_resid(ctx, e):
|
||
"""当日行 |identity_resid| ≤ 容差; 无行可判=绿(缺失归 is_report_ran 管径)."""
|
||
conn = _ro(ctx["pipeline_db"])
|
||
try:
|
||
rows = conn.execute(
|
||
"SELECT ABS(identity_resid) FROM is_daily WHERE day=?",
|
||
(ctx["today"],)).fetchall()
|
||
finally:
|
||
conn.close()
|
||
vals = [r[0] for r in rows if r[0] is not None]
|
||
if not vals:
|
||
return _done(e, "green", f"|resid|<={e['tol']}", "n/a", {"rows": 0})
|
||
worst = max(vals)
|
||
status = "green" if worst <= e["tol"] else e["severity"]
|
||
return _done(e, status, f"|resid|<={e['tol']}", worst,
|
||
{"rows": len(vals), "max_abs": worst})
|
||
|
||
|
||
def orders_leftover(ctx, e):
|
||
"""重放当日无 open/ack/partial 终态残留(submitted 单独存在=初始 open 态)."""
|
||
states = order_event_store.replay_order_states(ctx["main_db"],
|
||
day=ctx["today"])
|
||
leftover = sorted(k for k, s in states.items()
|
||
if s.get("status") in LEFTOVER_STATUS)
|
||
status = "green" if not leftover else e["severity"]
|
||
return _done(e, status, 0, len(leftover),
|
||
{"leftover": leftover[:20], "orders": len(states)})
|
||
|
||
|
||
def fills_vs_paper(ctx, e):
|
||
"""当日双轨配对逐对账: live 账户 vs 其 shadow 镜像(笔数+金额, 容差内)。
|
||
10-01 二次审计 P1-3 重设计——全表总量口径无账户配对, live 桌回归日两桌
|
||
各跑各的策略必总量不平=误红; 逐对账才是「同策略同参数分别接真/假 broker」
|
||
的双轨本义。无 running live 账户=n/a 绿(09-30 修订保留: shadow 桌本地撮合
|
||
订单不出门, LIVE=0 时代 shadow 独走日不构成对账对象); 配对内 0:1 也算不平
|
||
(影子镜像未跟=真双轨分歧)。柜台↔台账恒等式仍归 15:10 仪式(infra 域)。"""
|
||
pairs, unpaired_l, unpaired_s = _dual_track_pairs(ctx["main_db"])
|
||
if not pairs:
|
||
if not unpaired_l:
|
||
return _done(e, "green", "n/a", "no-live-desk",
|
||
{"pairs": 0,
|
||
"note": "无 running live 账户, 双轨对账不适用"})
|
||
# live 桌在但零配对(10-05 P3): 不再伪装成 no-live-desk——绿灯但点名
|
||
# (instance_id 漂移嫌疑; 灯色不动=10-08 纪律, 升级与否复市后再议)
|
||
return _done(e, "green", "n/a", f"live-desk-unpaired({len(unpaired_l)})",
|
||
{"pairs": 0, "unpaired_live": unpaired_l,
|
||
"unpaired_shadow": unpaired_s,
|
||
"note": "live 桌存在但零配对——instance_id 漂移嫌疑, 人工核对"})
|
||
bad = []
|
||
for lid, sid in pairs:
|
||
ln, lam = _day_fills(ctx["main_db"], "live_trades", lid, ctx["today"])
|
||
pn, pam = _day_fills(ctx["main_db"], "paper_trades", sid, ctx["today"])
|
||
if not (ln == pn and abs(lam - pam) <= e["amount_tol"]):
|
||
bad.append({"live": lid, "shadow": sid,
|
||
"live_count": ln, "live_amount": round(lam, 2),
|
||
"paper_count": pn, "paper_amount": round(pam, 2)})
|
||
status = "green" if not bad else e["severity"]
|
||
return _done(e, status, f"逐对count相等 且 金额±{e['amount_tol']}",
|
||
f"{len(pairs) - len(bad)}/{len(pairs)}对平",
|
||
{"pairs": len(pairs), "mismatch": bad[:10],
|
||
"unpaired_live": unpaired_l, "unpaired_shadow": unpaired_s})
|
||
|
||
|
||
def weekly_ran(ctx, e):
|
||
"""strategy_weekly 本 ISO 周行存在(任一 strategy_id 即算)."""
|
||
d = dt.date.fromisoformat(ctx["today"])
|
||
monday = (d - dt.timedelta(days=d.weekday())).isoformat()
|
||
row = _pipeline_one(
|
||
ctx, "SELECT 1 FROM strategy_weekly WHERE week_start>=? LIMIT 1",
|
||
(monday,))
|
||
status = "green" if row else e["severity"]
|
||
return _done(e, status, f"week_start>={monday}", bool(row),
|
||
{"week_start": monday})
|
||
|
||
|
||
def data_monitor_alive(ctx, e):
|
||
"""data-monitor health_latest.json 新鲜度(互为外盯, 10-01 审计 P1-10 另一半).
|
||
|
||
data-monitor 每日两班(08:05/23:50)原子写 {data_root}/monitor/health_latest.json
|
||
(节假日 schtask 照跑, 日历降级在进程内); >max_age_hours 无新写=其两班全死,
|
||
而它的 SCHTASKS 自检自己也死了无人执行=全链静默——本检在 strategy 两班
|
||
侧外盯兜住该结构洞(反向: data-monitor SCHTASKS 已盯 strategy 两班)."""
|
||
p = ctx["data_monitor_health"]
|
||
try:
|
||
mtime = dt.datetime.fromtimestamp(os.path.getmtime(p))
|
||
except OSError:
|
||
return _done(e, e["severity"], f"<={e['max_age_hours']}h", "missing",
|
||
{"file": p})
|
||
age_h = (ctx["now"] - mtime).total_seconds() / 3600
|
||
status = "green" if age_h <= e["max_age_hours"] else e["severity"]
|
||
return _done(e, status, f"<={e['max_age_hours']}h", f"{age_h:.0f}h",
|
||
{"file": p, "age_h": round(age_h, 1)})
|
||
|
||
|
||
CHECK_FNS = {
|
||
"engines_expected": engines_expected,
|
||
"engines_actual": engines_actual,
|
||
"engines_no_restart": engines_no_restart,
|
||
"logs_error_storm": logs_error_storm,
|
||
"logs_stale": logs_stale,
|
||
"is_report_ran": is_report_ran,
|
||
"is_identity_resid": is_identity_resid,
|
||
"orders_leftover": orders_leftover,
|
||
"fills_vs_paper": fills_vs_paper,
|
||
"weekly_ran": weekly_ran,
|
||
"data_monitor_alive": data_monitor_alive,
|
||
}
|