crypto_okx/lib/options/options_stats_lib.py

"""期权本地交易统计(胜率 / 盈亏 / 持仓时长)."""
from __future__ import annotations

from datetime import datetime
from typing import Any

from lib.instance.instance_embed_context_lib import profit_loss_ratio_from_averages
from lib.options.options_db import init_options_tables


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 _underlying_index_usdt(ex: Any, underly: str) -> float | None:
    """取标的 USDT 近似指数(币本位已平盈亏折 U).优先公开 ticker,避免私钥失败."""
    u = (underly or "ETH").strip().upper() or "ETH"
    pubs: list[Any] = []
    try:
        from lib.sim.hooks import _APP_MODULE, _sim_public_exchange

        pub = _sim_public_exchange(ex) if _APP_MODULE is not None else None
        if pub is not None:
            pubs.append(pub)
    except Exception:
        pass
    if ex is not None and ex not in pubs:
        pubs.append(ex)
    from lib.exchange.okx_options_lib import fetch_index_price

    for pub in pubs:
        try:
            if hasattr(pub, "public_get_market_ticker"):
                rows = (pub.public_get_market_ticker({"instId": f"{u}-USDT"}) or {}).get("data") or []
                if rows:
                    last = _safe_float(rows[0].get("last") or rows[0].get("lastPx"))
                    if last is not None and last > 0:
                        return float(last)
        except Exception:
            pass
        try:
            px = fetch_index_price(pub, f"{u}-USD")
            if px is not None and float(px) > 0:
                return float(px)
        except Exception:
            pass
        try:
            t = pub.fetch_ticker(f"{u}/USDT") or {}
            last = _safe_float(t.get("last") or t.get("close"))
            if last is not None and last > 0:
                return float(last)
        except Exception:
            continue
    return None


def history_pnl_to_usdt(history: list[dict[str, Any]], ex: Any = None) -> list[dict[str, Any]]:
    """
    统计用:币本位 realized_pnl(ETH/BTC) 按指数折成 U;USDC 原样.
    折算失败的币仓剔除盈亏字段,避免把「币数量」当成 U.
    """
    from lib.options.options_margin_mode_lib import margin_mode_from_inst_id, premium_ccy_for_mode

    idx_cache: dict[str, float | None] = {}
    out: list[dict[str, Any]] = []
    for row in history:
        r = dict(row)
        if r.get("status") == "open":
            out.append(r)
            continue
        pnl = _safe_float(r.get("realized_pnl"))
        if pnl is None:
            out.append(r)
            continue
        inst = str(r.get("inst_id") or "")
        underly = str(r.get("underlying") or (inst.split("-")[0] if inst else "ETH") or "ETH")
        ccy = str(r.get("premium_ccy") or "").strip().upper()
        if not ccy:
            ccy = premium_ccy_for_mode(margin_mode_from_inst_id(inst), underly)
        if ccy in ("ETH", "BTC"):
            if underly not in idx_cache:
                idx_cache[underly] = _underlying_index_usdt(ex, underly)
            idx = idx_cache.get(underly)
            if idx is None or idx <= 0:
                r["realized_pnl"] = None
            else:
                r["realized_pnl"] = round(float(pnl) * float(idx), 4)
        else:
            r["realized_pnl"] = round(float(pnl), 4)
        out.append(r)
    return out


def _parse_ts(raw: Any) -> datetime | None:
    if raw is None or raw == "":
        return None
    s = str(raw).strip().replace(" ", "T", 1)
    try:
        return datetime.fromisoformat(s)
    except (TypeError, ValueError):
        return None


def _hold_seconds(created_at: Any, closed_at: Any) -> float | None:
    start = _parse_ts(created_at)
    end = _parse_ts(closed_at)
    if start is None or end is None:
        return None
    sec = (end - start).total_seconds()
    return sec if sec >= 0 else None


def _avg_seconds(values: list[float]) -> float | None:
    if not values:
        return None
    return round(sum(values) / len(values), 1)


