group_xinghuo_jinrong/scripts/dev/verify_convert_seed.py

"""T-1 验证脚本：用 pymysql 复刻 `scripts/core/reset.ps1` 的 SQL 部分并跑 DoD 断言。

背景：本机 `mysql` 客户端不在 PATH，且 reset.ps1 用 `mysql -p`（交互式密码），
无法在非交互环境直接跑。本脚本用 pymysql 执行**同一组 SQL 文件**（00~09），
再对 T-1 的三条 DoD 判定阈值逐条断言，做到「判定阈值明确、非人工目测」。

用法：
    python scripts/dev/verify_convert_seed.py            # 重建 jinrong_core + 恢复演示态 + 全部断言
    python scripts/dev/verify_convert_seed.py --no-reset # 只跑断言（库已就绪时）

注意 1：会 DROP DATABASE jinrong_core 并重建 —— 该库是 Core 模拟底座，本就设计为可重置
（与 reset.ps1 行为一致）。**不触碰 jinrong_agent**。
注意 2：重建后会自动补跑 `scripts/demo/prepare_risk_demo.sql`（reset 的既定伴随步骤）——
不做的话 `03-seed-customers.sql` 写死的风评有效期会让真库集成模块整体 skip
（实测 516 → 505 passed）。
"""

from __future__ import annotations

import argparse
import re
import sys
from datetime import date
from pathlib import Path

import pymysql

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))

from app.config.settings import settings  # noqa: E402

CORE_DIR = ROOT / "scripts" / "core"
SQL_FILES = [
    "00-create-database.sql",
    "01-ddl.sql",
    "02-seed-base.sql",
    "03-seed-customers.sql",
    "04-seed-holdings.sql",
    "05-seed-trades.sql",
    "06-seed-nav.sql",
    "07-seed-fee-rule.sql",
    "08-seed-share-lot.sql",
    "09-seed-org.sql",
]

# 档位覆盖参照日：与该批种子中 as_of / 净值日一致（2026-09-04）
AS_OF = date(2026, 9, 4)
BANDS = [(0, 7, "0.0150"), (7, 30, "0.0100"), (30, 180, "0.0050"), (180, 365, "0.0025"), (365, None, "0.0000")]


def split_statements(sql: str) -> list[str]:
    """剥离 `--` 整行注释后按分号切分。种子文件均无存储过程/触发器，可安全切分。"""
    body = "\n".join(ln for ln in sql.splitlines() if not ln.strip().startswith("--"))
    return [s.strip() for s in body.split(";") if s.strip()]


def connect(with_db: bool = False) -> pymysql.connections.Connection:
    return pymysql.connect(
        host=settings.mysql_host,
        port=settings.mysql_port,
        user=settings.mysql_user,
        password=settings.mysql_password,
        database=settings.mysql_core_database if with_db else None,
        charset="utf8mb4",
        autocommit=True,
    )


def reset_core() -> None:
    with connect() as conn, conn.cursor() as cur:
        cur.execute(f"DROP DATABASE IF EXISTS `{settings.mysql_core_database}`")
        print(f"  DROP DATABASE {settings.mysql_core_database}")
        for name in SQL_FILES:
            path = CORE_DIR / name
            for stmt in split_statements(path.read_text(encoding="utf-8")):
                cur.execute(stmt)
            print(f"  ✓ {name}")


def run_demo_prepare() -> None:
    """把演示态刷回去（风评有效期）。

    重建 jinrong_core 会让 `03-seed-customers.sql` 写死的风评有效期回到「已过期」状态，
    于是 `tests/conftest.py::ensure_risk_demo_ready` 判定演示数据未就位，
    **整个真库集成模块被 skip**（实测 516 → 505 passed）。这是 reset 的既定伴随步骤
    （演示 SOP §2），reset.ps1 因不含 agent 库步骤而未内置，本脚本补上。
    """
    path = ROOT / "scripts" / "demo" / "prepare_risk_demo.sql"
    with connect(with_db=True) as conn, conn.cursor() as cur:
        for stmt in split_statements(path.read_text(encoding="utf-8")):
            cur.execute(stmt)
    print("  ✓ prepare_risk_demo.sql（演示客户风评刷为「剩余 275 天」）")


def band_of(hold_days: int) -> str:
    for lo, hi, rate in BANDS:
        if hold_days >= lo and (hi is None or hold_days < hi):
            return rate
    return "?"


