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


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