crypto_okx/lib/hedge_plan/hedge_plan_orders_lib.py

"""对冲计划开仓/平仓编排(可 dry_run 校验下单路径)."""
from __future__ import annotations

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


def _now() -> str:
    return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S")


def _env_bool(key: str, default: bool = False) -> bool:
    raw = (os.getenv(key) or "").strip().lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "on")


def open_order_mode() -> str:
    v = (os.getenv("HEDGE_PLAN_OPEN_ORDER") or "options_first").strip().lower()
    return v if v in ("options_first", "perp_first") else "options_first"


def manual_complete_on_partial() -> bool:
    """半腿失败后挂 partial 并手动补开(默认 true)."""
    return _env_bool("HEDGE_PLAN_MANUAL_COMPLETE_ON_PARTIAL", True)


def partial_auto_close_enabled() -> bool:
    """手动补开开启时强制关闭自动平,避免吃买卖价差."""
    if manual_complete_on_partial():
        return False
    return _env_bool("HEDGE_PLAN_PARTIAL_AUTO_CLOSE_OPTION", True)


def build_po_path_plan(body: dict[str, Any]) -> list[dict[str, Any]]:
    """永期下单路径清单(不交易)."""
    from lib.hedge_plan.hedge_plan_option_primary_lib import (
        is_option_primary,
        perp_direction_for_view,
    )

    opt_primary = is_option_primary(body)
    mode = "options_first" if opt_primary else open_order_mode()
    view = str(body.get("direction") or "long")
    perp_dir = perp_direction_for_view(view) if opt_primary else view
    opt = {
        "step": "options_buy_limit",
        "account": "options",
        "inst_id": body.get("opt_inst_id"),
        "sheets": float(body.get("sheets") or 1),
        "side": "buy",
        "price_hint": "ask",
    }
    perp = {
        "step": "perp_market_open",
        "account": "swap",
        "symbol": body.get("exchange_symbol"),
        "direction": perp_dir,
        "contracts": float(body.get("contracts") or 0),
        "tp": None if opt_primary else body.get("tp"),
        "sl": None if opt_primary else body.get("sl"),
        "attach_tpsl": False if opt_primary else True,
        "option_primary": opt_primary,
        "view_side": view,
    }
    return [opt, perp] if mode == "options_first" else [perp, opt]


def build_oo_path_plan(body: dict[str, Any]) -> list[dict[str, Any]]:
    return [
        {
            "step": "options_buy_limit",
            "account": "options",
            "leg": "a",
            "inst_id": (body.get("leg_a") or {}).get("inst_id"),
            "sheets": float((body.get("leg_a") or {}).get("sheets") or 1),
            "side": "buy",
            "price_hint": "ask",
        },
        {
            "step": "options_buy_limit",
            "account": "options",
            "leg": "b",
            "inst_id": (body.get("leg_b") or {}).get("inst_id"),
            "sheets": float((body.get("leg_b") or {}).get("sheets") or 1),
            "side": "buy",
            "price_hint": "ask",
        },
    ]


def _option_open_fill_timeout_sec() -> float:
    try:
        return max(2.0, float(os.getenv("OKX_OPTIONS_OPEN_FILL_TIMEOUT_SEC") or "12"))
    except (TypeError, ValueError):
        return 12.0


def _buy_option(
    cfg: dict[str, Any],
    *,
    inst_id: str,
    sheets: float,
    dry_run: bool,
) -> dict[str, Any]:
    from lib.exchange.okx_options_lib import (
        cap_option_buy_sheets_to_ask_depth,
        option_buy_liquidity_ok,
        wait_option_order_full_fill,
    )

    ex = cfg.get("exchange_options")
    quote_fn = cfg.get("quote_option_contract")
    place_fn = cfg.get("place_option_limit_order")
    td_buy = cfg.get("td_mode_for_option_buy")
    if not inst_id:
        return {"ok": False, "msg": "缺少期权合约"}
    if not callable(quote_fn) or ex is None:
        return {"ok": False, "msg": "期权报价能力未就绪"}
    q = quote_fn(ex, inst_id)
    if not q.get("ok"):
        return {"ok": False, "msg": q.get("msg") or "期权报价失败", "quote": q}
    ask = q.get("ask")
    ask_sz = q.get("ask_sz")
    can_open, block_msg = option_buy_liquidity_ok(ask, ask_sz)
    if not can_open:
        return {
            "ok": False,
            "msg": block_msg or q.get("open_block_msg") or "暂无卖一深度,无法买入",
            "quote": q,
            "mark": q.get("mark"),
            "ref_ask": q.get("ref_ask"),
            "can_open": False,
        }
    from lib.options.options_position_limit_lib import option_position_limit_block_msg

    pos_limit_msg = option_position_limit_block_msg(
        ex,
        opening_inst_id=inst_id,
        fetch_positions=cfg.get("fetch_option_positions"),
    )
    if pos_limit_msg:
        return {"ok": False, "msg": pos_limit_msg, "quote": q, "can_open": False}
    sheets_i = max(1, int(round(float(sheets))))
    requested_sheets = sheets_i
    capped, cap_msg = cap_option_buy_sheets_to_ask_depth(sheets_i, ask_sz, min_sz=1)
    if capped is None:
        return {"ok": False, "msg": cap_msg or "卖一深度不足,无法买入", "quote": q}
    if int(capped) < requested_sheets:
        return {
            "ok": False,
            "msg": f"卖一深度仅 {int(capped)} 张,不足请求 {requested_sheets} 张,拒绝缩量成交",
            "quote": q,
            "can_open": False,
            "requested_sheets": requested_sheets,
            "ask_sz": ask_sz,
        }
    sheets_i = int(capped)
    ct_mult = float(q.get("ct_mult") or 0.01)
    premium = float(ask) * sheets_i * ct_mult
    if dry_run:
        return {
            "ok": True,
            "dry_run": True,
            "inst_id": inst_id,
            "sheets": sheets_i,
            "ask": float(ask),
            "ask_sz": float(ask_sz),
            "premium": premium,
            "ct_mult": ct_mult,
            "tick_sz": q.get("tick_sz"),
            "meta": q.get("meta") or {},
            "strike": q.get("strike"),
            "exp_time": q.get("exp_time"),
            "opt_type": (q.get("meta") or {}).get("optType") or q.get("opt_type"),
            "can_open": True,
        }
    if not callable(place_fn):
        return {"ok": False, "msg": "期权限价下单未注入"}
    td = "isolated"
    if callable(td_buy):
        td = td_buy(cfg.get("options_td_mode") or "isolated")
    # IOC:能成交多少成交多少,剩余立即撤销;再校验是否完全成交
    order = place_fn(
        ex,
        inst_id=inst_id,
        side="buy",
        sheets=sheets_i,
        price=float(ask),
        td_mode=td,
        tick_sz=q.get("tick_sz"),
        ord_type="ioc",
    )
    if not order.get("ok"):
        return order
    ord_id = str((order.get("data") or {}).get("ordId") or "").strip()
    if not ord_id:
        return {"ok": False, "msg": "下单成功但未返回订单号", "order": order}
    fill = wait_option_order_full_fill(
        ex,
        inst_id=inst_id,
        ord_id=ord_id,
        need_sheets=sheets_i,
        timeout_sec=_option_open_fill_timeout_sec(),
        cancel_on_timeout=True,
    )
    if not fill.get("ok"):
        filled_n = int(fill.get("filled_sheets") or 0)
        orphan_close = None
        if filled_n > 0 and not dry_run:
            # 部分成交后撤单:尝试立刻平掉已成交,避免孤儿多头
            try:
                orphan_close = _sell_option(cfg, inst_id=inst_id, sheets=float(filled_n))
            except Exception as e:
                orphan_close = {"ok": False, "msg": str(e)}
        return {
            "ok": False,
            "msg": fill.get("msg") or "未完全成交,开仓失败",
            "inst_id": inst_id,
            "sheets": sheets_i,
            "ask": float(ask),
            "exchange_ord_id": ord_id,
            "filled_sheets": filled_n,
            "orphan_close": orphan_close,
            "order": order,
            "fill": fill,
            "can_open": False,
        }
    fill_px = float(fill.get("avg_px") or ask)
    filled_n = int(fill.get("filled_sheets") or sheets_i)
    premium = fill_px * filled_n * ct_mult
    return {
        "ok": True,
        "inst_id": inst_id,
        "sheets": filled_n,
        "ask": fill_px,
        "ask_sz": float(ask_sz),
        "premium": premium,
        "ct_mult": ct_mult,
        "tick_sz": q.get("tick_sz"),
        "meta": q.get("meta") or {},
        "strike": q.get("strike"),
        "exp_time": q.get("exp_time"),
        "opt_type": (q.get("meta") or {}).get("optType") or q.get("opt_type"),
        "exchange_ord_id": ord_id,
        "order": order,
        "fill": fill,
        "can_open": True,
    }


