group_xinghuo_jinrong/tests/test_convert_confirm.py

"""T-7 确认段（`confirm_service`）单测 · T+1 受理/确认分离模型的第二段。

**核心验证目标**

1. **折算与 PRD §5.3.2 示例逐字节一致** —— 同一个输入（30000+20000 份、T 日净值
   1.3604 / 1.9194、费率 0.30% / 0.80%）必须算出 68020.00 / 612.18 / 67407.82 /
   333.36 / 67074.46 / 34945.54 / -0.0049。这是「实现与文档不漂移」的硬锚点。
2. **T+1 相对 v1.0 的实质变更**：转出端逐批金额按 **T 日净值**（未知价法），
   而不是批次买入时的净值。
3. **确认事务的原子性**：扣批次 + 两条流水 + 转入批次 + 两端持仓 + 明细 +
   **受理单置 confirmed** 全在一个事务里，任何一步失败整体回滚。
4. **幂等**：重复确认（单笔 / 整批）不产生第二组流水。
5. **受理单状态不可逆**：终态单不再被批处理捞到。

sqlite 内存库；真库侧（ENUM 严校验 / DATETIME(3) / 死锁重试）由
`scripts/dev/verify_convert_confirm.py` 承担。
"""

from __future__ import annotations

import json
from datetime import date, datetime, timedelta
from decimal import Decimal

import pytest
from sqlalchemy import text

from app.gateway.convert_core_repository import ConvertCoreRepository
from app.repository.convert_repository import ConvertRepository
from app.repository.convert_request_repository import ConvertRequestRepository
from app.repository.core_ro import CoreReadOnlyRepository
from app.repository.risk_repository import RiskRepository
from app.service.convert.confirm_service import confirm_batch, confirm_one
from app.service.convert.convert_service import accept_convert, cancel_convert
from app.service.convert.errors import (
    CancelNotAllowed,
    ConfirmConflict,
    ConvertRequestNotFound,
)
from app.service.risk.rules import RiskThresholds

CUST = "CUST-CC"  # C3 客户 → 转入 R4 = 放行
CUST_LOW = "CUST-CCL"  # C1 客户 → 转入 R4 = 阻断（T+1 复核用）
PROD_OUT = "PROD-CC-OUT"
PROD_IN = "PROD-CC-IN"
COMPANY = "华夏模拟基金"
TA = "TA-CN-001"

T_DAY = date(2026, 9, 4)  # 受理日 T（周五）
T1_DAY = date(2026, 9, 7)  # 确认业务日 T+1（下周一）
SUBMIT_AT = datetime(2026, 9, 4, 10, 0)
CONFIRM_AT = datetime(2026, 9, 7, 9, 0)

# PRD §5.3.2 的输入（真实净值：PROD-110022 / PROD-003095 在 T=2026-09-09 的值）
OUT_NAV = Decimal("1.3604")
IN_NAV = Decimal("1.9194")
OUT_RATE = Decimal("0.0030")
IN_RATE = Decimal("0.0080")
FEE_TIERS = [
    (0, 7, "0.0150"), (7, 30, "0.0100"), (30, 180, "0.0050"),
    (180, 365, "0.0025"), (365, None, "0.0000"),
]


# ── 种子 ────────────────────────────────────────────────────────────
def _exec(engine, sql: str, **params) -> None:
    with engine.begin() as conn:
        conn.execute(text(sql), params)


def _rows(engine, sql: str, **params) -> list[dict]:
    with engine.connect() as conn:
        return [dict(r) for r in conn.execute(text(sql), params).mappings()]


def _seed_calendar(engine, start: date = date(2026, 9, 1), days: int = 60) -> None:
    cursor = start
    for _ in range(days):
        if cursor.weekday() < 5:  # 周一~周五开市；周末不插行 = 休市
            _exec(
                engine,
                "INSERT INTO core_trade_calendar (cal_date, is_open, remark) VALUES (:d, 1, 't')",
                d=cursor,
            )
        cursor += timedelta(days=1)


def _seed(engine, *, with_nav: bool = True, min_hold: str = "0") -> None:
    _exec(
        engine,
        "INSERT INTO core_customer (customer_id, display_name, age, is_active) "
        "VALUES (:c, 'CC客户', 40, 1)",
        c=CUST,
    )
    _exec(
        engine,
        "INSERT INTO core_customer (customer_id, display_name, age, is_active) "
        "VALUES (:c, 'CC低风险', 40, 1)",
        c=CUST_LOW,
    )
    for cid, code in ((CUST, "C3"), (CUST_LOW, "C1")):
        _exec(
            engine,
            "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at, expires_at) "
            "VALUES (:c, :r, :t, :e)",
            c=cid, r=code,
            t=SUBMIT_AT - timedelta(days=30), e=SUBMIT_AT + timedelta(days=300),
        )
    _exec(
        engine,
        "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type, "
        "can_subscribe, can_redeem, subscribe_fee_rate, min_redeem_qty, min_hold_qty, "
        "fund_company, ta_code) "
        "VALUES (:p, '转出债基', 'R2', 'bond', 1, 1, :r, 0, :mh, :co, :ta)",
        p=PROD_OUT, r=float(OUT_RATE), mh=float(min_hold), co=COMPANY, ta=TA,
    )
    _exec(
        engine,
        "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type, "
        "can_subscribe, can_redeem, subscribe_fee_rate, fund_company, ta_code) "
        "VALUES (:p, '转入股基', 'R4', 'stock', 1, 1, :r, :co, :ta)",
        p=PROD_IN, r=float(IN_RATE), co=COMPANY, ta=TA,
    )
    for mh, mh_max, rate in FEE_TIERS:
        _exec(
            engine,
            "INSERT INTO core_fee_rule (product_id, fee_type, min_hold_days, max_hold_days, rate) "
            "VALUES (:p, 'redeem', :a, :b, :r)",
            p=PROD_OUT, a=mh, b=mh_max, r=float(rate),
        )
    if with_nav:
        _exec(
            engine,
            "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) "
            "VALUES (:p, :n, 0, :d)",
            p=PROD_OUT, n=float(OUT_NAV), d=T_DAY,
        )
        _exec(
            engine,
            "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) "
            "VALUES (:p, :n, 0, :d)",
            p=PROD_IN, n=float(IN_NAV), d=T_DAY,
        )
    for cid, tag in ((CUST, "CC"), (CUST_LOW, "CCL")):
        # 批次 1：2026-08-01 确认 → 到 T 日持有 34 天 → 档位 (30,180) = 0.50%
        _exec(
            engine,
            "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty, remain_qty, "
            "nav, confirmed_at) VALUES (:l, :c, :p, 30000, 30000, 1.0300, :cat)",
            l=f"LOT-{tag}-1", c=cid, p=PROD_OUT, cat=datetime(2026, 8, 1, 10, 0, 0),
        )
        # 批次 2：2026-09-01 确认 → 到 T 日持有 3 天 → 档位 (0,7) = 1.50%
        _exec(
            engine,
            "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty, remain_qty, "
            "nav, confirmed_at) VALUES (:l, :c, :p, 20000, 20000, 1.0000, :cat)",
            l=f"LOT-{tag}-2", c=cid, p=PROD_OUT, cat=datetime(2026, 9, 1, 10, 0, 0),
        )
        _exec(
            engine,
            "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount, "
            "market_value, pnl_pct, as_of) VALUES (:c, :p, 50000, 50000, 51000, 0, :d)",
            c=cid, p=PROD_OUT, d=T_DAY,
        )
    _seed_calendar(engine)


