"""币本位单笔期权:买满 USDT→币 → 开满期权 → 平后卖回.""" from __future__ import annotations import os import time from typing import Any from lib.exchange.okx_options_lib import ( cap_option_buy_sheets_to_ask_depth, option_buy_liquidity_ok, td_mode_for_option_buy, wait_option_order_full_fill, ) from lib.options.options_margin_mode_lib import ( calc_sheets_from_coin_balance, compute_coin_budget_usdt, is_coin_margin_mode, margin_mode_from_inst_id, normalize_options_margin_mode, plan_coin_open_by_budget, premium_ccy_for_mode, ) from lib.options import options_spot_bridge_lib as _spot_bridge from lib.options.options_spot_bridge_lib import ( BRIDGE_BOUGHT, BRIDGE_HOLDING, ) def _sell_excess_bridge_coin( cfg: dict[str, Any], ex: Any, conn: Any, *, bridge_id: int, underlying: str, keep_coin: float, coin_bought: float, ) -> dict[str, Any] | None: """开仓后卖掉多买的币:卖量不超过 bridge 买入量,并保留权利金所需 keep_coin.""" try: keep = max(0.0, float(keep_coin)) except (TypeError, ValueError): keep = 0.0 try: bought = max(0.0, float(coin_bought)) except (TypeError, ValueError): bought = 0.0 excess_cap = max(0.0, bought - keep) if excess_cap <= 0: return None try: conn.commit() except Exception: pass avail = _spot_bridge.fetch_trading_coin_available(ex, underlying) if avail is None: return None sell_amt = min(float(avail), excess_cap) if not _spot_bridge.is_sellable_spot_coin(ex, underlying, sell_amt): return None return _spot_bridge.spot_market_sell_coin_to_usdt( ex, underlying=underlying, coin_amount=sell_amt ) def coin_budget_preview(cfg: dict[str, Any], ex: Any) -> dict[str, Any]: from lib.exchange.okx_options_lib import fetch_options_balances bal = cfg.get("fetch_options_balances")(ex, force=True) if callable(cfg.get("fetch_options_balances")) else fetch_options_balances(ex, force=True) trading = bal.get("trading_usdt_avail") if trading is None: trading = bal.get("trading_usdt") try: trading_f = float(trading or 0) except (TypeError, ValueError): trading_f = 0.0 buf = float(cfg.get("budget_buffer") or 0.95) return compute_coin_budget_usdt(trading_f, buffer=buf) def open_coin_option_buy_full( cfg: dict[str, Any], ex: Any, *, inst_id: str, signal_note: str = "", target_index: float | None = None, profit_exit_enabled: bool = False, profit_exit_mult: float = 1.0, target_sheets: int | None = None, ) -> dict[str, Any]: """先按最大可开张数估权利金×现货缓冲买币,再开对应张数(不全额兑换预算).""" from lib.options.options_db import init_options_tables from lib.options.options_position_limit_lib import ( compound_full_single_position_block_msg, option_position_limit_block_msg, ) if not is_coin_margin_mode(): return {"ok": False, "msg": "当前非币本位模式"} if margin_mode_from_inst_id(inst_id) != "coin": return {"ok": False, "msg": "合约不是币本位期权(请确认未选中 USD_UM 合约)"} # 解析标的 parts = inst_id.split("-") underlying = (parts[0] if parts else "ETH").upper() # 先清粉尘桥 + 空仓扫尾卖回(短连接),再开业务连接 try: _spot_bridge.clear_dust_pending_sell_bridges(cfg["get_db"], ex) except Exception: pass try: sweep_spot_after_options_flat(cfg, ex) except Exception: pass conn: Any = None conn = cfg["get_db"]() try: init_options_tables(conn) try: conn.commit() except Exception: pass block = _spot_bridge.bridge_blocks_new_open_msg(conn) if block: return {"ok": False, "msg": block, "can_open": False} # 网络检查前再提交,避免握锁拉持仓/行情 try: conn.commit() except Exception: pass compound_block = compound_full_single_position_block_msg( ex, fetch_positions=cfg.get("fetch_option_positions") ) if compound_block: return {"ok": False, "msg": compound_block, "can_open": False} 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, "can_open": False} budget_info = coin_budget_preview(cfg, ex) if not budget_info.get("ok"): return {"ok": False, "msg": budget_info.get("msg") or "USDT 预算无效", "budget": budget_info} budget_usdt = float(budget_info["budget_usdt"]) q = cfg["quote_option_contract"](ex, inst_id) if not q.get("ok"): return 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 "暂无卖一深度,无法买入", "can_open": False, } ct_mult = float(q.get("ct_mult") or 0.01) min_sz = int(q.get("min_sz") or 1) idx = None try: idx = float(q.get("index_px") or q.get("idxPx") or 0) except (TypeError, ValueError): idx = 0.0 if idx <= 0: try: from lib.exchange.okx_options_lib import fetch_index_price idx = float(fetch_index_price(ex, f"{underlying}-USD") or 0) except Exception: idx = 0.0 plan = plan_coin_open_by_budget( quote_per_unit=float(ask), ct_mult=ct_mult, min_sz=min_sz, budget_usdt=budget_usdt, index_px=float(idx), ask_sz=ask_sz, target_sheets=target_sheets, ) if not plan.get("ok"): return { "ok": False, "msg": plan.get("msg") or "无法规划买币张数", "plan": plan, "budget": budget_info, "can_open": False, } buy_usdt = float(plan["buy_usdt"]) sheets = int(plan["sheets"]) # 买币/读余额(模拟盘写钱包)前必须提交释放锁 try: conn.commit() except Exception: pass # 1) 仅买「权利金×现货缓冲」所需 USDT,不全额兑换预算 coin_before = _spot_bridge.fetch_trading_coin_available(ex, underlying) or 0.0 buy = _spot_bridge.spot_market_buy_coin_with_usdt(ex, underlying=underlying, usdt_amount=buy_usdt) if not buy.get("ok"): return { "ok": False, "msg": f"现货买入 {underlying} 失败: {buy.get('msg')}", "budget": budget_info, "plan": plan, } bridge_id = _spot_bridge.insert_bridge( conn, underlying=underlying, status=BRIDGE_BOUGHT, budget_usdt=buy_usdt, buy_ord_id=str(buy.get("ord_id") or ""), inst_id=inst_id, message="已买币,待开期权", ) try: conn.commit() except Exception: pass # 等余额落账 time.sleep(1.5) try: from lib.exchange.okx_options_lib import invalidate_options_balance_cache invalidate_options_balance_cache() except Exception: pass coin_after = _spot_bridge.fetch_trading_coin_available(ex, underlying) if coin_after is None: rb = _spot_bridge.rollback_bought_coin_to_usdt( conn, ex, bridge_id=bridge_id, underlying=underlying, reason="买币后读不到可用余额" ) return { "ok": False, "msg": "买币后读不到可用余额,已尝试卖回 USDT", "rollback": rb, "budget": budget_info, "plan": plan, } coin_bought = max(0.0, float(coin_after) - float(coin_before or 0)) if coin_bought <= 0: # 落账延迟:短重试取增量;绝不把整户余额当成「本次买入」 for _ in range(3): time.sleep(0.6) try: from lib.exchange.okx_options_lib import invalidate_options_balance_cache invalidate_options_balance_cache() except Exception: pass again = _spot_bridge.fetch_trading_coin_available(ex, underlying) if again is None: continue coin_bought = max(0.0, float(again) - float(coin_before or 0)) if coin_bought > 0: coin_after = again break if coin_bought <= 0: # 订单回执有数量则用之,否则回滚 try: hinted = float(buy.get("coin_bought") or 0) except (TypeError, ValueError): hinted = 0.0 if hinted > 0: coin_bought = hinted else: rb = _spot_bridge.rollback_bought_coin_to_usdt( conn, ex, bridge_id=bridge_id, underlying=underlying, reason="买入量无效" ) return { "ok": False, "msg": "买币后可用增量无效", "rollback": rb, "budget": budget_info, } _spot_bridge.update_bridge(conn, bridge_id, coin_bought=float(coin_bought)) try: conn.commit() except Exception: pass sizing = calc_sheets_from_coin_balance( quote_per_unit=float(ask), ct_mult=ct_mult, min_sz=min_sz, coin_available=float(coin_bought), ) if not sizing.get("ok"): rb = _spot_bridge.rollback_bought_coin_to_usdt( conn, ex, bridge_id=bridge_id, underlying=underlying, reason=sizing.get("msg") or "张数不足", coin_amount=float(coin_bought), ) return {"ok": False, "msg": sizing.get("msg"), "sizing": sizing, "rollback": rb, "budget": budget_info, "plan": plan} # 实盘以买到的币为准,但不超过规划张数 sheets = min(int(sizing["sheets"]), int(plan["sheets"])) capped, cap_msg = cap_option_buy_sheets_to_ask_depth(sheets, ask_sz, min_sz=min_sz) if capped is None: rb = _spot_bridge.rollback_bought_coin_to_usdt( conn, ex, bridge_id=bridge_id, underlying=underlying, reason=cap_msg or "深度不足" ) return {"ok": False, "msg": cap_msg or "卖一深度不足", "rollback": rb} if capped < sheets: sheets = int(capped) sizing = { "ok": True, "sheets": sheets, "eth_amount": round(sheets * ct_mult, 8), "coin_premium": round(sheets * float(ask) * ct_mult, 8), "ask_depth_capped": True, } else: sizing = { "ok": True, "sheets": sheets, "eth_amount": round(sheets * ct_mult, 8), "coin_premium": round(sheets * float(ask) * ct_mult, 8), } tick_sz = q.get("tick_sz") order = cfg["place_option_limit_order"]( ex, inst_id=inst_id, side="buy", sheets=sheets, price=float(ask), td_mode=td_mode_for_option_buy(cfg.get("td_mode")), tick_sz=tick_sz, ord_type="ioc", ) # 51008 时自动减半张数再试一次(买币已到位,避免整笔回滚) if (not order.get("ok")) and sheets > 1: msg_l = str(order.get("msg") or "").lower() if "51008" in str(order.get("raw") or "").lower() or "不足" in str(order.get("msg") or ""): sheets2 = max(1, sheets // 2) if sheets2 < sheets: order2 = cfg["place_option_limit_order"]( ex, inst_id=inst_id, side="buy", sheets=sheets2, price=float(ask), td_mode=td_mode_for_option_buy(cfg.get("td_mode")), tick_sz=tick_sz, ord_type="ioc", ) if order2.get("ok"): order = order2 sheets = sheets2 sizing = { "ok": True, "sheets": sheets, "eth_amount": round(sheets * ct_mult, 8), "coin_premium": round(sheets * float(ask) * ct_mult, 8), "retried_half": True, } if not order.get("ok"): rb = _spot_bridge.rollback_bought_coin_to_usdt( conn, ex, bridge_id=bridge_id, underlying=underlying, reason=order.get("msg") or "下单失败" ) return {"ok": False, "msg": order.get("msg") or "期权下单失败", "order": order, "rollback": rb} ord_id = str((order.get("data") or {}).get("ordId") or "").strip() if not ord_id: rb = _spot_bridge.rollback_bought_coin_to_usdt( conn, ex, bridge_id=bridge_id, underlying=underlying, reason="无订单号" ) return {"ok": False, "msg": "下单成功但未返回订单号", "rollback": rb} try: fill_timeout = max(2.0, float(os.getenv("OKX_OPTIONS_OPEN_FILL_TIMEOUT_SEC") or "12")) except (TypeError, ValueError): fill_timeout = 12.0 # 等待成交可能十余秒:先提交并释放业务连接,避免锁死中控/卖币 try: conn.commit() except Exception: pass try: conn.close() except Exception: pass conn = None fill = wait_option_order_full_fill( ex, inst_id=inst_id, ord_id=ord_id, need_sheets=int(sheets), timeout_sec=fill_timeout, cancel_on_timeout=True, ) conn = cfg["get_db"]() init_options_tables(conn) if not fill.get("ok"): filled_n = int(fill.get("filled_sheets") or 0) if filled_n <= 0: rb = _spot_bridge.rollback_bought_coin_to_usdt( conn, ex, bridge_id=bridge_id, underlying=underlying, reason=fill.get("msg") or "未成交", ) return {"ok": False, "msg": fill.get("msg") or "未完全成交", "fill": fill, "rollback": rb} sheets = filled_n eth_amount = round(float(sheets) * ct_mult, 8) premium_paid = round(float(ask) * eth_amount, 8) premium_ccy = premium_ccy_for_mode("coin", underlying) # 开仓手续费(折权利金单位);优先成交回包,否则查单 open_fee = 0.0 idx_open = None try: from lib.options.options_pnl_settle_lib import ( extract_order_fee, fee_to_premium_ccy, resolve_close_index_px, sum_option_order_fees_in_premium, ) idx_open = resolve_close_index_px( ex, inst_id=inst_id, underlying=underlying, hint=(q.get("index_px") if isinstance(q, dict) else None) or idx, ) fee_hint, fee_ccy_hint = extract_order_fee(order if isinstance(order, dict) else None) if fee_hint <= 0 and isinstance(fill, dict): fee_hint, fee_ccy_hint = extract_order_fee(fill.get("order") if isinstance(fill.get("order"), dict) else fill) if fee_hint > 0: open_fee = fee_to_premium_ccy( fee_hint, fee_ccy_hint or premium_ccy, premium_ccy=premium_ccy, index_px=idx_open ) if open_fee <= 0: open_fee = sum_option_order_fees_in_premium( ex, inst_id=inst_id, ord_ids=[ord_id], premium_ccy=premium_ccy, index_px=idx_open, fetch_order_fn=cfg.get("fetch_option_order"), ) except Exception: open_fee = 0.0 # 部分成交/减半张数后卖掉多余买币(按 bridge coin_bought 封顶) need_coin = float(premium_paid) try: _sell_excess_bridge_coin( cfg, ex, conn, bridge_id=bridge_id, underlying=underlying, keep_coin=need_coin, coin_bought=float(coin_bought), ) except Exception: pass _spot_bridge.update_bridge( conn, bridge_id, status=BRIDGE_HOLDING, inst_id=inst_id, coin_bought=float(need_coin), message="期权持仓中", ) trade_id = _insert_coin_trade( conn, inst_id=inst_id, underlying=underlying, opt_type=str(q.get("opt_type") or ""), strike=q.get("strike"), exp_time=q.get("exp_time"), sheets=int(sheets), eth_amount=eth_amount, open_quote=float(ask), premium_paid=premium_paid, signal_note=signal_note, exchange_ord_id=ord_id, bridge_id=bridge_id, budget_usdt=buy_usdt, premium_ccy=premium_ccy, open_fee=open_fee, open_index_px=idx_open, profit_exit_enabled=profit_exit_enabled, profit_exit_mult=profit_exit_mult, ) # 目标位 / 翻倍离场 — 复用现有逻辑若存在 try: if target_index is not None: from lib.options.options_target_lib import upsert_target_monitor upsert_target_monitor( conn, inst_id=inst_id, underlying=underlying, opt_type=str(q.get("opt_type") or ""), target_index=float(target_index), trade_id=trade_id, sheets=int(sheets), ) except Exception: pass try: from lib.options.options_notify_lib import notify_options_open notify_options_open( cfg, conn, trade_id=trade_id, inst_id=inst_id, underlying=underlying, opt_type=str(q.get("opt_type") or ""), sheets=int(sheets), premium_paid=premium_paid, open_quote=float(ask), target_index=target_index, signal_note=signal_note, premium_ccy=premium_ccy, ) except Exception: pass return { "ok": True, "msg": f"币本位开仓成功 {sheets} 张", "margin_mode": "coin", "budget": budget_info, "sizing": sizing, "sheets": sheets, "eth_amount": eth_amount, "premium_paid": premium_paid, "premium_ccy": premium_ccy, "bridge_id": bridge_id, "trade_id": trade_id, "order": order, "fill": fill, } finally: if conn is not None: try: conn.close() except Exception: pass def _insert_coin_trade(conn: Any, **kwargs: Any) -> int: from lib.options.options_time_lib import options_app_now_str pe = 1 if kwargs.get("profit_exit_enabled") else 0 pe_mult = float(kwargs.get("profit_exit_mult") or 1.0) pe_state = "active" if pe else "idle" open_fee = float(kwargs.get("open_fee") or 0.0) if open_fee < 0: open_fee = abs(open_fee) fee_ccy = str(kwargs.get("premium_ccy") or "ETH") open_index_px = kwargs.get("open_index_px") try: open_index_px = float(open_index_px) if open_index_px is not None else None if open_index_px is not None and open_index_px <= 0: open_index_px = None except (TypeError, ValueError): open_index_px = None cur = conn.execute( """ INSERT INTO options_trades( inst_id, underlying, opt_type, strike, exp_time, sheets, eth_amount, open_quote, premium_paid, status, signal_note, exchange_ord_id, margin_mode, premium_ccy, bridge_id, budget_usdt, open_fee, fee_ccy, open_index_px, profit_exit_enabled, profit_exit_mult, profit_exit_state, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?, ?, 'coin', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( kwargs["inst_id"], kwargs["underlying"], kwargs["opt_type"], kwargs.get("strike"), str(kwargs.get("exp_time") or ""), kwargs["sheets"], kwargs["eth_amount"], kwargs.get("open_quote"), kwargs.get("premium_paid"), kwargs.get("signal_note") or "", kwargs.get("exchange_ord_id"), kwargs.get("premium_ccy") or "ETH", kwargs.get("bridge_id"), kwargs.get("budget_usdt"), open_fee, fee_ccy, open_index_px, pe, pe_mult, pe_state, options_app_now_str(), ), ) conn.commit() return int(cur.lastrowid) def maybe_sell_spot_after_close( cfg: dict[str, Any], ex: Any, *, inst_id: str, close_result: dict[str, Any] | None = None, ) -> dict[str, Any] | None: """期权平仓后若该合约为币本位且已空仓,卖回本桥残留币. 持仓查询失败/空响应时不卖,等下次监控再查(与 sweep_spot_after_options_flat 一致). """ del close_result # 预留字段;卖回只认交易所持仓实况 if margin_mode_from_inst_id(inst_id) != "coin": return None try: rows = cfg["fetch_option_positions"](ex) except Exception as e: return {"ok": False, "msg": f"持仓查询失败,暂不卖币: {e}", "skipped": True} if rows is None: return {"ok": False, "msg": "持仓查询失败: 空响应,暂不卖币", "skipped": True} for p in rows or []: if str(p.get("instId") or p.get("inst_id") or "") != inst_id: continue try: if abs(float(p.get("pos") or 0)) > 1e-12: return {"ok": True, "skipped": True, "msg": "仍有持仓,暂不卖币"} except (TypeError, ValueError): pass parts = inst_id.split("-") underlying = (parts[0] if parts else "ETH").upper() return _spot_bridge.sell_residual_after_option_flat( cfg["get_db"], ex, underlying=underlying, inst_id=inst_id, ) def sweep_spot_after_options_flat( cfg: dict[str, Any], ex: Any | None = None, ) -> dict[str, Any]: """币本位:某标的已无期权持仓时,把交易户该标的残留币卖回 USDT. 覆盖到期结算、限价平仓后同步关仓等未走买一平仓卖回路径的情况。 """ if not is_coin_margin_mode(): return {"ok": True, "skipped": True, "msg": "非币本位"} exchange = ex if ex is not None else cfg.get("exchange_options") if exchange is None: return {"ok": False, "msg": "exchange 未就绪"} live_uly: set[str] = set() try: rows = cfg["fetch_option_positions"](exchange) except Exception as e: # 持仓查询失败:只清 pending 粉尘桥,绝不卖币 try: _spot_bridge.clear_dust_pending_sell_bridges(cfg["get_db"], exchange) except Exception: pass return {"ok": False, "msg": f"持仓查询失败: {e}"} if rows is None: try: _spot_bridge.clear_dust_pending_sell_bridges(cfg["get_db"], exchange) except Exception: pass return {"ok": False, "msg": "持仓查询失败: 空响应"} for p in rows or []: try: if abs(float(p.get("pos") or 0)) <= 1e-12: continue except (TypeError, ValueError): continue inst = str(p.get("instId") or p.get("inst_id") or "").strip() if not inst: continue live_uly.add(inst.split("-")[0].upper()) # 先关掉只剩粉尘的 pending 桥,避免拦开仓又看不到重试按钮 try: _spot_bridge.clear_dust_pending_sell_bridges(cfg["get_db"], exchange) except Exception: pass sold: list[dict[str, Any]] = [] for uly in ("BTC", "ETH"): if uly in live_uly: continue try: avail = _spot_bridge.fetch_trading_coin_available(exchange, uly) # 低于现货 minSz 的粉尘不扫、不重试 if avail is None or not _spot_bridge.is_sellable_spot_coin(exchange, uly, avail): continue except Exception: continue try: out = _spot_bridge.sell_residual_after_option_flat( cfg["get_db"], exchange, underlying=uly, inst_id=None, ) sold.append({"underlying": uly, **(out or {})}) except Exception as e: sold.append({"underlying": uly, "ok": False, "msg": str(e)}) return {"ok": True, "results": sold, "skipped": not sold}