def _open_perp(
    cfg: dict[str, Any],
    *,
    symbol: str,
    direction: str,
    contracts: float,
    leverage: int,
    tp: Optional[float],
    sl: Optional[float],
    dry_run: bool,
    attach_tpsl: bool = True,
) -> dict[str, Any]:
    if not symbol or contracts <= 0:
        return {"ok": False, "msg": "永续符号或张数无效"}
    amount = float(contracts)
    to_prec = cfg.get("amount_to_precision")
    ex = cfg.get("exchange")
    if callable(to_prec) and ex is not None:
        try:
            amount = float(to_prec(symbol, amount))
        except Exception:
            pass
    if amount <= 0:
        return {"ok": False, "msg": "张数经精度舍入后为 0"}
    use_tpsl = bool(attach_tpsl) and tp is not None and sl is not None
    tp_v = float(tp) if use_tpsl else None
    sl_v = float(sl) if use_tpsl else None
    if dry_run:
        return {
            "ok": True,
            "dry_run": True,
            "symbol": symbol,
            "direction": direction,
            "contracts": amount,
            "leverage": leverage,
            "tp": tp_v,
            "sl": sl_v,
            "attach_tpsl": use_tpsl,
        }
    ensure = cfg.get("ensure_okx_live_ready")
    if callable(ensure):
        ok, msg = ensure()
        if not ok:
            return {"ok": False, "msg": msg or "实盘未就绪"}
    place = cfg.get("place_exchange_order")
    if not callable(place):
        return {"ok": False, "msg": "永续下单函数未注入"}
    try:
        order = place(
            symbol,
            direction,
            amount,
            leverage,
            stop_loss=sl_v,
            take_profit=tp_v,
        )
    except Exception as e:
        return {"ok": False, "msg": f"永续开仓失败: {e}"}
    return {
        "ok": True,
        "symbol": symbol,
        "direction": direction,
        "contracts": amount,
        "leverage": leverage,
        "tp": tp_v,
        "sl": sl_v,
        "attach_tpsl": use_tpsl,
        "order": order,
        "exchange_ord_id": str((order or {}).get("id") or (order or {}).get("info", {}).get("ordId") or ""),
    }


def _close_perp(
    cfg: dict[str, Any],
    *,
    symbol: str,
    direction: str,
    contracts: float,
    dry_run: bool = False,
) -> dict[str, Any]:
    """市价平永续(reduce-only);优先用注入的 close_exchange_order."""
    if not symbol:
        return {"ok": False, "msg": "永续符号无效"}
    if dry_run:
        return {
            "ok": True,
            "dry_run": True,
            "symbol": symbol,
            "direction": direction,
            "contracts": float(contracts or 0),
        }
    close_fn = cfg.get("close_exchange_order")
    if callable(close_fn):
        try:
            order = close_fn(
                {
                    "exchange_symbol": symbol,
                    "direction": direction,
                    "order_amount": float(contracts or 0),
                    "symbol": symbol,
                }
            )
            return {"ok": True, "symbol": symbol, "direction": direction, "order": order}
        except Exception as e:
            return {"ok": False, "msg": f"永续平仓失败: {e}"}
    # 回退:对向市价 reduce-only(若注入了 place + 支持)
    place = cfg.get("place_exchange_order")
    if not callable(place):
        return {"ok": False, "msg": "永续平仓函数未注入"}
    try:
        # 无 TP/SL 的对向单;依赖交易所 reduceOnly 由 place 实现不保证,优先 close_exchange_order
        side_dir = "short" if str(direction).lower() == "long" else "long"
        order = place(symbol, side_dir, float(contracts or 0), int(cfg.get("alt_leverage") or 5), None, None)
        return {"ok": True, "symbol": symbol, "direction": direction, "order": order, "note": "fallback_place"}
    except Exception as e:
        return {"ok": False, "msg": f"永续平仓失败: {e}"}


