313 lines
14 KiB
Python
313 lines
14 KiB
Python
|
"""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())
|