"""期权本地交易统计(胜率 / 盈亏 / 持仓时长)."""
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_usdt(平仓当时口径,含手续费后净额);
    否则币本位用 close_index_px(再否则实时指数)折 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
        frozen = _safe_float(r.get("realized_pnl_usdt"))
        if frozen is not None:
            r["realized_pnl"] = round(float(frozen), 4)
            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"):
            idx = _safe_float(r.get("close_index_px"))
            if idx is None or idx <= 0:
                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"),
                "realized_pnl_usdt": _row_get(row, "realized_pnl_usdt"),
                "close_index_px": _row_get(row, "close_index_px"),
                "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


_CLOSED_STATS_SELECT = """
    SELECT realized_pnl, realized_pnl_usdt, close_index_px,
           created_at, closed_at, inst_id, underlying, premium_ccy
    FROM options_trades
    WHERE status = 'closed'
      AND (
            (COALESCE(pnl_settled, 0) = 1 AND realized_pnl_usdt IS NOT NULL)
         OR (trading_usdt_before IS NULL
             AND (realized_pnl IS NOT NULL OR realized_pnl_usdt IS NOT NULL))
      )
"""


def freeze_missing_realized_pnl_usdt(conn, ex: Any = None) -> int:
    """旧已平仓缺冻结 U 时按 close_index/当前指数写一次,之后不再跟盘变.

    注意:已有 realized_pnl 视为净额直接冻结,不再二次扣 open_fee/close_fee
    (避免对 OKX realizedPnl 或已扣费本地净额重复减费).
    """
    init_options_tables(conn)
    rows = conn.execute(
        """
        SELECT id, realized_pnl, realized_pnl_usdt, close_index_px,
               inst_id, underlying, premium_ccy
        FROM options_trades
        WHERE status = 'closed'
          AND realized_pnl_usdt IS NULL
          AND realized_pnl IS NOT NULL
          AND trading_usdt_before IS NULL
        ORDER BY id DESC
        LIMIT 300
        """
    ).fetchall()
    if not rows:
        return 0
    from lib.options.options_pnl_settle_lib import (
        freeze_pnl_to_usdt,
        resolve_close_index_px,
    )

    updated = 0
    for row in rows:
        ccy = str(_row_get(row, "premium_ccy") or "").strip().upper()
        idx = _safe_float(_row_get(row, "close_index_px"))
        if idx is None or idx <= 0:
            idx = resolve_close_index_px(
                ex,
                inst_id=str(_row_get(row, "inst_id") or ""),
                underlying=str(_row_get(row, "underlying") or ""),
            )
        pnl = _safe_float(_row_get(row, "realized_pnl"))
        if pnl is None:
            continue
        settled = freeze_pnl_to_usdt(pnl, premium_ccy=ccy or "USDC", index_px=idx)
        usdt = settled.get("realized_pnl_usdt")
        close_idx = settled.get("close_index_px")
        if usdt is None:
            continue
        conn.execute(
            """
            UPDATE options_trades
            SET realized_pnl_usdt = ?,
                close_index_px = COALESCE(close_index_px, ?)
            WHERE id = ? AND realized_pnl_usdt IS NULL
            """,
            (float(usdt), close_idx, int(row["id"])),
        )
        updated += 1
    return updated


