客户看板「今日盈亏」不再是硬编码 0：按行情/净值基准真实计算（含当日买卖）

- 口径：今日盈亏 = 今日市值 − 昨日持仓市值 − 今日买入金额 + 今日卖出金额（不含费用，与「持有盈亏」同口径） - 昨日持仓数量由当日成交反推，不需要新表新字段 - 基准优先场内行情最近两个交易日收盘价（与同页 latest_price/market_value 同源、客户可核对），行情只有一天时回退基金净值（15911/159991-159995 的行情本身就来自净值序列） - 只认 transaction_type ∈ {买入,卖出}：演示库里混进的风控场外申购/赎回（RISKDEMO-*）会把「昨日持仓数量」抬到 76 万份、今日盈亏从 -21.22 变成 -1612.72 - 接口字段与前端一行未改：HoldingItem.today_profit_loss / PortfolioSummary 两个字段数值变真 - 新增 tools/check_today_profit_loss.py：纯 SQL 独立复算并与接口逐只比对（实测 9001 = -132.70 / -0.2528%） - 新增 6 条单测；pytest tests/unit tests/contract → 1466 passed, 0 failed
2026-09-15 00:03:23 +08:00
parent 48b43cfad5
commit 9dd2802d67
7 changed files with 813 additions and 53 deletions
@@ -0,0 +1,186 @@
+"""复算并核对「今日盈亏」（只读，不改任何数据）。
+
+`app/service/trade_service.py` 的「今日盈亏」不是把库里某个字段读出来，而是**算**出来的：
+
+```
+今日盈亏 = 今日市值 − 昨日持仓市值 − 今日买入金额 + 今日卖出金额     （不含交易费用）
+昨日持仓数量 = 今日持仓数量 − 今日买入份额 + 今日卖出份额
+基准：优先场内行情最近两个交易日收盘价；行情只有一天则回退基金净值最近两个净值日
+```
+
+这个脚本用 **纯 SQL 独立复算**一遍，再和 `TradeService.get_account_dashboard()` 的输出逐只比对，
+用来证明"接口里的数字不是拍出来的"。任何一处不一致都会以 `!! 不一致` 标出并以退出码 1 结束。
+
+用法::
+
+    python tools/check_today_profit_loss.py            # 默认查演示客户 9001 10001 10002 12001
+    python tools/check_today_profit_loss.py 9101 9102  # 指定客户号
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+from decimal import ROUND_HALF_UP, Decimal
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.contracts import RequestContext
+from app.infrastructure.db import SessionFactory
+from app.service.trade_service import TradeService
+
+TWO_PLACES = Decimal("0.01")
+FOUR_PLACES = Decimal("0.0001")
+HUNDRED = Decimal("100")
+DEFAULT_CUSTOMERS = (9001, 10001, 10002, 12001)
+
+_HOLDINGS_SQL = text(
+    """
+    SELECT h.product_id, f.product_code, h.total_quantity
+    FROM fin_holding h JOIN fin_product f ON f.id = h.product_id
+    WHERE h.customer_id = :cid AND h.status = '持有中'
+    ORDER BY f.product_code
+    """
+)
+
+# 每个交易日只取一行（同一天可能被重复刷入），再取最近两天。
+_PRICE_SQL = text(
+    """
+    SELECT trade_date, close_price FROM (
+        SELECT trade_date, close_price,
+               ROW_NUMBER() OVER (PARTITION BY trade_date ORDER BY id DESC) rn
+        FROM fin_market_price WHERE product_id = :pid
+    ) t WHERE rn = 1
+    ORDER BY trade_date DESC LIMIT 2
+    """
+)
+
+_NAV_SQL = text(
+    """
+    SELECT nav_date, nav FROM (
+        SELECT nav_date, nav,
+               ROW_NUMBER() OVER (PARTITION BY nav_date ORDER BY id DESC) rn
+        FROM fin_nav_history WHERE product_id = :pid
+    ) t WHERE rn = 1
+    ORDER BY nav_date DESC LIMIT 2
+    """
+)
+
+# 只认本平台场内成交（买入/卖出）；场外申购/赎回即便被写进来也不算当日买卖。
+_FLOW_SQL = text(
+    """
+    SELECT
+      COALESCE(SUM(CASE WHEN order_side = 'buy'  THEN shares END), 0),
+      COALESCE(SUM(CASE WHEN order_side = 'sell' THEN shares END), 0),
+      COALESCE(SUM(CASE WHEN order_side = 'buy'  THEN gross_amount END), 0),
+      COALESCE(SUM(CASE WHEN order_side = 'sell' THEN gross_amount END), 0),
+      COUNT(*)
+    FROM fin_transaction
+    WHERE customer_id = :cid AND product_id = :pid
+      AND transaction_type IN ('买入', '卖出')
+      AND DATE(confirmed_at) = :ref
+    """
+)
+
+
+async def recompute(session: AsyncSession, customer_id: int) -> list[tuple[str, Decimal, Decimal, str]]:
+    """独立复算：返回 [(产品代码, 今日盈亏, 今日盈亏率分母, 口径说明)]。"""
+    out: list[tuple[str, Decimal, Decimal, str]] = []
+    holdings = (await session.execute(_HOLDINGS_SQL, {"cid": customer_id})).all()
+    for product_id, product_code, quantity in holdings:
+        prices = (await session.execute(_PRICE_SQL, {"pid": product_id})).all()
+        if len(prices) == 2:
+            ref_date, today_value, prev_value = prices[0][0], prices[0][1], prices[1][1]
+            note = f"行情 {ref_date} 昨收={prev_value} 今收={today_value}"
+        else:
+            navs = (await session.execute(_NAV_SQL, {"pid": product_id})).all()
+            if len(navs) < 2:
+                out.append((product_code, Decimal("0.00"), Decimal("0.00"), "无基准（记 0）"))
+                continue
+            ref_date, today_value, prev_value = navs[0][0], navs[0][1], navs[1][1]
+            note = f"净值 {ref_date} 昨日={prev_value} 今日={today_value}"
+
+        row = (
+            await session.execute(
+                _FLOW_SQL, {"cid": customer_id, "pid": product_id, "ref": ref_date}
+            )
+        ).all()[0]
+        buy_shares, sell_shares, buy_gross, sell_gross, txn_count = row
+        qty_at_open = quantity - buy_shares + sell_shares
+        if qty_at_open < 0:
+            out.append((product_code, Decimal("0.00"), Decimal("0.00"), "成交与持仓不符（记 0）"))
+            continue
+        today_pl = (
+            quantity * today_value
+            - qty_at_open * prev_value
+            - buy_gross
+            + sell_gross
+        ).quantize(TWO_PLACES, rounding=ROUND_HALF_UP)
+        base = (quantity * today_value - today_pl).quantize(TWO_PLACES, rounding=ROUND_HALF_UP)
+        out.append(
+            (
+                product_code,
+                today_pl,
+                base,
+                f"{note} 昨日持仓={qty_at_open} 当日成交={txn_count} 笔",
+            )
+        )
+    return out
+
+
+async def check(customer_id: int) -> bool:
+    async with SessionFactory() as session:
+        expected = await recompute(session, customer_id)
+    async with SessionFactory() as session:
+        service = TradeService(session)
+        response = await service.get_account_dashboard(
+            RequestContext(
+                user_id=str(customer_id),
+                trace_id=f"today-pl-check-{customer_id}",
+                roles=("customer",),
+                customer_ids=(str(customer_id),),
+            )
+        )
+
+    actual = {item.product_code: item.today_profit_loss for item in response.holdings}
+    print(f"\n===== 客户 {customer_id} =====")
+    ok = True
+    for product_code, today_pl, _base, note in expected:
+        got = actual.get(product_code)
+        same = got == today_pl
+        ok = ok and same
+        print(
+            f"  {product_code:8s} 复算={today_pl:>12} 接口={str(got):>12} "
+            f"{'OK' if same else '!! 不一致'}  {note}"
+        )
+
+    expected_total = sum((pl for _c, pl, _b, _n in expected), Decimal("0")).quantize(TWO_PLACES)
+    expected_base = sum((b for _c, _p, b, _n in expected), Decimal("0"))
+    expected_ratio = (
+        (expected_total / expected_base * HUNDRED).quantize(FOUR_PLACES, rounding=ROUND_HALF_UP)
+        if expected_base > 0
+        else Decimal("0")
+    )
+    summary = response.summary
+    totals_ok = (
+        summary.today_profit_loss == expected_total
+        and summary.today_profit_loss_ratio == expected_ratio
+    )
+    ok = ok and totals_ok
+    print(
+        f"  汇总：接口 {summary.today_profit_loss} / {summary.today_profit_loss_ratio}%  "
+        f"复算 {expected_total} / {expected_ratio}%  {'OK' if totals_ok else '!! 不一致'}"
+    )
+    return ok
+
+
+async def main() -> int:
+    customers = [int(arg) for arg in sys.argv[1:]] or list(DEFAULT_CUSTOMERS)
+    results = [await check(customer_id) for customer_id in customers]
+    print(f"\n结论：{'全部一致' if all(results) else '存在不一致，见上面 !! 行'}")
+    return 0 if all(results) else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(main()))