crypto_okx/lib/exchange/okx_ledger_lib.py

"""OKX:资金账户 asset bills + 交易账户 account bills;USDT + USDC."""
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,
)

OKX_LEDGER_CCYS = ("USDT", "USDC")


def _fetch_one(
    exchange,
    *,
    code: str,
    since: int,
    until: int,
    method: str,
    max_pages: int = 10,
) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    after = None
    for _ in range(max_pages):
        params: dict[str, Any] = {"method": method, "until": int(until)}
        if after:
            params["after"] = after
        try:
            batch = exchange.fetch_ledger(code, int(since), 100, params) or []
        except Exception:
            # archive / bills 窗口差异:失败则停
            break
        if not batch:
            break
        out.extend(batch)
        if len(batch) < 100:
            break
        # OKX 翻页用 billId
        last = batch[-1]
        info = last.get("info") if isinstance(last.get("info"), dict) else {}
        bid = last.get("id") or info.get("billId")
        if not bid:
            break
        after = str(bid)
    return out


def fetch_okx_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}")

    for ccy in OKX_LEDGER_CCYS:
        # 资金账户
        try:
            raw = _fetch_one(
                exchange,
                code=ccy,
                since=start_ms,
                until=end_ms,
                method="privateGetAssetBills",
            )
            for e in raw:
                n = from_ccxt_ledger_entry(e, account=ACCOUNT_FUNDING)
                if n:
                    rows.append(n)
        except Exception as e:
            errors.append(f"funding:{ccy}:{e}")

        # 交易账户:近 3 月 archive + 近 7 日 bills(去重靠 upsert)
        for method in ("privateGetAccountBillsArchive", "privateGetAccountBills"):
            try:
                raw = _fetch_one(
                    exchange,
                    code=ccy,
                    since=start_ms,
                    until=end_ms,
                    method=method,
                )
                for e in raw:
                    n = from_ccxt_ledger_entry(e, account=ACCOUNT_TRADING)
                    if n:
                        rows.append(n)
            except Exception as e:
                errors.append(f"trading:{ccy}:{method}:{e}")

    return rows, errors