def recompute_closed_realized_pnl_usdt(conn, ex: Any = None) -> int:
    """强制按「平仓收回×平仓指数 − 开仓成本U − 手续费」重算已平冻结 U.

    旧逻辑用 (收回−已付)×平仓指数,等于把开仓成本也按平仓价重估,ETH 上涨时会偏亏,
    导致合计盈亏与余额对不上.有 budget_usdt/open_index_px 时优先用真实开仓成本.
    """
    init_options_tables(conn)
    rows = conn.execute(
        """
        SELECT id, realized_pnl, realized_pnl_usdt, close_index_px, open_index_px,
               inst_id, underlying, premium_ccy, premium_paid, premium_received,
               open_fee, close_fee, fee_ccy, budget_usdt, close_quote, eth_amount,
               open_quote
        FROM options_trades
        WHERE status = 'closed'
          AND premium_paid IS NOT NULL
        ORDER BY id DESC
        LIMIT 300
        """
    ).fetchall()
    if not rows:
        return 0
    from lib.options.options_pnl_settle_lib import resolve_close_index_px, settle_closed_pnl
    from lib.options.options_pricing_lib import total_premium

    updated = 0
    for row in rows:
        paid = _safe_float(_row_get(row, "premium_paid"))
        recv = _safe_float(_row_get(row, "premium_received"))
        if recv is None:
            cq = _safe_float(_row_get(row, "close_quote"))
            eth = _safe_float(_row_get(row, "eth_amount"))
            if cq is not None and eth is not None and eth > 0:
                recv = float(total_premium(cq, eth))
        if paid is None or recv is None:
            continue
        ofee = abs(_safe_float(_row_get(row, "open_fee")) or 0.0)
        cfee = abs(_safe_float(_row_get(row, "close_fee")) or 0.0)
        ccy = str(_row_get(row, "premium_ccy") or "").strip().upper() or "ETH"
        close_idx = _safe_float(_row_get(row, "close_index_px"))
        open_idx = _safe_float(_row_get(row, "open_index_px"))
        budget = _safe_float(_row_get(row, "budget_usdt"))
        if close_idx is None or close_idx <= 0:
            close_idx = resolve_close_index_px(
                ex,
                inst_id=str(_row_get(row, "inst_id") or ""),
                underlying=str(_row_get(row, "underlying") or ""),
            )
        if (open_idx is None or open_idx <= 0) and budget and paid > 1e-12:
            open_idx = float(budget) / float(paid)
        settled = settle_closed_pnl(
            premium_paid=paid,
            premium_received=recv,
            open_fee=ofee,
            close_fee=cfee,
            premium_ccy=ccy,
            index_px=close_idx,
            open_index_px=open_idx,
            budget_usdt=budget,
            fee_ccy=str(_row_get(row, "fee_ccy") or ccy),
        )
        new_usdt = settled.get("realized_pnl_usdt")
        if new_usdt is None:
            continue
        old_usdt = _safe_float(_row_get(row, "realized_pnl_usdt"))
        new_pnl = settled.get("realized_pnl")
        # 有变化才写,避免无意义刷库
        if (
            old_usdt is not None
            and abs(float(old_usdt) - float(new_usdt)) < 1e-6
            and _safe_float(_row_get(row, "realized_pnl")) is not None
            and new_pnl is not None
            and abs(float(_row_get(row, "realized_pnl")) - float(new_pnl)) < 1e-10
        ):
            continue
        conn.execute(
            """
            UPDATE options_trades
            SET realized_pnl = COALESCE(?, realized_pnl),
                close_index_px = COALESCE(?, close_index_px),
                open_index_px = COALESCE(open_index_px, ?),
                realized_pnl_usdt = ?
            WHERE id = ?
            """,
            (
                new_pnl,
                settled.get("close_index_px"),
                settled.get("open_index_px"),
                float(new_usdt),
                int(row["id"]),
            ),
        )
        updated += 1
    return updated


def repair_double_counted_fee_pnl(conn, ex: Any = None) -> int:
    """兼容旧名:改为强制按开仓成本重算冻结 U."""
    return recompute_closed_realized_pnl_usdt(conn, ex)


