基金转换 T-11：工具与 SQL 汇总去重（FR-C15 / R-d）

convert 在 core_trade 落两条流水（转出 redeem + 转入 subscribe，共享 convert_group_id），
所有金额聚合读取方必须只计一次，否则同笔转换金额翻倍。

落地（改码 3 处 + 新增 1 脚本 + 测试 3 文件）
- rules.py：_amount_view → 公开 amount_view（提升而非复制别名，全仓唯一金额聚合口径）；
  docstring 补「跨模块共用」说明
- core_tools.py：query_recent_trades 的 sum_amount 改走 amount_view（FR-C15）；
  items / total_count 保持全量不变——两条流水是真实的两笔权益变动，读到两条是对的
- core_ro.py：list_holdings 加 h.qty > 0（convert 转出全部份额留下的 qty=0 归零行不是持仓）；
  sum_trades_on_date 加 convert 去重条件（R-d），组内只计转出端
- 新增 tests/test_core_tools.py（3 条：汇总不翻倍 / 跨口径一致性 / 持仓不返回 qty=0 行）
- tests/test_core_ro_sum.py 追加 1 条 convert 去重用例（既有断言零改动）
- 新增 scripts/dev/verify_convert_tools.py：真库验证脚本（MySQL 8.0.46）

执行期裁定 2 条（已留痕）
- _amount_view 提升为公开而非复制：一个函数两个名字会漂移（自检第 13 问）
- SQL 条件从 IS NULL 扩为 IS NULL OR = ''：Python 的 if not gid 把空串当无组，
  而 SQL 里 '' IS NULL 恒假。真库实证：只写 IS NULL → 合计数 400000（正确 450000，漏算 50000）。
  不加则 RISK-002 漏算，且 sqlite 单测若只造 NULL 数据永远发现不了
- 等价性边界：组内无 redeem 时 SQL 丢整组、amount_view 保留首条；由 R-b 保证不可达

验证
- pytest -q → 718 passed / 3 skipped（基线 714 加 4，零回归）
- 突变验证 3 组均精准命中：去掉去重条件（2 红）/ 去掉 qty>0（1 红）/ 汇总不走 amount_view（2 红）
- 真库 verify_convert_tools.py 14/14，隔离数据零残留
- 复跑受影响真库脚本零回归：T-8 31/31、T-7 35/35、T-10 20/20

tests/test_convert_engine.py