def _sell_option(
    cfg: dict[str, Any],
    *,
    inst_id: str,
    sheets: float,
    dry_run: bool = False,
) -> dict[str, Any]:
    """平期权:走买一限价 + 验仓;仅 fully_closed/already_flat 视为成功.

    对冲强平/目标平仓不启用 2× 回收门控(require_recycle_gate=False).
    """
    from lib.exchange.okx_options_lib import fetch_option_book_depth, fetch_option_positions
    from lib.options.options_close_exec_lib import close_option_by_bid1

    ex = cfg.get("exchange_options")
    quote_fn = cfg.get("quote_option_contract")
    if not inst_id:
        return {"ok": False, "msg": "缺少期权合约"}
    if not callable(quote_fn) or ex is None:
        return {"ok": False, "msg": "期权报价能力未就绪"}
    q = quote_fn(ex, inst_id)
    bid = q.get("bid") if q.get("ok") else None
    if bid is None or float(bid) <= 0:
        return {"ok": False, "msg": "暂无买一价,无法平期权"}
    sheets_i = max(1, int(round(float(sheets))))
    if dry_run:
        return {
            "ok": True,
            "dry_run": True,
            "inst_id": inst_id,
            "sheets": sheets_i,
            "bid": float(bid),
            "fully_closed": True,
        }
    if not callable(cfg.get("place_option_limit_order")):
        return {"ok": False, "msg": "期权平仓未注入"}
    close_cfg = dict(cfg)
    if not callable(close_cfg.get("fetch_option_positions")):
        close_cfg["fetch_option_positions"] = fetch_option_positions
    if not callable(close_cfg.get("fetch_option_book_depth")):
        close_cfg["fetch_option_book_depth"] = fetch_option_book_depth
    if "td_mode" not in close_cfg:
        close_cfg["td_mode"] = close_cfg.get("options_td_mode") or "isolated"
    result = close_option_by_bid1(
        close_cfg,
        ex,
        inst_id,
        sheets=sheets_i,
        require_recycle_gate=False,
    )
    out = dict(result or {})
    if out.get("already_flat"):
        # 二次验仓,避免一次空列表误判已平
        import time as _time

        _time.sleep(0.35)
        try:
            from lib.exchange.okx_options_lib import invalidate_option_positions_cache

            invalidate_option_positions_cache()
        except Exception:
            pass
        rows2 = close_cfg["fetch_option_positions"](ex)
        if rows2 is None:
            return {"ok": False, "msg": "二次验仓失败,未确认是否已平", "fully_closed": False}
        still = next((p for p in rows2 if str(p.get("instId")) == inst_id), None)
        still_sz = 0.0
        if still is not None:
            try:
                still_sz = abs(float(still.get("availPos") or still.get("pos") or 0))
            except (TypeError, ValueError):
                still_sz = 0.0
        if still is not None and still_sz >= 1:
            return {
                "ok": False,
                "msg": "二次验仓仍有持仓,拒绝 already_flat",
                "fully_closed": False,
            }
        out["ok"] = True
        out["fully_closed"] = True
        out.setdefault("bid", float(bid))
        return out
    if not out.get("ok"):
        out.setdefault("bid", float(bid))
        return out
    if not out.get("fully_closed"):
        return {
            "ok": False,
            "msg": out.get("msg") or "期权尚未完全平仓,将下轮重试",
            "bid": out.get("locked_bid_px") or float(bid),
            "fully_closed": False,
            "partial": True,
            "close": out,
        }
    out["bid"] = out.get("locked_bid_px") or float(bid)
    out["fully_closed"] = True
    return out


def _notify_partial(cfg: dict[str, Any], plan_type: str, msg: str, results: list[dict[str, Any]]) -> None:
    try:
        from lib.hedge_plan.hedge_plan_notify_lib import notify_partial_fail

        notify_partial_fail(cfg, plan_type=plan_type, msg=msg, results=results)
    except Exception:
        pass


def _park_partial(
    cfg: dict[str, Any],
    *,
    plan_type: str,
    body: dict[str, Any],
    missing_leg: str,
    msg: str,
    path: list[dict[str, Any]],
    results: list[dict[str, Any]],
    persist: Optional[Callable[..., Any]],
    dry_run: bool,
    **filled: Any,
) -> dict[str, Any]:
    """半腿失败:保留已成腿,挂 partial 供手动补开."""
    if not dry_run:
        _notify_partial(cfg, plan_type, msg, results)
    out: dict[str, Any] = {
        "ok": True,
        "partial": True,
        "status": "partial",
        "dry_run": dry_run,
        "plan_type": plan_type,
        "missing_leg": missing_leg,
        "msg": msg,
        "path": path,
        "results": results,
        "opened_at": _now(),
        **filled,
    }
    if persist and not dry_run:
        out["plan_id"] = persist(out, body)
    return out


def _hedge_budget_buffer(cfg: dict[str, Any] | None = None) -> float:
    """对冲专用预算缓冲;默认 0.95.与 OKX_OPTIONS_BUDGET_BUFFER 独立."""
    raw = None
    if cfg is not None:
        raw = cfg.get("budget_buffer")
    if raw is None or raw == "":
        raw = os.getenv("HEDGE_PLAN_BUDGET_BUFFER") or "0.95"
    try:
        buf = float(raw)
    except (TypeError, ValueError):
        buf = 0.95
    if buf <= 0:
        buf = 0.95
    if buf > 1:
        buf = 1.0
    return float(buf)