def _sim_net_usdt_converts_for_trade(
    conn,
    *,
    underlying: str,
    start_ts: str,
    end_ts: str,
) -> float | None:
    """从 sim_ledger 汇总该标的现货兑换的 USDT 净变动(买币为负、卖回为正)."""
    coin = (underlying or "").strip().upper()
    if coin not in ("ETH", "BTC") or not start_ts or not end_ts:
        return None
    try:
        rows = conn.execute(
            """
            SELECT amount, note FROM sim_ledger_entries
            WHERE kind = 'convert'
              AND UPPER(ccy) = 'USDT'
              AND account = 'trading'
              AND ts >= ?
              AND ts <= ?
            """,
            (str(start_ts), str(end_ts)),
        ).fetchall()
    except Exception:
        return None
    total = 0.0
    n = 0
    for row in rows or []:
        note = str(_row_get(row, "note") or "").upper()
        amt = _safe_float(_row_get(row, "amount"))
        if amt is None:
            continue
        related = (
            f"{coin}/USDT" in note
            or f"{coin}-USDT" in note
            or note.strip() == f"FROM {coin}"
        )
        if not related:
            continue
        total += float(amt)
        n += 1
    if n < 1:
        return None
    return round(total, 4)


def sync_sim_cash_realized_pnl_usdt(conn, ex: Any = None) -> int:
    """兼容入口:已按交易账户结算(pnl_settled=1)的永不改写.

    有 trading_usdt_before 的待结算单留给卖回结算;
    仅无账户快照的旧单才回退权利金×指数冻结.
    """
    del ex
    init_options_tables(conn)
    # 账户快照路径由 sell_residual / maybe_sell 结算,此处不覆盖
    return 0


def load_local_options_stats_history(conn, ex: Any = None) -> list[dict[str, Any]]:
    """本地已平(模拟盘口径) + 当前 open."""
    init_options_tables(conn)
    try:
        # 权利金净盈亏冻结 U(纠正旧版把买币预算差额当盈利的数据)
        sync_sim_cash_realized_pnl_usdt(conn, ex)
    except Exception:
        pass
    try:
        freeze_missing_realized_pnl_usdt(conn, ex)
    except Exception:
        pass
    closed_rows = conn.execute(_CLOSED_STATS_SELECT).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")})
    return history


def _load_exchange_pnl_freeze_map(conn) -> dict[str, dict[str, Any]]:
    init_options_tables(conn)
    out: dict[str, dict[str, Any]] = {}
    try:
        rows = conn.execute(
            """
            SELECT history_key, realized_pnl_usdt, close_index_px, realized_pnl
            FROM options_exchange_pnl_freeze
            """
        ).fetchall()
    except Exception:
        return out
    for row in rows or []:
        key = str(_row_get(row, "history_key") or "").strip()
        if not key:
            continue
        out[key] = {
            "realized_pnl_usdt": _row_get(row, "realized_pnl_usdt"),
            "close_index_px": _row_get(row, "close_index_px"),
            "realized_pnl": _row_get(row, "realized_pnl"),
        }
    return out


def apply_exchange_pnl_freeze(
    conn,
    history: list[dict[str, Any]],
    ex: Any = None,
) -> list[dict[str, Any]]:
    """写冻结表.行内已有 close_index_px / _freeze_index_hint 时不打网.

    若仍缺指数且传了 ex,会在持连接时补拉(仅兼容旧单测).生产请用
    load_live_exchange_stats_history(先网后短事务).
    """
    freeze_map = _load_exchange_pnl_freeze_map(conn)
    enriched: list[dict[str, Any]] = []
    need_idx: dict[str, str] = {}
    for row in history or []:
        r = dict(row)
        if r.get("status") == "open":
            enriched.append(r)
            continue
        key = str(r.get("history_key") or "").strip()
        pnl = _safe_float(r.get("realized_pnl"))
        if (
            key
            and pnl is not None
            and not (
                key in freeze_map
                and _safe_float(freeze_map[key].get("realized_pnl_usdt")) is not None
            )
            and _safe_float(r.get("close_index_px")) is None
            and _safe_float(r.get("_freeze_index_hint")) is None
        ):
            underly = str(r.get("underlying") or "").strip().upper()
            inst = str(r.get("inst_id") or "")
            if not underly and inst:
                underly = inst.split("-")[0].strip().upper()
            if underly:
                need_idx[underly] = inst
        enriched.append(r)

    if need_idx and ex is not None:
        from lib.options.options_pnl_settle_lib import resolve_close_index_px

        idx_cache: dict[str, float | None] = {}
        for underly, inst in need_idx.items():
            idx_cache[underly] = resolve_close_index_px(
                ex, inst_id=inst, underlying=underly
            )
        for r in enriched:
            if r.get("status") == "open":
                continue
            underly = str(r.get("underlying") or "").strip().upper()
            inst = str(r.get("inst_id") or "")
            if not underly and inst:
                underly = inst.split("-")[0].strip().upper()
            if underly in idx_cache and _safe_float(r.get("close_index_px")) is None:
                r["_freeze_index_hint"] = idx_cache.get(underly)

    return _apply_exchange_pnl_freeze_with_hints(conn, enriched, freeze_map)