def _services(engine) -> dict:
    """受理段 / 撤单段的仓储集合（不含写库网关 —— 这两段只读 + 落受理单）。"""
    return dict(
        core_ro=CoreReadOnlyRepository(engine=engine),
        risk_repo=RiskRepository(engine=engine),
        convert_repo=ConvertRepository(engine=engine),
        request_repo=ConvertRequestRepository(engine=engine),
    )


def _confirm_services(engine) -> dict:
    """确认段独有：多一个 `core_writer`（写库网关，确认事务的载体）。"""
    return dict(**_services(engine), core_writer=ConvertCoreRepository(engine=engine))


def _accept(engine, *, qty="50000", customer: str = CUST, cid=None) -> dict:
    req = {
        "customer_id": customer,
        "from_product_id": PROD_OUT,
        "to_product_id": PROD_IN,
        "qty": Decimal(qty),
    }
    if cid:
        req["client_request_id"] = cid
    return accept_convert(req, now=SUBMIT_AT, **_services(engine))


def _confirm(engine, gid: str, **kw) -> dict:
    return confirm_one(gid, now=CONFIRM_AT, as_of=T1_DAY, **_confirm_services(engine), **kw)


def _trades(engine, gid: str) -> list[dict]:
    return _rows(
        engine,
        "SELECT trade_id, trade_type, amount, qty FROM core_trade "
        "WHERE convert_group_id = :g ORDER BY trade_type",
        g=gid,
    )


def _req_row(engine, gid: str) -> dict:
    return _rows(
        engine, "SELECT * FROM core_convert_request WHERE convert_group_id = :g", g=gid
    )[0]


def _qty(engine, lot_id: str) -> Decimal:
    r = _rows(engine, "SELECT remain_qty FROM core_share_lot WHERE lot_id = :l", l=lot_id)[0]
    return Decimal(str(r["remain_qty"]))


# ── 1. 全流程 + PRD §5.3.2 逐字节对齐 ───────────────────────────────
def test_confirm_full_flow_matches_prd_example(sqlite_engine):
    """受理 50000 → 确认：折算值与 PRD §5.3.2 示例逐字节一致，且写全 6 类数据。"""
    _seed(sqlite_engine)
    accepted = _accept(sqlite_engine, cid="CC-REQ-1")
    gid = accepted["convert_group_id"]

    res = _confirm(sqlite_engine, gid)

    assert res["status"] == "confirmed"
    assert res["partial"] is False
    assert res["forced_full_transfer"] is False
    # ── 折算（PRD §5.3.2 示例数字，逐字节）──
    assert res["out_nav"] == "1.3604"
    assert res["nav_date"] == str(T_DAY)
    assert res["out_amount"] == "68020.00"
    assert res["redeem_fee"] == "612.18"
    assert res["convert_amount"] == "67407.82"
    assert res["diff_fee"] == "333.36"
    assert res["in_amount"] == "67074.46"
    assert res["in_qty"] == "34945.54"
    assert res["rounding_diff"] == "-0.0049"
    assert res["lot_count"] == 2
    # 逐批：批次 1 持 34 天 → 0.50%；批次 2 持 3 天 → 1.50%
    by_lot = {b["lot_id"]: b for b in res["lot_breakdown"]}
    assert by_lot["LOT-CC-1"]["hold_days"] == 34
    assert by_lot["LOT-CC-1"]["fee_rate"] == "0.0050"
    assert by_lot["LOT-CC-1"]["fee_amount"] == "204.06"
    assert by_lot["LOT-CC-2"]["hold_days"] == 3
    assert by_lot["LOT-CC-2"]["fee_rate"] == "0.0150"
    assert by_lot["LOT-CC-2"]["fee_amount"] == "408.12"

    # ── 两条流水（R-b：redeem / subscribe，同 gid）──
    trades = _trades(sqlite_engine, gid)
    assert [t["trade_type"] for t in trades] == ["redeem", "subscribe"]
    assert Decimal(str(trades[0]["amount"])) == Decimal("68020")
    assert Decimal(str(trades[0]["qty"])) == Decimal("50000")
    assert Decimal(str(trades[1]["amount"])) == Decimal("67074.46")
    assert Decimal(str(trades[1]["qty"])) == Decimal("34945.54")

    # ── 批次扣减 + 转入新批次 ──
    assert _qty(sqlite_engine, "LOT-CC-1") == Decimal("0")
    assert _qty(sqlite_engine, "LOT-CC-2") == Decimal("0")
    assert _qty(sqlite_engine, f"LOT-{gid}-IN") == Decimal("34945.54")

    # ── 两端持仓 ──
    out_h = _rows(
        sqlite_engine,
        "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p",
        c=CUST, p=PROD_OUT,
    )[0]
    in_h = _rows(
        sqlite_engine,
        "SELECT qty FROM core_holding WHERE customer_id = :c AND product_id = :p",
        c=CUST, p=PROD_IN,
    )
    assert Decimal(str(out_h["qty"])) == Decimal("0")  # 全部转出
    assert len(in_h) == 1 and Decimal(str(in_h[0]["qty"])) == Decimal("34945.54")

    # ── 受理单终态 + 回填 ──
    row = _req_row(sqlite_engine, gid)
    assert row["status"] == "confirmed"
    assert Decimal(str(row["actual_qty"])) == Decimal("50000")
    assert Decimal(str(row["coupon"])) == Decimal("333.36")
    assert row["confirmed_at"] is not None

    # ── 计费明细（补偿数据源）──
    assert len(_rows(
        sqlite_engine,
        "SELECT 1 FROM core_convert_lot_detail WHERE convert_group_id = :g", g=gid,
    )) == 2

    # ── agent 镜像 completed（含详情）──
    mirror = _rows(
        sqlite_engine,
        "SELECT status, out_trade_id, fee_amount, hold_days_min, hold_days_max "
        "FROM risk_convert_detail WHERE convert_group_id = :g",
        g=gid,
    )[0]
    assert mirror["status"] == "completed"
    assert mirror["out_trade_id"] == res["out_trade_id"]
    assert Decimal(str(mirror["fee_amount"])) == Decimal("612.18")
    assert (int(mirror["hold_days_min"]), int(mirror["hold_days_max"])) == (3, 34)

    audits = [
        r["decision"]
        for r in _rows(
            sqlite_engine,
            "SELECT decision FROM audit_log WHERE event_type = 'convert_request' "
            "AND input_summary LIKE :p",
            p=f"%{gid}%",
        )
    ]
    assert "confirmed" in audits


