"""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