def compute_options_stats(get_db, ex: Any = None) -> dict[str, Any]:
    """模拟盘:本地交易结算.实盘请用 compute_options_stats_live_exchange."""
    conn = get_db()
    try:
        history = load_local_options_stats_history(conn, ex)
        try:
            conn.commit()
        except Exception:
            pass
        return compute_options_stats_from_history(history_pnl_to_usdt(history, ex))
    finally:
        conn.close()


def load_live_exchange_stats_history(cfg: dict[str, Any], ex: Any) -> list[dict[str, Any]]:
    """实盘统计用历史:先拉交易所,再无连接时取指数,最后短事务写冻结.

    禁止在持有 DB 连接时打行情/历史网,否则会拖死持仓与中控轮询.
    """
    from lib.options.options_history_lib import load_options_history
    from lib.options.options_pnl_settle_lib import resolve_close_index_px

    history = load_options_history(ex, cfg) or []
    get_db = cfg["get_db"]
    conn = get_db()
    try:
        freeze_map = _load_exchange_pnl_freeze_map(conn)
        try:
            conn.commit()
        except Exception:
            pass
    finally:
        conn.close()

    need_idx: dict[str, str] = {}
    for row in history:
        if row.get("status") == "open":
            continue
        key = str(row.get("history_key") or "").strip()
        if not key or _safe_float(row.get("realized_pnl")) is None:
            continue
        if key in freeze_map and _safe_float(freeze_map[key].get("realized_pnl_usdt")) is not None:
            continue
        underly = str(row.get("underlying") or "").strip().upper()
        inst = str(row.get("inst_id") or "")
        if not underly and inst:
            underly = inst.split("-")[0].strip().upper()
        if underly and _safe_float(row.get("close_index_px")) is None:
            need_idx[underly] = inst

    idx_cache: dict[str, float | None] = {}
    for underly, inst in need_idx.items():
        idx_cache[underly] = resolve_close_index_px(ex, inst_id=inst, underlying=underly)

    enriched: list[dict[str, Any]] = []
    for row in history:
        r = dict(row)
        if r.get("status") != "open":
            underly = str(r.get("underlying") or "").strip().upper()
            inst = str(r.get("inst_id") or "")
            if not underly and inst:
                underly = inst.split("-")[0].strip().upper()
            key = str(r.get("history_key") or "").strip()
            if underly in idx_cache and _safe_float(r.get("close_index_px")) is None:
                if not (
                    key in freeze_map
                    and _safe_float(freeze_map[key].get("realized_pnl_usdt")) is not None
                ):
                    r["_freeze_index_hint"] = idx_cache.get(underly)
        enriched.append(r)

    conn = get_db()
    try:
        out = _apply_exchange_pnl_freeze_with_hints(conn, enriched, freeze_map)
        try:
            conn.commit()
        except Exception:
            pass
        return out
    finally:
        conn.close()


def compute_options_stats_live_exchange(cfg: dict[str, Any], ex: Any) -> dict[str, Any]:
    """实盘:与「期权历史」同一数据源(交易所 positions-history),盈亏以交易所为准."""
    history = load_live_exchange_stats_history(cfg, ex)
    return compute_options_stats_from_history(history_pnl_to_usdt(history, ex))


