"""实例数据看板:本户活跃监控 / 持仓只读聚合."""
from __future__ import annotations

from datetime import datetime, timezone
from typing import Any, Callable, Optional


def _row_dict(row: Any) -> dict[str, Any]:
    if row is None:
        return {}
    if isinstance(row, dict):
        return dict(row)
    try:
        return dict(row)
    except Exception:
        return {}


def _safe_float(v: Any) -> Optional[float]:
    try:
        if v is None or v == "":
            return None
        return float(v)
    except (TypeError, ValueError):
        return None


def _dir_label(direction: Any) -> str:
    d = str(direction or "").strip().lower()
    if d == "short":
        return "做空"
    if d == "long":
        return "做多"
    return str(direction or "-")


def _format_order_item(od: dict[str, Any]) -> dict[str, Any]:
    try:
        from lib.trade.trade_labels_lib import apply_order_monitor_source_labels

        od = apply_order_monitor_source_labels(od)
    except Exception:
        pass
    try:
        from lib.trade.entry_model_lib import enrich_entry_model_display

        enrich_entry_model_display(od)
    except Exception:
        pass
    sym = od.get("exchange_symbol") or od.get("symbol") or "-"
    direction = str(od.get("direction") or "long").lower()
    mt = od.get("monitor_type_display") or od.get("monitor_type") or ""
    kst = od.get("key_signal_type") or ""
    title = f"{sym} {_dir_label(direction)}"
    bits = [x for x in (mt, kst) if x]
    subtitle = " · ".join(bits) if bits else ""
    entry = _safe_float(od.get("trigger_price"))
    sl = _safe_float(od.get("stop_loss"))
    tp = _safe_float(od.get("take_profit"))
    return {
        "id": od.get("id"),
        "kind": "order",
        "tab": "trade",
        "title": title,
        "subtitle": subtitle,
        "symbol": sym,
        "price_symbol": od.get("symbol") or sym,
        "direction": direction,
        "direction_label": _dir_label(direction),
        "entry": entry,
        "mark_price": None,
        "contracts": _safe_float(od.get("order_amount")),
        "tp_profit": None,
        "float_pnl": None,
        "stop_loss": sl,
        "take_profit": tp,
        "status": od.get("status") or "active",
    }


OPTIONS_SOURCE_LABELS = {
    "option": "纯期权",
    "perp_options": "永期对冲",
    "options_options": "期期对冲",
}

HEDGE_ACTIVE_STATUSES = frozenset({"opening", "active", "partial"})


def _resolve_options_source(conn, inst_id: str) -> tuple[str, str, int | None]:
    """根据进行中对冲计划腿判定来源;默认纯期权. 返回 (source, label, plan_id)."""
    default = ("option", OPTIONS_SOURCE_LABELS["option"], None)
    if not inst_id or not _table_exists(conn, "hedge_plans") or not _table_exists(conn, "hedge_plan_legs"):
        return default
    try:
        row = conn.execute(
            """
            SELECT p.plan_type, p.id
            FROM hedge_plans p
            JOIN hedge_plan_legs l ON l.plan_id = p.id
            WHERE p.status IN ('opening', 'active', 'partial')
              AND l.status = 'open'
              AND l.inst_id = ?
            ORDER BY p.id DESC
            LIMIT 1
            """,
            (inst_id,),
        ).fetchone()
    except Exception:
        return default
    if not row:
        return default
    d = _row_dict(row)
    pt = str(d.get("plan_type") or "").strip()
    try:
        plan_id = int(d["id"]) if d.get("id") is not None else None
    except (TypeError, ValueError):
        plan_id = None
    if pt in OPTIONS_SOURCE_LABELS and pt != "option":
        return pt, OPTIONS_SOURCE_LABELS[pt], plan_id
    return default


def _format_profit_exit_mult(mult: Any) -> str:
    try:
        n = float(mult)
    except (TypeError, ValueError):
        return "1倍"
    if n <= 0:
        return "1倍"
    if abs(n - round(n)) < 1e-9:
        return f"{int(round(n))}倍"
    return f"{n:g}倍"