# ── 1.5 D27 产品级舍入（T-14）：转入份额位数按 core_share_rule 回退链 ──
# 4 位期望锚点：67074.46 ÷ 1.9194 = 34945.53506… → 4 位 34945.5351（2 位 34945.54）
_IN_QTY_4 = Decimal("34945.5351")


def _in_trade_qty(engine, gid: str) -> Decimal:
    return Decimal(str(
        [t for t in _trades(engine, gid) if t["trade_type"] == "subscribe"][0]["qty"]
    ))


def test_confirm_in_qty_four_places_via_share_rule(sqlite_engine):
    """T-14 DoD1（非降级）：转入产品 `convert` 规则行 4 位 → in_qty 量化到 4 位。

    同产品插一条 `subscribe` 型 2 位干扰行 —— 确认段必须取 `convert` 型
    （business_type 区分度；错取干扰行会得 34945.54 而红）。
    ⚠️ 4 位是 **D27 机制验证载体**（位数按产品合同配置，南方 2012 调整公告
    为行业依据），非真实业务常态 —— 联网核实（2026-09-12）：公募场外申购/
    转换份额主流 2 位，产品间真实差异在舍入方式（rounding_mode）而非位数；
    真库种子 11-seed 全 2 位（v1.0 的「指数基金 4 位」依据已证伪并纠偏）。
    """
    _seed(sqlite_engine)
    _exec(
        sqlite_engine,
        "INSERT INTO core_share_rule (product_id, business_type, share_digits, rounding_mode) "
        "VALUES (:p, 'convert', 4, 'half_up')",
        p=PROD_IN,
    )
    _exec(
        sqlite_engine,
        "INSERT INTO core_share_rule (product_id, business_type, share_digits, rounding_mode) "
        "VALUES (:p, 'subscribe', 2, 'half_up')",
        p=PROD_IN,
    )
    gid = _accept(sqlite_engine, cid="CC-REQ-D27A")["convert_group_id"]

    res = _confirm(sqlite_engine, gid)

    assert res["status"] == "confirmed"
    assert _in_trade_qty(sqlite_engine, gid) == _IN_QTY_4
    # 转入批次同值（库内 DECIMAL(18,4) 存 4 位量化结果）
    assert _qty(sqlite_engine, f"LOT-{gid}-IN") == _IN_QTY_4


def test_confirm_in_qty_falls_back_to_product_share_digits(sqlite_engine):
    """T-14 回退链第二层：无规则行 → `core_product.share_digits`（显式置 4）。"""
    _seed(sqlite_engine)
    _exec(
        sqlite_engine,
        "UPDATE core_product SET share_digits = 4 WHERE product_id = :p",
        p=PROD_IN,
    )
    gid = _accept(sqlite_engine, cid="CC-REQ-D27B")["convert_group_id"]

    res = _confirm(sqlite_engine, gid)

    assert res["status"] == "confirmed"
    assert _in_trade_qty(sqlite_engine, gid) == _IN_QTY_4


def test_confirm_in_qty_missing_rule_falls_back_default_two_places(sqlite_engine):
    """T-14 DoD2：缺规则产品回退 2 位（显式降级用例，非唯一覆盖 ——
    PRD §5.3.2 示例用例同样在无规则行下断言 2 位）。

    连 `core_product.share_digits` 也置 NULL → 回退链第三层默认 `PLACES=2`。
    """
    _seed(sqlite_engine)
    _exec(
        sqlite_engine,
        "UPDATE core_product SET share_digits = NULL WHERE product_id = :p",
        p=PROD_IN,
    )
    gid = _accept(sqlite_engine, cid="CC-REQ-D27C")["convert_group_id"]

    res = _confirm(sqlite_engine, gid)

    assert res["status"] == "confirmed"
    # 与不接 D27 时代的 PRD §5.3.2 示例逐字节一致（DoD3 链路级锚点）
    assert _in_trade_qty(sqlite_engine, gid) == Decimal("34945.54")