def _oo_bias_settings(cfg: dict[str, Any] | None = None) -> tuple[str, float]:
    from lib.hedge_plan.hedge_plan_calc_lib import _clamp_oo_bias_ratio, _normalize_oo_bias_split_by

    split = None
    ratio = None
    if cfg is not None:
        split = cfg.get("oo_bias_split_by")
        ratio = cfg.get("oo_bias_ratio")
    if split in (None, ""):
        split = os.getenv("HEDGE_PLAN_OO_BIAS_SPLIT_BY") or "budget"
    if ratio in (None, ""):
        ratio = os.getenv("HEDGE_PLAN_OO_BIAS_RATIO") or "0.7"
    return _normalize_oo_bias_split_by(split), _clamp_oo_bias_ratio(ratio)


def refresh_oo_sizing_before_start(cfg: dict[str, Any], body: dict[str, Any]) -> dict[str, Any]:
    """启动前再拉两腿卖一,按对冲预算缓冲重算张数;就地写回 body.leg_*.

    方案 A:成交价与张数均基于点击启动瞬间的最新卖一/余额.
    """
    from lib.exchange.okx_options_lib import fetch_options_trading_usdc, option_buy_liquidity_ok
    from lib.hedge_plan.hedge_plan_calc_lib import resolve_oo_budget_usdc, suggest_oo_sheets

    leg_a = dict(body.get("leg_a") or {})
    leg_b = dict(body.get("leg_b") or {})
    inst_a = str(leg_a.get("inst_id") or "").strip()
    inst_b = str(leg_b.get("inst_id") or "").strip()
    if not inst_a or not inst_b:
        return {"ok": False, "msg": "缺少期权合约"}
    quote_fn = cfg.get("quote_option_contract")
    ex = cfg.get("exchange_options")
    if not callable(quote_fn) or ex is None:
        return {"ok": False, "msg": "期权报价能力未就绪"}

    qa = quote_fn(ex, inst_a)
    if not qa.get("ok"):
        return {"ok": False, "msg": qa.get("msg") or "腿A报价失败", "quote_a": qa}
    qb = quote_fn(ex, inst_b)
    if not qb.get("ok"):
        return {"ok": False, "msg": qb.get("msg") or "腿B报价失败", "quote_b": qb}

    for tag, q in (("A", qa), ("B", qb)):
        can_open, block_msg = option_buy_liquidity_ok(q.get("ask"), q.get("ask_sz"))
        if not can_open:
            return {
                "ok": False,
                "msg": f"腿{tag}: {block_msg or '暂无卖一深度,无法买入'}",
                "quote_a": qa,
                "quote_b": qb,
            }

    trading = fetch_options_trading_usdc(ex)
    buf = _hedge_budget_buffer(cfg)
    budget_info = resolve_oo_budget_usdc(
        trading_usdc=trading,
        trade_budget_usdc=cfg.get("trade_budget_usdc"),
        buffer_ratio=buf,
    )
    if not budget_info.get("ok"):
        return {
            "ok": False,
            "msg": budget_info.get("msg") or "可用预算不足",
            "budget": budget_info,
            "quote_a": qa,
            "quote_b": qb,
        }

    mode = str(body.get("oo_sheets_mode") or "same_sheets")
    split_by, bias_ratio = _oo_bias_settings(cfg)
    opt_a = str(
        leg_a.get("opt_type")
        or (qa.get("meta") or {}).get("optType")
        or qa.get("opt_type")
        or ""
    )
    opt_b = str(
        leg_b.get("opt_type")
        or (qb.get("meta") or {}).get("optType")
        or qb.get("opt_type")
        or ""
    )
    sug = suggest_oo_sheets(
        mode=mode,
        budget_usdc=float(budget_info["budget_usdc"]),
        ask_a=float(qa["ask"]),
        ct_mult_a=float(qa.get("ct_mult") or leg_a.get("ct_mult") or 0.01),
        ask_sz_a=qa.get("ask_sz"),
        opt_type_a=opt_a,
        ask_b=float(qb["ask"]),
        ct_mult_b=float(qb.get("ct_mult") or leg_b.get("ct_mult") or 0.01),
        ask_sz_b=qb.get("ask_sz"),
        opt_type_b=opt_b,
        bias_split_by=split_by,
        bias_ratio=bias_ratio,
    )
    if not sug.get("ok"):
        return {
            "ok": False,
            "msg": sug.get("msg") or "按最新卖一无法建议张数",
            "sizing": sug,
            "budget": budget_info,
            "quote_a": qa,
            "quote_b": qb,
        }

    prev_a = leg_a.get("sheets")
    prev_b = leg_b.get("sheets")
    leg_a["sheets"] = int(sug["sheets_a"])
    leg_a["ask"] = float(qa["ask"])
    leg_a["ask_sz"] = qa.get("ask_sz")
    leg_a["ct_mult"] = float(qa.get("ct_mult") or leg_a.get("ct_mult") or 0.01)
    if opt_a:
        leg_a["opt_type"] = opt_a
    leg_b["sheets"] = int(sug["sheets_b"])
    leg_b["ask"] = float(qb["ask"])
    leg_b["ask_sz"] = qb.get("ask_sz")
    leg_b["ct_mult"] = float(qb.get("ct_mult") or leg_b.get("ct_mult") or 0.01)
    if opt_b:
        leg_b["opt_type"] = opt_b
    body["leg_a"] = leg_a
    body["leg_b"] = leg_b
    return {
        "ok": True,
        "buffer_ratio": buf,
        "budget": budget_info,
        "sizing": sug,
        "quote_a": qa,
        "quote_b": qb,
        "prev_sheets_a": prev_a,
        "prev_sheets_b": prev_b,
        "sheets_a": int(sug["sheets_a"]),
        "sheets_b": int(sug["sheets_b"]),
        "ask_a": float(qa["ask"]),
        "ask_b": float(qb["ask"]),
        "premium_est": sug.get("premium_est"),
        "msg": (
            f"已按最新卖一重算: A {sug['sheets_a']}张@{qa['ask']} + "
            f"B {sug['sheets_b']}张@{qb['ask']} · 预估 {sug.get('premium_est')}U"
        ),
    }


