sanguo_vnpy_v2/scripts/strategy_monitor/strategy_checks.py

# -*- coding: utf-8 -*-
"""策略域体检检查函数库 — 数据面 monitor_checks 的策略镜像(目录内自含).

判据全只读: 主库(paper_accounts/live_trades/paper_trades/dbbardata/order_events)
+ pipeline.db(is_daily/strategy_weekly) + logs/live_*.log + wmic 进程快照。
返回统一形状 {"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 为终态不算残留


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):
    conn = _ro(db_path)
    try:
        return int(conn.execute(
            "SELECT COUNT(*) FROM paper_accounts WHERE status='running'"
        ).fetchone()[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):
    """全 shadow 进程 CreationDate < 当日 00:00(盘前出生=昨夜/今晨被重启)."""
    shadows = shadow_processes(run_wmic())
    bad = [{"cmd": cmd, "created": c.isoformat() if c else None}
           for cmd, c in shadows if c is None or c >= ctx["midnight"]]
    status = "green" if not bad else e["severity"]
    return _done(e, status, f"<{ctx['midnight'].isoformat()}", len(bad),
                 {"violations": bad[:20], "procs": len(shadows)})


# ---------- 盘前班: 日志面 ----------

def logs_error_storm(ctx, e):
    """live_*.log 26h 窗(两班通用, 旧失败不重炒——数据面 _LOG_WINDOW_H 同值)
    内 ERROR/Traceback 行数 ≤ 阈值."""
    window = ctx["now"] - dt.timedelta(hours=e["window_hours"])
    total = files = 0
    for p in glob.glob(os.path.join(ctx["logs_dir"], "live_*.log")):
        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:
            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,
                  "window_hours": e["window_hours"]})


def logs_stale(ctx, e):
    """最新 live_*.log mtime 距今 ≤ max_age_days(连休容忍; 仅工作日班评)."""
    mtimes = []
    for p in glob.glob(os.path.join(ctx["logs_dir"], "live_*.log")):
        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_trades vs paper_trades(rejected=0 才是成交, 一表两态)笔数+金额
    双对账; 两表都空=绿."""
    conn = _ro(ctx["main_db"])
    try:
        live = conn.execute(
            "SELECT COUNT(*), COALESCE(SUM(price*volume),0) FROM live_trades "
            "WHERE traded_at LIKE ?", (ctx["today"] + "%",)).fetchone()
        paper = conn.execute(
            "SELECT COUNT(*), COALESCE(SUM(price*volume),0) FROM paper_trades "
            "WHERE rejected=0 AND bar_date=?", (ctx["today"],)).fetchone()
    finally:
        conn.close()
    ln, lam = int(live[0]), float(live[1] or 0)
    pn, pam = int(paper[0]), float(paper[1] or 0)
    if ln == 0 and pn == 0:
        return _done(e, "green", "match", "both-empty",
                     {"live_count": 0, "live_amount": 0.0,
                      "paper_count": 0, "paper_amount": 0.0})
    ok = ln == pn and abs(lam - pam) <= e["amount_tol"]
    status = "green" if ok else e["severity"]
    return _done(e, status, f"count相等 且 金额±{e['amount_tol']}",
                 f"{ln}:{pn} {lam:.2f}:{pam:.2f}",
                 {"live_count": ln, "live_amount": round(lam, 2),
                  "paper_count": pn, "paper_amount": round(pam, 2)})


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})


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,
}