def test_confirm_in_qty_truncate_mode_from_product(sqlite_engine):
    """T-14 开口②闭环（2026-09-12 用户拍板补接）：产品层 `rounding_mode='truncate'`
    （无规则行）→ 确认段转入份额走**舍去法**。

    基准值 34945.5351 天然分化：half_up = 34945.54（PRD §5.3.2 锚点），
    truncate = 34945.53 —— 错取 half_up 即红。
    舍去型真实出处：南方利众/宝元、邮储、华宝、申万菱信、东方基金等
    （联网核实 2026-09-12，PRD §2.5.3）；真库载体 = 09-seed-org.sql ⑦ 段
    PROD-005828（规则行未覆盖 → 恰好走 core_product 层回退）。
    """
    _seed(sqlite_engine)
    _exec(
        sqlite_engine,
        "UPDATE core_product SET rounding_mode = 'truncate' WHERE product_id = :p",
        p=PROD_IN,
    )
    gid = _accept(sqlite_engine, cid="CC-REQ-D27T")["convert_group_id"]

    res = _confirm(sqlite_engine, gid)

    assert res["status"] == "confirmed"
    assert _in_trade_qty(sqlite_engine, gid) == Decimal("34945.53")
    # 转入批次同值（库内 DECIMAL(18,4) 存量化结果）
    assert _qty(sqlite_engine, f"LOT-{gid}-IN") == Decimal("34945.53")


def test_get_share_rounding_fallback_chain_digits_and_mode(sqlite_engine):
    """回退链单点扩展（开口②）：`get_share_rounding` 位数与模式**逐层独立回退**。

    ① 全缺 → (2, 'half_up')；② product 层配 mode（无规则行）→ (2, 'truncate')；
    ③ 规则行存在 → digits+mode 同取规则行（规则行两列 NOT NULL，恒整体优先）；
    ④ product 层 mode 非法 → 读取即 ValueError（错配不留到算份额）。
    """
    _seed(sqlite_engine)
    ro = CoreReadOnlyRepository(engine=sqlite_engine)

    assert ro.get_share_rounding(PROD_IN, "convert") == (2, "half_up")

    _exec(
        sqlite_engine,
        "UPDATE core_product SET rounding_mode = 'truncate' WHERE product_id = :p",
        p=PROD_IN,
    )
    assert ro.get_share_rounding(PROD_IN, "convert") == (2, "truncate")

    _exec(
        sqlite_engine,
        "INSERT INTO core_share_rule (product_id, business_type, share_digits, rounding_mode) "
        "VALUES (:p, 'convert', 4, 'half_up')",
        p=PROD_IN,
    )
    assert ro.get_share_rounding(PROD_IN, "convert") == (4, "half_up")

    # ④ product 层 mode 非法 → 读取即 ValueError（先删规则行：规则行 mode
    # 恒整体优先，其存在会遮蔽 product 层错配 —— 本断言同时钉住遮蔽语义）
    _exec(
        sqlite_engine,
        "DELETE FROM core_share_rule WHERE product_id = :p AND business_type = 'convert'",
        p=PROD_IN,
    )
    _exec(
        sqlite_engine,
        "UPDATE core_product SET rounding_mode = 'bogus' WHERE product_id = :p",
        p=PROD_IN,
    )
    with pytest.raises(ValueError, match="舍入模式非法"):
        ro.get_share_rounding(PROD_IN, "convert")


# ── 2. 幂等 ─────────────────────────────────────────────────────────
def test_confirm_twice_is_idempotent(sqlite_engine):
    """重复确认：第二次 `skipped`，流水不重复（验收 24 后半 / 15）。"""
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]

    first = _confirm(sqlite_engine, gid)
    assert first["status"] == "confirmed"

    second = _confirm(sqlite_engine, gid)
    assert second["status"] == "skipped"
    assert second["current_status"] == "confirmed"
    assert len(_trades(sqlite_engine, gid)) == 2  # 没有第二组流水
    assert _qty(sqlite_engine, f"LOT-{gid}-IN") == Decimal("34945.54")


def test_confirm_batch_second_run_processes_nothing(sqlite_engine):
    """同 as_of 再跑批处理 → 0 处理（捞单 SQL 只取 accepted/nav_pending）。"""
    _seed(sqlite_engine)
    _accept(sqlite_engine, qty="20000")
    _accept(sqlite_engine, qty="30000")

    first = confirm_batch(T1_DAY, now=CONFIRM_AT, **_confirm_services(sqlite_engine))
    assert first["locked"] is True
    assert first["scanned"] == 2 and first["confirmed"] == 2

    second = confirm_batch(T1_DAY, now=CONFIRM_AT, **_confirm_services(sqlite_engine))
    assert second["scanned"] == 0 and second["confirmed"] == 0


# ── 3. 缺 T 日净值 → nav_pending → 补净值后确认 ─────────────────────
def test_confirm_missing_nav_then_retry(sqlite_engine):
    """缺 T 日净值：受理单挂 `nav_pending`、**无流水、份额不变**；补净值后重跑即确认。"""
    _seed(sqlite_engine, with_nav=False)
    gid = _accept(sqlite_engine)["convert_group_id"]

    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "nav_pending"
    assert set(res["missing_nav_products"]) == {PROD_OUT, PROD_IN}
    assert _req_row(sqlite_engine, gid)["status"] == "nav_pending"
    assert _trades(sqlite_engine, gid) == []
    assert _qty(sqlite_engine, "LOT-CC-1") == Decimal("30000")  # 份额一分未动
    # 占用仍保持（nav_pending 属在途态），超量申请照样被拒
    from app.service.convert.errors import InsufficientShares

    with pytest.raises(InsufficientShares):
        _accept(sqlite_engine, qty="30000")  # 可用仅 0（50000 全部在途）

    # 补 T 日净值 → 重跑（同一单，状态 nav_pending → confirmed）
    _exec(
        sqlite_engine,
        "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) "
        "VALUES (:p, :n, 0, :d)",
        p=PROD_OUT, n=float(OUT_NAV), d=T_DAY,
    )
    _exec(
        sqlite_engine,
        "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) "
        "VALUES (:p, :n, 0, :d)",
        p=PROD_IN, n=float(IN_NAV), d=T_DAY,
    )
    again = _confirm(sqlite_engine, gid)
    assert again["status"] == "confirmed"
    assert again["out_amount"] == "68020.00"