def _format_options_target(p: dict[str, Any]) -> str:
    hedge = p.get("hedge_plan_target") if isinstance(p.get("hedge_plan_target"), dict) else None
    opt_type = str(p.get("opt_type") or p.get("optType") or "").upper()
    if hedge:
        rr = _safe_float(hedge.get("profit_rr"))
        pid = hedge.get("plan_id")
        if rr is not None and rr > 0:
            return f"对冲#{pid} 盈亏比 {rr:g}" if pid is not None else f"盈亏比 {rr:g}"
        ot = str(hedge.get("opt_type") or opt_type).upper()
        side = "Put ≤" if ot == "P" else "Call ≥"
        tgt = _safe_float(hedge.get("target_index"))
        if tgt is not None:
            return f"对冲#{pid} {side} {tgt:g}" if pid is not None else f"{side} {tgt:g}"
    parts: list[str] = []
    tgt = _safe_float(p.get("target_index"))
    if tgt is not None and tgt > 0:
        side = "Put ≤" if opt_type == "P" else "Call ≥"
        parts.append(f"{side} {tgt:g}")
    if p.get("profit_exit_enabled"):
        parts.append(_format_profit_exit_mult(p.get("profit_exit_mult")))
    if parts:
        return " · ".join(parts)
    return "—"


def _format_options_item(p: dict[str, Any], *, conn=None) -> dict[str, Any]:
    inst = str(p.get("inst_id") or p.get("instId") or "-").strip() or "-"
    opt_type = str(p.get("opt_type") or p.get("optType") or "").upper()
    label = "Call" if opt_type == "C" else "Put" if opt_type == "P" else (opt_type or "OPT")
    # 看板期权列:优先买一净盈亏,残档回退交易所 upl
    pnl = None
    try:
        from lib.options.options_positions_lib import display_pnl_from_option_row

        pnl = display_pnl_from_option_row(p)
    except Exception:
        pnl = None
    pos = _safe_float(p.get("pos"))
    exp_ms = p.get("exp_time_ms")
    if exp_ms is None:
        exp_ms = p.get("exp_time")
    try:
        exp_ms = int(float(exp_ms)) if exp_ms not in (None, "") else None
    except (TypeError, ValueError):
        exp_ms = None
    if conn is not None:
        source_key, source_label, source_plan_id = _resolve_options_source(conn, inst)
    else:
        source_key, source_label, source_plan_id = "option", OPTIONS_SOURCE_LABELS["option"], None
    return {
        "id": inst,
        "kind": "options",
        "tab": "options",
        "title": f"{inst} {label}",
        "subtitle": f"张数 {pos if pos is not None else '-'}",
        "inst_id": inst,
        "opt_type": opt_type,
        "opt_type_label": label,
        "source": source_key,
        "source_label": source_label,
        "source_plan_id": source_plan_id,
        "pos": pos,
        "exp_time_ms": exp_ms,
        "target_monitor": _format_options_target(p),
        "pnl": round(pnl, 4) if pnl is not None else None,
    }


def _format_hedge_item(plan: dict[str, Any]) -> dict[str, Any]:
    pid = plan.get("id")
    underlying = plan.get("underlying") or "-"
    plan_type = plan.get("plan_type") or ""
    status = str(plan.get("status") or "")
    summary = plan.get("contracts_summary") or ""
    plan_type_label = OPTIONS_SOURCE_LABELS.get(plan_type, plan_type)
    active = status in HEDGE_ACTIVE_STATUSES
    status_label = "进行中" if active else (status or "—")
    return {
        "id": pid,
        "kind": "hedge_plan",
        "tab": "hedge_plan",
        "title": f"对冲 #{pid} {underlying}",
        "subtitle": " · ".join(x for x in (plan_type_label, status_label, summary) if x),
        "underlying": underlying,
        "plan_type": plan_type,
        "plan_type_label": plan_type_label,
        "status": status,
        "status_label": status_label,
        "status_active": active,
        "contracts_summary": summary,
    }


