crypto_monitor/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
from lib.options.options_margin_mode_lib import margin_mode_from_inst_id, premium_ccy_for_mode


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 _safe_float(v: Any) -> float | None:
    if v is None or v == "":
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def _row_premium_ccy(row: dict[str, Any]) -> str:
    ccy = str(row.get("premium_ccy") or "").strip().upper()
    if ccy:
        return ccy
    inst = str(row.get("inst_id") or "").strip()
    mode = str(row.get("margin_mode") or "").strip().lower()
    underly = str(row.get("underlying") or (inst.split("-")[0] if inst else "ETH") or "ETH")
    if mode:
        return premium_ccy_for_mode(mode, underly)
    if not inst:
        # 旧统计行无合约信息时按 USDC 口径,避免默认币本位把盈亏跳过
        return "USDC"
    return premium_ccy_for_mode(margin_mode_from_inst_id(inst), underly)


def _pnl_as_usdt(row: dict[str, Any], *, fallback_index: float | None = None) -> float | None:
    """已平/浮盈统一折算为 USDT(币本位×指数;USDC 原样)."""
    pnl = _safe_float(row.get("realized_pnl"))
    if pnl is None:
        pnl = _safe_float(row.get("upl"))
    if pnl is None:
        return None
    ccy = _row_premium_ccy(row)
    if ccy in ("ETH", "BTC"):
        px = _safe_float(row.get("idx_px") or row.get("idxPx") or row.get("index_px"))
        if px is None or px <= 0:
            px = fallback_index
        if px is None or px <= 0:
            return None
        return float(pnl) * float(px)
    return float(pnl)


def _history_index_px(history: list[dict[str, Any]]) -> float | None:
    for row in history:
        px = _safe_float(row.get("idx_px") or row.get("idxPx") or row.get("index_px"))
        if px is not None and px > 0:
            return px
    return None


def compute_options_stats_from_history(
    history: list[dict[str, Any]],
    *,
    index_px: float | None = None,
) -> dict[str, Any]:
    """基于期权历史列表计算统计;币本位盈亏按指数折算为 U."""
    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()
    fallback_idx = index_px if index_px is not None and index_px > 0 else _history_index_px(history)
    coinish = False

    for row in history:
        ccy = _row_premium_ccy(row)
        if ccy in ("ETH", "BTC"):
            coinish = True
        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 = _pnl_as_usdt(row, fallback_index=fallback_idx)
        if pnl is None:
            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),
        "pnl_unit": "U" if coinish else "USDC",
        "index_px": fallback_idx,
    }


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, inst_id, premium_ccy, margin_mode
            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()
    finally:
        conn.close()

    hist = []
    for row in closed_rows:
        hist.append(
            {
                "status": "closed",
                "realized_pnl": row["realized_pnl"],
                "created_at": row["created_at"],
                "closed_at": row["closed_at"],
                "inst_id": row["inst_id"] if "inst_id" in row.keys() else None,
                "premium_ccy": row["premium_ccy"] if "premium_ccy" in row.keys() else None,
                "margin_mode": row["margin_mode"] if "margin_mode" in row.keys() else None,
            }
        )
    for row in open_rows:
        hist.append({"status": "open", "created_at": row["created_at"]})
    return compute_options_stats_from_history(hist)