def compute_options_stats_from_history(history: list[dict[str, Any]]) -> dict[str, Any]:
    """基于期权历史列表(交易所)计算统计."""
    wins: list[float] = []
    losses: list[float] = []
    win_holds: list[float] = []
    loss_holds: list[float] = []
    all_holds: list[float] = []
    open_holds: list[float] = []
    now = datetime.now()

    for row in history:
        if row.get("status") == "open":
            start = _parse_ts(row.get("created_at"))
            if start is not None:
                sec = (now - start).total_seconds()
                if sec >= 0:
                    open_holds.append(sec)
            continue
        pnl_raw = row.get("realized_pnl")
        if pnl_raw is None:
            continue
        try:
            pnl = float(pnl_raw)
        except (TypeError, ValueError):
            continue
        hold = _hold_seconds(row.get("created_at"), row.get("closed_at"))
        if hold is not None:
            all_holds.append(hold)
        if pnl > 0:
            wins.append(pnl)
            if hold is not None:
                win_holds.append(hold)
        elif pnl < 0:
            losses.append(pnl)
            if hold is not None:
                loss_holds.append(hold)

    total_closed = len(wins) + len(losses)
    win_rate = round(len(wins) / total_closed * 100, 2) if total_closed else 0
    avg_win = sum(wins) / len(wins) if wins else None
    avg_loss = sum(losses) / len(losses) if losses else None

    total_profit = round(sum(wins), 4) if wins else 0.0
    total_loss = round(abs(sum(losses)), 4) if losses else 0.0
    net_realized = round(sum(wins) + sum(losses), 4)
    return {
        "total_closed": total_closed,
        "win_count": len(wins),
        "loss_count": len(losses),
        "win_rate": win_rate,
        "profit_loss_ratio": profit_loss_ratio_from_averages(avg_win, avg_loss),
        "avg_win": round(avg_win, 4) if avg_win is not None else None,
        "avg_loss": round(abs(avg_loss), 4) if avg_loss is not None else None,
        "total_profit": total_profit,
        "total_loss": total_loss,
        "net_realized_pnl": net_realized,
        "avg_hold_sec": _avg_seconds(all_holds),
        "avg_win_hold_sec": _avg_seconds(win_holds),
        "avg_loss_hold_sec": _avg_seconds(loss_holds),
        "open_count": len(open_holds),
        "avg_open_hold_sec": _avg_seconds(open_holds),
    }


_STATS_WINDOWS = ("today", "7d", "30d", "all")


def normalize_stats_window(raw: Any) -> str:
    w = str(raw or "7d").strip().lower()
    if w in ("1d", "day", "今日"):
        return "today"
    if w in ("7", "week"):
        return "7d"
    if w in ("30", "month"):
        return "30d"
    if w in ("0", "全部", "total"):
        return "all"
    return w if w in _STATS_WINDOWS else "7d"


def stats_window_bounds(window: str, *, now: datetime | None = None) -> tuple[datetime | None, datetime | None]:
    """返回闭区间起止(本地/业务时区近似用本地 now).all → (None, None)."""
    w = normalize_stats_window(window)
    end = now or datetime.now()
    if w == "all":
        return None, None
    if w == "today":
        start = end.replace(hour=0, minute=0, second=0, microsecond=0)
        return start, end
    days = 7 if w == "7d" else 30
    from datetime import timedelta

    start = end - timedelta(days=days)
    return start, end


def filter_history_for_stats_window(
    history: list[dict[str, Any]],
    window: str,
    *,
    now: datetime | None = None,
) -> list[dict[str, Any]]:
    """窗口内已平仓 + 当前持仓(open)."""
    start, end = stats_window_bounds(window, now=now)
    out: list[dict[str, Any]] = []
    for row in history or []:
        if row.get("status") == "open":
            out.append(row)
            continue
        if start is None:
            out.append(row)
            continue
        ts = _parse_ts(row.get("closed_at")) or _parse_ts(row.get("created_at"))
        if ts is None:
            continue
        if ts < start:
            continue
        if end is not None and ts > end:
            continue
        out.append(row)
    return out


