group_xinghuo_jinrong/scripts/dev/verify_convert_apply.py

"""T-6 真 MySQL 验证脚本：`convert_core_repository.apply_convert` 实跑 + DoD 断言。

**为什么 sqlite 单测全绿还不够**

T-6 的 `core_holding` UPDATE 有两条**只有 MySQL 才成立的语义依赖**，
sqlite 单测绿**证明不了**它们：

1. **SET 从左到右求值**（MySQL 特性；sqlite 恒用更新前的值）。
   本实现靠它让"成本/市值按**原 qty** 等比例结转、最后再减 qty"两库结果一致；
   若 MySQL 语义与预期不符，成本结转会算错（sqlite 侧完全看不出来）。
2. **`(:dq + 0.0)` 语法**（为绕开 sqlite 不转 TEXT 参数的坑）在 MySQL 必须同样合法。

此外真库还能验证 sqlite 测不到的：InnoDB 事务回滚、DECIMAL(18,2/18,4) 精度、
`UNIQUE(customer_id, product_id)` 冲突回退（R-a ②③步）。

用法：
    python scripts/dev/verify_convert_apply.py            # 建隔离数据 → 跑 4 组断言 → 清理

注意：
- 用 **T6M 前缀**的隔离数据（客户/产品/批次），跑完全部 DELETE，不碰既有 61 行种子；
- 建/清数据走 `role="admin"`（R-e：需 DELETE），事务本身走 `ConvertCoreRepository`
  默认的 `role="rw"`（账号未配时回退 `mysql_user`，与 T-0b 一致）。
"""

from __future__ import annotations

import sys
import threading
from datetime import date, datetime, time as dtime, timedelta
from decimal import ROUND_HALF_UP, Decimal
from pathlib import Path

from sqlalchemy import event, text

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

from app.config.settings import settings  # noqa: E402
from app.gateway.convert_core_repository import (  # noqa: E402
    ConvertApplyInput,
    ConvertCoreRepository,
    LotCharge,
)
from app.service.convert.calc import (  # noqa: E402
    convert_amount,
    diff_fee,
    hold_days,
    in_qty,
    lot_amount,
    lot_fee,
    plan_lots,
)
from app.repository.core_ro import CoreReadOnlyRepository  # noqa: E402
from app.service.convert.errors import LotConflict  # noqa: E402
from app.service.convert.trading_calendar import previous_biz_day  # noqa: E402
from app.service.convert.fee import pick_fee_rate  # noqa: E402
from app.service.convert.types import FeeRule, Lot  # noqa: E402
from app.utils.db import dispose_engines, get_engine  # noqa: E402

CUSTOMER = "CUST-T6M"
PROD_OUT = "PROD-T6MO"
PROD_IN = "PROD-T6MI"
# 造数日期在 main() 里按真库交易日历链式回推动态确定（T-12 模式：不猜自然日）。
# 相对关系与原固定基准（2026-09-04）一致 —— hold_days 是自然日语义，档位断言不变：
# TRADE_DATE = 回推 3 个交易日；A1 批次 = −34 自然日（档 [30,180)）；
# A2 批次 = −3 自然日（档 [0,7)）；转入批次 confirmed_at = +1 自然日。
TRADE_AT: "datetime | None" = None
TRADE_DATE: "date | None" = None
_A1_AT: "datetime | None" = None
_A2_AT: "datetime | None" = None
_IN_CONFIRMED_AT: "datetime | None" = None


def _init_dates(admin) -> None:
    """按真库交易日历把造数日期锚定到确定过去的历史开市日（必须在 seed 前调用）。"""
    global TRADE_AT, TRADE_DATE, _A1_AT, _A2_AT, _IN_CONFIRMED_AT
    is_open = CoreReadOnlyRepository(engine=admin).is_open
    d = date.today()
    for _ in range(3):
        d = previous_biz_day(d, is_open)
    TRADE_DATE = d
    TRADE_AT = datetime.combine(d, dtime(10, 0))
    _A1_AT = datetime.combine(d - timedelta(days=34), dtime(10, 0))
    _A2_AT = datetime.combine(d - timedelta(days=3), dtime(10, 0))
    _IN_CONFIRMED_AT = datetime.combine(d + timedelta(days=1), dtime(10, 0))
