|
|
@@ -332,6 +332,7 @@ async def read_rows(gateway: QueryGateway, request: ReadRequest) -> ReadResult:
|
|
|
trs_dataset_id = await _trs_available_dataset_id(gateway, request, missing)
|
|
|
await _load_user_ledgers(gateway, request, trs_users, accounts, user_ids, trs_dataset_id, missing)
|
|
|
await _load_fund_in_out(gateway, request, trs_users, accounts, user_ids, missing)
|
|
|
+ await _load_interest_collect(gateway, request, trs_users, accounts, user_ids, missing)
|
|
|
await _load_interest_payable(gateway, trs_users, accounts, user_ids, missing)
|
|
|
await _load_cleared_positions(gateway, trs_users, accounts, user_ids, trs_dataset_id, missing)
|
|
|
# 步骤节点另读 cleared 镜像和封存快照。TRS006 用的合并流水保持原样。
|
|
|
@@ -1704,8 +1705,18 @@ async def _load_user_ledgers(
|
|
|
_add_unique(_ensure_trs(users, account).setdefault("fund_ledger", []), parsed)
|
|
|
|
|
|
|
|
|
-# TRS009 按主键或 source_no 对出入金。列与 db_trs.fund_in_out_record 一致。
|
|
|
-_IN_OUT_COLUMNS = ("id", "user_id", "source_no", "direction", "amount", "business_date")
|
|
|
+# TRS009:报表 ref 对入金/出金单 id(source_id),兼容流水主键与 source_no。
|
|
|
+_IN_OUT_COLUMNS = ("id", "user_id", "source_id", "source_no", "direction", "amount", "business_date")
|
|
|
+# TRS007:利息收取事实按用户挂桶,与报表 period_start/period_end/settle_ccy 对齐。
|
|
|
+_INTEREST_COLLECT_COLUMNS = (
|
|
|
+ "id",
|
|
|
+ "user_id",
|
|
|
+ "settle_ccy",
|
|
|
+ "period_start",
|
|
|
+ "period_end",
|
|
|
+ "amount",
|
|
|
+ "business_date",
|
|
|
+)
|
|
|
|
|
|
|
|
|
async def _load_fund_in_out(
|
|
|
@@ -1742,6 +1753,40 @@ async def _load_fund_in_out(
|
|
|
_ensure_trs(users, account).setdefault("fund_in_out_record", []).append(dict(row))
|
|
|
|
|
|
|
|
|
+async def _load_interest_collect(
|
|
|
+ gateway: QueryGateway,
|
|
|
+ request: ReadRequest,
|
|
|
+ users: dict[str, dict[str, Any]],
|
|
|
+ accounts: Mapping[int, str],
|
|
|
+ user_ids: Sequence[int],
|
|
|
+ missing: list[str],
|
|
|
+) -> None:
|
|
|
+ """按营业日挂利息收取事实,供 TRS007 与报表对齐。"""
|
|
|
+
|
|
|
+ if not user_ids:
|
|
|
+ return
|
|
|
+ rows = await _select_in(
|
|
|
+ gateway,
|
|
|
+ "db_trs.fund_interest_collect",
|
|
|
+ _INTEREST_COLLECT_COLUMNS,
|
|
|
+ "business_date = %s",
|
|
|
+ (request.business_date,),
|
|
|
+ "user_id",
|
|
|
+ user_ids,
|
|
|
+ missing,
|
|
|
+ )
|
|
|
+ if "db_trs.fund_interest_collect" in missing:
|
|
|
+ return
|
|
|
+ wanted = {accounts[ident] for ident in user_ids if ident in accounts}
|
|
|
+ for account in wanted:
|
|
|
+ _ensure_trs(users, account).setdefault("fund_interest_collect", [])
|
|
|
+ for row in rows:
|
|
|
+ account = accounts.get(_int_id(row.get("user_id")) or -1)
|
|
|
+ if account is None:
|
|
|
+ continue
|
|
|
+ _ensure_trs(users, account).setdefault("fund_interest_collect", []).append(dict(row))
|
|
|
+
|
|
|
+
|
|
|
# TRS013 要知道未收取的应付利息有没有报表行。金额另按报表四列计算。
|
|
|
_PAYABLE_COLUMNS = (
|
|
|
"id",
|
|
|
@@ -2313,12 +2358,13 @@ async def _load_note_chain_rows(
|
|
|
if dataset_id is None:
|
|
|
return empty
|
|
|
input_where, input_args = with_scope("", (), dataset_id, "input", 0)
|
|
|
+ # 增额累计要把同日 sell_release 计入;主键集合仍只看 top_up。
|
|
|
topups = await _select(
|
|
|
gateway,
|
|
|
TOPUP_ARCHIVE_TABLE,
|
|
|
TOPUP_COLUMNS,
|
|
|
- f"{input_where} AND change_type = %s",
|
|
|
- input_args + ("top_up",),
|
|
|
+ f"{input_where} AND change_type IN (%s, %s)",
|
|
|
+ input_args + ("top_up", "sell_release"),
|
|
|
missing,
|
|
|
)
|
|
|
extensions = await _select(
|
|
|
@@ -2402,8 +2448,8 @@ async def _load_note_chain_rows(
|
|
|
gateway,
|
|
|
TOPUP_LIVE_TABLE,
|
|
|
TOPUP_COLUMNS,
|
|
|
- "change_type = %s AND business_date >= %s",
|
|
|
- ("top_up", request.business_date),
|
|
|
+ "change_type IN (%s, %s) AND business_date >= %s",
|
|
|
+ ("top_up", "sell_release", request.business_date),
|
|
|
"note_no",
|
|
|
note_nos,
|
|
|
missing,
|
|
|
@@ -3225,6 +3271,8 @@ def _freeze_user(bucket: Mapping[str, Any]) -> dict[str, Any]:
|
|
|
frozen["clr_user_position"] = tuple(bucket["clr_user_position"])
|
|
|
if "fund_in_out_record" in bucket:
|
|
|
frozen["fund_in_out_record"] = tuple(dict(row) for row in bucket["fund_in_out_record"])
|
|
|
+ if "fund_interest_collect" in bucket:
|
|
|
+ frozen["fund_interest_collect"] = tuple(dict(row) for row in bucket["fund_interest_collect"])
|
|
|
if "fund_interest_payable" in bucket:
|
|
|
frozen["fund_interest_payable"] = tuple(dict(row) for row in bucket["fund_interest_payable"])
|
|
|
for key in ("clr_trade_order", "clr_trade_fill", "clr_fund_account", "clr_fx_user"):
|