tradingSystem/scripts/check_db.py

# -*- coding: utf-8 -*-
"""
连通性与表结构自检 (实机, 需 .env 真实连接串)
==============================================
运行:  docker compose run --rm pms-web python scripts/check_db.py

检查项:
  1. 三个库连通性 (153 代理 / 因子库 / 指数库)
  2. pms_* 十三张表是否存在与当前行数
  3. 下游三表可读性 + 持仓数量列探测结果 (回填 QMT_INTERFACE_REQUIREMENTS A1/D1 用)
  4. 行情 Redis (db13) 连通性与样本键
  5. 运行参数表当前生效值
  6. ws 通道: 密钥配置 / 本端公钥 (带外交给 QMT) / 进程心跳 / seq 水位 / 出口队列
全部通过退出码 0; 任一 FAIL 退出码 1 (WARN 不影响退出码)。
"""
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

PMS_TABLES = ["pms_command", "pms_plan", "pms_position", "pms_lot", "pms_instruction",
              "pms_proposal", "pms_action_ledger", "pms_daily_report", "pms_industry_map",
              "pms_runtime_param",
              # ws 直连通道三表 (协议 QMT_WS_PROTOCOL.md V1.0)
              "pms_qmt_order", "pms_qmt_inbox", "pms_ws_state"]
DOWNSTREAM = ["trading_position", "trading_order", "trading_buy_plan"]

FAILED, WARNED = [], []


def line(tag, msg):
    print(f"  {tag:<6}{msg}")


def fail(msg):
    FAILED.append(msg)
    line("FAIL", msg)


def warn(msg):
    WARNED.append(msg)
    line("WARN", msg)


def check_ws_channel():
    """通道自检。密钥缺失只报 WARN —— 影子运行期本来就还没配, 不该把自检判成失败。"""
    from config.settings import settings
    from app.core import ws_codec as wsc
    from app.repo import qmt_repo
    from app.services import dispatcher, param_store

    mode = param_store.get("PMS_DISPATCH_MODE", "shadow")
    enabled = param_store.get_bool("PMS_QMT_WS_ENABLED", False)
    line("OK", f"下发模式 PMS_DISPATCH_MODE = {mode}; 通道开关 PMS_QMT_WS_ENABLED = {enabled}")
    line("", f"端点 {param_store.get('PMS_QMT_WS_URL', settings.PMS_QMT_WS_URL)}")

    seed = (settings.PMS_QMT_SIGN_SEED_HEX or "").strip()
    peer = (settings.PMS_QMT_PEER_PUBKEY_B64 or "").strip()
    if not seed:
        (warn if mode != "ws" else fail)(
            "PMS_QMT_SIGN_SEED_HEX 未配置 —— 无法签名, ws 通道不可用 (见 .env.example)")
    else:
        try:
            # 打印本端公钥: 部署清单 §10.1.1 第 2 条要把它带外交给 QMT 侧
            line("OK", f"本端 Ed25519 公钥 (交给 QMT 侧加白): {wsc.public_key_b64(seed)}")
        except Exception as e:
            fail(f"PMS_QMT_SIGN_SEED_HEX 非法: {e}")
    if not peer:
        (warn if mode != "ws" else fail)(
            "PMS_QMT_PEER_PUBKEY_B64 未配置 —— 无法验上行签名, 所有上行消息都会被丢弃")
    else:
        line("OK", f"对端公钥已配置 ({peer[:12]}...)")

    if param_store.get_bool("PMS_QMT_SIGN_SEED_HEX") or param_store.get("PMS_QMT_SIGN_SEED_HEX"):
        fail("私钥出现在 ParamStore 可读路径 —— 违反协议 §10.1.1, 请检查 SECRET_KEYS")

    try:
        ch = dispatcher.channel_status()
    except Exception as e:
        fail(f"通道状态读取失败: {type(e).__name__}: {e}")
        return
    alive = "在线" if ch["process_alive"] else "不在线"
    line("OK" if ch["process_alive"] else "INFO",
         f"pms-ws 进程 {alive} (心跳 {ch['heartbeat_age_sec']}s 前), 连接 {ch['conn_state']}")
    line("", f"seq 水位: 已落库 {ch['last_seq']} / 已确认 {ch['acked_seq']}; "
             f"待入账上行 {ch['inbox_pending']} 条")
    line("", f"出口队列: {ch['queue'] or '空'}")
    if ch.get("resync_required"):
        fail("resync_flag=1: 对端补发不全, 必须走全量对账后经页面清除标记 (协议 §6.2)")
    if mode == "ws" and not ch["online"]:
        warn("已切 ws 模式但通道未在线 —— 买入会被拒发 (卖出仍入队), 先起 pms-ws: "
             "docker compose --profile ws up -d pms-ws")
    if ch.get("error"):
        warn(ch["error"])
    stuck = (ch["queue"] or {}).get("SEND_FAILED", 0)
    if stuck:
        warn(f"{stuck} 张委托发送失败待人工处理 (pms_qmt_order.status = SEND_FAILED)")