def test_confirm_nav_pending_retry_does_not_false_conflict(sqlite_engine):
    """已是 `nav_pending` 的单重跑**不得**被判成并发冲突。

    真库陷阱：MySQL 的 `rowcount` 是 **changed rows**，
    `SET status='nav_pending' WHERE status='nav_pending'` 返回 0（sqlite 返回 1）
    —— 若无条件地复用 `transition_status`，真库上会把「正常的重试」误判成冲突。
    """
    _seed(sqlite_engine, with_nav=False)
    gid = _accept(sqlite_engine)["convert_group_id"]
    _confirm(sqlite_engine, gid)

    res = _confirm(sqlite_engine, gid)  # 第二次仍缺净值
    assert res["status"] == "nav_pending"
    assert res["transitioned"] is False  # 状态没变，如实回 False（而非报冲突）


# ── 4. T+1 适当性复核（D25 · 验收 26）───────────────────────────────
def test_confirm_suitability_recheck_rejects_and_releases(sqlite_engine):
    """受理通过、T+1 复核不通过 → `rejected` + 占用释放、份额不变、无流水。"""
    _seed(sqlite_engine)
    # 该客户在受理时**合规**（C3 且风评在有效期内）→ 受理通过；
    # 随后风评到期并降级为 C1（转入 R4 股基即不匹配），模拟 D25 的 T+1 复核场景。
    _exec(
        sqlite_engine,
        "UPDATE core_customer_risk SET risk_code = 'C3', evaluated_at = :t, expires_at = :e "
        "WHERE customer_id = :c",
        c=CUST_LOW, t=SUBMIT_AT - timedelta(days=30), e=SUBMIT_AT + timedelta(days=300),
    )
    gid = _accept(sqlite_engine, customer=CUST_LOW)["convert_group_id"]
    assert _req_row(sqlite_engine, gid)["status"] == "accepted"

    # T+1 受理日之前：风评到期 + 降级 → 复核不通过
    _exec(
        sqlite_engine,
        "UPDATE core_customer_risk SET risk_code = 'C1', expires_at = :e "
        "WHERE customer_id = :c",
        c=CUST_LOW, e=SUBMIT_AT - timedelta(days=1),
    )
    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "rejected"
    assert _req_row(sqlite_engine, gid)["status"] == "rejected"
    assert _trades(sqlite_engine, gid) == []
    assert _qty(sqlite_engine, "LOT-CCL-1") == Decimal("30000")
    # 占用释放 → 同客户可再次受理全量
    assert CoreReadOnlyRepository(engine=sqlite_engine).sum_inflight_qty(
        CUST_LOW, PROD_OUT
    ) == Decimal("0")


def test_confirm_rejects_when_product_disabled(sqlite_engine):
    """确认段复核产品可申赎状态（受理后产品被下架 → rejected）。"""
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]
    _exec(sqlite_engine, "UPDATE core_product SET can_redeem = 0 WHERE product_id = :p", p=PROD_OUT)

    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "rejected"
    assert res["reject_reason"] == "PRODUCT_NOT_REDEEMABLE"
    assert "PRODUCT_NOT_REDEEMABLE" in (_req_row(sqlite_engine, gid)["remark"] or "")


# ── 5. 部分成交（R-10 · 验收 30）────────────────────────────────────
def test_confirm_partial_fill_records_actual_and_releases(sqlite_engine):
    """T 日→T+1 之间份额被别处用掉 → 部分成交：`actual_qty < qty` + `remark='partial'`。"""
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]  # 申请 50000

    # 模拟：批次 2 的份额在确认前被普通赎回用掉 10000
    _exec(
        sqlite_engine,
        "UPDATE core_share_lot SET remain_qty = 10000 WHERE lot_id = 'LOT-CC-2'",
    )

    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "confirmed"
    assert res["partial"] is True
    assert res["requested_qty"] == "50000.00"
    assert res["actual_qty"] == "40000.00"
    row = _req_row(sqlite_engine, gid)
    assert row["remark"] == "partial"
    # 未确认部分（10000 份）占用随终态释放
    assert CoreReadOnlyRepository(engine=sqlite_engine).sum_inflight_qty(
        CUST, PROD_OUT
    ) == Decimal("0")


def test_confirm_rejects_when_no_share_left(sqlite_engine):
    """可用份额全被抢光 → `rejected`（无可确认份额，不清零占用以外的任何东西）。"""
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]
    _exec(sqlite_engine, "UPDATE core_share_lot SET remain_qty = 0 WHERE customer_id = :c", c=CUST)

    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "rejected"
    assert res["reject_reason"] == "NO_AVAILABLE_QTY"


# ── 6. 确认事务原子性（状态被别人改掉 → 整体回滚）────────────────────
def test_confirm_conflict_rolls_back_whole_transaction(sqlite_engine, monkeypatch):
    """读单（accepted）→ 事务内置位之间状态被改 → `ConfirmConflict` 且**全部回滚**。

    这是「状态与份额必须同生同死」的核心守护：若状态置位不在事务内，
    会出现「份额已扣、受理单仍 accepted」→ 下轮批处理二次扣减（超扣）。
    """
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]

    real_apply = ConvertCoreRepository.apply_convert

    def racing_apply(self, req):
        # 模拟并发：另一路先把受理单撤了
        _exec(
            sqlite_engine,
            "UPDATE core_convert_request SET status = 'cancelled' WHERE convert_group_id = :g",
            g=gid,
        )
        return real_apply(self, req)

    monkeypatch.setattr(ConvertCoreRepository, "apply_convert", racing_apply)

    with pytest.raises(ConfirmConflict):
        _confirm(sqlite_engine, gid)

    # 事务整体回滚：无流水、份额未动、无转入批次
    assert _trades(sqlite_engine, gid) == []
    assert _qty(sqlite_engine, "LOT-CC-1") == Decimal("30000")
    assert _qty(sqlite_engine, "LOT-CC-2") == Decimal("20000")
    assert _rows(
        sqlite_engine, "SELECT 1 FROM core_share_lot WHERE lot_id = :l", l=f"LOT-{gid}-IN"
    ) == []