OUT_RATE = Decimal("0.0030")
IN_RATE = Decimal("0.0080")
IN_NAV = Decimal("0.9500")
FEE_TIERS = [
    (0, 7, "0.0150"), (7, 30, "0.0100"), (30, 180, "0.0050"),
    (180, 365, "0.0025"), (365, None, "0.0000"),
]

_passed = 0
_failed = 0


def check(name: str, actual, expected) -> None:
    """逐条断言并打印（与 `verify_convert_seed.py` / `calc_convert_demo.py` 同款输出）。"""
    global _passed, _failed
    ok = actual == expected
    if ok:
        _passed += 1
    else:
        _failed += 1
    flag = "✅" if ok else "❌"
    print(f"  {flag} {name}: 实际 {actual!r}" + ("" if ok else f" / 期望 {expected!r}"))


def dec(value, places: str = "0.01") -> Decimal:
    """真库读回是 `Decimal`（MySQL DECIMAL）→ 统一量化后比较。"""
    return Decimal(str(value)).quantize(Decimal(places), rounding=ROUND_HALF_UP)


# ── 数据准备 / 清理 ─────────────────────────────────────────────────
def seed(engine) -> None:
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO core_customer (customer_id, display_name, open_date) "
                "VALUES (:c, 'T6真库验证', :d)"
            ),
            {"c": CUSTOMER, "d": TRADE_DATE},
        )
        for pid, name, ptype, rate in [
            (PROD_OUT, "T6转出基金", "bond", OUT_RATE),
            (PROD_IN, "T6转入基金", "stock", IN_RATE),
        ]:
            conn.execute(
                text(
                    "INSERT INTO core_product (product_id, product_name, min_risk_code, "
                    "product_type, can_subscribe, can_redeem, subscribe_fee_rate) "
                    "VALUES (:p, :n, 'R2', :t, 1, 1, :r)"
                ),
                {"p": pid, "n": name, "t": ptype, "r": str(rate)},
            )
        for mh, mh_max, rate in FEE_TIERS:
            conn.execute(
                text(
                    "INSERT INTO core_fee_rule (product_id, fee_type, min_hold_days, "
                    "max_hold_days, rate) VALUES (:p, 'redeem', :mh, :mh_max, :rate)"
                ),
                {"p": PROD_OUT, "mh": mh, "mh_max": mh_max, "rate": rate},
            )
        conn.execute(
            text(
                "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) "
                "VALUES (:p, :nav, 0, :d)"
            ),
            {"p": PROD_IN, "nav": str(IN_NAV), "d": TRADE_DATE},
        )
        for lot_id, qty, nav, confirmed in [
            ("LOT-T6M-A1", "100", "1.0300", _A1_AT),
            ("LOT-T6M-A2", "50", "1.0000", _A2_AT),
        ]:
            conn.execute(
                text(
                    "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty, "
                    "remain_qty, nav, confirmed_at) VALUES (:l, :c, :p, :q, :q, :nav, :cat)"
                ),
                {"l": lot_id, "c": CUSTOMER, "p": PROD_OUT, "q": qty, "nav": nav, "cat": confirmed},
            )
        conn.execute(
            text(
                "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount, "
                "market_value, pnl_pct, as_of) VALUES (:c, :p, 150, 150.00, 154.50, 0, :d)"
            ),
            {"c": CUSTOMER, "p": PROD_OUT, "d": TRADE_DATE},
        )