def compute_options_stats_for_window(
    history: list[dict[str, Any]],
    window: str,
    *,
    now: datetime | None = None,
) -> dict[str, Any]:
    w = normalize_stats_window(window)
    filtered = filter_history_for_stats_window(history, w, now=now)
    stats = compute_options_stats_from_history(filtered)
    start, end = stats_window_bounds(w, now=now)
    stats["window"] = w
    stats["window_start"] = start.strftime("%Y-%m-%d %H:%M:%S") if start else None
    stats["window_end"] = end.strftime("%Y-%m-%d %H:%M:%S") if end else None
    return stats


def _row_get(row: Any, key: str, default: Any = None) -> Any:
    try:
        if hasattr(row, "keys") and key in row.keys():
            return row[key]
    except Exception:
        pass
    if isinstance(row, dict):
        return row.get(key, default)
    return default


def _closed_trade_dicts(closed_rows) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for row in closed_rows or []:
        out.append(
            {
                "status": "closed",
                "realized_pnl": _row_get(row, "realized_pnl"),
                "created_at": _row_get(row, "created_at"),
                "closed_at": _row_get(row, "closed_at"),
                "inst_id": _row_get(row, "inst_id") or "",
                "underlying": _row_get(row, "underlying") or "",
                "premium_ccy": _row_get(row, "premium_ccy") or "",
            }
        )
    return out


def compute_options_stats(get_db, ex: Any = None) -> dict[str, Any]:
    conn = get_db()
    try:
        init_options_tables(conn)
        closed_rows = conn.execute(
            """
            SELECT realized_pnl, created_at, closed_at, inst_id, underlying, premium_ccy
            FROM options_trades
            WHERE status = 'closed' AND realized_pnl IS NOT NULL
            """
        ).fetchall()
        open_rows = conn.execute(
            """
            SELECT created_at FROM options_trades WHERE status = 'open'
            """
        ).fetchall()
        history = _closed_trade_dicts(closed_rows)
        for row in open_rows:
            history.append({"status": "open", "created_at": _row_get(row, "created_at")})
        # 币本位盈亏按指数折 U,再算胜率/盈亏比(与 /api/options/stats 一致)
        return compute_options_stats_from_history(history_pnl_to_usdt(history, ex))
    finally:
        conn.close()


def _stats_from_option_trade_rows(closed_rows, open_rows, *, ex: Any = None) -> dict[str, Any]:
    """兼容旧调用;内部统一折 U 后走 compute_options_stats_from_history."""
    history = _closed_trade_dicts(closed_rows)
    for row in open_rows or []:
        history.append({"status": "open", "created_at": _row_get(row, "created_at")})
    return compute_options_stats_from_history(history_pnl_to_usdt(history, ex))


def header_options_stats_for_window(
    conn,
    list_window: dict[str, Any],
    app_tz,
    ex: Any = None,
) -> dict[str, Any]:
    """顶栏总交易/胜率/盈亏比:纯期权(options_trades),全站导航共用,按列表窗过滤.

    币本位 realized_pnl 为币数量,须折 U 后再算盈亏比,否则 BTC/ETH 混算会严重失真.
    """
    from lib.common.history_window_lib import utc_window_to_bj_sql_strings

    init_options_tables(conn)
    start_bj, end_bj = utc_window_to_bj_sql_strings(
        list_window["start_utc"], list_window["end_utc"], app_tz
    )
    closed_rows = conn.execute(
        """
        SELECT realized_pnl, created_at, closed_at, inst_id, underlying, premium_ccy
        FROM options_trades
        WHERE status = 'closed'
          AND realized_pnl IS NOT NULL
          AND COALESCE(closed_at, created_at) >= ?
          AND COALESCE(closed_at, created_at) <= ?
        """,
        (start_bj, end_bj),
    ).fetchall()
    open_rows = conn.execute(
        "SELECT created_at FROM options_trades WHERE status = 'open'"
    ).fetchall()
    history = _closed_trade_dicts(closed_rows)
    for row in open_rows:
        history.append({"status": "open", "created_at": _row_get(row, "created_at")})
    stats = compute_options_stats_from_history(history_pnl_to_usdt(history, ex))
    return {
        "total": int(stats.get("total_closed") or 0),
        "rate": float(stats.get("win_rate") or 0),
        "profit_loss_ratio": stats.get("profit_loss_ratio"),
    }