@@ -1,7 +1,7 @@
 """T-8 规则引擎改造单测（开发计划 §7.1 DoD / 验收 5·6·7）。
 
 覆盖三层：
-1. `rules._amount_view` 纯函数 —— 同组只留转出端 / 无 gid 恒等 / 组内无 redeem 防御 / 顺序保持
+1. `rules.amount_view` 纯函数 —— 同组只留转出端 / 无 gid 恒等 / 组内无 redeem 防御 / 顺序保持
 2. `run_rules` 视图分流 —— RISK-002 与 RISK-005 走金额视图（**不翻倍**），
    RISK-001 / RISK-003 仍看**全量**（证明去重未删行，验收 6）
 3. `engine.process_convert_event` —— **一张单 + `payload.events` 两条**（验收 7）、
@@ -22,7 +22,7 @@ from sqlalchemy import text
 from app.repository.core_ro import CoreReadOnlyRepository
 from app.repository.risk_repository import RiskRepository
 from app.service.risk.engine import process_convert_event, process_trade_event
-from app.service.risk.rules import RiskThresholds, _amount_view, run_rules
+from app.service.risk.rules import RiskThresholds, amount_view, run_rules
 
 CUST = "CUST-T8"
 PROD_A = "PROD-T8A"  # 转出方
@@ -84,27 +84,27 @@ def _ids(hits) -> set[str]:
     return {h.rule_id for h in hits}
 
 
-# ── 1. `_amount_view` 纯函数 ─────────────────────────────────────────
+# ── 1. `amount_view` 纯函数 ─────────────────────────────────────────
 def test_amount_view_identity_without_group():
     """无 convert_group_id 的交易**原样通过**（非 convert 场景恒等 → 既有断言零影响）。"""
     raw = [
         _trade("T-1", "redeem", "1000"),
         _trade("T-2", "subscribe", "900", NOW + timedelta(seconds=1), product=PROD_B),
     ]
-    view = _amount_view(raw)
+    view = amount_view(raw)
     assert view == raw, "无 gid 输入必须恒等（内容与顺序都不变）"
     assert view is not raw, "返回独立列表，不得共享可变状态"
 
 
 def test_amount_view_keeps_redeem_when_subscribe_comes_first():
     out, inn = _convert_legs()
-    view = _amount_view([inn, out])
+    view = amount_view([inn, out])
     assert [t["trade_id"] for t in view] == ["TRD-T8-OUT"], "同组只留转出端，且位置不变"
 
 
 def test_amount_view_keeps_redeem_when_redeem_comes_first():
     out, inn = _convert_legs()
-    view = _amount_view([out, inn])
+    view = amount_view([out, inn])
     assert [t["trade_id"] for t in view] == ["TRD-T8-OUT"], "转入端不得覆盖已占位的转出端"
 
 
@@ -113,7 +113,7 @@ def test_amount_view_group_without_redeem_keeps_first_row():
     a = _trade("T-IN-1", "subscribe", "100", gid="CNV-T8-X", product=PROD_B)
     b = _trade("T-IN-2", "subscribe", "200", NOW + timedelta(seconds=1), gid="CNV-T8-X",
                product=PROD_B)
-    assert _amount_view([a, b]) == [a]
+    assert amount_view([a, b]) == [a]
 
 
 def test_amount_view_mixes_groups_and_plain_trades_in_order():
@@ -124,7 +124,7 @@ def test_amount_view_mixes_groups_and_plain_trades_in_order():
     out2 = _trade("TRD-T8-OUT2", "redeem", "500", NOW + timedelta(seconds=4), gid="CNV-T8-0002")
     in2 = _trade("TRD-T8-IN2", "subscribe", "480", NOW + timedelta(seconds=5),
                  gid="CNV-T8-0002", product=PROD_B)
-    view = _amount_view([n1, in1, out1, n2, in2, out2])
+    view = amount_view([n1, in1, out1, n2, in2, out2])
     assert [t["trade_id"] for t in view] == ["T-N1", "TRD-T8-OUT", "T-N2", "TRD-T8-OUT2"]
 
 

tests/test_core_ro_sum.py

@@ -15,13 +15,22 @@ from app.repository.core_ro import CoreReadOnlyRepository
 def repo():
     engine = create_sqlite_engine()  # DDL 单一事实源（B4 评审 P3-12）
 
-    def insert(trade_id, amount, traded_at, trade_type="subscribe", status="confirmed", cid="C1"):
+    def insert(
+        trade_id,
+        amount,
+        traded_at,
+        trade_type="subscribe",
+        status="confirmed",
+        cid="C1",
+        gid=None,
+    ):
+        """插一笔流水。`gid` = convert 组 ID：None → NULL、'' → 空串，二者均视为「非 convert」。"""
         with engine.begin() as conn:
             conn.execute(
                 text(
                     "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type,"
-                    " amount, trade_status, traded_at)"
-                    " VALUES (:tid, :cid, 'P1', :tt, :amt, :st, :at)"
+                    " amount, trade_status, traded_at, convert_group_id)"
+                    " VALUES (:tid, :cid, 'P1', :tt, :amt, :st, :at, :gid)"
                 ),
                 {
                     "tid": trade_id,
@@ -30,6 +39,7 @@ def repo():
                     "amt": amount,
                     "st": status,
                     "at": traded_at,
+                    "gid": gid,
                 },
             )
 
@@ -72,3 +82,19 @@ def test_sum_per_customer(repo):
     insert("T2", "500000", _at(9), cid="C2")
     assert r.sum_trades_on_date("C1", DAY) == Decimal("500000")
     assert r.sum_trades_on_date("C2", DAY) == Decimal("500000")
+
+
+def test_sum_dedupes_convert_group(repo):
+    """T-11 · R-d：一次转换只计**转出端**；空串 gid 与 NULL 同视为「非 convert」。
+
+    数据：G1 组两条（转出 300000 + 转入 300000，共享 gid）+ 普通赎回 100000
+    + 空串 gid 的申购 50000。
+    期望：300000 + 100000 + 50000 = **450000**。
+    **若未去重则为 750000（T1+T2 双计）** —— 断言用具体值，能区分对错。
+    """
+    r, insert = repo
+    insert("T1", "300000", _at(9), "redeem", gid="G1")       # convert 转出端 → 计入
+    insert("T2", "300000", _at(9), "subscribe", gid="G1")    # convert 转入端 → 同组，不计
+    insert("T3", "100000", _at(10), "redeem")                # 普通赎回（gid=NULL）→ 计入
+    insert("T4", "50000", _at(11), "subscribe", gid="")      # 空串 gid → 视为无组 → 计入
+    assert r.sum_trades_on_date("C1", DAY) == Decimal("450000")

tests/test_core_tools.py

@@ -0,0 +1,107 @@
+"""core_tools 查询口径单测（T-11 · FR-C15 + 归零行过滤）。
+
+自建 sqlite 完整种子（`core_product` / `core_holding` / `core_trade`），直调 Tool 函数。
+覆盖 T-11 两条 DoD：
+
+1. `query_recent_trades` 的 `sum_amount` 不因 convert 两条流水而**翻倍**
+   （走 `amount_view`；明细与 `total_count` 仍为全量），
+   并与 `core_ro.sum_trades_on_date` 做**跨口径一致性**断言（自检第 13 问：
+   同一口径不得有两份实现漂移）；
+2. `query_holdings` 不返回 `qty = 0` 的**归零行**（convert 转出全部份额后的台账留痕行）。
+
+种子时间刻意取**同一个 `now`**：既落在 `query_recent_trades` 的 [now−30d, now) 窗内，
+又保证属于 `now.date()` 这一天，使跨口径断言不受运行时刻影响。
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+from decimal import Decimal
+
+import pytest
+from sqlalchemy import text
+
+from _ddl import create_sqlite_engine
+
+from app.repository.core_ro import CoreReadOnlyRepository
+from app.tool.core_tools import query_holdings, query_recent_trades
+
+CUST = "CUST-T11"
+
+
+@pytest.fixture()
+def seed():
+    engine = create_sqlite_engine()
+    now = _dt.datetime.now()
+    with engine.begin() as conn:
+        conn.execute(
+            text(
+                "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type,"
+                " min_subscribe_amount, term_days) VALUES"
+                " ('PA', '甲基金', 'R3', 'mixed', 100, 0),"
+                " ('PB', '乙基金', 'R3', 'mixed', 100, 0)"
+            )
+        )
+        # PA 正常持有；PB 为 convert 转出全部后的**归零行**（qty = 0，行保留）
+        conn.execute(
+            text(
+                "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount,"
+                " market_value, pnl_pct, as_of) VALUES"
+                " (:cid, 'PA', 1000.00, 1000.00, 1200.00, 0.2000, :d),"
+                " (:cid, 'PB', 0.00, 0.00, 0.00, 0.0000, :d)"
+            ),
+            {"cid": CUST, "d": now.date()},
+        )
+        # 一次 convert 落两条（转出 redeem 300000 + 转入 subscribe 300000，共享 G1）
+        # + 1 笔普通赎回 100000（convert_group_id = NULL）
+        conn.execute(
+            text(
+                "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount,"
+                " qty, convert_group_id, trade_status, traded_at) VALUES"
+                " ('TX-OUT',   :cid, 'PA', 'redeem',    300000, 250.00, 'G1', 'confirmed', :t),"
+                " ('TX-IN',    :cid, 'PB', 'subscribe', 300000, 240.00, 'G1', 'confirmed', :t),"
+                " ('TX-PLAIN', :cid, 'PA', 'redeem',    100000,  80.00, NULL, 'confirmed', :t)"
+            ),
+            {"cid": CUST, "t": now},
+        )
+    yield engine
+    engine.dispose()
+
+
+def test_query_recent_trades_sum_not_doubled(seed):
+    """FR-C15：汇总只计转出端；**明细与条数保持全量**（一次转换两条是真实的）。"""
+    repo = CoreReadOnlyRepository(engine=seed)
+    res = query_recent_trades(CUST, days=30, core_ro=repo)
+
+    # 明细全量：convert 两条 + 普通赎回 1 条
+    assert res["total_count"] == 3
+    assert sorted(r["trade_type"] for r in res["items"]) == ["redeem", "redeem", "subscribe"]
+
+    # 汇总去重：300000（转出端）+ 100000（普通赎回）= 400000
+    # 若未去重则为 700000（TX-OUT + TX-IN 双计）—— 用具体值才区分得开
+    assert res["sum_amount"] == 400000.0
+
+
+def test_query_recent_trades_sum_matches_sum_trades_on_date(seed):
+    """跨口径一致性：Tool 汇总（Python 侧 amount_view）== 仓储 SQL 汇总（IS NULL OR =''）。
+
+    两处是同一口径的两种落地（一个 Python 一个 SQL），此处用断言锁死等价性，
+    任一侧口径漂移都会在此变红。
+    """
+    repo = CoreReadOnlyRepository(engine=seed)
+    tool_sum = query_recent_trades(CUST, days=30, core_ro=repo)["sum_amount"]
+    sql_sum = repo.sum_trades_on_date(CUST, _dt.date.today())
+    assert Decimal(str(tool_sum)) == sql_sum
+    assert sql_sum == Decimal("400000")
+
+
+def test_query_holdings_excludes_zero_qty(seed):
+    """归零行（convert 转出全部份额，`qty = 0` 台账留痕）不得作为持仓返回。"""
+    repo = CoreReadOnlyRepository(engine=seed)
+    res = query_holdings(CUST, core_ro=repo)
+
+    assert res["total_count"] == 1
+    assert [r["product_id"] for r in res["items"]] == ["PA"]
+    # 合计不含归零行（PB 的 market_value 为 0，即使计入也不变；
+    # 故断言**条数**才是真正能区分对错的判据）
+    assert res["sum_market_value"] == 1200.0