"""币本位期权:USDT↔标的币现货桥与本地状态."""
from __future__ import annotations

import sqlite3
import time
from typing import Any

from lib.options.options_margin_mode_lib import spot_quote_inst_id


BRIDGE_BOUGHT = "bought_pending_open"
BRIDGE_HOLDING = "holding"
BRIDGE_PENDING_SELL = "pending_sell_spot"
BRIDGE_CLOSED = "closed"

# 拉取 instruments 失败时的兜底(与 OKX 现货常见 minSz 对齐)
_SPOT_MIN_SZ_FALLBACK: dict[str, float] = {"ETH": 0.0001, "BTC": 0.00001}
_SPOT_MIN_SZ_CACHE: dict[str, tuple[float, float]] = {}  # ccy -> (min_sz, mono_ts)
_SPOT_MIN_SZ_TTL_SEC = 3600.0


def spot_min_sell_sz(ex: Any | None, underlying: str) -> float:
    """ETH/BTC-USDT 现货最小下单量(base ccy);粉尘低于此值不卖、不展示."""
    ccy = (underlying or "ETH").strip().upper() or "ETH"
    fb = float(_SPOT_MIN_SZ_FALLBACK.get(ccy) or _SPOT_MIN_SZ_FALLBACK["ETH"])
    now = time.monotonic()
    cached = _SPOT_MIN_SZ_CACHE.get(ccy)
    if cached and (now - cached[1]) < _SPOT_MIN_SZ_TTL_SEC:
        return float(cached[0])
    min_sz = fb
    if ex is not None:
        try:
            inst_id = spot_quote_inst_id(ccy)
            rows = ex.public_get_public_instruments({"instType": "SPOT", "instId": inst_id}) or {}
            data = rows.get("data") if isinstance(rows, dict) else None
            if not data and hasattr(ex, "publicGetPublicInstruments"):
                rows = ex.publicGetPublicInstruments({"instType": "SPOT", "instId": inst_id}) or {}
                data = rows.get("data") if isinstance(rows, dict) else None
            row = (data or [None])[0] if data else None
            if isinstance(row, dict):
                raw = _safe_float(row.get("minSz"))
                if raw is not None and raw > 0:
                    min_sz = float(raw)
        except Exception:
            pass
    _SPOT_MIN_SZ_CACHE[ccy] = (min_sz, now)
    return min_sz


def is_sellable_spot_coin(ex: Any | None, underlying: str, amount: Any) -> bool:
    """数量是否达到现货最小可卖(达到才重试/展示)."""
    try:
        amt = float(amount or 0)
    except (TypeError, ValueError):
        return False
    if amt <= 0:
        return False
    return amt + 1e-15 >= spot_min_sell_sz(ex, underlying)


def display_sellable_coin(ex: Any | None, underlying: str, amount: Any) -> float:
    """展示用:不够 minSz 的粉尘归零."""
    try:
        amt = float(amount or 0)
    except (TypeError, ValueError):
        return 0.0
    if not is_sellable_spot_coin(ex, underlying, amt):
        return 0.0
    return amt


def ensure_bridge_table(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS options_spot_bridge (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            underlying TEXT NOT NULL,
            status TEXT NOT NULL,
            budget_usdt REAL,
            buy_ord_id TEXT,
            coin_bought REAL,
            sell_ord_id TEXT,
            coin_sold REAL,
            usdt_recovered REAL,
            inst_id TEXT,
            message TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            closed_at TIMESTAMP
        )
        """
    )
    conn.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_options_spot_bridge_status
        ON options_spot_bridge(status)
        """
    )


def list_open_bridges(conn: sqlite3.Connection) -> list[dict[str, Any]]:
    ensure_bridge_table(conn)
    cur = conn.execute(
        """
        SELECT id, underlying, status, budget_usdt, buy_ord_id, coin_bought,
               sell_ord_id, coin_sold, usdt_recovered, inst_id, message,
               created_at, updated_at, closed_at
        FROM options_spot_bridge
        WHERE status IN (?, ?, ?)
        ORDER BY id DESC
        """,
        (BRIDGE_BOUGHT, BRIDGE_HOLDING, BRIDGE_PENDING_SELL),
    )
    cols = [d[0] for d in cur.description]
    return [dict(zip(cols, row)) for row in cur.fetchall()]


