"""期权持仓监控:浮盈翻倍微信提醒 + 平仓/到期状态同步."""
from __future__ import annotations

import os
import sqlite3
import time
from datetime import datetime, timezone
from typing import Any, Callable
from zoneinfo import ZoneInfo

from lib.exchange.okx_options_lib import normalize_option_exp_ms, resolve_option_close_from_history

_APP_TZ = ZoneInfo((os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai")


def _safe_float(v: Any) -> float | None:
    if v is None:
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def build_profit_alert_message(
    *,
    account_label: str,
    inst_id: str,
    premium_paid: float,
    upl: float,
    upl_ratio: float | None,
    bid: float | None,
) -> str:
    pct = f"{upl_ratio * 100:.1f}%" if upl_ratio is not None else "—"
    bid_txt = f"{bid:.4f}" if bid is not None else "—"
    return "\n".join(
        [
            "【OKX期权·翻倍提醒】",
            f"账户:{account_label}",
            f"合约:{inst_id}",
            f"已付权利金:{premium_paid:.4f} USDC",
            f"未实现盈亏:{upl:+.4f} USDC({pct})",
            f"当前买一:{bid_txt}(可考虑限价平仓锁利)",
        ]
    )


def run_options_profit_alerts(
    conn: sqlite3.Connection,
    positions: list[dict[str, Any]],
    *,
    profit_ratio: float,
    send_wechat: Callable[[str], None],
    account_label: str,
    ticker_bid_fn: Callable[[str], float | None],
) -> int:
    """
    对比 DB 中 open 记录与交易所持仓;达到阈值发微信.
    返回发送条数.
    """
    sent = 0
    pos_by_inst = {str(p.get("inst_id") or p.get("instId") or ""): p for p in positions}
    rows = conn.execute(
        """
        SELECT id, inst_id, premium_paid, profit_alert_sent
        FROM options_trades
        WHERE status = 'open'
        ORDER BY id ASC
        """
    ).fetchall()
    # 同合约多腿加仓:按合约汇总权利金,整仓只告警一次
    by_inst: dict[str, dict[str, Any]] = {}
    for row in rows:
        inst_id = str(row["inst_id"] or "")
        if not inst_id:
            continue
        bucket = by_inst.setdefault(
            inst_id,
            {"ids": [], "premium": 0.0, "all_sent": True, "has_prem": False},
        )
        bucket["ids"].append(int(row["id"]))
        prem = _safe_float(row["premium_paid"])
        if prem is not None:
            bucket["premium"] += float(prem)
            bucket["has_prem"] = True
        if not int(row["profit_alert_sent"] or 0):
            bucket["all_sent"] = False

    for inst_id, bucket in by_inst.items():
        if bucket["all_sent"] or not bucket["has_prem"] or bucket["premium"] <= 0:
            continue
        pos = pos_by_inst.get(inst_id)
        if not pos:
            continue
        prem = float(bucket["premium"])
        upl = _safe_float(pos.get("upl"))
        upl_ratio = _safe_float(pos.get("upl_ratio_pct"))
        if upl_ratio is not None:
            ratio = upl_ratio / 100.0
        elif upl is not None:
            ratio = upl / prem
        else:
            continue
        if ratio < float(profit_ratio):
            continue
        bid = ticker_bid_fn(inst_id)
        msg = build_profit_alert_message(
            account_label=account_label,
            inst_id=inst_id,
            premium_paid=prem,
            upl=upl or 0.0,
            upl_ratio=ratio,
            bid=bid,
        )
        try:
            send_wechat(msg)
            conn.execute(
                f"UPDATE options_trades SET profit_alert_sent = 1 WHERE id IN ({','.join('?' * len(bucket['ids']))})",
                tuple(bucket["ids"]),
            )
            sent += 1
        except Exception:
            pass
    return sent


def _created_at_ms(created_at: Any) -> int | None:
    """墙钟 created_at → UTC ms.库内时间为业务时区(默认 Asia/Shanghai),不可当 UTC."""
    if not created_at:
        return None
    raw = str(created_at).strip()
    if not raw:
        return None
    for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M:%f", 26), ("%Y-%m-%d %H:%M", 16)):
        try:
            dt = datetime.strptime(raw[:ln], fmt).replace(tzinfo=_APP_TZ)
            return int(dt.timestamp() * 1000)
        except ValueError:
            continue
    return None


def _group_key_for_closed_trade(row: Any) -> str:
    inst = str(row["inst_id"] or "").strip()
    ord_id = str(row["close_ord_id"] or "").strip() if "close_ord_id" in row.keys() else ""
    if ord_id:
        return f"{inst}|ord:{ord_id}"
    closed = str(row["closed_at"] or "").strip()
    return f"{inst}|close:{(closed[:16] if closed else '')}"


def backfill_closed_options_realized_pnl_from_history(
    conn: sqlite3.Connection,
    hist_rows: list[dict[str, Any]],
    *,
    trade_limit: int = 200,
) -> int:
    """
    用 OKX positions-history 的 realizedPnl 覆盖本地已平记录.
    同一次平仓多笔本地 open(加仓)按权利金占比分摊交易所总盈亏.
    """
    by_inst: dict[str, list[dict[str, Any]]] = {}
    for raw in hist_rows or []:
        if not isinstance(raw, dict):
            continue
        inst = str(raw.get("instId") or "").strip()
        if not inst:
            continue
        by_inst.setdefault(inst, []).append(raw)

    rows = conn.execute(
        """
        SELECT id, inst_id, sheets, premium_paid, realized_pnl, created_at, closed_at, close_ord_id
        FROM options_trades
        WHERE status = 'closed'
        ORDER BY id DESC
        LIMIT ?
        """,
        (int(trade_limit),),
    ).fetchall()
    if not rows:
        return 0

    groups: dict[str, list[Any]] = {}
    for row in rows:
        inst = str(row["inst_id"] or "").strip()
        if not inst or inst not in by_inst:
            continue
        groups.setdefault(_group_key_for_closed_trade(row), []).append(row)

    updated = 0
    for group in groups.values():
        inst = str(group[0]["inst_id"] or "").strip()
        open_candidates = [_created_at_ms(r["created_at"]) for r in group]
        open_ms = min((x for x in open_candidates if x is not None), default=None)
        close_info = resolve_option_close_from_history(by_inst.get(inst) or [], open_ms=open_ms)
        if not close_info:
            continue
        ex_pnl = _safe_float(close_info.get("realized_pnl"))
        if ex_pnl is None:
            continue
        close_quote = _safe_float(close_info.get("close_quote"))
        total_paid = 0.0
        for r in group:
            total_paid += float(_safe_float(r["premium_paid"]) or 0.0)
        allocated = 0.0
        for i, r in enumerate(group):
            paid = float(_safe_float(r["premium_paid"]) or 0.0)
            if i == len(group) - 1:
                share = round(float(ex_pnl) - allocated, 4)
            elif total_paid > 0:
                share = round(float(ex_pnl) * (paid / total_paid), 4)
                allocated += share
            else:
                share = round(float(ex_pnl) / len(group), 4)
                allocated += share
            local = _safe_float(r["realized_pnl"])
            if local is not None and abs(local - share) < 1e-6:
                continue
            prem_recv = round(paid + share, 4)
            conn.execute(
                """
                UPDATE options_trades
                SET realized_pnl = ?,
                    premium_received = ?,
                    close_quote = COALESCE(?, close_quote)
                WHERE id = ?
                """,
                (share, prem_recv, close_quote, int(r["id"])),
            )
            updated += 1
    return updated


def sync_open_options_trades(
    conn: sqlite3.Connection,
    *,
    live_inst_ids: set[str],
    fetch_history_fn: Callable[[str], list[dict[str, Any]]],
    notify_cfg: dict[str, Any] | None = None,
) -> int:
    """
    交易所已无持仓时,将本地 open 记录同步为 closed.
    优先用 positions-history 回填盈亏;否则到期后按归零处理.
    """
    rows = conn.execute(
        """
        SELECT id, inst_id, premium_paid, exp_time, created_at
        FROM options_trades
        WHERE status = 'open'
        """
    ).fetchall()
    updated = 0
    now_ms = int(time.time() * 1000)
    for row in rows:
        inst_id = str(row["inst_id"] or "")
        if not inst_id or inst_id in live_inst_ids:
            continue
        paid = _safe_float(row["premium_paid"]) or 0.0
        open_ms = _created_at_ms(row["created_at"])
        exp_ms = normalize_option_exp_ms(row["exp_time"], inst_id)
        close_quote: float | None = None
        prem_recv: float | None = None
        realized_pnl: float | None = None
        close_ord_id: str | None = None
        closed_at: str | None = None
        close_reason = "exchange"

        close_info = resolve_option_close_from_history(
            fetch_history_fn(inst_id),
            open_ms=open_ms,
        )
        if close_info:
            close_quote = close_info.get("close_quote")
            realized_pnl = close_info.get("realized_pnl")
            close_ord_id = close_info.get("pos_id")
            if realized_pnl is not None:
                prem_recv = round(paid + float(realized_pnl), 4)
            close_ms = close_info.get("close_ms")
            if close_ms:
                closed_at = datetime.fromtimestamp(int(close_ms) / 1000, tz=timezone.utc).strftime(
                    "%Y-%m-%d %H:%M:%S"
                )
        elif exp_ms is not None and now_ms >= int(exp_ms):
            close_reason = "expired"
            close_quote = 0.0
            prem_recv = 0.0
            realized_pnl = round(-paid, 4)
            if exp_ms:
                closed_at = datetime.fromtimestamp(int(exp_ms) / 1000, tz=timezone.utc).strftime(
                    "%Y-%m-%d %H:%M:%S"
                )
        else:
            continue

        conn.execute(
            """
            UPDATE options_trades
            SET status = 'closed',
                close_quote = ?,
                premium_received = ?,
                realized_pnl = ?,
                close_ord_id = COALESCE(?, close_ord_id),
                closed_at = COALESCE(?, closed_at, CURRENT_TIMESTAMP),
                signal_note = CASE
                    WHEN ? = 'expired' AND (signal_note IS NULL OR TRIM(signal_note) = '')
                    THEN '到期结算'
                    ELSE signal_note
                END
            WHERE id = ?
            """,
            (
                close_quote,
                prem_recv,
                realized_pnl,
                close_ord_id,
                closed_at,
                close_reason,
                int(row["id"]),
            ),
        )
        updated += 1
        if notify_cfg is not None:
            try:
                from lib.options.options_notify_lib import notify_options_close

                reason = "到期结算" if close_reason == "expired" else "交易所平仓"
                notify_options_close(
                    notify_cfg,
                    conn,
                    inst_id=inst_id,
                    reason=reason,
                    trade_id=int(row["id"]),
                    premium_paid=paid,
                    premium_received=prem_recv,
                    realized_pnl=realized_pnl,
                    close_quote=close_quote,
                )
            except Exception:
                pass
    return updated


def reconcile_live_open_trades(
    conn: sqlite3.Connection,
    *,
    live_inst_ids: set[str],
) -> int:
    """交易所有持仓但本地误标 closed 时恢复为 open."""
    fixed = 0
    for inst_id in live_inst_ids:
        if not inst_id:
            continue
        open_row = conn.execute(
            "SELECT id FROM options_trades WHERE inst_id = ? AND status = 'open' LIMIT 1",
            (inst_id,),
        ).fetchone()
        if open_row:
            continue
        row = conn.execute(
            """
            SELECT id, close_ord_id, realized_pnl
            FROM options_trades
            WHERE inst_id = ? AND status = 'closed'
            ORDER BY id DESC LIMIT 1
            """,
            (inst_id,),
        ).fetchone()
        if not row:
            continue
        if row["close_ord_id"]:
            continue
        if row["realized_pnl"] is not None:
            continue
        conn.execute(
            """
            UPDATE options_trades
            SET status = 'open',
                close_quote = NULL,
                premium_received = NULL,
                realized_pnl = NULL,
                closed_at = NULL,
                signal_note = CASE
                    WHEN signal_note = '到期结算' THEN NULL
                    ELSE signal_note
                END
            WHERE id = ?
            """,
            (int(row["id"]),),
        )
        fixed += 1
    return fixed


def options_monitor_loop(
    *,
    enabled: bool,
    poll_seconds: float,
    get_db: Callable[[], sqlite3.Connection],
    fetch_positions: Callable[[], list[dict[str, Any]]],
    ticker_bid_fn: Callable[[str], float | None],
    send_wechat: Callable[[str], None],
    account_label: str,
    profit_ratio: float,
    sync_trades_fn: Callable[[sqlite3.Connection], int] | None = None,
    target_close_fn: Callable[[str], dict[str, Any]] | None = None,
    stale_pending_fn: Callable[[], dict[str, Any]] | None = None,
    stop_event: Any = None,
) -> None:
    if not enabled:
        return
    while True:
        if stop_event is not None and getattr(stop_event, "is_set", lambda: False)():
            break
        try:
            conn = get_db()
            try:
                positions = fetch_positions()
                run_options_profit_alerts(
                    conn,
                    positions,
                    profit_ratio=profit_ratio,
                    send_wechat=send_wechat,
                    account_label=account_label,
                    ticker_bid_fn=ticker_bid_fn,
                )
                if target_close_fn is not None:
                    from lib.options.options_target_lib import run_options_target_closes

                    run_options_target_closes(
                        conn,
                        positions,
                        close_fn=target_close_fn,
                        send_wechat=send_wechat,
                        account_label=account_label,
                        cfg={"send_wechat": send_wechat, "account_label": account_label},
                    )
                if sync_trades_fn is not None:
                    sync_trades_fn(conn)
                conn.commit()
            finally:
                conn.close()
            # 平仓限价挂单超时撤单(独立于 DB 事务)
            if stale_pending_fn is not None:
                try:
                    stale_pending_fn()
                except Exception:
                    pass
        except Exception:
            pass
        time.sleep(max(5.0, float(poll_seconds)))