def _format_key_item(kd: dict[str, Any]) -> dict[str, Any]:
    sym = kd.get("exchange_symbol") or kd.get("symbol") or "-"
    direction = str(kd.get("direction") or "long").lower()
    signal = kd.get("signal_type") or kd.get("key_signal_type") or kd.get("monitor_type") or ""
    upper = _safe_float(kd.get("upper"))
    lower = _safe_float(kd.get("lower"))
    subtitle_parts = []
    if signal:
        subtitle_parts.append(str(signal))
    if upper is not None or lower is not None:
        subtitle_parts.append(
            f"上{upper if upper is not None else '-'} / 下{lower if lower is not None else '-'}"
        )
    return {
        "id": kd.get("id"),
        "kind": "key",
        "tab": "key_monitor",
        "title": f"{sym} {_dir_label(direction)}",
        "subtitle": " · ".join(subtitle_parts),
        "symbol": sym,
        "direction": direction,
        "direction_label": _dir_label(direction),
        "upper": upper,
        "lower": lower,
        "status": kd.get("status") or "active",
    }


def _format_trend_item(td: dict[str, Any]) -> dict[str, Any]:
    sym = td.get("exchange_symbol") or td.get("symbol") or "-"
    direction = str(td.get("direction") or "long").lower()
    status = td.get("status") or "active"
    entry = _safe_float(td.get("entry_price") or td.get("trigger_price"))
    return {
        "id": td.get("id"),
        "kind": "trend",
        "tab": "strategy",
        "title": f"趋势回调 {sym} {_dir_label(direction)}",
        "subtitle": f"状态 {status}",
        "symbol": sym,
        "direction": direction,
        "direction_label": _dir_label(direction),
        "entry": entry,
        "status": status,
    }


def _format_roll_item(rd: dict[str, Any]) -> dict[str, Any]:
    sym = rd.get("exchange_symbol") or rd.get("symbol") or "-"
    direction = str(rd.get("direction") or "long").lower()
    status = rd.get("status") or "active"
    return {
        "id": rd.get("id"),
        "kind": "roll",
        "tab": "strategy",
        "title": f"顺势加仓 {sym} {_dir_label(direction)}",
        "subtitle": f"状态 {status}",
        "symbol": sym,
        "direction": direction,
        "direction_label": _dir_label(direction),
        "status": status,
    }


def _table_exists(conn, name: str) -> bool:
    try:
        row = conn.execute(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1",
            (name,),
        ).fetchone()
        return bool(row)
    except Exception:
        return False


def collect_orders(conn) -> list[dict[str, Any]]:
    if not _table_exists(conn, "order_monitors"):
        return []
    rows = conn.execute(
        "SELECT * FROM order_monitors WHERE status='active' ORDER BY id DESC"
    ).fetchall()
    return [_format_order_item(_row_dict(r)) for r in rows]


def collect_keys(conn) -> list[dict[str, Any]]:
    if not _table_exists(conn, "key_monitors"):
        return []
    rows = conn.execute("SELECT * FROM key_monitors ORDER BY id DESC").fetchall()
    return [_format_key_item(_row_dict(r)) for r in rows]


def collect_trends(conn) -> list[dict[str, Any]]:
    if not _table_exists(conn, "trend_pullback_plans"):
        return []
    try:
        rows = conn.execute(
            "SELECT * FROM trend_pullback_plans WHERE status='active' ORDER BY id DESC"
        ).fetchall()
    except Exception:
        return []
    return [_format_trend_item(_row_dict(r)) for r in rows]


def collect_rolls(conn) -> list[dict[str, Any]]:
    if not _table_exists(conn, "roll_groups") or not _table_exists(conn, "order_monitors"):
        return []
    try:
        rows = conn.execute(
            """SELECT g.* FROM roll_groups g
               INNER JOIN order_monitors m ON m.id = g.order_monitor_id AND m.status='active'
               WHERE g.status='active' ORDER BY g.id DESC"""
        ).fetchall()
    except Exception:
        return []
    return [_format_roll_item(_row_dict(r)) for r in rows]


