crypto_monitor/lib/options/options_positions_lib.py

"""期权持仓展示(实例页 / 中控快照共用)."""
from __future__ import annotations

from typing import Any

from lib.options.options_db import init_options_tables, sum_open_premium_paid
from lib.options.options_history_lib import enrich_position_row_display
from lib.options.options_close_gate_lib import clear_close_gate, is_close_gate_passed, update_close_gate
from lib.options.options_pricing_lib import estimate_close_by_bids, intrinsic_px_per_unit


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 attach_close_preview(
    cfg: dict[str, Any],
    ex: Any,
    row: dict[str, Any],
    *,
    sheets: int | None = None,
    premium_paid: float | None = None,
) -> dict[str, Any]:
    inst_id = str(row.get("inst_id") or row.get("instId") or "").strip()
    if not inst_id:
        return row
    ct_mult = float(row.get("ct_mult") or 0.01)
    target_sheets = int(sheets) if sheets is not None else int(abs(_safe_float(row.get("pos")) or 0))
    paid = premium_paid if premium_paid is not None else _safe_float(row.get("premium_paid"))
    book = cfg["fetch_option_book_depth"](ex, inst_id, 5)
    row["bid_depth"] = book.get("bids") or []
    row["ask_depth"] = book.get("asks") or []
    mark_px = _safe_float(row.get("mark_px") or row.get("markPx"))
    intrinsic = intrinsic_px_per_unit(
        row.get("opt_type") or row.get("optType"),
        _safe_float(row.get("strike") or row.get("stk")),
        _safe_float(row.get("idx_px") or row.get("idxPx")),
    )
    # 与实盘一致:只按买一估算本轮可平
    preview = estimate_close_by_bids(
        row["bid_depth"],
        target_sheets,
        ct_mult=ct_mult,
        premium_paid=paid,
        mark_px=mark_px,
        intrinsic_px=intrinsic,
        max_levels=1,
    )
    # 残档时不累计 2×门控;有效买一时刷新计时(仅自动平仓需要)
    if preview.get("bid_invalid") or preview.get("auto_close_blocked"):
        gate = update_close_gate(inst_id, recycle_usdc=None, premium_paid=paid)
        preview["close_gate"] = gate
        preview["close_gate_blocked"] = True
        preview["close_gate_msg"] = preview.get("bid_invalid_reason") or gate.get("msg")
        preview["manual_close_blocked"] = True
        preview["liquidity_ok"] = False
    else:
        gate = update_close_gate(
            inst_id,
            recycle_usdc=_safe_float(preview.get("total_received")),
            premium_paid=paid,
        )
        passed = bool(gate.get("passed") or is_close_gate_passed(inst_id) or gate.get("ready"))
        preview["close_gate"] = gate
        preview["close_gate_blocked"] = not passed
        preview["close_gate_msg"] = gate.get("msg")
        preview["manual_close_blocked"] = False
        preview["liquidity_ok"] = True
        if not passed:
            preview["auto_close_blocked"] = True
    row["close_preview"] = preview
    return row


def forget_close_gate_for_inst(inst_id: str) -> None:
    clear_close_gate(inst_id)


def net_pnl_from_display_row(row: dict[str, Any]) -> float | None:
    """与持仓卡「净盈亏」同口径:买一可回收 − 权利金;残档买一则无净值."""
    preview = row.get("close_preview") if isinstance(row.get("close_preview"), dict) else {}
    if preview.get("bid_invalid"):
        return None
    net = preview.get("estimated_pnl")
    if net is not None:
        try:
            return float(net)
        except (TypeError, ValueError):
            pass
    recv = _safe_float(preview.get("total_received"))
    paid = _safe_float(row.get("premium_paid"))
    if recv is not None and paid is not None:
        return round(recv - paid, 4)
    return None


def sum_options_net_pnl_usdc(
    cfg: dict[str, Any],
    ex: Any,
    raw_positions: list[dict[str, Any]] | None = None,
) -> float | None:
    """
    期权浮盈合计(USDC),与顶栏实时盈亏/中控口径对齐为「净盈亏」:
    各仓买一可回收 − 权利金之和.获取失败返回 None;无持仓返回 0.
    """
    raw = raw_positions
    if raw is None:
        raw = cfg["fetch_option_positions"](ex)
    if raw is None:
        return None
    if not raw:
        return 0.0
    positions = build_display_option_positions(cfg, ex, raw)
    total = 0.0
    found = False
    for p in positions:
        net = net_pnl_from_display_row(p)
        if net is None:
            continue
        found = True
        total += float(net)
    return round(total, 4) if found else (0.0 if not positions else None)


def build_display_option_positions(
    cfg: dict[str, Any],
    ex: Any,
    raw_positions: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """与实例 /api/options/positions 相同 enrichment + close_preview."""
    meta_cache: dict[str, dict[str, Any] | None] = {}
    rows: list[dict[str, Any]] = []
    conn = cfg["get_db"]()
    try:
        init_options_tables(conn)
        for p in raw_positions:
            inst = str(p.get("instId") or "").strip()
            premium_override = sum_open_premium_paid(conn, inst) if inst else None
            row = enrich_position_row_display(
                cfg,
                ex,
                p,
                meta_cache=meta_cache,
                premium_override=premium_override,
            )
            attach_close_preview(cfg, ex, row, premium_paid=_safe_float(row.get("premium_paid")))
            rows.append(row)
    finally:
        conn.close()
    return rows