xinghuo/group_xinghuo_jinrong
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

基金转换 T-6：apply_convert 阶段一单事务（core 库唯一写入口）

Browse Source 
新增 app/gateway/convert_core_repository.py：两条流水（redeem/subscribe，
R-b 不使用 ENUM 的 convert）+ N×条件 UPDATE 批次扣减 + 转入新批次
+ 两端 core_holding + N×计费明细，全部收口在 with engine.begin() 单事务。

实现级要点（与开发计划 §6.1 的三处差异，已回写该节执行记录）：
- core_holding 改用「基于列当前值的增量 UPDATE」，入参不带持仓快照：
  MySQL 的 UPDATE 是当前读，并发两笔自然累加；用快照算绝对值会互相覆盖。
- 数值参数参与算术时写成 (:x + 0.0)：实测 sqlite 在 UPDATE 算术表达式中
  不把 TEXT 绑定参数转数值（传 '120' 时 c 不变），加 +0.0 后两库一致。
- pnl_pct 拆成独立 UPDATE 重算，避开 MySQL「SET 从左到右」的顺序坑。

验证：pytest 639 passed / 3 skipped（基线 634 +5，零回归）；
新增 scripts/dev/verify_convert_apply.py 真 MySQL 验证 24/24 一致，
含真并发两笔首次转入同一产品的终态断言（MySQL RR 的 gap lock 实测留痕）。
This commit is contained in:
GaoYiYuan_0626
2026-09-10 16:26:42 +08:00
parent 4fd2ec05c8
commit 82c9467fb7
4 changed files with 1314 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_convert_core.py
+451
View File
@@ -0,0 +1,451 @@
"""T-6 `convert_core_repository.apply_convert` 单测（开发计划 §6.1 DoD）。
sqlite 内存库（`conftest.sqlite_engine` 单一事实源建表），数据自建。
**金额一律由生产 `calc.py` / `fee.py` 算出，不手算**（自检第 13 问：
同一规则只留一个副本 —— 测试里手算一遍等于给公式造第二个副本，
生产改了测试还在"绿"，是假证据）。
覆盖：
1. 正常路径：2 条流水（R-b：`redeem`/`subscribe`、同组）+ N 条明细 + 两端持仓
2. 冲突路径：`remain_qty < 本次扣减` → `LotConflict`(409) 且**无残留**
3. 回滚路径：明细写入注入异常 → `core_trade`/`core_share_lot`/`core_holding` **全无残留**
4. 转出端归零**保留行**；转入端首次 INSERT、再次 UPDATE（累加）
5. R-a 并发首次建行：另一笔先落行 → 本笔 INSERT 撞 UNIQUE → 回退增量 UPDATE
→ 只有一行、金额为两笔之和、本笔不报错
"""
from __future__ import annotations
from datetime import date, datetime
from decimal import ROUND_HALF_UP, Decimal
import pytest
from sqlalchemy import text
from app.gateway.convert_core_repository import (
ConvertApplyInput,
ConvertCoreRepository,
LotCharge,
)
from app.service.convert.calc import (
convert_amount,
diff_fee,
hold_days,
in_qty,
lot_amount,
lot_fee,
plan_lots,
)
from app.service.convert.errors import LotConflict
from app.service.convert.fee import pick_fee_rate
from app.service.convert.types import FeeRule, Lot
CUSTOMER = "CUST-T6"
PROD_OUT = "PROD-T6A" # 转出（债券型，申购费率 0.0030）
PROD_IN = "PROD-T6B" # 转入（股票型，申购费率 0.0080）
TRADE_AT = datetime(2026, 9, 4, 10, 0, 0)
TRADE_DATE = date(2026, 9, 4)
OUT_RATE = Decimal("0.0030")
IN_RATE = Decimal("0.0080")
IN_NAV = Decimal("0.9500")
# 赎回费五档（与 07-seed-fee-rule.sql、fee.py docstring 三处一致）
FEE_TIERS = [
(0, 7, "0.0150"),
(7, 30, "0.0100"),
(30, 180, "0.0050"),
(180, 365, "0.0025"),
(365, None, "0.0000"),
]
def _dec(value: object, places: str = "0.01") -> Decimal:
"""DB 读回值（sqlite float / MySQL Decimal）→ 统一 2 位 Decimal 再比较。"""
return Decimal(str(value)).quantize(Decimal(places), rounding=ROUND_HALF_UP)
# ── 种子 ────────────────────────────────────────────────────────────
def _seed_base(engine) -> None:
with engine.begin() as conn:
conn.execute(
text("INSERT INTO core_customer (customer_id, display_name) VALUES (:c, 'T6')"),
{"c": CUSTOMER},
)
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, '转出基金', 'R2', 'bond', 1, 1, :r)"
),
{"p": PROD_OUT, "r": float(OUT_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, '转入基金', 'R4', 'stock', 1, 1, :r)"
),
{"p": PROD_IN, "r": float(IN_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": float(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": float(IN_NAV), "d": TRADE_DATE},
)
def _seed_lot(engine, lot_id: str, qty: str, nav: str, confirmed_at: datetime) -> None:
with engine.begin() as conn:
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": float(qty), "nav": float(nav), "cat": confirmed_at,
},
)
def _seed_holding(engine, pid: str, qty: str, cost: str, mv: str) -> None:
with engine.begin() as conn:
conn.execute(
text(
"INSERT INTO core_holding (customer_id, product_id, qty, cost_amount, "
"market_value, pnl_pct, as_of) VALUES (:c, :p, :q, :cost, :mv, 0, :d)"
),
{
"c": CUSTOMER, "p": pid, "q": float(qty),
"cost": float(cost), "mv": float(mv), "d": TRADE_DATE,
},
)
def _read_lots(engine) -> list[Lot]:
from app.repository.core_ro import CoreReadOnlyRepository
return [
Lot.from_row(r)
for r in CoreReadOnlyRepository(engine=engine).list_share_lots(CUSTOMER, PROD_OUT)
]
def _read_rules(engine) -> list[FeeRule]:
from app.repository.core_ro import CoreReadOnlyRepository
return [
FeeRule.from_row(r)
for r in CoreReadOnlyRepository(engine=engine).get_redeem_fee_rules(PROD_OUT)
]
# ── 折算（全部走生产纯函数，与 T-7 的调用姿势一致）──────────────────────
def _build_input(
engine,
*,
group_id: str,
requested: str,
in_lot_id: str = "LOT-IN-1",
out_trade_id: str | None = None,
in_trade_id: str | None = None,
) -> ConvertApplyInput:
"""跑一遍「选批 → 逐批计费 → 折算」，构造阶段一入参。"""
lots = _read_lots(engine)
rules = _read_rules(engine)
plan = plan_lots(lots, Decimal(requested))
charges: list[LotCharge] = []
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"))
redeem_fee = sum((c.fee_amount for c in charges), Decimal("0"))
conv = convert_amount(out_amount, redeem_fee)
gap = diff_fee(conv, OUT_RATE, IN_RATE, "amount_diff")
in_amount = conv - gap
shares = in_qty(in_amount, IN_NAV)
gid = group_id
return ConvertApplyInput(
convert_group_id=gid,
out_trade_id=out_trade_id or f"{gid}-OUT",
in_trade_id=in_trade_id or f"{gid}-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=shares,
in_amount=in_amount,
in_nav=IN_NAV,
in_nav_date=TRADE_DATE,
in_lot_id=in_lot_id,
in_confirmed_at=datetime(2026, 9, 5, 10, 0, 0), # T+1（D14）
charges=tuple(charges),
)
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 _holding(engine, pid: str) -> dict | None:
rows = _rows(
engine,
"SELECT * FROM core_holding WHERE customer_id = :c AND product_id = :p",
c=CUSTOMER, p=pid,
)
return rows[0] if rows else None
# ── 1. 正常路径 ─────────────────────────────────────────────────────
def test_apply_convert_happy_path_writes_trades_details_and_holdings(sqlite_engine):
_seed_base(sqlite_engine)
# 持有 34 天（30–180 档 0.0050）与 3 天（<7 档 0.0150）两个批次
_seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
_seed_lot(sqlite_engine, "LOT-A2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0))
_seed_holding(sqlite_engine, PROD_OUT, "150", "150.00", "154.50")
req = _build_input(sqlite_engine, group_id="CNV-T6-1", requested="120")
ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)
# ① 两条流水：R-b 必须是 redeem / subscribe（**不是 convert**），同组
trades = _rows(
sqlite_engine,
"SELECT * FROM core_trade WHERE convert_group_id = :g ORDER BY trade_type",
g="CNV-T6-1",
)
assert len(trades) == 2
assert [t["trade_type"] for t in trades] == ["redeem", "subscribe"]
assert {t["product_id"] for t in trades} == {PROD_OUT, PROD_IN}
out_trade = next(t for t in trades if t["trade_type"] == "redeem")
in_trade = next(t for t in trades if t["trade_type"] == "subscribe")
assert _dec(out_trade["amount"]) == _dec(req.out_amount)
assert _dec(in_trade["amount"]) == _dec(req.in_amount)
assert _dec(out_trade["qty"]) == _dec(req.out_qty)
assert _dec(in_trade["qty"]) == _dec(req.in_qty)
# ② 批次：LOT-A1 扣满归零但**保留行**，LOT-A2 剩 30，转入新批次已建
remain = {
r["lot_id"]: _dec(r["remain_qty"])
for r in _rows(
sqlite_engine,
"SELECT lot_id, remain_qty FROM core_share_lot WHERE customer_id = :c",
c=CUSTOMER,
)
}
assert remain["LOT-A1"] == Decimal("0.00") # 归零保留行（D9/P2）
assert remain["LOT-A2"] == Decimal("30.00")
assert remain["LOT-IN-1"] == _dec(req.in_qty)
new_lot = _rows(
sqlite_engine,
"SELECT * FROM core_share_lot WHERE lot_id = 'LOT-IN-1'",
)[0]
assert new_lot["product_id"] == PROD_IN
assert _dec(new_lot["nav"]) == _dec(IN_NAV)
assert str(new_lot["confirmed_at"])[:10] == "2026-09-05" # T+1（D14）
assert new_lot["source_trade_id"] == req.in_trade_id
# ③ 明细：每批一行，含 nav / nav_date（D6 补偿源）
details = _rows(
sqlite_engine,
"SELECT * FROM core_convert_lot_detail WHERE convert_group_id = :g ORDER BY lot_id",
g="CNV-T6-1",
)
assert [d["lot_id"] for d in details] == ["LOT-A1", "LOT-A2"]
assert [d["hold_days"] for d in details] == [34, 3]
assert [_dec(d["fee_rate"], "0.0001") for d in details] == [
Decimal("0.0050"), Decimal("0.0150")
]
# ④ 两端持仓：转出等比例结转成本、转入按净值建仓
out_h = _holding(sqlite_engine, PROD_OUT)
assert _dec(out_h["qty"]) == Decimal("30.00") # 150 − 120
assert _dec(out_h["cost_amount"]) == Decimal("30.00") # 150 × (1 − 120/150)
assert _dec(out_h["market_value"]) == Decimal("30.90") # 154.50 × 0.2
assert _dec(out_h["pnl_pct"], "0.0001") == Decimal("0.0300")
assert str(out_h["as_of"])[:10] == "2026-09-04"
in_h = _holding(sqlite_engine, PROD_IN)
assert _dec(in_h["qty"]) == _dec(req.in_qty)
assert _dec(in_h["cost_amount"]) == _dec(req.in_amount)
# ── 2. 冲突路径（并发哨兵）───────────────────────────────────────────
def test_apply_convert_raises_lot_conflict_and_leaves_nothing(sqlite_engine):
_seed_base(sqlite_engine)
_seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
_seed_holding(sqlite_engine, PROD_OUT, "100", "100.00", "103.00")
# 申请 100 拿到正常入参，再把扣减量改大到超过剩余 → 条件 UPDATE rowcount=0
req = _build_input(sqlite_engine, group_id="CNV-T6-C", requested="100")
bigger = tuple(
LotCharge(
lot_id=c.lot_id, qty=Decimal("120"), 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 req.charges
)
req = ConvertApplyInput(**{**req.__dict__, "charges": bigger})
with pytest.raises(LotConflict) as exc:
ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)
assert exc.value.status_code == 409
# 冲突即回滚：流水与明细都是 0 行
assert _rows(sqlite_engine, "SELECT * FROM core_trade") == []
assert _rows(sqlite_engine, "SELECT * FROM core_convert_lot_detail") == []
# ── 3. 回滚路径（同事务的硬证据）──────────────────────────────────────
def test_apply_convert_rolls_back_everything_when_detail_insert_fails(
sqlite_engine, monkeypatch,
):
_seed_base(sqlite_engine)
_seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
_seed_lot(sqlite_engine, "LOT-A2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0))
_seed_holding(sqlite_engine, PROD_OUT, "150", "150.00", "154.50")
def _boom(self, conn, req): # noqa: ANN001
raise RuntimeError("注入故障：明细写入失败")
monkeypatch.setattr(ConvertCoreRepository, "_insert_lot_details", _boom)
req = _build_input(sqlite_engine, group_id="CNV-T6-R", requested="120")
with pytest.raises(RuntimeError):
ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)
# 全部无残留 = 阶段一确实在同一个事务里（架构 §5.1）
assert _rows(sqlite_engine, "SELECT * FROM core_trade") == []
assert _rows(sqlite_engine, "SELECT * FROM core_convert_lot_detail") == []
assert _dec(
_rows(
sqlite_engine,
"SELECT remain_qty FROM core_share_lot WHERE lot_id = 'LOT-A1'",
)[0]["remain_qty"]
) == Decimal("100.00")
assert _dec(_holding(sqlite_engine, PROD_OUT)["qty"]) == Decimal("150.00")
assert _rows(
sqlite_engine,
"SELECT * FROM core_share_lot WHERE lot_id = 'LOT-IN-1'",
) == []
assert _holding(sqlite_engine, PROD_IN) is None
# ── 4. 转出归零保留行 / 转入端首次 INSERT 与再次 UPDATE ────────────────
def test_second_convert_zeroes_out_holding_and_accumulates_in_holding(sqlite_engine):
_seed_base(sqlite_engine)
_seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
_seed_lot(sqlite_engine, "LOT-A2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0))
_seed_holding(sqlite_engine, PROD_OUT, "150", "150.00", "154.50")
repo = ConvertCoreRepository(engine=sqlite_engine)
first = _build_input(sqlite_engine, group_id="CNV-T6-F", requested="120")
repo.apply_convert(first)
assert _dec(_holding(sqlite_engine, PROD_OUT)["qty"]) == Decimal("30.00")
# 第二次：转出剩余 30（全转）→ 归零保留行；转入端已存在 → 累加而非覆盖
second = _build_input(
sqlite_engine, group_id="CNV-T6-S", requested="30", in_lot_id="LOT-IN-2",
)
repo.apply_convert(second)
out_h = _holding(sqlite_engine, PROD_OUT)
assert out_h is not None # 归零**保留行**，不删（D9/P2，`xh_core_rw` 也无 DELETE）
assert _dec(out_h["qty"]) == Decimal("0.00")
in_h = _holding(sqlite_engine, PROD_IN)
assert _dec(in_h["qty"]) == _dec(first.in_qty + second.in_qty)
assert _dec(in_h["cost_amount"]) == _dec(first.in_amount + second.in_amount)
# 持仓行数仍为 1（UNIQUE(customer_id, product_id)）
assert len(
_rows(
sqlite_engine,
"SELECT * FROM core_holding WHERE customer_id = :c",
c=CUSTOMER,
)
) == 2 # 转出 + 转入各一行
# ── 5. R-a 并发首次建行（IntegrityError → 回退增量 UPDATE）─────────────
def test_concurrent_first_insert_falls_back_to_incremental_update(
sqlite_engine, monkeypatch,
):
"""两笔并发首次转入同一 (客户, 产品)：只有一行、金额为两笔之和、两笔都成功。
模拟方式：注入点选在 `_insert_holding_row`（**UPDATE 之后、INSERT 之前**）——
本笔的 UPDATE 已命中 0 行，此刻另一笔并发转换提交了同一 (客户, 产品) 的行，
于是本笔 INSERT 撞 `UNIQUE(customer_id, product_id)` → 回退增量 UPDATE。
回退若写成"绝对值 UPDATE"就会覆盖另一笔，故本用例是 R-a 的硬证据。
"""
_seed_base(sqlite_engine)
_seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
_seed_holding(sqlite_engine, PROD_OUT, "100", "100.00", "103.00")
original = ConvertCoreRepository._insert_holding_row
concurrent_qty = Decimal("50")
concurrent_cost = Decimal("50")
hit = {"fallback": False} # 记录回退分支是否真的被走到（防止用例假绿）
def _with_concurrent_row(self, conn, insert_sql, update_sql, params): # noqa: ANN001
if params["pid"] == PROD_IN:
# 模拟另一笔并发转换已提交：本事务内可见，INSERT 必然撞 UNIQUE
conn.execute(
text(
"INSERT INTO core_holding (customer_id, product_id, qty, "
"cost_amount, market_value, pnl_pct, as_of) "
"VALUES (:c, :p, :q, :cost, :mv, 0, :d)"
),
{
"c": params["cid"], "p": PROD_IN,
"q": str(concurrent_qty), "cost": str(concurrent_cost),
"mv": str(concurrent_cost), "d": params["as_of"],
},
)
hit["fallback"] = True
original(self, conn, insert_sql, update_sql, params)
monkeypatch.setattr(
ConvertCoreRepository, "_insert_holding_row", _with_concurrent_row
)
req = _build_input(sqlite_engine, group_id="CNV-T6-X", requested="100")
ConvertCoreRepository(engine=sqlite_engine).apply_convert(req) # 不得抛 5xx
assert hit["fallback"] is True # 注入点确实被走到，否则本用例无意义
rows = _rows(
sqlite_engine,
"SELECT * FROM core_holding WHERE customer_id = :c AND product_id = :p",
c=CUSTOMER, p=PROD_IN,
)
assert len(rows) == 1 # 只有一行
assert _dec(rows[0]["qty"]) == _dec(concurrent_qty + req.in_qty)
assert _dec(rows[0]["cost_amount"]) == _dec(concurrent_cost + req.in_amount)