def refresh_po_option_quote_before_start(cfg: dict[str, Any], body: dict[str, Any]) -> dict[str, Any]:
    """永期启动前再拉卖一;保险模式张数沿用页面;期权为主时按权利金×0.95重算定仓."""
    from lib.exchange.okx_options_lib import option_buy_liquidity_ok
    from lib.hedge_plan.hedge_plan_option_primary_lib import is_option_primary, size_from_premium

    inst = str(body.get("opt_inst_id") or "").strip()
    if not inst:
        return {"ok": False, "msg": "缺少期权合约"}
    quote_fn = cfg.get("quote_option_contract")
    ex = cfg.get("exchange_options")
    if not callable(quote_fn) or ex is None:
        return {"ok": False, "msg": "期权报价能力未就绪"}
    q = quote_fn(ex, inst)
    if not q.get("ok"):
        return {"ok": False, "msg": q.get("msg") or "期权报价失败", "quote": q}
    can_open, block_msg = option_buy_liquidity_ok(q.get("ask"), q.get("ask_sz"))
    if not can_open:
        return {
            "ok": False,
            "msg": block_msg or "暂无卖一深度,无法买入",
            "quote": q,
        }
    body["ask"] = float(q["ask"])
    body["ask_sz"] = q.get("ask_sz")
    if q.get("ct_mult") is not None:
        body["ct_mult"] = float(q.get("ct_mult") or 0.01)
    if is_option_primary(body):
        cs = float(body.get("contract_size") or 0.01)
        get_cs = cfg.get("get_contract_size")
        sym = str(body.get("exchange_symbol") or "")
        if callable(get_cs) and sym:
            try:
                cs = float(get_cs(sym) or cs)
            except Exception:
                pass
        sized = size_from_premium(
            premium_budget=float(body.get("premium_budget") or 0),
            ask=float(body["ask"]),
            ct_mult=float(body.get("ct_mult") or 0.01),
            ratio=float(body.get("option_perp_ratio") or 2),
            contract_size=cs,
        )
        if not sized.get("ok"):
            return {"ok": False, "msg": sized.get("msg") or "定仓失败", "quote": q, "sizing": sized}
        body["sheets"] = sized["sheets"]
        body["contracts"] = sized["contracts"]
        body["eth_qty"] = sized["eth_qty"]
        body["contract_size"] = cs
        # 深度不足则缩量
        ask_sz = float(q.get("ask_sz") or 0)
        if ask_sz > 0 and float(body["sheets"]) > ask_sz:
            body["sheets"] = float(int(ask_sz))
            if body["sheets"] <= 0:
                return {"ok": False, "msg": "卖一深度不足 1 张", "quote": q, "sizing": sized}
            eth = round(float(body["sheets"]) * float(body.get("ct_mult") or 0.01), 2)
            body["eth_qty"] = eth
            body["contracts"] = (eth / float(body.get("option_perp_ratio") or 2)) / cs
        return {
            "ok": True,
            "ask": float(q["ask"]),
            "ask_sz": q.get("ask_sz"),
            "sheets": body.get("sheets"),
            "contracts": body.get("contracts"),
            "eth_qty": body.get("eth_qty"),
            "sizing": sized,
            "quote": q,
            "msg": (
                f"期权为主定仓: 权利金×0.95→{body.get('eth_qty')}ETH / "
                f"{body.get('sheets')}张期权 / {float(body.get('contracts') or 0):.4f}张永续 @{q['ask']}"
            ),
        }
    return {
        "ok": True,
        "ask": float(q["ask"]),
        "ask_sz": q.get("ask_sz"),
        "sheets": body.get("sheets"),
        "quote": q,
        "msg": f"已按最新卖一: {body.get('sheets')}张@{q['ask']}",
    }


def execute_perp_options_start(
    cfg: dict[str, Any],
    body: dict[str, Any],
    *,
    dry_run: bool = False,
    persist: Optional[Callable[..., Any]] = None,
) -> dict[str, Any]:
    refresh = refresh_po_option_quote_before_start(cfg, body)
    if not refresh.get("ok"):
        return {"ok": False, "msg": refresh.get("msg") or "启动前刷新卖一失败", "refresh": refresh}
    path = build_po_path_plan(body)
    results: list[dict[str, Any]] = []
    opt_res: Optional[dict[str, Any]] = None
    perp_res: Optional[dict[str, Any]] = None
    for step in path:
        if step["step"] == "options_buy_limit":
            opt_res = _buy_option(
                cfg,
                inst_id=str(body.get("opt_inst_id") or ""),
                sheets=float(body.get("sheets") or 1),
                dry_run=dry_run,
            )
            results.append({"step": step["step"], **opt_res})
            if not opt_res.get("ok"):
                # 永续已成、期权失败 → 可挂 partial 等补开期权
                if (
                    perp_res
                    and perp_res.get("ok")
                    and not dry_run
                    and manual_complete_on_partial()
                    and persist
                ):
                    return _park_partial(
                        cfg,
                        plan_type="perp_options",
                        body=body,
                        missing_leg="option_hedge",
                        msg="永续已开、期权失败。计划已挂半腿待补，请在「进行中」补开期权",
                        path=path,
                        results=results,
                        persist=persist,
                        dry_run=dry_run,
                        option=None,
                        perp=perp_res,
                    )
                return {"ok": False, "msg": opt_res.get("msg") or "期权开仓失败", "path": path, "results": results}
        else:
            from lib.hedge_plan.hedge_plan_option_primary_lib import (
                is_option_primary,
                perp_direction_for_view,
            )

            opt_primary = is_option_primary(body)
            view = str(body.get("direction") or "long")
            perp_dir = str(step.get("direction") or (
                perp_direction_for_view(view) if opt_primary else view
            ))
            attach = bool(step.get("attach_tpsl", not opt_primary))
            tp_v = None if not attach else body.get("tp")
            sl_v = None if not attach else body.get("sl")
            if attach:
                tp_v = float(body["tp"])
                sl_v = float(body["sl"])
            perp_res = _open_perp(
                cfg,
                symbol=str(body.get("exchange_symbol") or ""),
                direction=perp_dir,
                contracts=float(body.get("contracts") or 0),
                leverage=int(body.get("leverage") or (100 if opt_primary else 10)),
                tp=tp_v,
                sl=sl_v,
                dry_run=dry_run,
                attach_tpsl=attach,
            )
            results.append({"step": step["step"], **perp_res})
            if not perp_res.get("ok"):
                if opt_res and opt_res.get("ok") and not dry_run and partial_auto_close_enabled():
                    close_r = _sell_option(
                        cfg,
                        inst_id=str(opt_res.get("inst_id") or body.get("opt_inst_id") or ""),
                        sheets=float(opt_res.get("sheets") or body.get("sheets") or 1),
                    )
                    results.append({"step": "options_auto_close_on_perp_fail", **close_r})
                    msg = perp_res.get("msg") or "永续开仓失败"
                    _notify_partial(cfg, "perp_options", msg, results)
                    return {
                        "ok": False,
                        "msg": msg,
                        "path": path,
                        "results": results,
                        "partial": True,
                    }
                if (
                    opt_res
                    and opt_res.get("ok")
                    and not dry_run
                    and manual_complete_on_partial()
                    and persist
                ):
                    return _park_partial(
                        cfg,
                        plan_type="perp_options",
                        body=body,
                        missing_leg="perp",
                        msg="期权已开、永续失败。计划已挂半腿待补，请在「进行中」补开永续",
                        path=path,
                        results=results,
                        persist=persist,
                        dry_run=dry_run,
                        option=opt_res,
                        perp=None,
                    )
                msg = perp_res.get("msg") or "永续开仓失败"
                if not dry_run:
                    _notify_partial(cfg, "perp_options", msg, results)
                return {
                    "ok": False,
                    "msg": msg,
                    "path": path,
                    "results": results,
                    "partial": True,
                }

    out = {
        "ok": True,
        "dry_run": dry_run,
        "plan_type": "perp_options",
        "path": path,
        "results": results,
        "option": opt_res,
        "perp": perp_res,
        "refresh": refresh,
        "opened_at": _now(),
    }
    if persist and not dry_run:
        out["plan_id"] = persist(out, body)
    return out


