"""对冲计划开仓/平仓编排(可 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 underlying = str(inst_id).split("-")[0].upper() or "ETH" spot_bridge = None 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, "premium_ccy": underlying, "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, "margin_mode": "coin", } if not callable(place_fn): return {"ok": False, "msg": "期权限价下单未注入"} # 币本位:开仓前按权利金×缓冲用 USDT 买足标的币 try: from lib.options.options_margin_mode_lib import ( normalize_coin_spot_buy_buffer, plan_coin_open_by_budget, ) from lib.options.options_spot_bridge_lib import ( fetch_trading_coin_available, spot_market_buy_coin_with_usdt, ) idx = float( q.get("index_px") or q.get("idx_px") or (q.get("meta") or {}).get("idxPx") or q.get("mark") or 0 ) if idx <= 0: return {"ok": False, "msg": "缺少指数价,无法估算买币 USDT", "quote": q} # 权利金币×缓冲买币;默认 +5%滑点(1.05),可经 cfg.spot_buy_buffer / env 覆盖 buf = normalize_coin_spot_buy_buffer( cfg.get("spot_buy_buffer") if cfg.get("spot_buy_buffer") is not None else 1.05 ) need_coin = float(premium) * float(buf) avail = float(fetch_trading_coin_available(ex, underlying) or 0) if avail + 1e-12 < need_coin: plan = plan_coin_open_by_budget( quote_per_unit=float(ask), ct_mult=ct_mult, min_sz=1, budget_usdt=need_coin * idx * 1.05, index_px=idx, ask_sz=float(ask_sz) if ask_sz is not None else None, spot_buy_buffer=buf, target_sheets=sheets_i, ) if not plan.get("ok"): return { "ok": False, "msg": plan.get("msg") or "币本位买币计划失败", "quote": q, "plan": plan, } buy_usdt = float(plan.get("buy_usdt") or 0) if buy_usdt > 0: buy = spot_market_buy_coin_with_usdt( ex, underlying=underlying, usdt_amount=buy_usdt ) spot_bridge = {"buy_usdt": buy_usdt, "buy": buy, "plan": plan} if not buy.get("ok"): return { "ok": False, "msg": buy.get("msg") or "买币失败", "quote": q, "spot_bridge": spot_bridge, } except Exception as e: return {"ok": False, "msg": f"币本位买币准备失败: {e}", "quote": q} 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, "premium_ccy": underlying, "margin_mode": "coin", "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, "spot_bridge": spot_bridge, "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_balances, 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, } # CoinOX 期期预算改用交易户 USDT(resolve 函数名仍 usdc,数值按 USDT) bal = fetch_options_balances(ex, force=True) or {} trading = bal.get("trading_usdt_avail") if trading is None: trading = bal.get("trading_usdt") 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 market = body.get("market_order") in (True, 1, "1", "true", "yes", "on") ct = float(body.get("ct_mult") or q.get("ct_mult") or 0.01) ratio = float(body.get("option_perp_ratio") or 2) body["contract_size"] = cs if market and float(body.get("sheets") or 0) > 0: # 手动下单:以张数为准,永续按比例 sheets_i = float(int(float(body["sheets"]))) eth = round(sheets_i * ct, 2) body["sheets"] = sheets_i body["eth_qty"] = eth body["contracts"] = (eth / ratio) / cs if ratio > 0 and cs > 0 else 0 sized = { "ok": True, "sheets": sheets_i, "eth_qty": eth, "contracts": body["contracts"], "ratio": ratio, } else: sized = size_from_premium( premium_budget=float(body.get("premium_budget") or 0), ask=float(body["ask"]), ct_mult=ct, ratio=ratio, 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"] # 深度不足则缩量 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"]) * ct, 2) body["eth_qty"] = eth body["contracts"] = (eth / ratio) / cs if ratio > 0 and cs > 0 else 0 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"手动定仓: {body.get('eth_qty')}币 / " f"{body.get('sheets')}张期权 / {float(body.get('contracts') or 0):.4f}张永续 @{q['ask']}" if market else ( 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]: # 买币滑点缓冲:手动单默认 +5% if body.get("spot_buy_buffer") is not None: cfg = dict(cfg) cfg["spot_buy_buffer"] = body.get("spot_buy_buffer") elif body.get("market_order") in (True, 1, "1", "true", "yes", "on"): cfg = dict(cfg) cfg["spot_buy_buffer"] = 1.05 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 # 手动下单(market_order)或显式 watch_entry=0 → 即开校验;否则盯盘 market = body.get("market_order") in (True, 1, "1", "true", "yes", "on") watch = body.get("watch_entry") watch_off = watch in (False, 0, "0", "false", "no", "off") if market or watch_off: return validate_option_primary_start(body) return validate_option_primary_watch(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]: """仅结束计划记录(不平仓).盯盘/无持仓时用;有持仓请走 execute_manual_close_plan.""" 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, } def execute_manual_close_plan(cfg: dict[str, Any], conn: Any, plan_id: int) -> dict[str, Any]: """人工平仓:先期权(须买一流动性)→永续→卖币回 USDT,再结束计划.""" from lib.exchange.okx_options_lib import cancel_option_order 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.hedge_plan.hedge_plan_option_primary_lib import ( option_bid_liquidity_ok, perp_direction_for_view, ) from lib.hedge_plan.hedge_plan_settle_lib import resolve_option_leg_realized_pnl from lib.options.options_spot_bridge_lib import spot_market_sell_coin_to_usdt 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 '—'} 不可平仓"} # 盯盘无持仓:直接结束 if st == "watching": return execute_manual_end_plan(cfg, conn, plan_id) notes = reconcile_unfilled_option_legs(cfg, conn, int(plan_id)) legs = get_plan_legs(conn, int(plan_id)) ex = cfg.get("exchange_options") # 取消待补委托 for leg in legs: if str(leg.get("status") or "").lower() != "pending": continue inst = str(leg.get("inst_id") or "").strip() oid = str(leg.get("exchange_ord_id") or "").strip() if ex is not None and inst and oid: try: cancel_option_order(ex, inst_id=inst, ord_id=oid) except Exception: pass update_leg( conn, int(leg["id"]), status="cancelled", close_reason="manual_close", closed_at=_now(), avg_open=None, premium=0, ) notes.append(f"{inst or leg.get('leg_role')} 待补→cancelled") legs = get_plan_legs(conn, int(plan_id)) opt = next( ( x for x in legs if str(x.get("leg_role") or "") in ("option_hedge", "option_a", "option_b") and str(x.get("status") or "").lower() == "open" ), None, ) perp = next( ( x for x in legs if str(x.get("leg_role") or "") == "perp" and str(x.get("status") or "").lower() == "open" ), None, ) if not opt and not perp: update_plan( conn, int(plan_id), status="closed", close_reason="manual_close", closed_at=_now(), note=((plan.get("note") or "") + " · 人工平仓(无持仓腿)").strip(" ·")[:500], ) return {"ok": True, "plan_id": int(plan_id), "msg": "无持仓腿,计划已结束", "notes": notes} view = str(plan.get("direction") or "long").lower() uly = str(plan.get("underlying") or "ETH").upper() opt_pnl = 0.0 perp_pnl = 0.0 close_detail: dict[str, Any] = {} # 1) 先平期权 if opt: inst_id = str(opt.get("inst_id") or "") sheets = float(opt.get("size") or 1) quote_fn = cfg.get("quote_option_contract") bid = None bid_sz = None if callable(quote_fn) and ex is not None and inst_id: try: q = quote_fn(ex, inst_id) or {} if q.get("ok"): bid = q.get("bid") bid_sz = q.get("bid_sz") except Exception: pass liq_ok, liq_msg = option_bid_liquidity_ok(bid, bid_sz, need_sheets=sheets) if not liq_ok: return { "ok": False, "msg": liq_msg or "买一流动性不足,无法平期权", "plan_id": int(plan_id), "need_liquidity": True, } close_r = _sell_option(cfg, inst_id=inst_id, sheets=sheets, dry_run=False) close_detail["option"] = close_r if not close_r.get("already_flat") and (not close_r.get("ok") or not close_r.get("fully_closed", True)): return { "ok": False, "msg": close_r.get("msg") or "期权平仓失败", "plan_id": int(plan_id), "close": close_r, } # 盈亏:优先交易所历史,否则用买一估算 ct = float(opt.get("ct_mult") or 0.01) avg = float(opt.get("avg_open") or 0) premium = float(opt.get("premium") or (avg * sheets * ct)) bid_f = float(close_r.get("bid") or bid or 0) fallback = bid_f * sheets * ct - premium try: pnl, _src = resolve_option_leg_realized_pnl(ex=ex, leg=opt, fallback=fallback) opt_pnl = float(pnl if pnl is not None else fallback) except Exception: opt_pnl = float(fallback) update_leg( conn, int(opt["id"]), status="closed", close_reason="manual_close", closed_at=_now(), realized_pnl=round(opt_pnl, 4), ) notes.append(f"期权已平 {inst_id}") # 2) 再平永续 if perp: symbol = str(perp.get("symbol") or plan.get("exchange_symbol") or "") perp_dir = str(perp.get("side") or plan.get("perp_direction") or perp_direction_for_view(view)).lower() contracts = float(perp.get("size") or plan.get("perp_size") or 0) # 尽量用实盘张数 get_live = cfg.get("get_live_position_contracts") if callable(get_live) and symbol: try: live_sz = get_live(symbol, perp_dir) if live_sz is not None and float(live_sz) > 0: contracts = float(live_sz) except Exception: pass if contracts <= 0: update_leg( conn, int(perp["id"]), status="closed", close_reason="manual_close", closed_at=_now(), realized_pnl=0, ) notes.append("永续无仓/已平") else: perp_close = _close_perp( cfg, symbol=symbol, direction=perp_dir, contracts=contracts, dry_run=False ) close_detail["perp"] = perp_close if not perp_close.get("ok"): # 期权可能已平:计划挂 partial 提示 update_plan( conn, int(plan_id), close_reason="manual_close_perp_pending", note=((plan.get("note") or "") + " · 人工平仓:期权已平,永续失败待补").strip(" ·")[:500], ) return { "ok": False, "msg": perp_close.get("msg") or "永续平仓失败(期权可能已平)", "plan_id": int(plan_id), "partial_closed": True, "close": close_detail, "notes": notes, } # 粗估永续盈亏 try: entry = float(perp.get("avg_open") or plan.get("entry_mark") or 0) get_m = cfg.get("get_live_position_exchange_metrics") mark = entry if callable(get_m): m = get_m(symbol, perp_dir) or {} mark = float(m.get("mark_price") or m.get("mark") or entry) cs = 0.01 get_cs = cfg.get("get_contract_size") if callable(get_cs): cs = float(get_cs(symbol) or 0.01) coins = contracts * cs if perp_dir == "short": perp_pnl = (entry - mark) * coins else: perp_pnl = (mark - entry) * coins except Exception: perp_pnl = 0.0 update_leg( conn, int(perp["id"]), status="closed", close_reason="manual_close", closed_at=_now(), realized_pnl=round(perp_pnl, 4), ) notes.append(f"永续已平 {symbol}") # 3) 卖币回 USDT sell_r: dict[str, Any] = {"ok": True, "skipped": True} try: sell_ex = cfg.get("exchange_options") or cfg.get("exchange") if sell_ex is not None: sell_r = spot_market_sell_coin_to_usdt(sell_ex, underlying=uly) except Exception as e: sell_r = {"ok": False, "msg": str(e)} close_detail["sell_coin"] = sell_r if sell_r.get("ok"): notes.append("已卖币回 USDT" if not sell_r.get("skipped") else "卖币跳过") else: notes.append("卖币失败: " + str(sell_r.get("msg") or "")) total = float(opt_pnl) + float(perp_pnl) # 期权币本位盈亏换算到 U(粗):用指数 try: fetch_idx = cfg.get("fetch_index_price") idx = float(fetch_idx(uly) or 0) if callable(fetch_idx) else 0.0 # opt_pnl 可能已是币;若为币且绝对值很小则 × 指数 # resolve_option_leg_realized_pnl 对币本位通常返回币;统一展示用 preview 风格 if abs(opt_pnl) < 10 and idx > 0 and abs(opt_pnl * idx) > abs(opt_pnl): # 更可能是币单位 opt_pnl_u = opt_pnl * idx else: opt_pnl_u = opt_pnl except Exception: opt_pnl_u = opt_pnl update_plan( conn, int(plan_id), status="closed", close_reason="manual_close", realized_pnl_perp=round(perp_pnl, 4), realized_pnl_options=round(opt_pnl_u if isinstance(opt_pnl_u, float) else opt_pnl, 4), realized_pnl_total=round(float(opt_pnl_u) + float(perp_pnl), 4) if isinstance(opt_pnl_u, float) else round(total, 4), 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, "close": close_detail, "realized_pnl_total": plan2.get("realized_pnl_total") if plan2 else total, }