crypto_monitor/hub_position_metrics.py

"""ccxt 持仓标记价解析（实例 price_snapshot 与中控子代理共用）。"""
from __future__ import annotations

import math
from typing import Any, Callable


def _finite_or_none(x: Any) -> float | None:
    try:
        f = float(x)
        return f if math.isfinite(f) else None
    except (TypeError, ValueError):
        return None


def _coerce_float(*values: Any) -> float | None:
    for v in values:
        if v is None or v == "":
            continue
        px = _finite_or_none(v)
        if px is not None and px > 0:
            return px
    return None


def position_contracts(p: dict[str, Any]) -> float:
    raw = p.get("contracts")
    if raw is not None:
        try:
            return float(raw)
        except (TypeError, ValueError):
            pass
    info = p.get("info") or {}
    if not isinstance(info, dict):
        info = {}
    for k in ("positionAmt", "positionamt", "pos", "size"):
        if k in info:
            try:
                v = float(info[k])
                if v != 0:
                    return v
            except (TypeError, ValueError):
                pass
    return 0.0


def position_side_from_ccxt(p: dict[str, Any], contracts: float | None = None) -> str:
    s = (p.get("side") or "").lower()
    if s in ("long", "short"):
        return s
    c = contracts if contracts is not None else position_contracts(p)
    if c > 0:
        return "long"
    if c < 0:
        return "short"
    return "long"


def parse_position_unrealized_pnl(p: dict[str, Any]) -> float | None:
    """四所 ccxt 持仓统一解析未实现盈亏（Gate 常在 info.unrealised_pnl）。"""
    if not isinstance(p, dict):
        return None
    info = p.get("info") or {}
    if not isinstance(info, dict):
        info = {}
    for key in (
        p.get("unrealizedPnl"),
        p.get("unrealisedPnl"),
        p.get("unrealized_pnl"),
        p.get("unrealised_pnl"),
        info.get("unrealised_pnl"),
        info.get("unrealized_pnl"),
        info.get("unrealisedPnl"),
        info.get("unrealizedPnl"),
    ):
        px = _finite_or_none(key)
        if px is not None:
            return px
    return None


def parse_position_mark_price(p: dict[str, Any]) -> float | None:
    """四所 ccxt 持仓统一解析标记价（与 crypto_monitor_* parse_ccxt_position_metrics 口径一致）。"""
    if not isinstance(p, dict):
        return None
    info = p.get("info") or {}
    if not isinstance(info, dict):
        info = {}
    mark = _coerce_float(
        p.get("markPrice"),
        p.get("mark_price"),
        p.get("mark"),
        info.get("markPx"),
        info.get("mark_price"),
        info.get("markPrice"),
    )
    if mark is not None:
        return mark
    contracts = position_contracts(p)
    if abs(contracts) >= 1e-12:
        notional = _finite_or_none(p.get("notional"))
        if notional is not None and abs(notional) > 0:
            return abs(notional) / abs(contracts)
    return None


def build_position_marks_list(
    positions: list,
    *,
    format_mark_display: Callable[[str, float], str] | None = None,
) -> list[dict[str, Any]]:
    """从 fetch_positions 结果生成 position_marks，供 price_snapshot / 中控合并。"""
    out: list[dict[str, Any]] = []
    for p in positions or []:
        if not isinstance(p, dict):
            continue
        c = position_contracts(p)
        if abs(c) < 1e-12:
            continue
        mark = parse_position_mark_price(p)
        if mark is None or mark <= 0:
            continue
        sym = (p.get("symbol") or "").strip()
        side = position_side_from_ccxt(p, c)
        row: dict[str, Any] = {
            "symbol": sym,
            "side": side,
            "mark_price": mark,
        }
        if format_mark_display and sym:
            try:
                row["mark_price_display"] = format_mark_display(sym, mark)
            except Exception:
                row["mark_price_display"] = f"{mark:g}"
        else:
            row["mark_price_display"] = f"{mark:g}"
        out.append(row)
    return out