def has_unfinished_bridge(conn: sqlite3.Connection) -> bool:
    return bool(list_open_bridges(conn))


def insert_bridge(
    conn: sqlite3.Connection,
    *,
    underlying: str,
    status: str,
    budget_usdt: float | None = None,
    buy_ord_id: str | None = None,
    coin_bought: float | None = None,
    inst_id: str | None = None,
    message: str | None = None,
) -> int:
    ensure_bridge_table(conn)
    cur = conn.execute(
        """
        INSERT INTO options_spot_bridge(
            underlying, status, budget_usdt, buy_ord_id, coin_bought, inst_id, message, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
        """,
        (
            (underlying or "ETH").upper(),
            status,
            budget_usdt,
            buy_ord_id,
            coin_bought,
            inst_id,
            message,
        ),
    )
    conn.commit()
    return int(cur.lastrowid)


def add_bridge_usdt_recovered(
    conn: sqlite3.Connection,
    bridge_id: int,
    delta_usdt: float | None,
) -> float | None:
    """累加桥上卖币收回的 USDT(开仓多余卖回 + 平仓后卖回)."""
    ensure_bridge_table(conn)
    try:
        d = float(delta_usdt or 0)
    except (TypeError, ValueError):
        return None
    if d <= 0 or bridge_id is None:
        return None
    row = conn.execute(
        "SELECT usdt_recovered FROM options_spot_bridge WHERE id=?",
        (int(bridge_id),),
    ).fetchone()
    if not row:
        return None
    prev = _safe_float(row["usdt_recovered"] if hasattr(row, "keys") else row[0]) or 0.0
    new_v = round(float(prev) + d, 8)
    conn.execute(
        """
        UPDATE options_spot_bridge
        SET usdt_recovered = ?, updated_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (new_v, int(bridge_id)),
    )
    return new_v


def cash_pnl_usdt_from_bridge(conn: sqlite3.Connection, bridge_id: int | None) -> float | None:
    """桥上真盈亏(U)=累计卖回 USDT − 开仓买币预算.

    与账户 after−before 等价(一次一仓、无中途划转时);可作口径 A 回退.
    不是期权腿币盈亏(口径 B).
    """
    if bridge_id is None:
        return None
    ensure_bridge_table(conn)
    row = conn.execute(
        """
        SELECT budget_usdt, usdt_recovered
        FROM options_spot_bridge WHERE id = ?
        """,
        (int(bridge_id),),
    ).fetchone()
    if not row:
        return None
    spent = _safe_float(row["budget_usdt"] if hasattr(row, "keys") else row[0])
    recovered = _safe_float(row["usdt_recovered"] if hasattr(row, "keys") else row[1])
    if spent is None or recovered is None:
        return None
    return round(float(recovered) - float(spent), 4)


def update_bridge(
    conn: sqlite3.Connection,
    bridge_id: int,
    *,
    status: str | None = None,
    buy_ord_id: str | None = None,
    coin_bought: float | None = None,
    sell_ord_id: str | None = None,
    coin_sold: float | None = None,
    usdt_recovered: float | None = None,
    inst_id: str | None = None,
    message: str | None = None,
    close: bool = False,
) -> None:
    ensure_bridge_table(conn)
    fields: list[str] = ["updated_at=CURRENT_TIMESTAMP"]
    vals: list[Any] = []
    if status is not None:
        fields.append("status=?")
        vals.append(status)
    if buy_ord_id is not None:
        fields.append("buy_ord_id=?")
        vals.append(buy_ord_id)
    if coin_bought is not None:
        fields.append("coin_bought=?")
        vals.append(coin_bought)
    if sell_ord_id is not None:
        fields.append("sell_ord_id=?")
        vals.append(sell_ord_id)
    if coin_sold is not None:
        fields.append("coin_sold=?")
        vals.append(coin_sold)
    if usdt_recovered is not None:
        fields.append("usdt_recovered=?")
        vals.append(usdt_recovered)
    if inst_id is not None:
        fields.append("inst_id=?")
        vals.append(inst_id)
    if message is not None:
        fields.append("message=?")
        vals.append(message)
    if close or status == BRIDGE_CLOSED:
        fields.append("closed_at=CURRENT_TIMESTAMP")
    vals.append(int(bridge_id))
    conn.execute(
        f"UPDATE options_spot_bridge SET {', '.join(fields)} WHERE id=?",
        vals,
    )
    conn.commit()


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 fetch_trading_coin_available(ex: Any, ccy: str) -> float | None:
    """交易账户标的币可用."""
    from lib.exchange.okx_options_lib import _extract_ccy_free, _safe_float as _sf

    ccy_u = (ccy or "").upper()
    if not ccy_u:
        return None
    try:
        bal = ex.fetch_balance(params={"type": "trading"})
        free = _extract_ccy_free(bal, ccy_u)
        if free is not None:
            return float(free)
        # 部分账户结构只有 total
        from lib.exchange.okx_options_lib import _extract_ccy_balance

        tot = _extract_ccy_balance(bal, ccy_u)
        return float(tot) if tot is not None else None
    except Exception:
        return None


def spot_market_buy_coin_with_usdt(
    ex: Any,
    *,
    underlying: str,
    usdt_amount: float,
) -> dict[str, Any]:
    """交易账户:用 USDT 市价买入标的币."""
    if usdt_amount <= 0:
        return {"ok": False, "msg": "USDT 数量须大于 0"}
    inst_id = spot_quote_inst_id(underlying)
    try:
        body = {
            "instId": inst_id,
            "tdMode": "cash",
            "side": "buy",
            "ordType": "market",
            "sz": str(usdt_amount),
            "tgtCcy": "quote_ccy",
        }
        resp = ex.private_post_trade_order(body)
        data = (resp or {}).get("data") or []
        if data and str(data[0].get("sCode")) == "0":
            return {
                "ok": True,
                "inst_id": inst_id,
                "ord_id": str(data[0].get("ordId") or ""),
                "data": data[0],
                "raw": resp,
            }
        from lib.exchange.okx_options_lib import _okx_trade_error_message

        return {"ok": False, "msg": _okx_trade_error_message(resp=resp), "raw": resp}
    except Exception as e:
        from lib.exchange.okx_options_lib import _okx_trade_error_message

        return {"ok": False, "msg": _okx_trade_error_message(e)}


def spot_market_sell_coin_to_usdt(
    ex: Any,
    *,
    underlying: str,
    coin_amount: float | None = None,
) -> dict[str, Any]:
    """交易账户:市价卖出标的币换 USDT.

    coin_amount 有值时按 min(可用, coin_amount) 卖,避免误卖无关余额;
    未传时卖光该币可用(空仓扫尾).
    """
    ccy = (underlying or "ETH").upper()
    avail = fetch_trading_coin_available(ex, ccy)
    if avail is None or float(avail) <= 0:
        return {"ok": False, "msg": f"交易账户无可用 {ccy}"}
    amt = float(avail)
    if coin_amount is not None:
        try:
            lim = float(coin_amount)
        except (TypeError, ValueError):
            lim = 0.0
        if lim > 0:
            amt = min(amt, lim)
    if amt <= 0:
        return {"ok": False, "msg": f"{ccy} 数量须大于 0"}
    min_sz = spot_min_sell_sz(ex, ccy)
    if amt + 1e-15 < min_sz:
        return {"ok": False, "msg": f"{ccy} 可卖数量过小(低于现货最小 {min_sz:g})"}
    # 留一点粉尘避免精度拒单
    sell_sz = max(0.0, amt * 0.999)
    if sell_sz + 1e-15 < min_sz:
        return {"ok": False, "msg": f"{ccy} 可卖数量过小(低于现货最小 {min_sz:g})"}
    inst_id = spot_quote_inst_id(ccy)
    try:
        # 现货卖出数量精度:截到 8 位
        sz = f"{sell_sz:.8f}".rstrip("0").rstrip(".")
        if not sz or float(sz) <= 0:
            return {"ok": False, "msg": f"{ccy} 可卖数量过小"}
        body = {
            "instId": inst_id,
            "tdMode": "cash",
            "side": "sell",
            "ordType": "market",
            "sz": sz,
            "tgtCcy": "base_ccy",
        }
        resp = ex.private_post_trade_order(body)
        data = (resp or {}).get("data") or []
        if data and str(data[0].get("sCode")) == "0":
            return {
                "ok": True,
                "inst_id": inst_id,
                "ord_id": str(data[0].get("ordId") or ""),
                "coin_sold": float(sz),
                "data": data[0],
                "raw": resp,
            }
        from lib.exchange.okx_options_lib import _okx_trade_error_message

        return {"ok": False, "msg": _okx_trade_error_message(resp=resp), "raw": resp}
    except Exception as e:
        from lib.exchange.okx_options_lib import _okx_trade_error_message

        return {"ok": False, "msg": _okx_trade_error_message(e)}


def rollback_bought_coin_to_usdt(
    conn: sqlite3.Connection,
    ex: Any,
    *,
    bridge_id: int,
    underlying: str,
    reason: str = "",
    coin_amount: float | None = None,
) -> dict[str, Any]:
    """买币后开期权失败:按本桥买入量封顶卖回 USDT 并关闭桥."""
    ensure_bridge_table(conn)
    cap = coin_amount
    if cap is None:
        try:
            row = conn.execute(
                "SELECT coin_bought FROM options_spot_bridge WHERE id=?",
                (int(bridge_id),),
            ).fetchone()
            if row is not None:
                cap = _safe_float(row["coin_bought"] if hasattr(row, "keys") else row[0])
        except Exception:
            cap = None
    # 卖币(模拟盘)会另开连接写钱包,须先释放本连接事务锁
    try:
        conn.commit()
    except Exception:
        pass
    sell = spot_market_sell_coin_to_usdt(
        ex,
        underlying=underlying,
        coin_amount=float(cap) if cap is not None and float(cap) > 0 else None,
    )
    if not sell.get("ok"):
        update_bridge(
            conn,
            bridge_id,
            status=BRIDGE_PENDING_SELL,
            message=(reason or "") + " | 回滚卖币失败: " + str(sell.get("msg") or ""),
        )
        return {"ok": False, "msg": sell.get("msg") or "回滚卖币失败", "bridge_status": BRIDGE_PENDING_SELL}
    update_bridge(
        conn,
        bridge_id,
        status=BRIDGE_CLOSED,
        sell_ord_id=str(sell.get("ord_id") or ""),
        coin_sold=_safe_float(sell.get("coin_sold")),
        message=reason or "开仓失败已卖回 USDT",
        close=True,
    )
    return {"ok": True, "sell": sell, "bridge_status": BRIDGE_CLOSED}


def sell_residual_after_option_flat(
    conn_or_get_db: Any,
    ex: Any,
    *,
    underlying: str,
    inst_id: str | None = None,
) -> dict[str, Any]:
    """期权已平:卖光交易户该标的可卖币(一次一仓,含盈利币).

    规格:平仓卖回禁止按开仓权利金 coin_bought 封顶;仍有可卖则不关桥、不结算.
    可传 get_db 可调用对象(推荐)或已打开的 Connection。
    卖币(尤其模拟盘写钱包)期间绝不握着业务连接,否则会 database is locked。
    """
    get_db = conn_or_get_db if callable(conn_or_get_db) else None
    bid: int | None = None
    want_inst = str(inst_id or "").strip()

    def _pick_bridge(bridges: list[dict[str, Any]]) -> dict[str, Any] | None:
        uly_u = (underlying or "").upper()
        matched: list[dict[str, Any]] = []
        for b in bridges:
            if str(b.get("status")) not in (BRIDGE_HOLDING, BRIDGE_PENDING_SELL, BRIDGE_BOUGHT):
                continue
            if uly_u and str(b.get("underlying") or "").upper() != uly_u:
                continue
            matched.append(b)
        if not matched:
            return None
        if want_inst:
            for b in matched:
                if str(b.get("inst_id") or "").strip() == want_inst:
                    return b
        return matched[0]

    if get_db is not None:
        conn = get_db()
        try:
            ensure_bridge_table(conn)
            target = _pick_bridge(list_open_bridges(conn))
            if target is not None:
                bid = int(target["id"])
            try:
                conn.commit()
            except Exception:
                pass
        finally:
            try:
                conn.close()
            except Exception:
                pass
    else:
        conn = conn_or_get_db
        ensure_bridge_table(conn)
        target = _pick_bridge(list_open_bridges(conn))
        if target is not None:
            bid = int(target["id"])
        try:
            conn.commit()
        except Exception:
            pass

    # 粉尘:不够现货最小下单量则忽略,不再重试;有桥则直接关闭
    try:
        avail_chk = fetch_trading_coin_available(ex, underlying)
    except Exception:
        avail_chk = None
    if avail_chk is not None and not is_sellable_spot_coin(ex, underlying, avail_chk):
        dust_msg = "粉尘低于现货最小可卖,已忽略"
        if bid is None:
            return {"ok": True, "msg": dust_msg, "skipped": True, "dust": True, "sell_complete": True}
        def _close_dust(c: sqlite3.Connection) -> dict[str, Any]:
            update_bridge(
                c,
                bid,
                status=BRIDGE_CLOSED,
                inst_id=inst_id,
                message=dust_msg,
                close=True,
            )
            return {
                "ok": True,
                "msg": dust_msg,
                "skipped": True,
                "dust": True,
                "sell_complete": True,
                "bridge_id": bid,
                "bridge_status": BRIDGE_CLOSED,
            }
        if get_db is not None:
            conn_d = get_db()
            try:
                return _close_dust(conn_d)
            finally:
                try:
                    conn_d.close()
                except Exception:
                    pass
        return _close_dust(conn_or_get_db)

    # 一次一仓空仓后卖光可用(含盈利币);禁止用 coin_bought/权利金封顶
    # 此处无业务连接:模拟盘 convert 可安全另开写钱包
    sell = spot_market_sell_coin_to_usdt(
        ex,
        underlying=underlying,
        coin_amount=None,
    )
    if bid is None:
        if not sell.get("ok"):
            msg = str(sell.get("msg") or "")
            if "无可用" in msg or "过小" in msg:
                return {"ok": True, "msg": "无残留币需卖回", "skipped": True, "sell_complete": True}
            return {"ok": False, "msg": msg, "bridge_status": BRIDGE_PENDING_SELL, "sell_complete": False}
        # 再确认无可卖残留
        try:
            left = fetch_trading_coin_available(ex, underlying)
        except Exception:
            left = None
        if left is not None and is_sellable_spot_coin(ex, underlying, left):
            return {
                "ok": False,
                "msg": f"卖回未完成,仍有可卖 {str(underlying or '').upper()}",
                "incomplete": True,
                "sell_complete": False,
                "sell": sell,
            }
        return {"ok": True, "sell": sell, "bridge_status": None, "sell_complete": True}

    def _apply_bridge(c: sqlite3.Connection) -> dict[str, Any]:
        if not sell.get("ok"):
            msg = str(sell.get("msg") or "卖回 USDT 失败")
            # 交易所拒单「过小」也按粉尘关闭桥,避免一直 pending 重试
            if "过小" in msg or "minimum" in msg.lower() or "min sz" in msg.lower():
                update_bridge(
                    c,
                    bid,
                    status=BRIDGE_CLOSED,
                    inst_id=inst_id,
                    message="粉尘低于现货最小可卖,已忽略",
                    close=True,
                )
                return {
                    "ok": True,
                    "msg": "粉尘低于现货最小可卖,已忽略",
                    "skipped": True,
                    "dust": True,
                    "sell_complete": True,
                    "bridge_id": bid,
                    "bridge_status": BRIDGE_CLOSED,
                }
            update_bridge(
                c,
                bid,
                status=BRIDGE_PENDING_SELL,
                inst_id=inst_id,
                message=msg,
            )
            return {
                "ok": False,
                "msg": msg,
                "bridge_id": bid,
                "bridge_status": BRIDGE_PENDING_SELL,
                "sell_complete": False,
            }
        recovered = _safe_float(sell.get("usdt_recovered"))
        if recovered is not None and recovered > 0:
            add_bridge_usdt_recovered(c, bid, recovered)
        # 卖后仍有可卖 → 不关桥、不结算(盈利币未卖回时曾误冻亏损)
        try:
            left = fetch_trading_coin_available(ex, underlying)
        except Exception:
            left = None
        if left is not None and is_sellable_spot_coin(ex, underlying, left):
            left_msg = f"卖回未完成,仍有可卖 {str(underlying or '').upper()}({float(left):g})"
            update_bridge(
                c,
                bid,
                status=BRIDGE_PENDING_SELL,
                sell_ord_id=str(sell.get("ord_id") or ""),
                coin_sold=_safe_float(sell.get("coin_sold")),
                inst_id=inst_id,
                message=left_msg,
            )
            return {
                "ok": False,
                "msg": left_msg,
                "incomplete": True,
                "sell_complete": False,
                "sell": sell,
                "bridge_id": bid,
                "bridge_status": BRIDGE_PENDING_SELL,
            }
        update_bridge(
            c,
            bid,
            status=BRIDGE_CLOSED,
            sell_ord_id=str(sell.get("ord_id") or ""),
            coin_sold=_safe_float(sell.get("coin_sold")),
            inst_id=inst_id,
            message="期权已平,币已卖回 USDT",
            close=True,
        )
        return {
            "ok": True,
            "sell": sell,
            "bridge_id": bid,
            "bridge_status": BRIDGE_CLOSED,
            "sell_complete": True,
        }

    if get_db is not None:
        conn2 = get_db()
        try:
            return _apply_bridge(conn2)
        finally:
            try:
                conn2.close()
            except Exception:
                pass
    return _apply_bridge(conn_or_get_db)


def clear_dust_pending_sell_bridges(
    conn_or_get_db: Any,
    ex: Any | None,
) -> int:
    """待卖回桥若交易户只剩粉尘(或已空),关闭桥并返回关闭条数.

    推荐传 get_db:余额查询(模拟盘会另开连接)绝不能套在未关闭的业务连接上。
    """
    if ex is None:
        return 0
    get_db = conn_or_get_db if callable(conn_or_get_db) else None

    def _pending(c: sqlite3.Connection) -> list[dict[str, Any]]:
        # pending_sell 始终可清粉尘;holding 仅在无对应标的期权仓时清(避免误关持仓中桥)
        return [
            b
            for b in list_open_bridges(c)
            if str(b.get("status") or "") in (BRIDGE_PENDING_SELL, BRIDGE_HOLDING)
        ]

    live_uly: set[str] = set()
    positions_ok = False
    try:
        from lib.exchange.okx_options_lib import fetch_option_positions

        raw_pos = fetch_option_positions(ex)
        if raw_pos is not None:
            positions_ok = True
            for p in raw_pos or []:
                try:
                    if abs(float(p.get("pos") or 0)) <= 1e-12:
                        continue
                except (TypeError, ValueError):
                    continue
                inst = str(p.get("instId") or p.get("inst_id") or "").strip()
                if inst:
                    live_uly.add(inst.split("-")[0].upper())
    except Exception:
        positions_ok = False
        live_uly = set()

    if get_db is not None:
        conn = get_db()
        try:
            pending = _pending(conn)
            try:
                conn.commit()
            except Exception:
                pass
        finally:
            try:
                conn.close()
            except Exception:
                pass
    else:
        conn = conn_or_get_db
        pending = _pending(conn)
        try:
            conn.commit()
        except Exception:
            pass

    to_close: list[dict[str, Any]] = []
    for b in pending:
        st = str(b.get("status") or "")
        uly = str(b.get("underlying") or "ETH").upper() or "ETH"
        # holding 仅在「确认无该标的期权仓」时才可按粉尘关闭;持仓查询失败绝不关 holding
        if st == BRIDGE_HOLDING and (not positions_ok or uly in live_uly):
            continue
        try:
            avail = fetch_trading_coin_available(ex, uly)
        except Exception:
            continue
        # 余额查不到时:用桥上记录的买入量判断粉尘,避免永久拦开仓
        if avail is None:
            cb = _safe_float(b.get("coin_bought"))
            if cb is None or is_sellable_spot_coin(ex, uly, cb):
                continue
            to_close.append(b)
            continue
        if is_sellable_spot_coin(ex, uly, avail):
            continue
        to_close.append(b)
    if not to_close:
        return 0

    def _close_all(c: sqlite3.Connection) -> int:
        n = 0
        for b in to_close:
            try:
                update_bridge(
                    c,
                    int(b["id"]),
                    status=BRIDGE_CLOSED,
                    message="粉尘低于现货最小可卖,已忽略",
                    close=True,
                )
                n += 1
            except Exception:
                pass
        return n

    if get_db is not None:
        conn2 = get_db()
        try:
            return _close_all(conn2)
        finally:
            try:
                conn2.close()
            except Exception:
                pass
    return _close_all(conn_or_get_db)


def bridge_blocks_new_open_msg(
    conn: sqlite3.Connection,
    ex: Any | None = None,
) -> str | None:
    _ = ex  # 清粉尘请调用方先 clear_dust_pending_sell_bridges(get_db, ex)
    bridges = list_open_bridges(conn)
    if not bridges:
        return None
    st = str(bridges[0].get("status") or "")
    if st == BRIDGE_PENDING_SELL:
        return "存在待卖回 USDT 的币本位桥残留,请先到期权页重试卖回后再开仓"
    if st == BRIDGE_BOUGHT:
        return "存在已买币未完成开仓的桥流程,请等待回滚或联系处理后重试"
    if st == BRIDGE_HOLDING:
        return "币本位桥仍在持仓中(一次仅一笔),请先平仓并卖回 USDT"
    return "存在未完成的币本位资金桥,暂不可开仓"


def mode_switch_block_msg(
    conn: sqlite3.Connection,
    ex: Any | None = None,
    *,
    get_db: Any | None = None,
) -> str | None:
    """有单笔期权仓或未完成桥时禁止切换本位."""
    try:
        if get_db is not None:
            clear_dust_pending_sell_bridges(get_db, ex)
    except Exception:
        pass
    if has_unfinished_bridge(conn):
        return "存在未完成的币本位资金桥,禁止切换期权本位模式"
    if ex is not None:
        try:
            from lib.exchange.okx_options_lib import fetch_option_positions

            rows = fetch_option_positions(ex) or []
            for p in rows:
                try:
                    pos = float(p.get("pos") or 0)
                except (TypeError, ValueError):
                    pos = 0.0
                if abs(pos) > 1e-12:
                    return "存在未平期权持仓,禁止切换期权本位模式"
        except Exception:
            pass
    # 本地 open 交易记录
    try:
        row = conn.execute(
            "SELECT COUNT(*) FROM options_trades WHERE status='open'"
        ).fetchone()
        n = int(row[0] if not isinstance(row, dict) else row.get("COUNT(*)") or list(row.values())[0])
        if n > 0:
            return "本地仍有未平期权记录,禁止切换期权本位模式"
    except Exception:
        pass
    return None