crypto_monitor/lib/hedge_plan/hedge_plan_settle_lib.py

"""对冲计划结算辅助:到期内在价值与期权腿收口."""
from __future__ import annotations

import os
import time
from datetime import datetime
from typing import Any, Callable, Optional
from zoneinfo import ZoneInfo

from lib.exchange.okx_options_lib import normalize_option_exp_ms, resolve_option_close_from_history
from lib.hedge_plan.hedge_plan_calc_lib import option_expiry_pnl

_APP_TZ = ZoneInfo((os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai")


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


def leg_exp_ms(leg: dict[str, Any]) -> Optional[int]:
    return normalize_option_exp_ms(leg.get("exp_time"), str(leg.get("inst_id") or ""))


def leg_is_expired(leg: dict[str, Any], *, now_ms: Optional[int] = None) -> bool:
    exp = leg_exp_ms(leg)
    if exp is None:
        return False
    now = int(now_ms if now_ms is not None else time.time() * 1000)
    return now >= int(exp)


def settle_option_leg_at_spot(leg: dict[str, Any], spot: float) -> float:
    """按到期结算口径估算腿盈亏(USDC)."""
    premium = float(leg.get("premium") or 0)
    strike = _sf(leg.get("strike"))
    if strike is None:
        return -premium
    sheets = float(leg.get("size") or 1)
    # ct_mult 未入库时默认 0.01
    ct = float(leg.get("ct_mult") or 0.01)
    return float(
        option_expiry_pnl(
            opt_type=str(leg.get("opt_type") or "P"),
            strike=float(strike),
            spot=float(spot),
            sheets=sheets,
            ct_mult=ct,
            premium_paid=premium,
        )
    )


def all_option_legs_expired(legs: list[dict[str, Any]], *, now_ms: Optional[int] = None) -> bool:
    opts = [
        x
        for x in legs
        if str(x.get("leg_role") or "").startswith("option")
        and str(x.get("status") or "") in ("open", "hold_to_expiry")
    ]
    if not opts:
        return False
    return all(leg_is_expired(x, now_ms=now_ms) for x in opts)


def _parse_opened_ms(raw: Any) -> Optional[int]:
    """墙钟开仓时间 → UTC ms.库内时间为业务时区(默认 Asia/Shanghai),不可当 UTC."""
    if raw is None or raw == "":
        return None
    s = str(raw).strip()
    if not s:
        return None
    for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M:%f", 26), ("%Y-%m-%d %H:%M", 16)):
        try:
            dt = datetime.strptime(s[:ln], fmt).replace(tzinfo=_APP_TZ)
            return int(dt.timestamp() * 1000)
        except ValueError:
            continue
    return None


def resolve_option_leg_realized_pnl(
    *,
    ex: Any = None,
    leg: dict[str, Any],
    fallback: Optional[float] = None,
    fetch_history_fn: Optional[Callable[[str], list[dict[str, Any]]]] = None,
    hist_rows: Optional[list[dict[str, Any]]] = None,
) -> tuple[Optional[float], str]:
    """
    期权腿已实现盈亏:优先 OKX positions-history realizedPnl.
    返回 (pnl, source) source=exchange|fallback|none.
    """
    inst_id = str(leg.get("inst_id") or "").strip()
    open_ms = _parse_opened_ms(leg.get("opened_at"))
    rows = hist_rows
    if rows is None and inst_id:
        try:
            if callable(fetch_history_fn):
                rows = fetch_history_fn(inst_id)
            elif ex is not None:
                from lib.exchange.okx_options_lib import fetch_option_position_history

                rows = fetch_option_position_history(ex, inst_id)
        except Exception:
            rows = None
    if rows:
        close_ms = _parse_opened_ms(leg.get("closed_at"))
        sheets = _sf(leg.get("size")) or _sf(leg.get("sheets"))
        info = resolve_option_close_from_history(
            rows, open_ms=open_ms, close_ms=close_ms, sheets=sheets
        )
        pnl = _sf((info or {}).get("realized_pnl")) if info else None
        if pnl is not None:
            return round(float(pnl), 4), "exchange"
    if fallback is not None:
        return round(float(fallback), 4), "fallback"
    return None, "none"


def backfill_hedge_option_legs_realized_pnl(
    conn: Any,
    hist_rows: list[dict[str, Any]],
    *,
    update_plan_fn: Optional[Callable[..., Any]] = None,
) -> dict[str, int]:
    """用交易所历史覆盖已平期权腿盈亏,并重算已结束计划合计."""
    from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, update_plan

    by_inst: dict[str, list[dict[str, Any]]] = {}
    for raw in hist_rows or []:
        if not isinstance(raw, dict):
            continue
        inst = str(raw.get("instId") or "").strip()
        if inst:
            by_inst.setdefault(inst, []).append(raw)

    legs = conn.execute(
        """
        SELECT * FROM hedge_plan_legs
        WHERE status = 'closed'
          AND inst_id IS NOT NULL AND TRIM(inst_id) != ''
          AND (leg_role LIKE 'option%' OR opt_type IS NOT NULL)
        ORDER BY id DESC
        LIMIT 400
        """
    ).fetchall()
    updated_legs = 0
    touched_plans: set[int] = set()
    for row in legs:
        leg = dict(row)
        inst = str(leg.get("inst_id") or "").strip()
        if not inst or inst not in by_inst:
            continue
        pnl, src = resolve_option_leg_realized_pnl(
            leg=leg,
            hist_rows=by_inst[inst],
            fallback=None,
        )
        if src != "exchange" or pnl is None:
            continue
        local = _sf(leg.get("realized_pnl"))
        if local is not None and abs(local - pnl) < 1e-6:
            continue
        conn.execute(
            "UPDATE hedge_plan_legs SET realized_pnl=? WHERE id=?",
            (pnl, int(leg["id"])),
        )
        updated_legs += 1
        touched_plans.add(int(leg["plan_id"]))

    updated_plans = 0
    updater = update_plan_fn or update_plan
    for pid in touched_plans:
        plan = get_plan(conn, pid)
        if not plan or str(plan.get("status") or "") != "closed":
            continue
        plan_legs = get_plan_legs(conn, pid)
        opt_sum = 0.0
        for lg in plan_legs:
            role = str(lg.get("leg_role") or "")
            if not (role.startswith("option") or lg.get("opt_type")):
                continue
            if str(lg.get("status") or "") != "closed":
                continue
            opt_sum += float(_sf(lg.get("realized_pnl")) or 0.0)
        perp = float(_sf(plan.get("realized_pnl_perp")) or 0.0)
        ptype = str(plan.get("plan_type") or "")
        if ptype == "options_options":
            total = opt_sum
            kwargs: dict[str, Any] = {
                "realized_pnl_options": round(opt_sum, 4),
                "realized_pnl_total": round(total, 4),
            }
        else:
            total = perp + opt_sum
            kwargs = {
                "realized_pnl_perp": round(perp, 4),
                "realized_pnl_options": round(opt_sum, 4),
                "realized_pnl_total": round(total, 4),
            }
        old_total = _sf(plan.get("realized_pnl_total"))
        old_opts = _sf(plan.get("realized_pnl_options"))
        if (
            old_total is not None
            and abs(old_total - total) < 1e-6
            and old_opts is not None
            and abs(old_opts - opt_sum) < 1e-6
        ):
            continue
        updater(conn, pid, **kwargs)
        updated_plans += 1
    return {"legs": updated_legs, "plans": updated_plans}