def execute_options_options_start(
    cfg: dict[str, Any],
    body: dict[str, Any],
    *,
    dry_run: bool = False,
    persist: Optional[Callable[..., Any]] = None,
) -> dict[str, Any]:
    refresh = refresh_oo_sizing_before_start(cfg, body)
    if not refresh.get("ok"):
        return {"ok": False, "msg": refresh.get("msg") or "启动前刷新卖一/张数失败", "refresh": refresh}
    path = build_oo_path_plan(body)
    results: list[dict[str, Any]] = []
    leg_a = body.get("leg_a") or {}
    leg_b = body.get("leg_b") or {}
    inst_a = str(leg_a.get("inst_id") or "")
    inst_b = str(leg_b.get("inst_id") or "")
    from lib.options.options_position_limit_lib import option_position_limit_block_msg

    pos_limit_msg = option_position_limit_block_msg(
        cfg.get("exchange_options"),
        opening_inst_ids=[inst_a, inst_b],
        fetch_positions=cfg.get("fetch_option_positions"),
    )
    if pos_limit_msg:
        return {
            "ok": False,
            "msg": pos_limit_msg,
            "path": path,
            "results": [],
            "refresh": refresh,
        }
    a_res = _buy_option(cfg, inst_id=inst_a, sheets=float(leg_a.get("sheets") or 1), dry_run=dry_run)
    results.append({"step": "options_buy_limit", "leg": "a", **a_res})
    if not a_res.get("ok"):
        return {
            "ok": False,
            "msg": a_res.get("msg") or "腿A开仓失败",
            "path": path,
            "results": results,
            "refresh": refresh,
        }
    b_res = _buy_option(cfg, inst_id=inst_b, sheets=float(leg_b.get("sheets") or 1), dry_run=dry_run)
    results.append({"step": "options_buy_limit", "leg": "b", **b_res})
    if not b_res.get("ok"):
        if not dry_run and partial_auto_close_enabled():
            close_r = _sell_option(cfg, inst_id=str(a_res.get("inst_id") or ""), sheets=float(a_res.get("sheets") or 1))
            results.append({"step": "options_auto_close_leg_a", **close_r})
            msg = b_res.get("msg") or "腿B开仓失败"
            _notify_partial(cfg, "options_options", msg, results)
            return {
                "ok": False,
                "msg": msg,
                "path": path,
                "results": results,
                "partial": True,
                "refresh": refresh,
            }
        if not dry_run and manual_complete_on_partial() and persist:
            out_p = _park_partial(
                cfg,
                plan_type="options_options",
                body=body,
                missing_leg="option_b",
                msg="腿A已开、腿B失败。计划已挂半腿待补，请在「进行中」补开腿B",
                path=path,
                results=results,
                persist=persist,
                dry_run=dry_run,
                leg_a=a_res,
                leg_b=None,
            )
            out_p["refresh"] = refresh
            return out_p
        msg = b_res.get("msg") or "腿B开仓失败"
        if not dry_run:
            _notify_partial(cfg, "options_options", msg, results)
        return {
            "ok": False,
            "msg": msg,
            "path": path,
            "results": results,
            "partial": True,
            "refresh": refresh,
        }
    out = {
        "ok": True,
        "dry_run": dry_run,
        "plan_type": "options_options",
        "path": path,
        "results": results,
        "leg_a": a_res,
        "leg_b": b_res,
        "refresh": refresh,
        "opened_at": _now(),
    }
    if persist and not dry_run:
        out["plan_id"] = persist(out, body)
    return out