def collect_hedge_plans(conn) -> list[dict[str, Any]]:
    if not _table_exists(conn, "hedge_plans"):
        return []
    try:
        from lib.hedge_plan.hedge_plan_db import attach_legs_to_plans, list_plans

        rows: list[dict[str, Any]] = []
        for status in ("opening", "active", "partial"):
            rows.extend(list_plans(conn, status=status, limit=80))
        rows.sort(key=lambda row: int(row.get("id") or 0), reverse=True)
        plans = attach_legs_to_plans(conn, rows)
        return [_format_hedge_item(p) for p in plans]
    except Exception:
        return []


def collect_options_items(
    fetch_options_positions: Optional[Callable[[], list[dict[str, Any]]]] = None,
    *,
    conn=None,
) -> list[dict[str, Any]]:
    if not callable(fetch_options_positions):
        return []
    try:
        raw = fetch_options_positions() or []
    except Exception:
        return []
    pe_map: dict[str, dict[str, Any]] = {}
    tgt_map: dict[str, dict[str, Any]] = {}
    hedge_map: dict[str, dict[str, Any]] = {}
    if conn is not None:
        try:
            from lib.options.options_profit_exit_lib import profit_exit_by_inst
            from lib.options.options_target_lib import targets_by_inst
            from lib.hedge_plan.hedge_plan_db import active_options_targets_by_inst

            pe_map = profit_exit_by_inst(conn)
            tgt_map = targets_by_inst(conn)
            hedge_map = active_options_targets_by_inst(conn)
        except Exception:
            pe_map, tgt_map, hedge_map = {}, {}, {}
    out: list[dict[str, Any]] = []
    for p in raw:
        if not isinstance(p, dict):
            continue
        row = dict(p)
        inst = str(row.get("inst_id") or row.get("instId") or "").strip()
        mon = tgt_map.get(inst)
        if mon:
            row["target_index"] = mon.get("target_index")
        pe = pe_map.get(inst)
        if pe:
            row["profit_exit_enabled"] = pe.get("profit_exit_enabled")
            row["profit_exit_mult"] = pe.get("profit_exit_mult")
        hedge = hedge_map.get(inst)
        if hedge:
            row["hedge_plan_target"] = hedge
            if not mon:
                row["target_index"] = hedge.get("target_index")
        out.append(_format_options_item(row, conn=conn))
    return out


def _swap_symbol_candidates(row: dict[str, Any]) -> list[str]:
    """优先永续 symbol(含 settle),避免用现货 BTC/USDT 查到 contractSize=1."""
    raw: list[str] = []
    for key in ("symbol", "exchange_symbol", "price_symbol"):
        s = str(row.get(key) or "").strip()
        if s and s not in raw:
            raw.append(s)
    swapish: list[str] = []
    others: list[str] = []
    for s in raw:
        if ":" in s:
            swapish.append(s)
            continue
        others.append(s)
        if "/" in s:
            base, quote = s.split("/", 1)
            q = quote.split(":")[0].strip()
            if base and q:
                swapish.append(f"{base}/{q}:{q}")
    out: list[str] = []
    for s in swapish + others:
        if s and s not in out:
            out.append(s)
    return out


def _resolve_contract_size(
    row_or_sym: Any,
    *,
    get_contract_size: Optional[Callable[[str], Any]] = None,
) -> float:
    if not callable(get_contract_size):
        return 1.0
    if isinstance(row_or_sym, dict):
        candidates = _swap_symbol_candidates(row_or_sym)
    else:
        sym = str(row_or_sym or "").strip()
        candidates = _swap_symbol_candidates({"symbol": sym}) if sym else []
    for sym in candidates:
        try:
            cs = float(get_contract_size(sym) or 0)
            if cs > 0:
                return cs
        except Exception:
            continue
    return 1.0


