lzf_0626
9dd2802d67
- 口径：今日盈亏 = 今日市值 − 昨日持仓市值 − 今日买入金额 + 今日卖出金额（不含费用，与「持有盈亏」同口径）
- 昨日持仓数量由当日成交反推，不需要新表新字段
- 基准优先场内行情最近两个交易日收盘价（与同页 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
187 lines
7.1 KiB
Python
187 lines
7.1 KiB
Python
"""复算并核对「今日盈亏」（只读，不改任何数据）。
|
||
|
||
`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()))
|