crypto_okx/lib/options/options_pnl_settle_lib.py

"""已实现盈亏结算:扣手续费,并在平仓时按指数冻结 U 口径."""
from __future__ import annotations

from typing import Any


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


def fee_amount_abs(fee: Any) -> float:
    """OKX 扣费多为负;统一为正数「已付手续费」."""
    f = _safe_float(fee)
    if f is None:
        return 0.0
    return abs(float(f))


def fee_to_premium_ccy(
    fee_amt: Any,
    fee_ccy: Any,
    *,
    premium_ccy: str,
    index_px: float | None,
) -> float:
    """把手续费折成权利金币种数量(与 premium_paid/received 同单位)."""
    amt = fee_amount_abs(fee_amt)
    if amt <= 0:
        return 0.0
    fc = str(fee_ccy or "").strip().upper()
    pc = str(premium_ccy or "").strip().upper()
    if not fc or not pc or fc == pc:
        return round(amt, 8)
    # 稳定币互兑近似 1:1
    stables = {"USDT", "USDC", "USD"}
    if fc in stables and pc in stables:
        return round(amt, 8)
    idx = _safe_float(index_px)
    if idx is None or idx <= 0:
        return 0.0
    if fc in stables and pc in ("ETH", "BTC"):
        return round(amt / float(idx), 8)
    if fc in ("ETH", "BTC") and pc in stables:
        return round(amt * float(idx), 8)
    if fc in ("ETH", "BTC") and pc in ("ETH", "BTC") and fc != pc:
        # 跨币种无可靠汇率时放弃折算,避免脏数据
        return 0.0
    return round(amt, 8)


def extract_order_fee(order: dict[str, Any] | None) -> tuple[float, str]:
    """从下单/查单结果取手续费(原始币种,正数)."""
    if not isinstance(order, dict):
        return 0.0, ""
    raw = order.get("raw") if isinstance(order.get("raw"), dict) else {}
    info = order.get("info") if isinstance(order.get("info"), dict) else {}
    fee = order.get("fee")
    if fee is None:
        fee = raw.get("fee")
    if fee is None:
        fee = info.get("fee")
    ccy = (
        order.get("fee_ccy")
        or raw.get("feeCcy")
        or raw.get("fee_ccy")
        or info.get("fee_ccy")
        or info.get("feeCcy")
        or ""
    )
    return fee_amount_abs(fee), str(ccy or "").strip().upper()


def sum_option_order_fees_in_premium(
    ex: Any,
    *,
    inst_id: str,
    ord_ids: list[str] | str | None,
    premium_ccy: str,
    index_px: float | None,
    fetch_order_fn: Any = None,
) -> float:
    """按订单号汇总手续费并折成权利金单位."""
    if isinstance(ord_ids, str):
        ids = [p.strip() for p in ord_ids.split(",") if p.strip()]
    else:
        ids = [str(x).strip() for x in (ord_ids or []) if str(x).strip()]
    if not ids:
        return 0.0
    if fetch_order_fn is None:
        from lib.exchange.okx_options_lib import fetch_option_order

        fetch_order_fn = fetch_option_order
    total = 0.0
    seen: set[str] = set()
    for oid in ids:
        if oid in seen:
            continue
        seen.add(oid)
        try:
            info = fetch_order_fn(ex, inst_id=inst_id, ord_id=oid)
        except Exception:
            continue
        if not isinstance(info, dict) or not info.get("ok"):
            continue
        fee_amt, fee_ccy = extract_order_fee(info)
        total += fee_to_premium_ccy(
            fee_amt, fee_ccy, premium_ccy=premium_ccy, index_px=index_px
        )
    return round(total, 8)


def settle_closed_pnl(
    *,
    premium_paid: Any,
    premium_received: Any,
    open_fee: Any = 0,
    close_fee: Any = 0,
    premium_ccy: str,
    index_px: float | None,
) -> dict[str, float | None]:
    """
    净盈亏 = 收回 − 已付 − 开/平手续费(权利金单位).
    币本位再乘平仓指数得到冻结 U;USDC/USDT 直接冻结.
    """
    paid = float(_safe_float(premium_paid) or 0.0)
    recv = float(_safe_float(premium_received) or 0.0)
    ofee = float(_safe_float(open_fee) or 0.0)
    cfee = float(_safe_float(close_fee) or 0.0)
    if ofee < 0:
        ofee = abs(ofee)
    if cfee < 0:
        cfee = abs(cfee)
    net = round(recv - paid - ofee - cfee, 8)
    pc = str(premium_ccy or "").strip().upper()
    idx = _safe_float(index_px)
    if pc in ("ETH", "BTC"):
        if idx is None or idx <= 0:
            return {
                "realized_pnl": net,
                "close_index_px": None,
                "realized_pnl_usdt": None,
                "total_fee": round(ofee + cfee, 8),
            }
        return {
            "realized_pnl": net,
            "close_index_px": float(idx),
            "realized_pnl_usdt": round(net * float(idx), 4),
            "total_fee": round(ofee + cfee, 8),
        }
    # USDC / USDT 权利金
    return {
        "realized_pnl": net,
        "close_index_px": float(idx) if idx is not None and idx > 0 else None,
        "realized_pnl_usdt": round(net, 4),
        "total_fee": round(ofee + cfee, 8),
    }


def resolve_close_index_px(
    ex: Any,
    *,
    inst_id: str = "",
    underlying: str = "",
    hint: Any = None,
) -> float | None:
    """平仓冻结用指数:优先调用方传入,否则拉标的近似价."""
    h = _safe_float(hint)
    if h is not None and h > 0:
        return float(h)
    underly = (underlying or "").strip().upper()
    if not underly and inst_id:
        underly = str(inst_id).split("-")[0].strip().upper() or "ETH"
    if not underly:
        underly = "ETH"
    try:
        from lib.options.options_stats_lib import _underlying_index_usdt

        return _underlying_index_usdt(ex, underly)
    except Exception:
        return None