crypto_monitor/lib/account_ledger/account_ledger_normalize.py

"""账户流水:交易所原始记录 → 统一行模型."""
from __future__ import annotations

from typing import Any, Optional


ACCOUNT_FUNDING = "funding"
ACCOUNT_TRADING = "trading"
VALID_ACCOUNTS = frozenset({ACCOUNT_FUNDING, ACCOUNT_TRADING})

PAGE_SIZE = 10


def _safe_float(v: Any) -> Optional[float]:
    if v is None or v == "":
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def _safe_int(v: Any) -> Optional[int]:
    if v is None or v == "":
        return None
    try:
        n = float(v)
        if n > 1e12:
            return int(n)
        if n > 1e9:
            return int(n)
        return int(n)
    except (TypeError, ValueError):
        return None


def _ts_ms(v: Any) -> Optional[int]:
    if v is None or v == "":
        return None
    try:
        n = float(v)
    except (TypeError, ValueError):
        return None
    if n > 1e12:
        return int(n)
    if n > 1e10:
        return int(n)
    return int(n * 1000.0)


def kind_from_raw(raw_type: str, amount: Optional[float] = None) -> str:
    t = (raw_type or "").strip().lower()
    if not t:
        return "other"
    if "deposit" in t or t in ("1", "funding_deposit"):
        return "deposit"
    if "withdraw" in t or "withdrawal" in t:
        return "withdraw"
    if "transfer" in t or "dnw" in t or t in ("2", "18", "19"):
        if amount is not None and amount < 0:
            return "transfer_out"
        if amount is not None and amount > 0:
            return "transfer_in"
        return "transfer"
    if "funding" in t and "fee" in t:
        return "funding_fee"
    if t in ("funding_fee", "fundingfee", "8"):
        return "funding_fee"
    if "commission" in t or "fee" in t or t in ("commission", "5", "fee"):
        return "commission"
    if "realiz" in t or "pnl" in t or t in ("realized_pnl", "realizedpnl", "3"):
        return "realized_pnl"
    if "liqui" in t:
        return "liquidate"
    return "other"


def kind_label_zh(kind: str) -> str:
    return {
        "deposit": "充值",
        "withdraw": "提现",
        "transfer": "划转",
        "transfer_in": "划入",
        "transfer_out": "划出",
        "realized_pnl": "已实现盈亏",
        "funding_fee": "资金费",
        "commission": "手续费",
        "liquidate": "强平",
        "other": "其他",
    }.get((kind or "").strip().lower(), "其他")


def make_ref_id(*parts: Any) -> str:
    bits = []
    for p in parts:
        if p is None:
            continue
        s = str(p).strip()
        if s:
            bits.append(s)
    return "|".join(bits) if bits else ""


def normalize_row(
    *,
    account: str,
    ccy: str,
    amount: Any,
    ts_ms: Any,
    ref_id: str,
    raw_type: str = "",
    balance_after: Any = None,
    symbol: str = "",
    note: str = "",
    kind: str = "",
) -> Optional[dict[str, Any]]:
    acc = (account or "").strip().lower()
    if acc not in VALID_ACCOUNTS:
        return None
    ccy_u = (ccy or "").strip().upper()
    if not ccy_u:
        return None
    amt = _safe_float(amount)
    if amt is None:
        return None
    ts = _ts_ms(ts_ms)
    if ts is None or ts <= 0:
        return None
    rid = (ref_id or "").strip() or make_ref_id(acc, ccy_u, ts, amt, raw_type)
    k = (kind or "").strip().lower() or kind_from_raw(raw_type, amt)
    bal = _safe_float(balance_after)
    return {
        "account": acc,
        "ccy": ccy_u,
        "amount": amt,
        "balance_after": bal,
        "kind": k,
        "kind_label": kind_label_zh(k),
        "raw_type": (raw_type or "").strip()[:120],
        "symbol": (symbol or "").strip()[:80],
        "ref_id": rid[:200],
        "ts_ms": int(ts),
        "note": (note or "").strip()[:240],
    }


def from_ccxt_ledger_entry(entry: dict[str, Any], *, account: 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")
    if amount is None:
        amount = entry.get("change")
    if amount is None:
        amount = info.get("balChg") or info.get("change") or info.get("income") or info.get("amount")
    ts = entry.get("timestamp") or entry.get("datetime")
    if ts is None:
        ts = info.get("time") or info.get("uTime") or info.get("ts") or info.get("create_time") or info.get("createDate")
    ccy = entry.get("currency") or info.get("ccy") or info.get("asset") or info.get("currency") or "USDT"
    raw_type = (
        entry.get("type")
        or entry.get("status")
        or info.get("type")
        or info.get("incomeType")
        or info.get("change_type")
        or info.get("subType")
        or ""
    )
    if isinstance(raw_type, (int, float)):
        raw_type = str(raw_type)
    balance_after = entry.get("balance") or info.get("bal") or info.get("balance")
    symbol = entry.get("symbol") or info.get("instId") or info.get("symbol") or info.get("contract") or ""
    ref = (
        entry.get("id")
        or info.get("billId")
        or info.get("tranId")
        or info.get("id")
        or info.get("trade_id")
        or ""
    )
    note = entry.get("description") or info.get("info") or info.get("text") or ""
    return normalize_row(
        account=account,
        ccy=str(ccy),
        amount=amount,
        ts_ms=ts,
        ref_id=str(ref) if ref != "" else make_ref_id(account, ccy, ts, amount, raw_type),
        raw_type=str(raw_type),
        balance_after=balance_after,
        symbol=str(symbol or ""),
        note=str(note or ""),
    )