CoinOX/lib/hedge_plan/hedge_plan_book_lib.py

"""永续盘口:按平仓方向扫档估 VWAP(持仓净利/出场判定与真平仓同口径)."""
from __future__ import annotations

from typing import Any, Optional


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


def _levels_from_book(book: dict[str, Any] | None, side: str) -> list[tuple[float, float]]:
    """side=bids|asks → [(px, sz_contracts), ...]. ccxt 多为 [px, amount]."""
    if not isinstance(book, dict):
        return []
    raw = book.get(side) or []
    out: list[tuple[float, float]] = []
    for row in raw:
        px = sz = None
        if isinstance(row, (list, tuple)) and len(row) >= 2:
            px, sz = _sf(row[0]), _sf(row[1])
        elif isinstance(row, dict):
            px = _sf(row.get("px") or row.get("price") or row.get(0))
            sz = _sf(row.get("sz") or row.get("size") or row.get("amount") or row.get(1))
        if px is None or px <= 0 or sz is None or sz <= 0:
            continue
        out.append((float(px), float(sz)))
    return out


def walk_book_vwap(
    levels: list[tuple[float, float]],
    *,
    need_contracts: float,
) -> dict[str, Any]:
    """从第一档往下扫 need_contracts 张,返回 VWAP 与流动性结果."""
    need = float(need_contracts or 0)
    if need <= 0:
        return {
            "ok": False,
            "liq_ok": False,
            "msg": "平仓张数无效",
            "vwap": None,
            "top_px": None,
            "top_sz": None,
            "filled_contracts": 0.0,
            "need_contracts": need,
        }
    if not levels:
        return {
            "ok": False,
            "liq_ok": False,
            "msg": "盘口为空",
            "vwap": None,
            "top_px": None,
            "top_sz": None,
            "filled_contracts": 0.0,
            "need_contracts": need,
        }
    top_px, top_sz = levels[0][0], levels[0][1]
    remain = need
    notional = 0.0
    filled = 0.0
    for px, sz in levels:
        take = min(remain, sz)
        if take <= 0:
            continue
        notional += take * px
        filled += take
        remain -= take
        if remain <= 1e-12:
            break
    if filled <= 1e-12:
        return {
            "ok": False,
            "liq_ok": False,
            "msg": "盘口深度为 0",
            "vwap": None,
            "top_px": top_px,
            "top_sz": top_sz,
            "filled_contracts": 0.0,
            "need_contracts": need,
        }
    vwap = notional / filled
    liq_ok = remain <= 1e-12
    msg = "" if liq_ok else f"盘口深度不足(需 {need:g} 张,可扫 {filled:g})"
    return {
        "ok": True,
        "liq_ok": liq_ok,
        "msg": msg,
        "vwap": round(float(vwap), 8),
        "top_px": float(top_px),
        "top_sz": float(top_sz),
        "filled_contracts": round(float(filled), 8),
        "need_contracts": need,
    }


def perp_close_book_side(position_side: str) -> str:
    """持仓方向 → 平仓吃的盘口侧:平多吃 bids,平空吃 asks."""
    return "bids" if str(position_side or "").lower() == "long" else "asks"


def fetch_perp_close_vwap(
    exchange: Any,
    *,
    symbol: str,
    position_side: str,
    contracts: float,
    fallback_px: Optional[float] = None,
    limit: int = 20,
) -> dict[str, Any]:
    """拉永续盘口并按平仓方向扫档估 VWAP.

    返回含: vwap, top_px, top_sz, book_side, liq_ok, msg, filled_contracts.
    深度不足时仍返回已扫部分均价(ok=True,liq_ok=False);完全无盘口则回退 fallback_px.
    """
    sym = str(symbol or "").strip()
    side = str(position_side or "").lower()
    book_side = perp_close_book_side(side)
    need = float(contracts or 0)
    base: dict[str, Any] = {
        "ok": False,
        "liq_ok": False,
        "msg": "",
        "vwap": None,
        "top_px": None,
        "top_sz": None,
        "filled_contracts": 0.0,
        "need_contracts": need,
        "book_side": book_side,
        "symbol": sym,
        "position_side": side,
    }
    if not sym or need <= 0:
        base["msg"] = "符号或张数无效"
        if fallback_px and float(fallback_px) > 0:
            base.update({"ok": True, "vwap": float(fallback_px), "msg": "无盘口,回退标记/开仓价"})
        return base

    book = None
    if exchange is not None and hasattr(exchange, "fetch_order_book"):
        try:
            book = exchange.fetch_order_book(sym, limit=int(limit or 20))
        except Exception as e:
            base["msg"] = f"拉盘口失败: {e}"
            book = None

    levels = _levels_from_book(book, book_side)
    walked = walk_book_vwap(levels, need_contracts=need)
    base.update(walked)
    base["book_side"] = book_side
    if base.get("vwap") is not None:
        return base

    # 回退:ticker 单档
    try:
        if exchange is not None and hasattr(exchange, "fetch_ticker"):
            t = exchange.fetch_ticker(sym) or {}
            if book_side == "bids":
                px = _sf(t.get("bid") or (t.get("info") or {}).get("bidPx"))
            else:
                px = _sf(t.get("ask") or (t.get("info") or {}).get("askPx"))
            if px and px > 0:
                base.update(
                    {
                        "ok": True,
                        "liq_ok": False,
                        "vwap": float(px),
                        "top_px": float(px),
                        "top_sz": None,
                        "msg": "仅单档行情,深度未知",
                        "filled_contracts": need,
                    }
                )
                return base
    except Exception:
        pass

    if fallback_px and float(fallback_px) > 0:
        base.update(
            {
                "ok": True,
                "liq_ok": False,
                "vwap": float(fallback_px),
                "msg": base.get("msg") or "无盘口,回退标记/开仓价",
            }
        )
    return base