# ── 7. 批处理：串行 + 锁 + 汇总 ─────────────────────────────────────
def test_confirm_batch_summary_and_serial_order(sqlite_engine):
    """批处理按受理先后串行处理，汇总计数正确（FR-C23）。"""
    _seed(sqlite_engine)
    a = _accept(sqlite_engine, qty="20000")["convert_group_id"]  # 先受理
    b = _accept(sqlite_engine, qty="30000")["convert_group_id"]

    res = confirm_batch(T1_DAY, now=CONFIRM_AT, **_confirm_services(sqlite_engine))

    assert res["scanned"] == 2
    assert res["confirmed"] == 2
    assert [r["convert_group_id"] for r in res["results"]] == [a, b]  # requested_at 升序
    for gid in (a, b):
        assert _req_row(sqlite_engine, gid)["status"] == "confirmed"


def test_confirm_batch_skips_when_lock_not_acquired(sqlite_engine, monkeypatch):
    """未抢到批处理锁 → 整轮让路（R-1 防双跑），不处理任何单。"""
    _seed(sqlite_engine)
    _accept(sqlite_engine)

    import app.service.convert.confirm_service as cs

    monkeypatch.setattr(cs, "run_locked", lambda key, fn: fn(False))
    res = confirm_batch(T1_DAY, now=CONFIRM_AT, **_confirm_services(sqlite_engine))

    assert res["locked"] is False
    assert res["scanned"] == 0
    assert _rows(sqlite_engine, "SELECT 1 FROM core_trade") == []  # 一单未处理


def test_confirm_batch_survives_single_failure(sqlite_engine, monkeypatch):
    """单笔异常不中断整批（其余单继续确认）。"""
    _seed(sqlite_engine)
    a = _accept(sqlite_engine, qty="20000")["convert_group_id"]
    b = _accept(sqlite_engine, qty="30000")["convert_group_id"]

    import app.service.convert.confirm_service as cs

    real_one = cs.confirm_one

    def flaky(gid, **kw):
        if gid == a:
            raise RuntimeError("模拟单笔故障")
        return real_one(gid, **kw)

    monkeypatch.setattr(cs, "confirm_one", flaky)
    res = confirm_batch(T1_DAY, now=CONFIRM_AT, **_confirm_services(sqlite_engine))

    assert res["failed"] == 1
    assert res["confirmed"] == 1
    assert _req_row(sqlite_engine, b)["status"] == "confirmed"
    assert _req_row(sqlite_engine, a)["status"] == "accepted"  # 未被破坏，下轮可重试


# ── 8. 引擎恰好一次（FR-C28 · 验收 5/7/29）──────────────────────────
def test_engine_runs_exactly_once_on_confirm(sqlite_engine):
    """引擎只在确认段跑一次；重复确认不再跑（受理段不跑）。"""
    _seed(sqlite_engine)
    calls: list[str] = []

    def hook(out_trade, in_trade):
        calls.append(out_trade["convert_group_id"])
        return {"triggered_rules": ["RISK-002"], "alert_ids": [], "aml_hit": False}

    gid = _accept(sqlite_engine)["convert_group_id"]
    assert calls == []  # 受理段不跑引擎

    first = _confirm(sqlite_engine, gid, engine_hook=hook)
    assert calls == [gid]
    assert first["triggered_rules"] == ["RISK-002"]

    _confirm(sqlite_engine, gid, engine_hook=hook)  # 第二次 skipped
    assert calls == [gid]  # 仍然只跑过一次


def test_engine_exception_does_not_block_confirmation(sqlite_engine):
    """引擎异常不阻断已成立的交易（D17）——但必须留痕 + 标记（T-12 三链路对齐）。

    v1.0 阶段 1.5 与 trade_gateway 的引擎异常都落 `decision='engine_error'` 审计
    且响应 `engine_error=True`；T+1 确认段（主链路）此前只 log 不留痕，导致补偿侧
    `has_engine_error_audit` 的「人工核对」保护对主链路失效（T-12 真库脚本暴露）。
    """
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]

    def boom(out_trade, in_trade):
        raise RuntimeError("引擎炸了")

    res = _confirm(sqlite_engine, gid, engine_hook=boom)

    assert res["status"] == "confirmed"  # 不阻断（D17）
    assert len(_trades(sqlite_engine, gid)) == 2
    assert res["engine_error"] is True  # 异常必须标记（不再伪装成正常）
    audit = _rows(
        sqlite_engine,
        "SELECT decision, input_summary FROM audit_log WHERE input_summary LIKE :p",
        p=f"%{gid}%",
    )
    assert [a["decision"] for a in audit].count("engine_error") == 1, audit


