dekun
102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""账户流水归一化与分页查询."""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import unittest
|
|
|
|
from lib.account_ledger.account_ledger_db import (
|
|
ensure_account_ledger_tables,
|
|
query_entries,
|
|
upsert_entries,
|
|
)
|
|
from lib.account_ledger.account_ledger_normalize import (
|
|
ACCOUNT_FUNDING,
|
|
ACCOUNT_TRADING,
|
|
PAGE_SIZE,
|
|
from_ccxt_ledger_entry,
|
|
kind_from_raw,
|
|
normalize_row,
|
|
)
|
|
|
|
|
|
class TestAccountLedgerNormalize(unittest.TestCase):
|
|
def test_kind_from_raw(self):
|
|
self.assertEqual(kind_from_raw("REALIZED_PNL", 1.0), "realized_pnl")
|
|
self.assertEqual(kind_from_raw("FUNDING_FEE", -0.1), "funding_fee")
|
|
self.assertEqual(kind_from_raw("TRANSFER", 5.0), "transfer_in")
|
|
self.assertEqual(kind_from_raw("TRANSFER", -5.0), "transfer_out")
|
|
|
|
def test_normalize_row_and_ccxt(self):
|
|
row = normalize_row(
|
|
account=ACCOUNT_TRADING,
|
|
ccy="usdt",
|
|
amount="-1.25",
|
|
ts_ms=1_700_000_000_000,
|
|
ref_id="abc",
|
|
raw_type="COMMISSION",
|
|
)
|
|
self.assertIsNotNone(row)
|
|
assert row is not None
|
|
self.assertEqual(row["ccy"], "USDT")
|
|
self.assertEqual(row["kind"], "commission")
|
|
self.assertEqual(row["account"], "trading")
|
|
|
|
entry = {
|
|
"id": "9",
|
|
"currency": "USDC",
|
|
"amount": 2.5,
|
|
"timestamp": 1_700_000_000_000,
|
|
"type": "transfer",
|
|
"info": {"billId": "9", "bal": "10"},
|
|
}
|
|
n = from_ccxt_ledger_entry(entry, account=ACCOUNT_FUNDING)
|
|
self.assertIsNotNone(n)
|
|
assert n is not None
|
|
self.assertEqual(n["ccy"], "USDC")
|
|
self.assertEqual(n["balance_after"], 10.0)
|
|
|
|
def test_upsert_and_paginate(self):
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.row_factory = sqlite3.Row
|
|
ensure_account_ledger_tables(conn)
|
|
rows = []
|
|
for i in range(25):
|
|
rows.append(
|
|
normalize_row(
|
|
account=ACCOUNT_FUNDING,
|
|
ccy="USDT",
|
|
amount=1.0 + i,
|
|
ts_ms=1_700_000_000_000 + i * 1000,
|
|
ref_id=f"r{i}",
|
|
raw_type="deposit",
|
|
)
|
|
)
|
|
self.assertEqual(upsert_entries(conn, [r for r in rows if r]), 25)
|
|
page1 = query_entries(
|
|
conn,
|
|
account=ACCOUNT_FUNDING,
|
|
start_ms=0,
|
|
end_ms=9_000_000_000_000,
|
|
page=1,
|
|
page_size=PAGE_SIZE,
|
|
)
|
|
self.assertEqual(page1["total"], 25)
|
|
self.assertEqual(len(page1["items"]), 10)
|
|
self.assertEqual(page1["pages"], 3)
|
|
# 倒序:最新在前
|
|
self.assertGreater(page1["items"][0]["ts_ms"], page1["items"][1]["ts_ms"])
|
|
page3 = query_entries(
|
|
conn,
|
|
account=ACCOUNT_FUNDING,
|
|
start_ms=0,
|
|
end_ms=9_000_000_000_000,
|
|
page=3,
|
|
page_size=PAGE_SIZE,
|
|
)
|
|
self.assertEqual(len(page3["items"]), 5)
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|