def execute_complete_missing_leg(
    cfg: dict[str, Any],
    plan: dict[str, Any],
    legs: list[dict[str, Any]],
    start_body: dict[str, Any],
    *,
    dry_run: bool = False,
) -> dict[str, Any]:
    """对 partial 计划补开缺失腿;成功后由调用方把计划升为 active."""
    missing = None
    for leg in legs:
        if str(leg.get("status") or "").lower() == "pending":
            missing = leg
            break
    if not missing:
        return {"ok": False, "msg": "没有待补开的腿"}
    role = str(missing.get("leg_role") or "")
    results: list[dict[str, Any]] = []
    if role == "perp":
        from lib.hedge_plan.hedge_plan_option_primary_lib import (
            is_option_primary,
            perp_direction_for_view,
        )

        opt_primary = is_option_primary(start_body)
        view = str(start_body.get("direction") or "long")
        perp_dir = perp_direction_for_view(view) if opt_primary else view
        attach = not opt_primary
        res = _open_perp(
            cfg,
            symbol=str(start_body.get("exchange_symbol") or missing.get("symbol") or ""),
            direction=perp_dir,
            contracts=float(start_body.get("contracts") or missing.get("size") or 0),
            leverage=int(start_body.get("leverage") or (100 if opt_primary else 10)),
            tp=None if not attach else float(start_body["tp"]),
            sl=None if not attach else float(start_body["sl"]),
            dry_run=dry_run,
            attach_tpsl=attach,
        )
        results.append({"step": "perp_market_open", "complete": True, **res})
        if not res.get("ok"):
            return {"ok": False, "msg": res.get("msg") or "补开永续失败", "results": results, "leg_role": role}
        return {
            "ok": True,
            "leg_role": role,
            "leg_id": missing.get("id"),
            "results": results,
            "fill": res,
            "opened_at": _now(),
        }
    if role in ("option_hedge", "option_b", "option_a"):
        if role == "option_b":
            src = start_body.get("leg_b") or {}
            inst = str(src.get("inst_id") or missing.get("inst_id") or "")
            sheets = float(src.get("sheets") or missing.get("size") or 1)
        elif role == "option_a":
            src = start_body.get("leg_a") or {}
            inst = str(src.get("inst_id") or missing.get("inst_id") or "")
            sheets = float(src.get("sheets") or missing.get("size") or 1)
        else:
            inst = str(start_body.get("opt_inst_id") or missing.get("inst_id") or "")
            sheets = float(start_body.get("sheets") or missing.get("size") or 1)
        res = _buy_option(cfg, inst_id=inst, sheets=sheets, dry_run=dry_run)
        results.append({"step": "options_buy_limit", "complete": True, "leg_role": role, **res})
        if not res.get("ok"):
            return {"ok": False, "msg": res.get("msg") or "补开期权失败", "results": results, "leg_role": role}
        return {
            "ok": True,
            "leg_role": role,
            "leg_id": missing.get("id"),
            "results": results,
            "fill": res,
            "opened_at": _now(),
        }
    return {"ok": False, "msg": f"未知待补腿: {role}"}


def validate_start_body(plan_type: str, body: dict[str, Any]) -> Optional[str]:
    pt = (plan_type or "").strip().lower()
    if pt == "perp_options":
        from lib.hedge_plan.hedge_plan_option_primary_lib import (
            is_option_primary,
            validate_option_primary_start,
        )

        if is_option_primary(body):
            from lib.hedge_plan.hedge_plan_option_primary_lib import validate_option_primary_watch

            # 以期权为主默认盯盘启动(非现场开仓);显式 watch_entry=0 才走即开校验
            watch = body.get("watch_entry")
            if watch in (None, "", True, 1, "1", "true", "yes", "on"):
                return validate_option_primary_watch(body)
            return validate_option_primary_start(body)
        need = ("direction", "entry", "tp", "sl", "contracts", "opt_inst_id", "sheets", "exchange_symbol")
        for k in need:
            if body.get(k) in (None, ""):
                return f"缺少字段: {k}"
        try:
            if float(body["contracts"]) <= 0 or float(body["sheets"]) <= 0:
                return "张数必须大于 0"
            entry = float(body["entry"])
            tp = float(body["tp"])
            sl = float(body["sl"])
            if tp <= 0 or sl <= 0 or entry <= 0:
                return "止盈/止损/入场无效"
        except (TypeError, ValueError):
            return "数值字段无效"
        direction = str(body.get("direction") or "").strip().lower()
        if direction not in ("long", "short"):
            return "方向须为 long 或 short"
        opt_type = str(body.get("opt_type") or "").strip().upper()
        if not opt_type:
            # 允许从合约名推断 ETH-USD-...-P / -C
            inst = str(body.get("opt_inst_id") or "")
            if inst.upper().endswith("-P"):
                opt_type = "P"
            elif inst.upper().endswith("-C"):
                opt_type = "C"
        if opt_type not in ("P", "C"):
            return "缺少期权类型(Put/Call)"
        if direction == "long" and opt_type != "P":
            return "做多永期对冲须用 Put"
        if direction == "short" and opt_type != "C":
            return "做空永期对冲须用 Call"
        if direction == "long" and not (sl < entry < tp):
            return "做多须满足 止损 < 入场 < 止盈"
        if direction == "short" and not (tp < entry < sl):
            return "做空须满足 止盈 < 入场 < 止损"
        from lib.hedge_plan.hedge_plan_moneyness_lib import (
            parse_strike_from_inst,
            validate_po_option_moneyness,
        )

        strike = body.get("strike")
        if strike in (None, ""):
            strike = parse_strike_from_inst(str(body.get("opt_inst_id") or ""))
        index_px = body.get("index_px")
        if index_px in (None, ""):
            index_px = entry
        money_err = validate_po_option_moneyness(
            opt_type=opt_type,
            strike=strike,
            index_px=index_px,
            ask=body.get("ask"),
            hours_to_expiry=body.get("hours_to_expiry"),
        )
        if money_err:
            return money_err
        return None
    if pt == "options_options":
        a = body.get("leg_a") or {}
        b = body.get("leg_b") or {}
        if not a.get("inst_id") or not b.get("inst_id"):
            return "请选用两条期权腿"
        rr_raw = body.get("profit_rr")
        if rr_raw not in (None, ""):
            try:
                rr = float(rr_raw)
            except (TypeError, ValueError):
                return "盈亏比无效"
            if rr <= 0:
                return "盈亏比须大于0"
        else:
            # 兼容旧上/下破
            up = body.get("target_price_up")
            down = body.get("target_price_down")
            legacy = body.get("target_price")
            if up in (None, "") and legacy not in (None, ""):
                up = legacy
            if down in (None, "") and legacy not in (None, ""):
                down = legacy
            if up in (None, "") or down in (None, ""):
                return "请填写盈亏比"
            try:
                if float(up) <= float(down):
                    return "上破目标价必须大于下破目标价"
            except (TypeError, ValueError):
                return "目标价无效"
        from lib.hedge_plan.hedge_plan_moneyness_lib import (
            parse_strike_from_inst,
            validate_oo_legs_moneyness,
        )

        def _leg_for_money(leg: dict) -> dict:
            strike = leg.get("strike")
            if strike in (None, ""):
                strike = parse_strike_from_inst(str(leg.get("inst_id") or ""))
            opt_type = leg.get("opt_type")
            if not opt_type:
                inst = str(leg.get("inst_id") or "").upper()
                if inst.endswith("-C"):
                    opt_type = "C"
                elif inst.endswith("-P"):
                    opt_type = "P"
            return {"opt_type": opt_type, "strike": strike}

        index_px = body.get("index_px")
        money_err = validate_oo_legs_moneyness(
            _leg_for_money(a),
            _leg_for_money(b),
            index_px=index_px,
        )
        if money_err:
            return money_err
        return None
    return "未知计划类型"