def test_real_engine_on_confirm_dedupes_daily_total(sqlite_engine):
    """确认链路接**真引擎**（不注入 hook）：一张单、两条事件、RISK-002 只计一次。

    验收 5/6/7 的端到端版本 —— 本文件其余用例用 `engine_hook` 注入计数（验证时机），
    这条走真实 `process_convert_event`，证明确认段确实在事务提交后把两条流水
    投给了引擎，且金额聚合去重生效。

    折算值：转出端 68020.00 / 转入端 67074.46（和 135094.46）。阈值夹在
    「单条」与「两条之和」之间 → 去重后不命中；若聚合未去重则 RISK-002 误报。
    """
    _seed(sqlite_engine)
    base = dict(
        large_amount=Decimal("1000000"),  # 关掉 RISK-001，避免与 RISK-002 混淆
        freq_count=99,
        probe_window_minutes=5,
        probe_count=99,
        probe_amount=Decimal("99999999"),
        small_amount=Decimal("1"),
        small_count=99,
        concentration_threshold=1.01,
    )

    # ① 正证：68020 < 100000 < 135094.46 → 去重后不命中
    gid = _accept(sqlite_engine, cid="CC-ENG-1")["convert_group_id"]
    res = _confirm(
        sqlite_engine, gid, thresholds=RiskThresholds(daily_total=Decimal("100000"), **base)
    )
    assert res["status"] == "confirmed"
    assert res["triggered_rules"] == [], "金额聚合未去重时 RISK-002 会误报"
    assert res["alert_ids"] == []
    assert _rows(sqlite_engine, "SELECT 1 FROM risk_alert") == []

    # ② 反证：阈值降到 60000（< 单条 68020）→ 命中，证明引擎确实在确认段跑过
    _exec(
        sqlite_engine,
        "UPDATE core_share_lot SET remain_qty = qty "
        "WHERE customer_id = :c AND product_id = :p",
        c=CUST, p=PROD_OUT,
    )
    gid2 = _accept(sqlite_engine, cid="CC-ENG-2")["convert_group_id"]
    res2 = _confirm(
        sqlite_engine, gid2, thresholds=RiskThresholds(daily_total=Decimal("60000"), **base)
    )
    assert res2["status"] == "confirmed"
    assert "RISK-002" in res2["triggered_rules"]

    alerts = _rows(sqlite_engine, "SELECT * FROM risk_alert")
    assert len(alerts) == 1, "一次转换只出一张单"
    payload = json.loads(alerts[0]["payload"])
    assert len(payload["events"]) == 2, "一张单承载两条事件"
    assert [e["trade_type"] for e in payload["events"]] == ["redeem", "subscribe"]
    assert alerts[0]["trade_id"] == res2["out_trade_id"], "主流水口径取转出端"

    # 去重不删行：两单各 2 条流水仍在（验收 6）
    assert len(_trades(sqlite_engine, gid)) == 2
    assert len(_trades(sqlite_engine, gid2)) == 2


# ── 9. 资金流中转（T-16 / R-11 / D29：确认事务写 2 条 core_cash_flow）──
def test_confirm_writes_exactly_two_cash_flows(sqlite_engine):
    """确认事务写**恰 2 条** `core_cash_flow`（验收 2/24 资金流侧）：

    - 转出端 `out`/`redeem`，金额 = 费前转出金额（与 redeem 流水 `amount` 同源）；
    - 转入端 `in`/`subscribe`，金额 = 净转入金额（与 subscribe 流水同源）；
    - `remark` 带 `convert_group_id`（对账锚点）；`occurred_at` = 确认时刻。
    （T-7~T-15 期间确认段不写该表；T-16 起本方法 = 全仓唯一写入点，反转原
    「表不存在」断言。）
    """
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]
    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "confirmed"

    flows = _rows(
        sqlite_engine,
        "SELECT flow_type, flow_subtype, amount, remark, occurred_at FROM core_cash_flow",
    )
    assert len(flows) == 2, f"确认后应恰 2 条资金流，实际 {len(flows)}"
    by_sub = {f["flow_subtype"]: f for f in flows}
    assert set(by_sub) == {"redeem", "subscribe"}
    assert by_sub["redeem"]["flow_type"] == "out"
    assert Decimal(str(by_sub["redeem"]["amount"])) == Decimal("68020.00")
    assert by_sub["subscribe"]["flow_type"] == "in"
    assert Decimal(str(by_sub["subscribe"]["amount"])) == Decimal("67074.46")
    assert by_sub["redeem"]["remark"] == f"convert:{gid}"
    assert by_sub["subscribe"]["remark"] == f"convert:{gid}"

    # 幂等：重复确认（skipped）不产生第 3 条
    _confirm(sqlite_engine, gid)
    assert len(_rows(sqlite_engine, "SELECT 1 FROM core_cash_flow")) == 2


# ── 10. 撤单（PRD §7.0.3 / R-9）────────────────────────────────────
def test_cancel_within_window_releases_occupancy(sqlite_engine):
    """窗口内撤单：`cancelled` + 占用释放 + 审计留痕。"""
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine, qty="50000")["convert_group_id"]
    repo = CoreReadOnlyRepository(engine=sqlite_engine)
    assert repo.sum_inflight_qty(CUST, PROD_OUT) == Decimal("50000")

    res = cancel_convert(
        gid, now=datetime(2026, 9, 4, 14, 59), **_services(sqlite_engine)
    )

    assert res["status"] == "cancelled"
    assert res["released_qty"] == "50000.00"
    assert repo.sum_inflight_qty(CUST, PROD_OUT) == Decimal("0")  # 占用已释放
    row = _req_row(sqlite_engine, gid)
    assert row["status"] == "cancelled"
    # 撤单只动 Core 状态，镜像保留原进度 + 留痕撤单时刻
    mirror = _rows(
        sqlite_engine,
        "SELECT status, cancelled_at FROM risk_convert_detail WHERE convert_group_id = :g",
        g=gid,
    )[0]
    assert mirror["status"] == "pending"
    assert mirror["cancelled_at"] is not None
    audits = [
        r["decision"]
        for r in _rows(
            sqlite_engine,
            "SELECT decision FROM audit_log WHERE input_summary LIKE :p",
            p=f"%{gid}%",
        )
    ]
    assert "convert_cancelled" in audits


def test_cancel_after_deadline_rejected(sqlite_engine):
    """过 15:00 截点 → `CancelNotAllowed`（验收 23）。"""
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]
    with pytest.raises(CancelNotAllowed):
        cancel_convert(gid, now=datetime(2026, 9, 4, 15, 0, 1), **_services(sqlite_engine))
    assert _req_row(sqlite_engine, gid)["status"] == "accepted"


def test_cancel_confirmed_request_rejected(sqlite_engine):
    """已确认的单不可撤（终态不可逆）。"""
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]
    _confirm(sqlite_engine, gid)

    with pytest.raises(CancelNotAllowed):
        cancel_convert(gid, now=datetime(2026, 9, 4, 14, 0), **_services(sqlite_engine))