def _fill_order_pnl_fields(row: dict[str, Any], *, mark: Optional[float], contract_size: float) -> None:
    """按线性 U 本位补看板「盈利金额 / 浮盈」."""
    direction = str(row.get("direction") or "long").lower()
    entry = _safe_float(row.get("entry"))
    contracts = _safe_float(row.get("contracts"))
    tp = _safe_float(row.get("take_profit"))
    if entry is None or contracts is None or contracts <= 0:
        return
    cs = float(contract_size) if contract_size and contract_size > 0 else 1.0
    if mark is not None:
        try:
            from lib.market.position_metrics_lib import estimate_linear_swap_upnl_usdt

            upnl = estimate_linear_swap_upnl_usdt(direction, entry, mark, contracts, cs)
            if upnl is not None:
                row["float_pnl"] = upnl
        except Exception:
            pass
    if tp is not None and tp > 0:
        try:
            try:
                d = (direction or "long").lower()
                e, t, c, cs_f = float(entry), float(tp), float(contracts), float(cs)
                profit = (t - e) * c * cs_f if d == "long" else (e - t) * c * cs_f
            except Exception:
                profit = None
            if profit is not None:
                row["tp_profit"] = round(float(profit), 2)
        except Exception:
            pass


def enrich_order_items_with_marks(
    items: list[dict[str, Any]],
    *,
    get_price: Optional[Callable[[str], Any]] = None,
    get_contract_size: Optional[Callable[[str], Any]] = None,
) -> list[dict[str, Any]]:
    """后台聚合时补标记价,并按张数×合约面值估算盈利金额/浮盈."""
    if not items:
        return items
    if not callable(get_price) and not callable(get_contract_size):
        return items
    out: list[dict[str, Any]] = []
    for it in items:
        row = dict(it)
        # 标记价:先试 price_symbol,再试永续候选
        mark = _safe_float(row.get("mark_price"))
        if callable(get_price):
            ordered: list[str] = []
            for s in [str(row.get("price_symbol") or "").strip()] + _swap_symbol_candidates(row):
                if s and s not in ordered:
                    ordered.append(s)
            for sym in ordered:
                try:
                    px = get_price(sym)
                except Exception:
                    px = None
                mark = _safe_float(px)
                if mark is not None:
                    row["mark_price"] = mark
                    break
        cs = _resolve_contract_size(row, get_contract_size=get_contract_size)
        _fill_order_pnl_fields(row, mark=mark, contract_size=cs)
        out.append(row)
    return out


def build_instance_dashboard_payload(
    conn,
    *,
    fetch_options_positions: Optional[Callable[[], list[dict[str, Any]]]] = None,
    hedge_enabled: bool = False,
) -> dict[str, Any]:
    orders = collect_orders(conn)
    keys = collect_keys(conn)
    options_items = collect_options_items(fetch_options_positions, conn=conn)
    hedge_items = collect_hedge_plans(conn)  # 始终展示进行中计划,与当前交易模式无关
    # hedge_enabled 仅影响「新建」入口,不隐藏已有仓
    now = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S")
    return {
        "ok": True,
        "updated_at": now,
        "orders": {
            "title": "实盘下单",
            "count": 0,
            "items": [],
            "tab": "options",
            "removed": True,
        },
        "keys": {"title": "关键位监控", "count": len(keys), "items": keys, "tab": "key_monitor"},
        "strategy": {
            "title": "策略交易",
            "count": 0,
            "items": [],
            "trends": [],
            "rolls": [],
            "tab": "strategy",
            "removed": True,
        },
        "options": {
            "title": "期权持仓",
            "count": len(options_items),
            "items": options_items,
            "visible": len(options_items) > 0,
            "tab": "options",
        },
        "hedge_plan": {
            "title": "对冲计划",
            "count": len(hedge_items),
            "items": hedge_items,
            "visible": len(hedge_items) > 0,
            "tab": "hedge_plan",
        },
    }