def main():
    from app.db import session as dbs

    print("\n[1] 数据库连通性")
    for name, label in (("proxy", "153 代理 (pms_* / trading_* / 因子分表)"),
                        ("index", "大盘指数库")):
        r = dbs.ping(name)
        if r["ok"]:
            line("OK", f"{label}")
        elif name == "proxy":
            fail(f"{label} 连接失败: {r['error']}")
        else:
            warn(f"{label} 连接失败 (非致命): {r['error']}")
    rf = dbs.ping("factor")
    line("OK" if rf["ok"] else "INFO",
         "因子库直连 (SOURCE_DB_EXT_DSN) —— PMS 当前**不使用**: 因子分表经 "
         "ShardingSphere 动态分片, 必须走 153 代理读 (沿用 bionic 结论), "
         f"故此项连不通不影响任何功能。当前: {'通' if rf['ok'] else rf['error'].splitlines()[0]}")

    print("\n[2] PMS 自有表")
    missing_tables = []
    for t in PMS_TABLES:
        try:
            r = dbs.fetch_one(f"SELECT COUNT(*) AS n FROM {t}")
            line("OK", f"{t:<20} {r['n']} 行")
        except Exception as e:
            msg = str(e)
            if "does not exist" in msg or "doesn't exist" in msg:
                missing_tables.append(t)
                fail(f"{t} 不存在")
            else:
                fail(f"{t} 不可用: {type(e).__name__}: {e}")
    if missing_tables:
        line("", "→ 建表: docker compose run --rm pms-web python scripts/init_db.py --yes")

    print("\n[3] 下游只读表 (完整列定义, 可直接回填 QMT 需求清单 D1)")
    from app.repo import downstream_repo
    for t in DOWNSTREAM:
        try:
            cols = downstream_repo.describe(t)
            line("OK", f"{t} —— {len(cols)} 列")
            for c in cols:
                line("", f"    {str(c.get('Field')):<26} {str(c.get('Type')):<18} "
                         f"{'NULL' if str(c.get('Null')).upper() == 'YES' else 'NOT NULL':<9}"
                         f"{str(c.get('Comment') or '')[:40]}")
        except Exception as e:
            warn(f"{t} 读取失败: {type(e).__name__}: {e}")
    try:
        ds = downstream_repo.fetch_positions()
        if ds["raw_count"] and not ds["columns"].get("qty"):
            warn(f"未识别出持仓数量列 {ds['columns']} —— 把真实列名补进 "
                 f"downstream_repo.QTY_CANDIDATES")
        else:
            line("OK", f"持仓列映射: {ds['columns']} (共 {ds['raw_count']} 行)")
            sample = ds["rows"][:3]
            for s in sample:
                line("", f"    {s['ts_code']:<12} 持仓 {s['qty']} 可用 {s['avail_qty']} "
                         f"成本 {s['cost']} 现价 {s['price']}")
    except Exception as e:
        warn(f"持仓快照读取失败: {e}")

    print("\n[4] 行情 Redis (db13)")
    try:
        from app.services import market
        pos = []
        try:
            from app.repo import pms_repo
            pos = [p["ts_code"] for p in pms_repo.list_positions(only_open=True)][:3]
        except Exception:
            pass
        if not pos:
            try:
                pos = [r["ts_code"] for r in downstream_repo.fetch_positions()["rows"][:3]]
            except Exception:
                pos = []
        probe = pos or ["600000.SH"]
        got = {c: market.get_price(c) for c in probe}
        if any(v for v in got.values()):
            line("OK", f"实时价样本: {got}")
        else:
            warn(f"取不到实时价样本 {got} (非交易时段属正常; 若报 unknown command HELLO, "
                 f"说明 redis-py 版本过新走了 RESP3, 重新 build 镜像即可 —— "
                 f"requirements 已锁 redis<6)")
    except Exception as e:
        warn(f"Redis 行情不可用: {type(e).__name__}: {e}")

    print("\n[5] 运行参数 (表值优先于 settings 初值)")
    try:
        from app.services import param_store
        snap = param_store.snapshot()
        from_table = [p for p in snap["params"] if p["source"] == "table"]
        line("OK", f"可调参数 {len(snap['params'])} 项, 其中页面已改写 {len(from_table)} 项")
        for k in ("PMS_TOTAL_SCALE", "PMS_PORTFOLIO_CAP", "PMS_STOCK_CAP", "PMS_AUTONOMY",
                  "PMS_SECTOR_SOURCE"):
            line("", f"{k:<26} = {param_store.get(k)}")
        if snap.get("source_error"):
            warn(f"参数表读取异常: {snap['source_error']}")
    except Exception as e:
        fail(f"参数中心不可用: {type(e).__name__}: {e}")

    print("\n[6] ws 直连通道 (协议 QMT_WS_PROTOCOL.md V1.0)")
    check_ws_channel()

    print("\n" + "-" * 62)
    if FAILED:
        print(f"FAILED: {len(FAILED)} 项致命问题, {len(WARNED)} 项告警")
        for m in FAILED:
            print("  - " + m)
        sys.exit(1)
    print(f"ALL OK ({len(WARNED)} 项告警)" if WARNED else "ALL OK")


if __name__ == "__main__":
    main()