crypto_monitor/lib/exchange/binance_ledger_lib.py

"""Binance:交易账户 futures income;资金账户 deposits/withdrawals/transfers.USDT."""
from __future__ import annotations

from typing import Any, Callable, Optional

from lib.account_ledger.account_ledger_normalize import (
    ACCOUNT_FUNDING,
    ACCOUNT_TRADING,
    from_ccxt_ledger_entry,
    kind_from_raw,
    make_ref_id,
    normalize_row,
)


def _paginate_income(exchange, *, start_ms: int, end_ms: int, max_pages: int = 15) -> list[dict]:
    out: list[dict] = []
    cursor = int(start_ms)
    end = int(end_ms)
    for _ in range(max_pages):
        try:
            if hasattr(exchange, "fapiPrivateGetIncome"):
                batch = exchange.fapiPrivateGetIncome(
                    {"startTime": cursor, "endTime": end, "limit": 1000}
                )
            else:
                batch = exchange.fetch_ledger(
                    "USDT", cursor, 1000, {"type": "swap", "until": end}
                )
                # already unified
                return batch or []
        except Exception:
            break
        if not batch:
            break
        out.extend(batch)
        if len(batch) < 1000:
            break
        last_t = batch[-1].get("time") or batch[-1].get("timestamp")
        try:
            last_i = int(float(last_t))
        except Exception:
            break
        if last_i >= end:
            break
        cursor = last_i + 1
    return out


def _income_to_row(raw: dict) -> Optional[dict[str, Any]]:
    if not isinstance(raw, dict):
        return None
    # raw fapi income
    if "income" in raw or "incomeType" in raw:
        amt = raw.get("income")
        ts = raw.get("time")
        ccy = raw.get("asset") or "USDT"
        raw_type = str(raw.get("incomeType") or "")
        ref = str(raw.get("tranId") or raw.get("tradeId") or "")
        return normalize_row(
            account=ACCOUNT_TRADING,
            ccy=str(ccy),
            amount=amt,
            ts_ms=ts,
            ref_id=ref or make_ref_id("trading", ccy, ts, amt, raw_type),
            raw_type=raw_type,
            symbol=str(raw.get("symbol") or ""),
            note=str(raw.get("info") or ""),
            kind=kind_from_raw(raw_type, float(amt) if amt is not None else None),
        )
    return from_ccxt_ledger_entry(raw, account=ACCOUNT_TRADING)


def _dep_wd_to_row(entry: dict, *, kind: str) -> Optional[dict[str, Any]]:
    if not isinstance(entry, dict):
        return None
    info = entry.get("info") if isinstance(entry.get("info"), dict) else {}
    amount = entry.get("amount")
    ts = entry.get("timestamp") or info.get("insertTime") or info.get("applyTime")
    ccy = entry.get("currency") or info.get("coin") or "USDT"
    status = entry.get("status") or info.get("status") or ""
    ref = str(entry.get("id") or info.get("txId") or info.get("id") or "")
    amt = amount
    try:
        af = float(amount)
        if kind == "withdraw" and af > 0:
            af = -af
            amt = af
    except Exception:
        pass
    return normalize_row(
        account=ACCOUNT_FUNDING,
        ccy=str(ccy),
        amount=amt,
        ts_ms=ts,
        ref_id=ref or make_ref_id("funding", kind, ccy, ts, amount),
        raw_type=kind,
        note=str(status),
        kind=kind,
    )


def _transfer_to_row(entry: dict) -> Optional[dict[str, Any]]:
    if not isinstance(entry, dict):
        return None
    info = entry.get("info") if isinstance(entry.get("info"), dict) else {}
    amount = entry.get("amount")
    ts = entry.get("timestamp") or info.get("timestamp")
    ccy = entry.get("currency") or info.get("asset") or "USDT"
    ref = str(entry.get("id") or info.get("tranId") or info.get("id") or "")
    frm = str(entry.get("fromAccount") or info.get("from") or "")
    to = str(entry.get("toAccount") or info.get("to") or "")
    try:
        amt = float(amount)
    except Exception:
        return None
    # 资金侧视角:从资金转出为负,转入为正(粗分)
    note = f"{frm}->{to}".strip("->")
    raw_type = "transfer"
    return normalize_row(
        account=ACCOUNT_FUNDING,
        ccy=str(ccy),
        amount=amt,
        ts_ms=ts,
        ref_id=ref or make_ref_id("funding", "transfer", ccy, ts, amt),
        raw_type=raw_type,
        note=note,
        kind=kind_from_raw("transfer", amt),
    )


def fetch_binance_account_ledger(
    exchange,
    *,
    start_ms: int,
    end_ms: int,
    ensure_markets: Optional[Callable[[], None]] = None,
) -> tuple[list[dict[str, Any]], list[str]]:
    errors: list[str] = []
    rows: list[dict[str, Any]] = []
    if ensure_markets:
        try:
            ensure_markets()
        except Exception as e:
            errors.append(f"markets:{e}")

    # 交易账户
    try:
        raw = _paginate_income(exchange, start_ms=start_ms, end_ms=end_ms)
        for e in raw:
            n = _income_to_row(e)
            if n and n["ccy"] == "USDT":
                rows.append(n)
    except Exception as e:
        errors.append(f"trading:{e}")

    # 资金账户:充提 + 划转
    for label, fn, kind in (
        ("deposits", "fetch_deposits", "deposit"),
        ("withdrawals", "fetch_withdrawals", "withdraw"),
    ):
        try:
            meth = getattr(exchange, fn, None)
            if not callable(meth):
                continue
            batch = meth("USDT", int(start_ms), 1000, {"until": int(end_ms)}) or []
            for e in batch:
                n = _dep_wd_to_row(e, kind=kind)
                if n:
                    rows.append(n)
        except Exception as e:
            errors.append(f"{label}:{e}")

    try:
        if hasattr(exchange, "fetch_transfers"):
            batch = (
                exchange.fetch_transfers("USDT", int(start_ms), 1000, {"until": int(end_ms)})
                or []
            )
            for e in batch:
                n = _transfer_to_row(e)
                if n:
                    rows.append(n)
    except Exception as e:
        errors.append(f"transfers:{e}")

    return rows, errors