# ── T-15（D28 · 验收 30）：撤单与确认竞态 ───────────────────────────
def test_confirm_after_cancel_conflicts(sqlite_engine):
    """撤单与确认竞态（确认侧后到）：撤单置 `cancelled` 后 confirm_one
    **幂等跳过**（`skipped` + `current_status='cancelled'`），零流水、
    占用保持已释放态 —— 批处理单笔容错语义（确认段不因已撤单中断整批）。

    时序口径（T-15 留痕）：「撤单先落库、确认后到」= 状态闸门①的 skipped
    分支；「确认读到 accepted 后、事务内置位前状态被并发改掉」才是
    `ConfirmConflict` 回滚路径（`_confirm_request` 哨兵，见
    `test_confirm_conflict_rolls_back_whole_transaction`）。反向
    （确认后撤单）见 `test_cancel_confirmed_request_rejected`。
    """
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine)["convert_group_id"]
    cancel_convert(gid, now=datetime(2026, 9, 4, 14, 0), **_services(sqlite_engine))

    res = _confirm(sqlite_engine, gid)

    assert res["status"] == "skipped"
    assert res["current_status"] == "cancelled"
    # 单保持 cancelled（确认未夺走终态）+ 零流水 + 占用保持释放
    assert _req_row(sqlite_engine, gid)["status"] == "cancelled"
    assert _trades(sqlite_engine, gid) == []
    assert CoreReadOnlyRepository(engine=sqlite_engine).sum_inflight_qty(
        CUST, PROD_OUT
    ) == Decimal("0")


def test_cancel_unknown_request_raises_not_found(sqlite_engine):
    _seed(sqlite_engine)
    with pytest.raises(ConvertRequestNotFound):
        cancel_convert("CNV-NOPE", now=SUBMIT_AT, **_services(sqlite_engine))


def test_cancel_then_accept_again_full_amount(sqlite_engine):
    """撤单释放占用后，同量可再次受理（占用确实还原）。"""
    _seed(sqlite_engine)
    gid = _accept(sqlite_engine, qty="50000")["convert_group_id"]
    cancel_convert(gid, now=datetime(2026, 9, 4, 14, 0), **_services(sqlite_engine))

    again = _accept(sqlite_engine, qty="50000")
    assert again["accepted"] is True
    assert again["convert_group_id"] != gid


# ── 11. 强制全转在确认段的表现 ──────────────────────────────────────
def test_confirm_forced_full_transfer_records_actual(sqlite_engine):
    """强制全转：受理申请 48000（留 2000 < min_hold 5000）→ 实际全转 50000。

    关键契约：**强制全转是受理段的决策**，确认段只能继承 —— 受理单 `qty` 已
    收敛为 50000，确认段若重新判最低持有会因 `leftover = 0` 得出 False（误报）。
    """
    _seed(sqlite_engine, min_hold="5000")
    accepted = _accept(sqlite_engine, qty="48000")
    gid = accepted["convert_group_id"]
    assert accepted["forced_full_transfer"] is True  # 受理响应已如实标注
    row = _req_row(sqlite_engine, gid)
    assert row["qty"] == 50000  # 受理时已收敛为实际全转量
    assert row["remark"] == "full_transfer"  # 决策落库，供确认段继承

    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "confirmed"
    assert res["forced_full_transfer"] is True
    assert res["min_hold_action"] == "force_transfer"
    assert res["actual_qty"] == "50000.00"
    # 确认后 remark 仍保留受理标记（未被确认段覆盖）
    assert _req_row(sqlite_engine, gid)["remark"] == "full_transfer"
    assert _qty(sqlite_engine, "LOT-CC-1") == Decimal("0")
    assert _qty(sqlite_engine, "LOT-CC-2") == Decimal("0")


def test_confirm_partial_mark_coexists_with_full_transfer(sqlite_engine):
    """受理标记与确认标记共存：`full_transfer;partial`（`;` 连接，顺序固定）。"""
    _seed(sqlite_engine, min_hold="5000")
    gid = _accept(sqlite_engine, qty="48000")["convert_group_id"]
    assert _req_row(sqlite_engine, gid)["remark"] == "full_transfer"

    # 抢走批次 1 的一半 → T+1 可用 35000 < 受理 50000 → 部分成交
    _exec(
        sqlite_engine,
        "UPDATE core_share_lot SET remain_qty = 15000 WHERE lot_id = 'LOT-CC-1'",
    )
    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "confirmed"
    assert res["partial"] is True
    assert res["forced_full_transfer"] is True
    assert res["actual_qty"] == "35000.00"
    assert _req_row(sqlite_engine, gid)["remark"] == "full_transfer;partial"
    assert _req_row(sqlite_engine, gid)["actual_qty"] == 35000


def test_confirm_does_not_promote_to_full_transfer(sqlite_engine):
    """确认段**不得**因可用量/阈值变化新触发强制全转（受理未标 → 确认也不标）。

    受理时 `min_hold=0`（可用 50000、申请 48000 → 留 2000 不触发，remark 为空）；
    T+1 之前把阈值调大到 30000。若确认段重判最低持有，就会得出「留 2000 < 30000」
    而擅自把客户指令扩大成全转 —— 该用例锁死「只继承、不重判」的语义边界。
    """
    _seed(sqlite_engine, min_hold="0")
    accepted = _accept(sqlite_engine, qty="48000")
    gid = accepted["convert_group_id"]
    assert accepted["forced_full_transfer"] is False
    assert _req_row(sqlite_engine, gid)["remark"] is None

    # T+1 之前产品参数变更（阈值调大）
    _exec(
        sqlite_engine,
        "UPDATE core_product SET min_hold_qty = 30000 WHERE product_id = :p",
        p=PROD_OUT,
    )
    res = _confirm(sqlite_engine, gid)
    assert res["status"] == "confirmed"
    assert res["forced_full_transfer"] is False  # 未被"追认"
    assert res["actual_qty"] == "48000.00"
    assert _req_row(sqlite_engine, gid)["remark"] is None