def run_assertions() -> bool:
    ok = True
    with connect(with_db=True) as conn, conn.cursor() as cur:
        # ---- DoD ① 赎回费 5 档，且每只产品各 5 档 ----
        cur.execute("SELECT COUNT(DISTINCT min_hold_days) FROM core_fee_rule WHERE fee_type='redeem'")
        n_bands = cur.fetchone()[0]
        cur.execute(
            "SELECT product_id, COUNT(*) FROM core_fee_rule WHERE fee_type='redeem' "
            "GROUP BY product_id HAVING COUNT(*) <> 5"
        )
        uneven = cur.fetchall()
        cur.execute("SELECT COUNT(*), COUNT(DISTINCT product_id) FROM core_fee_rule WHERE fee_type='redeem'")
        total_rows, n_products = cur.fetchone()
        p1 = n_bands == 5 and not uneven
        print(f"\n① 赎回费档位：DISTINCT min_hold_days={n_bands}（期望 5）· "
              f"{n_products} 只产品 × {total_rows // max(n_products, 1)} 档 · 非 5 档产品={uneven or '无'}")
        if not p1:
            print("   ✗ 不满足 5 档 / 每产品 5 档")
        ok &= p1

        # ---- DoD ② Σ remain_qty 逐行等于 core_holding.qty ----
        cur.execute(
            """
            SELECT h.customer_id, h.product_id, h.qty, COALESCE(SUM(l.remain_qty), 0) AS lot_sum
            FROM core_holding h
            LEFT JOIN core_share_lot l
              ON l.customer_id = h.customer_id AND l.product_id = h.product_id
            GROUP BY h.customer_id, h.product_id, h.qty
            HAVING ABS(lot_sum - h.qty) > 0.0001
            """
        )
        mismatch = cur.fetchall()
        cur.execute("SELECT COUNT(*) FROM core_holding")
        n_holding = cur.fetchone()[0]
        cur.execute("SELECT COUNT(*), COUNT(DISTINCT CONCAT(customer_id, '|', product_id)) FROM core_share_lot")
        n_lots, n_pairs = cur.fetchone()
        p2 = not mismatch
        print(f"② 批次守恒：{n_holding} 行持仓 vs {n_lots} 行批次（{n_pairs} 个客户-产品对）· "
              f"失配={mismatch or '无'}")
        if not p2:
            print("   ✗ 存在 Σ remain_qty ≠ qty 的持仓行")
        ok &= p2

        # ---- DoD ③ 三列无 NULL 且存在不同费率对 ----
        cur.execute(
            "SELECT COUNT(*) FROM core_product "
            "WHERE fund_company IS NULL OR ta_code IS NULL OR subscribe_fee_rate IS NULL"
        )
        n_null = cur.fetchone()[0]
        cur.execute("SELECT COUNT(DISTINCT subscribe_fee_rate) FROM core_product")
        n_rates = cur.fetchone()[0]
        cur.execute(
            "SELECT product_id, subscribe_fee_rate FROM core_product "
            "WHERE product_id IN ('PROD-110022','PROD-003095','PROD-005827','PROD-510300','PROD-000001') "
            "ORDER BY product_id"
        )
        demo = cur.fetchall()
        p3 = n_null == 0 and n_rates > 1
        print(f"③ 机构/费率：NULL 行数={n_null}（期望 0）· DISTINCT 费率={n_rates}（期望 >1）")
        print(f"   演示费率对：{[(p, str(r)) for p, r in demo]}")
        if not p3:
            print("   ✗ 存在 NULL 或不存在不同费率对")
        ok &= p3

        # ---- 附加：批次实际覆盖的档位（以 AS_OF 为参照） ----
        cur.execute("SELECT lot_id, confirmed_at FROM core_share_lot")
        rows = cur.fetchall()
        seen: dict[str, int] = {}
        for _lot, ts in rows:
            b = band_of((AS_OF - ts.date()).days)
            seen[b] = seen.get(b, 0) + 1
        p4 = len(seen) == 5
        print(f"④ 批次档位覆盖（参照 {AS_OF}）：{dict(sorted(seen.items()))}")
        if not p4:
            print("   ✗ 未覆盖全部 5 档 → 费率档演示无数据")
        ok &= p4

        # ---- 附加：主演示客户 CUST-9527 确为跨批次 ----
        cur.execute(
            "SELECT product_id, COUNT(*) FROM core_share_lot WHERE customer_id='CUST-9527' "
            "GROUP BY product_id ORDER BY product_id"
        )
        c9527 = cur.fetchall()
        p5 = any(n >= 2 for _p, n in c9527)
        print(f"⑤ CUST-9527 批次数：{c9527}（至少一只产品 ≥2 批）")
        if not p5:
            print("   ✗ 主演示客户无跨批次数据")
        ok &= p5

        # ---- DoD ⑥ 向后兼容：不带新列的老 INSERT 仍可执行（架构风险 #9）----
        cur.execute("START TRANSACTION")
        try:
            cur.execute(
                "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount, traded_at) "
                "VALUES ('TRD-BC-TEST', 'CUST-9527', 'PROD-110022', 'subscribe', 100.00, NOW(3))"
            )
            cur.execute("SELECT convert_group_id FROM core_trade WHERE trade_id = 'TRD-BC-TEST'")
            gid = cur.fetchone()[0]
            cur.execute(
                "INSERT INTO core_product (product_id, product_name, product_type, min_risk_code, "
                "min_subscribe_amount, requires_disclosure) VALUES "
                "('PROD-BC-TEST', '向后兼容测试', 'bond', 'R1', 100.00, 0)"
            )
            cur.execute(
                "SELECT can_subscribe, can_redeem, min_hold_action, subscribe_fee_rate, fund_company "
                "FROM core_product WHERE product_id = 'PROD-BC-TEST'"
            )
            row = cur.fetchone()
            p6 = (
                gid is None
                and row[0] == 1 and row[1] == 1
                and row[2] == "force_transfer" and row[3] == 0 and row[4] is None
            )
        finally:
            cur.execute("ROLLBACK")
        print(f"⑥ 老 INSERT 向后兼容：core_trade.convert_group_id={gid}（期望 None）· "
              f"core_product 新列默认值={row}（期望 1/1/force_transfer/0/None）")
        if not p6:
            print("   ✗ 不带新列的老 INSERT 失败或默认值不符")
        ok &= p6

    # ---- DoD ⑦ risk_convert_detail 建表即含 5 个 status 值（免二期 ALTER）----
    doc = (ROOT / "docs" / "项目框架设计" / "表设计" / "02-mysql-agent专用.sql").read_text(encoding="utf-8")
    m = re.search(r"CREATE TABLE risk_convert_detail \(.*?\n\) ENGINE=InnoDB[^;]*;", doc, re.S)
    assert m, "02-mysql-agent专用.sql 中未找到 risk_convert_detail 建表语句"
    tmp_db = "jinrong_tmp_check"
    with connect() as conn, conn.cursor() as cur:
        cur.execute(f"DROP DATABASE IF EXISTS {tmp_db}")
        cur.execute(f"CREATE DATABASE {tmp_db}")
        cur.execute(f"USE {tmp_db}")
        cur.execute(m.group(0))
        cur.execute(
            "SELECT COLUMN_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s "
            "AND TABLE_NAME = 'risk_convert_detail' AND COLUMN_NAME = 'status'",
            (tmp_db,),
        )
        col_type = cur.fetchone()[0]
        cur.execute(f"DROP DATABASE {tmp_db}")
    vals = set(re.findall(r"'(\w+)'", col_type))
    p7 = vals == {"pending", "completed", "failed", "cancelled", "expired"}
    print(f"⑦ risk_convert_detail.status = {col_type}")
    if not p7:
        print(f"   ✗ ENUM 值不全（实际 {vals}）")
    ok &= p7

    # ---- DoD ⑧ 费率档必须与 product_type 匹配（自检第 12 问的机器化）----
    # 22 号文 §8 的费率档由产品类型决定：stock ≤0.80% / mixed ≤0.50% /
    # bond·index ≤0.30% / money = 0。任何「升档凑数值差」都会在此暴露。
    CAPS = {"stock": 0.0080, "mixed": 0.0050, "bond": 0.0030, "index": 0.0030, "money": 0.0}
    # 非公募主体：不适用 22 号文公募费率规定，本项目建模为费率 0 且不与公募互转
    NOT_APPLICABLE = {"wealth_mgmt", "private_fund"}
    with connect(with_db=True) as conn, conn.cursor() as cur:
        cur.execute("SELECT product_id, product_type, subscribe_fee_rate FROM core_product")
        rows = cur.fetchall()
        offenders = [
            (pid, ptype, float(rate))
            for pid, ptype, rate in rows
            if ptype in CAPS and float(rate) > CAPS[ptype] + 1e-9
        ]
        unmatched = [r for r in rows if r[1] not in CAPS and r[1] not in NOT_APPLICABLE]
        na_offenders = [
            (pid, ptype, float(rate))
            for pid, ptype, rate in rows
            if ptype in NOT_APPLICABLE and float(rate) != 0
        ]
        # 主示例两端必须同主体（同管理人 + 同 TA），否则转换前置约束不成立
        cur.execute(
            "SELECT DISTINCT fund_company, ta_code FROM core_product "
            "WHERE product_id IN ('PROD-110022', 'PROD-003095') AND fund_company IS NOT NULL"
        )
        entities = cur.fetchall()
        # 主演示组内费率档必须多样（否则补差费/同费率对照无数据）
        cur.execute(
            "SELECT COUNT(DISTINCT subscribe_fee_rate) FROM core_product "
            "WHERE fund_company = '华夏模拟基金' AND ta_code = 'TA-CN-001'"
        )
        main_group_rates = cur.fetchone()[0]
    p8 = (
        not offenders and not unmatched and not na_offenders
        and len(entities) == 1 and main_group_rates >= 3
    )
    print(f"⑧ 费率档 ↔ product_type 匹配：越档={offenders or '无'} · 未归类={unmatched or '无'} · "
          f"非公募费率非 0={na_offenders or '无'}")
    print(f"   主示例两端主体={entities}（须唯一）· 华夏组费率档数={main_group_rates}（须 ≥3）")
    if not p8:
        print("   ✗ 存在「按更宽松档次取费率」或主示例两端非同主体 / 主组费率档不足")
    ok &= p8

    return ok


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--no-reset", action="store_true", help="跳过重建，只跑断言")
    args = ap.parse_args()

    if not args.no_reset:
        print("== 重建 jinrong_core（等价 reset.ps1 的 SQL 部分）==")
        reset_core()
        print("== 恢复演示态 ==")
        run_demo_prepare()

    print("\n== T-1 DoD 断言 ==")
    ok = run_assertions()
    print("\n" + ("全部 PASS ✅" if ok else "存在 FAIL ❌"))
    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(main())