def _apply_exchange_pnl_freeze_with_hints(
    conn,
    history: list[dict[str, Any]],
    freeze_map: dict[str, dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
    """短事务写冻结;指数只用来自行内 _freeze_index_hint / 已有 close_index_px."""
    from lib.options.options_pnl_settle_lib import settle_exchange_realized_pnl
    from lib.options.options_time_lib import options_app_now_str

    init_options_tables(conn)
    fmap = freeze_map if freeze_map is not None else _load_exchange_pnl_freeze_map(conn)
    out: list[dict[str, Any]] = []
    for row in history or []:
        r = dict(row)
        hint = r.pop("_freeze_index_hint", None)
        if r.get("status") == "open":
            out.append(r)
            continue
        key = str(r.get("history_key") or "").strip()
        pnl = _safe_float(r.get("realized_pnl"))
        if not key or pnl is None:
            out.append(r)
            continue
        frozen = fmap.get(key)
        if frozen is not None and _safe_float(frozen.get("realized_pnl_usdt")) is not None:
            r["realized_pnl_usdt"] = frozen.get("realized_pnl_usdt")
            r["close_index_px"] = frozen.get("close_index_px")
            out.append(r)
            continue
        ccy = str(r.get("premium_ccy") or "").strip().upper()
        idx = _safe_float(r.get("close_index_px")) or _safe_float(hint)
        settled = settle_exchange_realized_pnl(
            realized_pnl=pnl,
            premium_ccy=ccy or "ETH",
            index_px=idx,
        )
        usdt = settled.get("realized_pnl_usdt")
        close_idx = settled.get("close_index_px")
        if usdt is not None:
            try:
                conn.execute(
                    """
                    INSERT INTO options_exchange_pnl_freeze(
                        history_key, realized_pnl, realized_pnl_usdt, close_index_px,
                        premium_ccy, updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?)
                    ON CONFLICT(history_key) DO UPDATE SET
                        realized_pnl = excluded.realized_pnl,
                        realized_pnl_usdt = excluded.realized_pnl_usdt,
                        close_index_px = excluded.close_index_px,
                        premium_ccy = excluded.premium_ccy,
                        updated_at = excluded.updated_at
                    """,
                    (
                        key,
                        float(pnl),
                        float(usdt),
                        close_idx,
                        ccy or "",
                        options_app_now_str(),
                    ),
                )
            except Exception:
                pass
            r["realized_pnl_usdt"] = float(usdt)
            r["close_index_px"] = close_idx
        out.append(r)
    return out


def compute_options_stats_auto(cfg: dict[str, Any], ex: Any = None) -> dict[str, Any]:
    """模拟盘走本地结算;实盘走交易所期权历史原生盈亏."""
    get_db = cfg.get("get_db")
    try:
        from lib.sim.mode_lib import is_sim_mode

        if callable(get_db) and is_sim_mode(get_db):
            return compute_options_stats(get_db, ex)
    except Exception:
        pass
    if ex is None:
        ex = cfg.get("exchange_options")
    return compute_options_stats_live_exchange(cfg, ex)


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]:
    """顶栏总交易/胜率/盈亏比:与期权统计同一结算口径.

    新单(有 trading_usdt_before):仅 pnl_settled=1 的账户差分计入;
    旧单(无 before):仍按原 realized 兼容.币数量须折 U 后再算盈亏比.
    """
    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
    )
    try:
        freeze_missing_realized_pnl_usdt(conn, ex)
    except Exception:
        pass
    closed_rows = conn.execute(
        """
        SELECT realized_pnl, realized_pnl_usdt, close_index_px,
               created_at, closed_at, inst_id, underlying, premium_ccy
        FROM options_trades
        WHERE status = 'closed'
          AND (
                (COALESCE(pnl_settled, 0) = 1 AND realized_pnl_usdt IS NOT NULL)
             OR (trading_usdt_before IS NULL
                 AND (realized_pnl IS NOT NULL OR realized_pnl_usdt 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"),
    }