def dump_preview(preview: Any) -> str:
    try:
        return json.dumps(preview, ensure_ascii=False)[:8000]
    except Exception:
        return ""


def _live_option_pos_sheets(ex: Any, inst_id: str) -> float:
    from lib.exchange.okx_options_lib import fetch_option_positions

    inst_id = (inst_id or "").strip()
    if not inst_id or ex is None:
        return 0.0
    rows = fetch_option_positions(ex)
    if rows is None:
        return -1.0  # API 失败:未知
    for r in rows:
        if str(r.get("instId") or "").strip() != inst_id:
            continue
        try:
            return abs(float(r.get("pos") or 0))
        except (TypeError, ValueError):
            return 0.0
    return 0.0


def _sync_plan_status_after_leg_fix(conn: Any, plan_id: int) -> None:
    """腿状态校正后:有 open + pending → partial."""
    from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, update_plan

    plan = get_plan(conn, int(plan_id))
    if not plan:
        return
    pst = str(plan.get("status") or "")
    if pst not in ("opening", "active", "partial"):
        return
    legs = get_plan_legs(conn, int(plan_id))
    statuses = [str(l.get("status") or "").lower() for l in legs]
    n_open = sum(1 for s in statuses if s == "open")
    n_pending = sum(1 for s in statuses if s == "pending")
    if n_pending and n_open:
        update_plan(conn, int(plan_id), status="partial", close_reason="partial_fail")


def reconcile_unfilled_option_legs(cfg: dict[str, Any], conn: Any, plan_id: int) -> list[str]:
    """未成交却标 open 的期权腿 → pending(可补开);不显示成持仓."""
    from lib.hedge_plan.hedge_plan_db import get_plan_legs, update_leg
    from lib.exchange.okx_options_lib import fetch_option_order

    ex = cfg.get("exchange_options")
    notes: list[str] = []
    legs = get_plan_legs(conn, int(plan_id))
    for leg in legs:
        role = str(leg.get("leg_role") or "")
        if not role.startswith("option"):
            continue
        st = str(leg.get("status") or "").lower()
        if st != "open":
            continue
        inst = str(leg.get("inst_id") or "").strip()
        oid = str(leg.get("exchange_ord_id") or "").strip()
        leg_id = int(leg["id"])
        sheets = _live_option_pos_sheets(ex, inst)
        if sheets < 0:
            continue  # 查仓失败不改
        if sheets >= 1:
            continue
        # 无实仓:再看订单是否已成交(仍挂单只改 pending,不撤单)
        if ex is not None and inst and oid:
            od = fetch_option_order(ex, inst_id=inst, ord_id=oid)
            if od.get("ok"):
                acc = float(od.get("acc_fill_sz") or 0)
                ostate = str(od.get("state") or "")
                if acc >= 1 or ostate == "filled":
                    continue  # 有成交但仓位暂未同步,暂不改
        update_leg(
            conn,
            leg_id,
            status="pending",
            close_reason=None,
            closed_at=None,
            avg_open=None,
            premium=0,
        )
        notes.append(f"{inst} 无成交却标open→pending")
    if notes:
        _sync_plan_status_after_leg_fix(conn, int(plan_id))
    return notes


def execute_manual_end_plan(cfg: dict[str, Any], conn: Any, plan_id: int) -> dict[str, Any]:
    """人工结束进行中计划:不自动平仓;未成交腿标 cancelled."""
    from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, update_leg, update_plan
    from lib.hedge_plan.hedge_plan_notify_lib import notify_plan_end
    from lib.exchange.okx_options_lib import cancel_option_order

    plan = get_plan(conn, int(plan_id))
    if not plan:
        return {"ok": False, "msg": "计划不存在"}
    st = str(plan.get("status") or "")
    if st not in ("opening", "active", "partial", "watching"):
        return {"ok": False, "msg": f"当前状态 {st or '—'} 不可结束"}

    notes = reconcile_unfilled_option_legs(cfg, conn, int(plan_id))
    ex = cfg.get("exchange_options")
    legs = get_plan_legs(conn, int(plan_id))
    for leg in legs:
        lst = str(leg.get("status") or "").lower()
        inst = str(leg.get("inst_id") or "").strip()
        oid = str(leg.get("exchange_ord_id") or "").strip()
        if lst == "pending":
            if ex is not None and inst and oid:
                cancel_option_order(ex, inst_id=inst, ord_id=oid)
            update_leg(
                conn,
                int(leg["id"]),
                status="cancelled",
                close_reason="manual_end",
                closed_at=_now(),
                avg_open=None,
                premium=0,
            )
            notes.append(f"{inst or leg.get('leg_role')} 待补→cancelled")

    update_plan(
        conn,
        int(plan_id),
        status="closed",
        close_reason="manual",
        closed_at=_now(),
        note=((plan.get("note") or "") + " · 人工结束(不平仓)").strip(" ·")[:500],
    )
    plan2 = get_plan(conn, int(plan_id))
    if plan2:
        try:
            notify_plan_end(cfg, conn, plan2)
        except Exception:
            pass
    return {
        "ok": True,
        "plan_id": int(plan_id),
        "msg": "计划已结束(未自动平仓;有持仓请自行平掉)",
        "notes": notes,
    }