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 compute_options_stats(get_db) -> dict[str, Any]:
    conn = get_db()
    try:
        init_options_tables(conn)
        closed_rows = conn.execute(
            """
            SELECT realized_pnl, created_at, closed_at
            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()
        return _stats_from_option_trade_rows(closed_rows, open_rows)
    finally:
        conn.close()


def _stats_from_option_trade_rows(closed_rows, open_rows) -> dict[str, Any]:
    wins: list[float] = []
    losses: list[float] = []
    win_holds: list[float] = []
    loss_holds: list[float] = []
    all_holds: list[float] = []
    now = datetime.now()

    for row in closed_rows:
        pnl = float(row["realized_pnl"])
        hold = _hold_seconds(row["created_at"], row["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)

    open_holds: list[float] = []
    for row in open_rows:
        start = _parse_ts(row["created_at"])
        if start is None:
            continue
        sec = (now - start).total_seconds()
        if sec >= 0:
            open_holds.append(sec)

    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),
    }


def header_options_stats_for_window(conn, list_window: dict[str, Any], app_tz) -> dict[str, Any]:
    """顶栏总交易/胜率/盈亏比:纯期权(options_trades),全站导航共用,按列表窗过滤."""
    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
        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()
    stats = _stats_from_option_trade_rows(closed_rows, open_rows)
    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"),
    }