def cleanup(engine) -> None:
    """倒序删（子表 → 父表），只删 T6M 前缀数据，不碰既有种子。"""
    with engine.begin() as conn:
        for sql, params in [
            ("DELETE FROM core_convert_lot_detail WHERE convert_group_id LIKE 'CNV-T6M%'", {}),
            ("DELETE FROM core_trade WHERE customer_id = :c", {"c": CUSTOMER}),
            ("DELETE FROM core_share_lot WHERE customer_id = :c", {"c": CUSTOMER}),
            ("DELETE FROM core_holding WHERE customer_id = :c", {"c": CUSTOMER}),
            ("DELETE FROM core_fee_rule WHERE product_id LIKE 'PROD-T6M%'", {}),
            ("DELETE FROM core_product_nav WHERE product_id LIKE 'PROD-T6M%'", {}),
            ("DELETE FROM core_product WHERE product_id LIKE 'PROD-T6M%'", {}),
            ("DELETE FROM core_customer WHERE customer_id = :c", {"c": CUSTOMER}),
        ]:
            conn.execute(text(sql), params)


# ── 折算（全部走生产纯函数）────────────────────────────────────────────
def build_input(engine, group_id: str, requested: str, in_lot_id: str = "LOT-T6M-IN") -> ConvertApplyInput:
    with engine.connect() as conn:
        lot_rows = [
            Lot.from_row(r)
            for r in conn.execute(
                text(
                    "SELECT * FROM core_share_lot WHERE customer_id = :c AND product_id = :p "
                    "ORDER BY confirmed_at ASC, lot_id ASC"
                ),
                {"c": CUSTOMER, "p": PROD_OUT},
            ).mappings()
        ]
        rules = [
            FeeRule.from_row(r)
            for r in conn.execute(
                text(
                    "SELECT * FROM core_fee_rule WHERE product_id = :p AND fee_type = 'redeem' "
                    "ORDER BY min_hold_days ASC"
                ),
                {"p": PROD_OUT},
            ).mappings()
        ]

    plan = plan_lots(lot_rows, Decimal(requested))
    charges = []
    for alloc in plan.allocations:
        days = hold_days(TRADE_DATE, alloc.confirmed_at)
        rate = pick_fee_rate(rules, days, product_id=PROD_OUT)
        amount = lot_amount(alloc.qty, alloc.nav)
        charges.append(
            LotCharge(
                lot_id=alloc.lot_id, qty=alloc.qty, hold_days=days, amount=amount,
                fee_rate=rate, fee_amount=lot_fee(amount, rate), nav=alloc.nav,
                nav_date=TRADE_DATE,
            )
        )
    out_amount = sum((c.amount for c in charges), Decimal("0"))
    conv = convert_amount(out_amount, sum((c.fee_amount for c in charges), Decimal("0")))
    in_amount = conv - diff_fee(conv, OUT_RATE, IN_RATE, "amount_diff")
    return ConvertApplyInput(
        convert_group_id=group_id,
        out_trade_id=f"{group_id}-OUT",
        in_trade_id=f"{group_id}-IN",
        customer_id=CUSTOMER,
        from_product_id=PROD_OUT,
        to_product_id=PROD_IN,
        traded_at=TRADE_AT,
        out_qty=plan.actual_qty,
        out_amount=out_amount,
        in_qty=in_qty(in_amount, IN_NAV),
        in_amount=in_amount,
        in_nav=IN_NAV,
        in_nav_date=TRADE_DATE,
        in_lot_id=in_lot_id,
        in_confirmed_at=_IN_CONFIRMED_AT,
        charges=tuple(charges),
    )


def q1(engine, sql: str, **params):
    with engine.connect() as conn:
        return conn.execute(text(sql), params).scalar()


def main() -> int:
    admin = get_engine(settings.mysql_core_database, "admin")
    _init_dates(admin)
    repo = ConvertCoreRepository()  # 真实路径：role="rw"（账号未配时回退 mysql_user）
    try:
        cleanup(admin)
        seed(admin)

        # ── A. 正常路径（含 MySQL 专属语义：SET 顺序 + (:x + 0.0)）────────
        print("\n【A】正常路径：2 流水 + 2 明细 + 两端持仓")
        req = build_input(admin, "CNV-T6M-1", "120")
        repo.apply_convert(req)

        check("流水条数", q1(admin, "SELECT COUNT(*) FROM core_trade WHERE convert_group_id = 'CNV-T6M-1'"), 2)
        check(
            "R-b trade_type",
            sorted(r[0] for r in _rows(admin, "SELECT trade_type FROM core_trade WHERE convert_group_id = 'CNV-T6M-1'")),
            ["redeem", "subscribe"],
        )
        check("LOT-A1 扣满归零", dec(q1(admin, "SELECT remain_qty FROM core_share_lot WHERE lot_id = 'LOT-T6M-A1'")), Decimal("0.00"))
        check("LOT-A2 剩 30", dec(q1(admin, "SELECT remain_qty FROM core_share_lot WHERE lot_id = 'LOT-T6M-A2'")), Decimal("30.00"))
        check("转入新批次份额", dec(q1(admin, "SELECT remain_qty FROM core_share_lot WHERE lot_id = 'LOT-T6M-IN'")), dec(req.in_qty))
        check("明细条数", q1(admin, "SELECT COUNT(*) FROM core_convert_lot_detail WHERE convert_group_id = 'CNV-T6M-1'"), 2)
        check(
            "转出端 qty",
            dec(q1(admin, "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_OUT)),
            Decimal("30.00"),
        )
        # ★ MySQL SET 从左到右语义的验证点：成本按**原 qty** 等比例结转
        check(
            "转出端 cost（等比例结转）",
            dec(q1(admin, "SELECT cost_amount FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_OUT)),
            Decimal("30.00"),
        )
        check(
            "转出端 pnl_pct",
            dec(q1(admin, "SELECT pnl_pct FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_OUT), "0.0001"),
            Decimal("0.0300"),
        )
        check(
            "转入端 qty",
            dec(q1(admin, "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_IN)),
            dec(req.in_qty),
        )
        check(
            "转入端 cost",
            dec(q1(admin, "SELECT cost_amount FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_IN)),
            dec(req.in_amount),
        )

        # ── B. 冲突路径（真实 InnoDB 事务回滚）────────────────────────
        print("\n【B】冲突路径：remain_qty 不足 → LotConflict + 无残留")
        conflict = build_input(admin, "CNV-T6M-2", "30")
        conflict = ConvertApplyInput(
            **{
                **conflict.__dict__,
                "charges": tuple(
                    LotCharge(
                        lot_id=c.lot_id, qty=Decimal("999"), hold_days=c.hold_days,
                        amount=c.amount, fee_rate=c.fee_rate, fee_amount=c.fee_amount,
                        nav=c.nav, nav_date=c.nav_date,
                    )
                    for c in conflict.charges
                ),
            }
        )
        raised = None
        try:
            repo.apply_convert(conflict)
        except LotConflict as exc:
            raised = exc.status_code
        check("抛 LotConflict(409)", raised, 409)
        check("冲突后无流水残留", q1(admin, "SELECT COUNT(*) FROM core_trade WHERE convert_group_id = 'CNV-T6M-2'"), 0)
        check("冲突后无明细残留", q1(admin, "SELECT COUNT(*) FROM core_convert_lot_detail WHERE convert_group_id = 'CNV-T6M-2'"), 0)
        check("批次未被误扣", dec(q1(admin, "SELECT remain_qty FROM core_share_lot WHERE lot_id = 'LOT-T6M-A2'")), Decimal("30.00"))

        # ── C. 转入端再次转换：增量累加（非覆盖）───────────────────────
        print("\n【C】第二笔转换：转出端归零保留行 + 转入端累加")
        before_in = dec(q1(admin, "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_IN))
        second = build_input(admin, "CNV-T6M-3", "30", in_lot_id="LOT-T6M-IN2")
        repo.apply_convert(second)
        check(
            "转出端归零保留行",
            dec(q1(admin, "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_OUT)),
            Decimal("0.00"),
        )
        check(
            "转入端 qty 累加",
            dec(q1(admin, "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_IN)),
            dec(before_in + second.in_qty),
        )
        check(
            "转入端只有一行",
            q1(admin, "SELECT COUNT(*) FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_IN),
            1,
        )

        # ── D. R-a 前提：真库约束是否真的存在（回退分支依赖它们）──────────
        print("\n【D】R-a 依赖的约束在真库是否存在")
        check(
            "core_holding 有 UNIQUE(customer_id, product_id)",
            q1(
                admin,
                # information_schema.statistics 按「索引×列」各一行，
                # 复合唯一索引会有 2 行 → 用 DISTINCT index_name 计数
                "SELECT COUNT(DISTINCT index_name) FROM information_schema.statistics "
                "WHERE table_schema = :db AND table_name = 'core_holding' "
                "AND index_name = 'uk_cust_product' AND non_unique = 0",
                db=settings.mysql_core_database,
            ),
            1,
        )
        check(
            "core_trade 有 idx_convert_group（T-7 重试判定依赖）",
            q1(
                admin,
                "SELECT COUNT(*) FROM information_schema.statistics "
                "WHERE table_schema = :db AND table_name = 'core_trade' "
                "AND index_name = 'idx_convert_group'",
                db=settings.mysql_core_database,
            ),
            1,
        )

        # ── E. 真并发两笔首次转入同一产品（gap lock / 死锁下的终态正确性）──
        print("\n【E】真并发：两笔同时首次转入同一产品 → 终态只有一行、失败笔无残留")
        cleanup(admin)
        seed(admin)
        with admin.begin() as conn:
            # 批次加到 200 份，让两笔都"理论可扣"，把冲突点逼到持仓首次建行
            conn.execute(
                text("UPDATE core_share_lot SET qty = 200, remain_qty = 200 WHERE lot_id = 'LOT-T6M-A1'"),
            )
            conn.execute(
                text("UPDATE core_holding SET qty = 200, cost_amount = 200 WHERE product_id = :p"),
                {"p": PROD_OUT},
            )
        first_a = build_input(admin, "CNV-T6M-5", "100", in_lot_id="LOT-T6M-IN5")
        first_b = build_input(admin, "CNV-T6M-6", "100", in_lot_id="LOT-T6M-IN6")
        # 会话级锁等待超时：真并发撞 gap lock 时快速失败，不等默认 50s
        core_engine = get_engine(settings.mysql_core_database, "rw")

        def _set_lock_timeout(dbapi_conn, _rec):  # noqa: ANN001
            dbapi_conn.cursor().execute("SET SESSION innodb_lock_wait_timeout = 3")

        event.listen(core_engine, "connect", _set_lock_timeout)

        results: dict[str, str | None] = {}

        def _run(gid: str, req: ConvertApplyInput) -> None:
            try:
                ConvertCoreRepository().apply_convert(req)
                results[gid] = None
            except Exception as exc:  # noqa: BLE001
                results[gid] = f"{type(exc).__name__}"

        threads = [
            threading.Thread(target=_run, args=("CNV-T6M-5", first_a)),
            threading.Thread(target=_run, args=("CNV-T6M-6", first_b)),
        ]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        ok_groups = [g for g, err in results.items() if err is None]
        print(f"      并发结果：{ {g: (e or '成功') for g, e in results.items()} }")
        check(
            "转入端只有一行（无论成败）",
            q1(admin, "SELECT COUNT(*) FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_IN),
            1,
        )
        if ok_groups:
            check(
                f"qty = 成功笔数 × 单笔份额（成功 {len(ok_groups)} 笔）",
                dec(q1(admin, "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p", c=CUSTOMER, p=PROD_IN)),
                dec(first_a.in_qty * len(ok_groups)),
            )
        for gid, err in results.items():
            if err is None:
                check(f"{gid} 成功 → 落 2 条流水", q1(admin, "SELECT COUNT(*) FROM core_trade WHERE convert_group_id = :g", g=gid), 2)
            else:
                check(f"{gid} 失败({err}) → 流水无残留", q1(admin, "SELECT COUNT(*) FROM core_trade WHERE convert_group_id = :g", g=gid), 0)
    finally:
        cleanup(admin)
        dispose_engines()

    print(f"\n{'=' * 60}")
    print(f"真库验证：{_passed} 项一致 / {_failed} 项不一致")
    return 1 if _failed else 0


def _rows(engine, sql: str, **params):
    with engine.connect() as conn:
        return conn.execute(text(sql), params).fetchall()


if __name__ == "__main__":
    sys.exit(main())