"""OKX USDⓈ 期权 API 封装(主账户 exchange_options 专用).""" from __future__ import annotations import json import math import re import threading import time from typing import Any, Callable import ccxt from lib.options.options_pricing_lib import ( expiry_breakeven_from_ask, idx_distance_to_be, intrinsic_px_per_unit, is_shallow_itm, option_moneyness, option_moneyness_label, strike_distance_to_be, ) _OKX_OPTION_ERR_ZH: dict[str, str] = { "51008": "可用余额或保证金不足(币本位请确认交易账户 ETH/BTC 足够;USDC 模式请确认 USDC 足够)", "51018": "期权账户不能持有净空头头寸", "51019": "期权买入须使用逐仓模式(全仓模式下不能持有多头净头寸)", } def _okx_trade_error_message(exc: BaseException | None = None, resp: Any = None) -> str: row: dict[str, Any] | None = None if isinstance(resp, dict): data = resp.get("data") or [] if data and isinstance(data[0], dict): row = data[0] if row is None and exc is not None: text = str(exc) match = re.search(r"\{.*\}", text, re.DOTALL) if match: try: payload = json.loads(match.group(0)) data = payload.get("data") or [] if data and isinstance(data[0], dict): row = data[0] except json.JSONDecodeError: pass if row: code = str(row.get("sCode") or "") msg = str(row.get("sMsg") or "").strip() low = msg.lower() if code == "51008": # 勿写死「资金账户 USDT」:USDC 模式常因交易户 USDC 不足;币本位则是标的币不足 if "usdc" in low: return "交易账户 USDC 可用余额不足" if "usdt" in low: return "USDT 可用余额不足" try: from lib.options.options_margin_mode_lib import is_coin_margin_mode if is_coin_margin_mode(): return "可用余额或保证金不足(币本位请确认交易账户 ETH/BTC 足够,或减少张数)" except Exception: pass return _OKX_OPTION_ERR_ZH["51008"] zh = _OKX_OPTION_ERR_ZH.get(code) if zh: return zh if msg: return msg if exc is not None: text = str(exc).strip() if text.lower().startswith("okx "): text = text[4:].strip() return text or "下单失败" return "下单失败" _OPTIONS_BALANCE_CACHE: dict[str, Any] = {"updated_at": 0.0, "data": None} # public/instruments 全族缓存:合约列表变化慢,限频时用旧数据保活 _OPTION_INSTRUMENTS_CACHE: dict[str, dict[str, Any]] = {} _OPTION_INSTRUMENTS_CACHE_LOCK = threading.Lock() _OPTION_INSTRUMENTS_CACHE_TTL = 90.0 _OPTION_INSTRUMENTS_STALE_MAX = 600.0 def invalidate_options_balance_cache() -> None: _OPTIONS_BALANCE_CACHE["updated_at"] = 0.0 _OPTIONS_BALANCE_CACHE["data"] = None def invalidate_option_instruments_cache(inst_family: str | None = None) -> None: with _OPTION_INSTRUMENTS_CACHE_LOCK: if inst_family: _OPTION_INSTRUMENTS_CACHE.pop(str(inst_family), None) else: _OPTION_INSTRUMENTS_CACHE.clear() def td_mode_for_option_buy(configured: str | None = None) -> str: """OKX 买入期权(多头)必须使用逐仓.""" mode = (configured or "isolated").strip().lower() return "isolated" if mode == "cross" else mode or "isolated" def create_options_exchange( api_key: str = "", api_secret: str = "", passphrase: str = "", proxies: dict[str, str] | None = None, ) -> ccxt.okx: """创建 option 客户端.未传密钥时读 OKX_API_*(与永续同源).""" import os key = (api_key or os.getenv("OKX_API_KEY") or "").strip() secret = (api_secret or os.getenv("OKX_API_SECRET") or "").strip() password = (passphrase or os.getenv("OKX_API_PASSPHRASE") or "").strip() ex = ccxt.okx( { "apiKey": key, "secret": secret, "password": password, "enableRateLimit": True, "options": {"defaultType": "option"}, } ) if proxies: ex.proxies = proxies return ex def _safe_float(v: Any) -> float | None: if v is None or v == "": return None try: return float(v) except (TypeError, ValueError): return None def round_option_px(px: float, tick_sz: Any, side: str) -> float: """按 OKX tickSz 对齐:买入向上取整,卖出向下取整.""" tick = _safe_float(tick_sz) if tick is None or tick <= 0 or px <= 0: return px steps = px / tick side_l = (side or "").lower() if side_l == "buy": return math.ceil(steps - 1e-12) * tick return math.floor(steps + 1e-12) * tick def format_option_px(px: float, tick_sz: Any) -> str: tick = _safe_float(tick_sz) if tick is None or tick <= 0: # 无 tick 时裁到 4 位并去尾零,避免 482.4881990066513 这类浮点毛刺 s = f"{float(px):.4f}".rstrip("0").rstrip(".") return s or "0" if tick < 1: decimals = max(0, -int(round(math.log10(tick)))) return f"{px:.{decimals}f}".rstrip("0").rstrip(".") or "0" # tick>=1(如 BTC 期权 tickSz=5):只按整数展示,禁止 rstrip('0') 把 1370 变成 137 if "." in str(tick): decimals = len(str(tick).split(".")[-1]) return f"{px:.{decimals}f}".rstrip("0").rstrip(".") or "0" return str(int(round(float(px)))) def format_usdc_amount(v: float | None) -> str | None: """USDC 金额展示(权利金/回收等,固定 2 位小数).""" if v is None: return None return f"{float(v):.2f}" def format_premium_amount(v: float | None, *, ccy: str | None = "USDC") -> str | None: """权利金/回收金额文案:USDC 2 位;币本位 ETH/BTC 最多 8 位去尾零.""" if v is None: return None try: n = float(v) except (TypeError, ValueError): return None unit = (ccy or "USDC").strip().upper() or "USDC" if unit in ("ETH", "BTC"): txt = f"{n:.8f}".rstrip("0").rstrip(".") return txt or "0" return f"{n:.2f}" def is_option_full_close_history(raw: dict[str, Any]) -> bool: """仅保留 OKX 历史仓位中的「全部平仓/强平/ADL 全平」记录,排除部分平仓.""" close_type = str(raw.get("type") or "").strip() return close_type in ("2", "3", "6") def option_history_row_key( *, source: str, inst_id: str = "", pos_id: str | None = None, close_ms: int | None = None, ) -> str: inst_id = (inst_id or "").strip() pos_id = (pos_id or "").strip() if source == "live": return f"live:{inst_id}:{pos_id or close_ms or '0'}" # OKX 可能对同合约多次开平复用 posId,必须带上平仓时间区分 if pos_id: if close_ms: return f"ex:{pos_id}:{int(close_ms)}" return f"ex:{pos_id}" return f"ex:{inst_id}:{close_ms or 0}" def _ms_to_iso(ms: Any) -> str | None: val = _safe_float(ms) if val is None or val <= 0: return None try: from datetime import datetime, timezone dt = datetime.fromtimestamp(int(val) / 1000.0, tz=timezone.utc).astimezone() return dt.strftime("%Y-%m-%d %H:%M:%S") except (TypeError, ValueError, OSError): return None def option_instrument_meta_cached( ex: ccxt.okx, inst_id: str, cache: dict[str, dict[str, Any] | None] | None = None, ) -> dict[str, Any] | None: inst_id = (inst_id or "").strip() if not inst_id: return None if cache is not None and inst_id in cache: return cache[inst_id] meta = fetch_option_instrument_meta(ex, inst_id) if cache is not None: cache[inst_id] = meta return meta def tick_sz_and_ct_mult( ex: ccxt.okx, inst_id: str, cache: dict[str, dict[str, Any] | None] | None = None, ) -> tuple[Any, float]: meta = option_instrument_meta_cached(ex, inst_id, cache) tick_sz = meta.get("tickSz") if meta else None ct_mult = _safe_float(meta.get("ctMult")) if meta else None return tick_sz, ct_mult or 0.01 def _resolve_chain_quote( *, ticker: dict[str, Any], meta: dict[str, Any], opt_type: str, strike: float, index_px: float, inst_id: str | None = None, ) -> dict[str, Any]: """链列表报价:卖一缺失时用标记价/内在价值估算(深度实值常见无卖一).""" tick_sz = meta.get("tickSz") ask = _safe_float(ticker.get("askPx")) bid = _safe_float(ticker.get("bidPx")) mark = _safe_float(ticker.get("markPx")) ask_sz = _safe_float(ticker.get("askSz")) bid_sz = _safe_float(ticker.get("bidSz")) ask_estimated = False iid = (inst_id or str(meta.get("instId") or "")).strip() if ask is None and mark is not None and mark > 0: ask = round_option_px(mark, tick_sz, "buy") ask_estimated = True if ask is None: intrinsic = intrinsic_px_per_unit(opt_type, strike, index_px, inst_id=iid or None) if intrinsic is not None and intrinsic > 0: ask = round_option_px(intrinsic, tick_sz, "buy") ask_estimated = True if bid is None and mark is not None and mark > 0: bid = round_option_px(mark, tick_sz, "sell") if bid is None: intrinsic = intrinsic_px_per_unit(opt_type, strike, index_px, inst_id=iid or None) if intrinsic is not None and intrinsic > 0: bid = round_option_px(intrinsic, tick_sz, "sell") if ask_estimated: ask_sz = None return { "ask": ask, "bid": bid, "ask_sz": ask_sz, "bid_sz": bid_sz, "mark_px": mark, "ask_estimated": ask_estimated, } def _fetch_book_bid_ask(ex: ccxt.okx, inst_id: str) -> tuple[float | None, float | None]: bid, ask, _, _ = _fetch_book_top(ex, inst_id) return bid, ask def _normalize_book_levels(rows: list[Any], depth: int) -> list[dict[str, float]]: levels: list[dict[str, float]] = [] for row in rows[: max(0, int(depth))]: if not isinstance(row, (list, tuple)) or len(row) < 2: continue px = _safe_float(row[0]) sz = _safe_float(row[1]) if px is None or sz is None or px <= 0 or sz <= 0: continue levels.append({"px": px, "sz": sz}) return levels def fetch_option_book_depth(ex: ccxt.okx, inst_id: str, depth: int = 5) -> dict[str, list[dict[str, float]]]: """获取期权盘口深度,sz 为 OKX 返回的张数口径.""" inst_id = (inst_id or "").strip() if not inst_id: return {"bids": [], "asks": []} try: sz = str(max(1, min(int(depth), 10))) rows = ex.public_get_market_books({"instId": inst_id, "sz": sz}).get("data") or [] if not rows: return {"bids": [], "asks": []} row = rows[0] return { "bids": _normalize_book_levels(row.get("bids") or [], int(depth)), "asks": _normalize_book_levels(row.get("asks") or [], int(depth)), } except Exception: return {"bids": [], "asks": []} def _fetch_book_top( ex: ccxt.okx, inst_id: str ) -> tuple[float | None, float | None, float | None, float | None]: try: rows = ex.public_get_market_books({"instId": inst_id, "sz": "1"}).get("data") or [] if not rows: return None, None, None, None row = rows[0] asks = row.get("asks") or [] bids = row.get("bids") or [] ask = _safe_float(asks[0][0]) if asks else None bid = _safe_float(bids[0][0]) if bids else None ask_sz = _safe_float(asks[0][1]) if asks and len(asks[0]) > 1 else None bid_sz = _safe_float(bids[0][1]) if bids and len(bids[0]) > 1 else None return bid, ask, bid_sz, ask_sz except Exception: return None, None, None, None def _pos_side_from_position(pos: dict[str, Any] | None) -> str | None: if not pos: return None ps = str(pos.get("posSide") or "").strip().lower() if ps in ("long", "short", "net"): return ps sheets = _safe_float(pos.get("pos")) or 0.0 if sheets > 0: return "long" if sheets < 0: return "short" return "net" def inst_family_from_inst_id(inst_id: str) -> str | None: """从 instId 解析 instFamily,如 ETH-USD_UM-260707-1790-C → ETH-USD_UM.""" parts = (inst_id or "").strip().split("-") if len(parts) < 4: return None return "-".join(parts[:-3]) def option_fields_from_inst_id(inst_id: str) -> tuple[str | None, float | None]: """从 instId 解析 optType 与 strike,如 ETH-USD_UM-260709-1700-P.""" parts = (inst_id or "").strip().split("-") if len(parts) < 2: return None, None tail = parts[-1].upper() opt_type = tail if tail in ("C", "P") else None strike = _safe_float(parts[-2]) if len(parts) >= 2 else None return opt_type, strike def expiry_ms_from_inst_id(inst_id: str) -> int | None: """从 instId 日期段解析到期时刻(OKX 期权默认 08:00 UTC).""" parts = (inst_id or "").strip().split("-") if len(parts) < 3: return None date_part = parts[-3] if not re.fullmatch(r"\d{6}", date_part): return None try: from datetime import datetime, timezone yy, mm, dd = int(date_part[0:2]), int(date_part[2:4]), int(date_part[4:6]) dt = datetime(2000 + yy, mm, dd, 8, 0, 0, tzinfo=timezone.utc) return int(dt.timestamp() * 1000) except (ValueError, OSError): return None def normalize_option_exp_ms(exp_time: Any, inst_id: str = "") -> int | None: """统一期权到期毫秒时间戳(优先 API expTime,否则从 instId 推算).""" raw = _safe_float(exp_time) if raw is not None and raw > 0: ms = int(raw) if ms < 10_000_000_000: ms *= 1000 return ms return expiry_ms_from_inst_id(inst_id) def _is_okx_rate_limit(err: BaseException) -> bool: text = str(err) or "" name = err.__class__.__name__ return "50011" in text or "Too Many Requests" in text or "RateLimit" in name def _meta_from_inst_id_fallback(inst_id: str) -> dict[str, Any]: """行情在但 instruments 限频时,用合约 ID 拼最小 meta,避免误报「合约不存在」.""" family = inst_family_from_inst_id(inst_id) or "" opt_type, strike = option_fields_from_inst_id(inst_id) uly = family.replace("_UM", "") if family else "" return { "instId": inst_id, "instFamily": family, "uly": uly, "optType": opt_type, "stk": strike, "ctMult": 0.01, "minSz": "1", "tickSz": "0.0001", "state": "live", } def fetch_option_instrument_meta(ex: ccxt.okx, inst_id: str) -> dict[str, Any] | None: family = inst_family_from_inst_id(inst_id) if not family: return None # 优先从全族缓存取,避免每选一腿再打 instruments try: cached_rows = fetch_option_instruments(ex, family, allow_stale=True) for r in cached_rows: if isinstance(r, dict) and str(r.get("instId")) == inst_id: return r except Exception: pass last_err: BaseException | None = None for attempt in range(2): try: rows = ex.public_get_public_instruments( {"instType": "OPTION", "instFamily": family, "instId": inst_id} ).get("data") or [] if rows and isinstance(rows[0], dict): return rows[0] rows = fetch_option_instruments(ex, family, allow_stale=True) for r in rows: if isinstance(r, dict) and str(r.get("instId")) == inst_id: return r return None except Exception as e: last_err = e if _is_okx_rate_limit(e) and attempt < 1: time.sleep(1.2) continue break if last_err is not None and _is_okx_rate_limit(last_err): try: t_rows = ex.public_get_market_ticker({"instId": inst_id}).get("data") or [] if t_rows: return _meta_from_inst_id_fallback(inst_id) except Exception: pass return None def _extract_ccy_free(balance: dict[str, Any], ccy: str) -> float | None: ccy = (ccy or "").upper() if not isinstance(balance, dict): return None info = balance.get(ccy) if isinstance(info, dict): v = _safe_float(info.get("free")) if v is not None: return v free_map = balance.get("free") or {} if isinstance(free_map, dict): return _safe_float(free_map.get(ccy)) return None def _extract_ccy_balance(balance: dict[str, Any], ccy: str) -> float | None: ccy = (ccy or "").upper() if not isinstance(balance, dict): return None info = balance.get(ccy) if isinstance(info, dict): for k in ("free", "total", "eq"): v = _safe_float(info.get(k)) if v is not None: return v total_map = balance.get("total") or {} if isinstance(total_map, dict): v = _safe_float(total_map.get(ccy)) if v is not None: return v free_map = balance.get("free") or {} if isinstance(free_map, dict): v = _safe_float(free_map.get(ccy)) if v is not None: return v return None def fetch_account_balances_by_type( ex: ccxt.okx, account_type: str, ) -> tuple[dict[str, float | None], dict[str, float | None]]: out: dict[str, float | None] = {"USDT": None, "USDC": None, "USDG": None, "ETH": None, "BTC": None} avail: dict[str, float | None] = {"USDT": None, "USDC": None, "USDG": None, "ETH": None, "BTC": None} try: bal = ex.fetch_balance(params={"type": account_type}) for c in out: out[c] = _extract_ccy_balance(bal, c) avail[c] = _extract_ccy_free(bal, c) except Exception: pass return out, avail def fetch_funding_balances_via_asset_api( ex: ccxt.okx, ) -> tuple[dict[str, float | None], dict[str, float | None]]: """OKX 资金账户余额(GET /api/v5/asset/balances),比 ccxt fetch_balance 更准确.""" out: dict[str, float | None] = {"USDT": None, "USDC": None, "USDG": None, "ETH": None, "BTC": None} avail: dict[str, float | None] = {"USDT": None, "USDC": None, "USDG": None, "ETH": None, "BTC": None} try: resp = ex.private_get_asset_balances({}) for row in (resp or {}).get("data") or []: if not isinstance(row, dict): continue ccy = str(row.get("ccy") or "").upper() if ccy not in out: continue a = _safe_float(row.get("availBal")) b = _safe_float(row.get("bal")) or _safe_float(row.get("eq")) avail[ccy] = a out[ccy] = b if b is not None else a except Exception: pass return out, avail def _merge_balance_maps( primary: dict[str, float | None], secondary: dict[str, float | None], ) -> dict[str, float | None]: merged = dict(primary) for ccy, val in secondary.items(): if merged.get(ccy) is None and val is not None: merged[ccy] = val return merged def fetch_subaccount_asset_balances(ex: ccxt.okx, sub_acct: str) -> dict[str, float | None]: """子账户各币种可用余额(主/子划转「全部」用).""" sub = (sub_acct or "").strip() out: dict[str, float | None] = {"USDT": None, "USDC": None} if not sub: return out try: resp = ex.private_get_asset_subaccount_balances({"subAcct": sub}) for row in (resp or {}).get("data") or []: if not isinstance(row, dict): continue ccy = str(row.get("ccy") or "").upper() if ccy not in out: continue out[ccy] = _safe_float(row.get("availBal")) or _safe_float(row.get("bal")) except Exception: pass return out def fetch_options_balances( ex: ccxt.okx, *, force: bool = False, scope: str = "main", sub_acct: str = "", ) -> dict[str, Any]: import os if (scope or "").strip().lower() == "sub": sub_bal = fetch_subaccount_asset_balances(ex, sub_acct) return { "scope": "sub", "funding_usdt": sub_bal.get("USDT"), "funding_usdc": sub_bal.get("USDC"), "funding_usdt_avail": sub_bal.get("USDT"), "funding_usdc_avail": sub_bal.get("USDC"), "trading_usdt": sub_bal.get("USDT"), "trading_usdc": sub_bal.get("USDC"), "trading_usdt_avail": sub_bal.get("USDT"), "trading_usdc_avail": sub_bal.get("USDC"), } ttl = float(os.getenv("OKX_OPTIONS_BALANCE_REFRESH_SEC", "30")) now = time.time() cached = _OPTIONS_BALANCE_CACHE.get("data") if not force and cached is not None and now - float(_OPTIONS_BALANCE_CACHE.get("updated_at") or 0) < ttl: return dict(cached) funding, funding_avail = fetch_account_balances_by_type(ex, "funding") asset_funding, asset_funding_avail = fetch_funding_balances_via_asset_api(ex) funding = _merge_balance_maps(funding, asset_funding) funding_avail = _merge_balance_maps(funding_avail, asset_funding_avail) trading, trading_avail = fetch_account_balances_by_type(ex, "trading") # OKX 统一账户:option 客户端拉 type=trading 常缺 USDT/币;用 swap 补齐缺失项 if any(trading.get(c) is None for c in ("USDT", "USDC", "ETH", "BTC")): swap_bal, swap_avail = fetch_account_balances_by_type(ex, "swap") for ccy in ("USDT", "USDC", "USDG", "ETH", "BTC"): if trading.get(ccy) is None and swap_bal.get(ccy) is not None: trading[ccy] = swap_bal[ccy] if trading_avail.get(ccy) is None and swap_avail.get(ccy) is not None: trading_avail[ccy] = swap_avail[ccy] result = { "scope": "main", "funding_usdt": funding.get("USDT"), "funding_usdc": funding.get("USDC"), "funding_usdg": funding.get("USDG"), "funding_eth": funding.get("ETH"), "funding_btc": funding.get("BTC"), "funding_usdt_avail": funding_avail.get("USDT"), "funding_usdc_avail": funding_avail.get("USDC"), "funding_eth_avail": funding_avail.get("ETH"), "funding_btc_avail": funding_avail.get("BTC"), "trading_usdt": trading.get("USDT"), "trading_usdc": trading.get("USDC"), "trading_usdg": trading.get("USDG"), "trading_eth": trading.get("ETH"), "trading_btc": trading.get("BTC"), "trading_usdt_avail": trading_avail.get("USDT"), "trading_usdc_avail": trading_avail.get("USDC"), "trading_eth_avail": trading_avail.get("ETH"), "trading_btc_avail": trading_avail.get("BTC"), } _OPTIONS_BALANCE_CACHE["updated_at"] = now _OPTIONS_BALANCE_CACHE["data"] = result return result def options_header_balances( ex: ccxt.okx, *, force: bool = False, ) -> tuple[float | None, float | None, float | None, float | None]: """顶栏期权两格用 USDC;顺带返回同账户 USDT(调用方勿再计入总资金,避免与永续栏重复). 返回:(trading_usdc, funding_usdc, funding_usdt, trading_usdt) """ pack = options_header_balance_pack(ex, force=force) return ( pack.get("trading_usdc"), pack.get("funding_usdc"), pack.get("funding_usdt"), pack.get("trading_usdt"), ) def options_header_balance_pack( ex: ccxt.okx, *, force: bool = False, ) -> dict[str, Any]: """顶栏/快照用期权资金包(含币本位 ETH/BTC).""" import os bal = fetch_options_balances(ex, force=force) def _round(v: Any, nd: int = 2) -> float | None: if v is None: return None try: return round(float(v), nd) except (TypeError, ValueError): return None def _round_coin(v: Any) -> float | None: if v is None: return None try: return round(float(v), 8) except (TypeError, ValueError): return None try: from lib.options.options_margin_mode_lib import normalize_options_margin_mode margin_mode = normalize_options_margin_mode() except Exception: margin_mode = "usdc" underly = (os.getenv("OKX_OPTIONS_DEFAULT_UNDERLY") or "ETH").strip().upper() or "ETH" coin_key = "btc" if underly == "BTC" else "eth" return { "trading_usdc": _round(bal.get("trading_usdc")), "funding_usdc": _round(bal.get("funding_usdc")), "funding_usdt": _round(bal.get("funding_usdt")), "trading_usdt": _round(bal.get("trading_usdt")), "funding_eth": _round_coin(bal.get("funding_eth")), "trading_eth": _round_coin(bal.get("trading_eth")), "funding_btc": _round_coin(bal.get("funding_btc")), "trading_btc": _round_coin(bal.get("trading_btc")), "options_margin_mode": margin_mode, "options_underly": underly, "funding_coin": _round_coin(bal.get(f"funding_{coin_key}")), "trading_coin": _round_coin(bal.get(f"trading_{coin_key}")), } def fetch_index_price(ex: ccxt.okx, uly: str) -> float | None: inst = f"{uly}" if "-" in uly else f"{uly}-USD" try: rows = ex.public_get_market_index_tickers({"instId": inst}).get("data") or [] if rows: return _safe_float(rows[0].get("idxPx")) except Exception: pass return None def fetch_option_instruments( ex: ccxt.okx, inst_family: str, *, force: bool = False, allow_stale: bool = True, ) -> list[dict[str, Any]]: """拉取 OPTION instruments;进程内缓存,50011 时回退旧列表.""" family = str(inst_family or "").strip() if not family: return [] now = time.time() with _OPTION_INSTRUMENTS_CACHE_LOCK: entry = _OPTION_INSTRUMENTS_CACHE.get(family) if ( not force and entry is not None and entry.get("rows") is not None and now - float(entry.get("updated_at") or 0) < _OPTION_INSTRUMENTS_CACHE_TTL ): return list(entry["rows"]) try: rows = ex.public_get_public_instruments( {"instType": "OPTION", "instFamily": family} ).get("data") or [] live = [r for r in rows if isinstance(r, dict) and r.get("state") == "live"] with _OPTION_INSTRUMENTS_CACHE_LOCK: _OPTION_INSTRUMENTS_CACHE[family] = {"updated_at": now, "rows": live} return list(live) except Exception as e: if allow_stale: with _OPTION_INSTRUMENTS_CACHE_LOCK: entry = _OPTION_INSTRUMENTS_CACHE.get(family) if entry is not None and entry.get("rows") is not None: age = now - float(entry.get("updated_at") or 0) if age <= _OPTION_INSTRUMENTS_STALE_MAX: return list(entry["rows"]) raise def fetch_option_tickers(ex: ccxt.okx, inst_family: str) -> dict[str, dict[str, Any]]: out: dict[str, dict[str, Any]] = {} try: rows = ex.public_get_market_tickers( {"instType": "OPTION", "instFamily": inst_family} ).get("data") or [] for r in rows: if isinstance(r, dict) and r.get("instId"): out[str(r["instId"])] = r except Exception: pass return out def build_option_chain( ex: ccxt.okx, underlying: str, *, max_dte_days: float = 2.0, itm_only: bool = True, itm_max_dist_usd: float = 30.0, index_px: float | None = None, margin_mode: str | None = None, inst_family: str | None = None, ) -> dict[str, Any]: u = (underlying or "ETH").upper() if inst_family: family = str(inst_family).strip() else: try: from lib.options.options_margin_mode_lib import inst_family_for_underlying family = inst_family_for_underlying(u, margin_mode=margin_mode) except Exception: family = f"{u}-USD_UM" uly = f"{u}-USD" idx = index_px if index_px is not None else fetch_index_price(ex, uly) chain_margin = "usdc" if "_UM" in family.upper() else "coin" now_ms = time.time() * 1000 max_ms = now_ms + max_dte_days * 86400 * 1000 instruments_err = "" instruments: list[dict[str, Any]] = [] try: instruments = fetch_option_instruments(ex, family) if not instruments: # 空列表可能是瞬时空;短退避后强制再拉一次(非 50011) time.sleep(0.5) instruments = fetch_option_instruments(ex, family, force=True) if not instruments: instruments_err = "期权合约列表为空" except Exception as e: instruments = [] instruments_err = str(e) or e.__class__.__name__ # 限频:再等一下用 stale/缓存,不要连打 if _is_okx_rate_limit(e): time.sleep(1.5) try: instruments = fetch_option_instruments(ex, family, allow_stale=True) if instruments: instruments_err = "" except Exception as e2: instruments_err = str(e2) or e2.__class__.__name__ tickers = fetch_option_tickers(ex, family) expiries: dict[str, list[dict[str, Any]]] = {} skipped_no_index = 0 for meta in instruments: try: exp_ms = int(meta.get("expTime") or 0) except (TypeError, ValueError): continue if exp_ms <= now_ms or exp_ms > max_ms: continue opt_type = str(meta.get("optType") or "") strike = _safe_float(meta.get("stk")) if strike is None: continue if idx is None: skipped_no_index += 1 continue if itm_only and not is_shallow_itm( opt_type=opt_type, strike=strike, index_px=idx, max_dist_usd=itm_max_dist_usd, ): continue inst_id = str(meta.get("instId") or "") t = tickers.get(inst_id) or {} q = _resolve_chain_quote( ticker=t, meta=meta, opt_type=opt_type, strike=strike, index_px=idx, inst_id=inst_id, ) ask = q["ask"] bid = q["bid"] mark = q["mark_px"] ask_sz = q["ask_sz"] bid_sz = q["bid_sz"] expiry_be = expiry_breakeven_from_ask( opt_type=opt_type, strike=strike, ask_px=ask, mark_px=mark, inst_id=inst_id, margin_mode=chain_margin, ) mny = option_moneyness(opt_type=opt_type, strike=strike, index_px=idx) exp_key = str(exp_ms) expiries.setdefault(exp_key, []).append( { "inst_id": inst_id, "strike": strike, "opt_type": opt_type, "exp_time": exp_ms, "ask": ask, "bid": bid, "ask_sz": ask_sz, "bid_sz": bid_sz, "mark_px": mark, "ask_estimated": q["ask_estimated"], "expiry_be_px": expiry_be, "dist_expiry_be": strike_distance_to_be(strike, expiry_be, opt_type=opt_type), "moneyness": mny, "moneyness_label": option_moneyness_label(mny), "ct_mult": _safe_float(meta.get("ctMult")) or 0.01, "tick_sz": meta.get("tickSz"), "min_sz": int(_safe_float(meta.get("minSz")) or 1), } ) exp_list = [] for exp_ms_str, contracts in sorted(expiries.items(), key=lambda x: int(x[0])): contracts.sort(key=lambda c: (c["opt_type"], c["strike"])) exp_list.append({"exp_time": int(exp_ms_str), "contracts": contracts}) out: dict[str, Any] = { "underlying": u, "index_px": idx, "inst_family": family, "margin_mode": "usdc" if "_UM" in family.upper() else "coin", "premium_ccy": "USDC" if "_UM" in family.upper() else u, "expiries": exp_list, "instruments_count": len(instruments), } if not exp_list: if instruments_err: out["chain_error"] = f"拉取期权合约失败: {instruments_err}" elif idx is None: out["chain_error"] = "指数价获取失败,无法构建期权链" elif skipped_no_index: out["chain_error"] = "指数价缺失,合约已跳过" elif instruments: out["chain_error"] = f"近 {max_dte_days:g} 日内无可用到期(已过滤 {len(instruments)} 个合约)" else: out["chain_error"] = "期权合约列表为空,请稍后刷新" return out def option_buy_liquidity_ok(ask: Any, ask_sz: Any) -> tuple[bool, str]: """开仓仅认真实卖一价+卖一深度;不接受标记价/内在价值顶包.""" a = _safe_float(ask) s = _safe_float(ask_sz) if a is None or a <= 0: return False, "暂无卖一价,无法买入" if s is None or s <= 0: return False, "暂无卖一深度,无法买入" return True, "" def cap_option_buy_sheets_to_ask_depth( sheets: int, ask_sz: Any, *, min_sz: int = 1, ) -> tuple[int | None, str]: """将买入张数限制在卖一深度内(向下取整).""" depth = _safe_float(ask_sz) if depth is None or depth <= 0: return None, "暂无卖一深度,无法买入" max_sheets = int(math.floor(depth + 1e-12)) need = max(1, int(min_sz or 1)) if max_sheets < need: return None, f"卖一深度不足 {need} 张(当前 {depth:g})" want = max(0, int(sheets)) capped = min(want, max_sheets) if capped < need: return None, f"卖一深度不足 {need} 张(当前 {depth:g})" return capped, "" def quote_option_contract(ex: ccxt.okx, inst_id: str) -> dict[str, Any]: inst_id = (inst_id or "").strip() if not inst_id: return {"ok": False, "msg": "缺少 inst_id"} try: meta = fetch_option_instrument_meta(ex, inst_id) t_rows: list[Any] = [] ticker_err: BaseException | None = None for attempt in range(3): try: t_rows = ex.public_get_market_ticker({"instId": inst_id}).get("data") or [] ticker_err = None break except Exception as e: ticker_err = e if _is_okx_rate_limit(e) and attempt < 2: time.sleep(0.45 * (attempt + 1)) continue break if not meta and t_rows: meta = _meta_from_inst_id_fallback(inst_id) if not meta: if ticker_err is not None and _is_okx_rate_limit(ticker_err): return {"ok": False, "msg": "行情限频,请稍后重试"} return {"ok": False, "msg": "合约不存在"} t = t_rows[0] if t_rows else {} # 开仓用真实盘口卖一;绝不把标记价写入 ask ask = _safe_float(t.get("askPx")) bid = _safe_float(t.get("bidPx")) ask_sz = _safe_float(t.get("askSz")) bid_sz = _safe_float(t.get("bidSz")) if ask is None or bid is None or ask_sz is None or bid_sz is None: book_bid, book_ask, book_bid_sz, book_ask_sz = _fetch_book_top(ex, inst_id) if ask is None: ask = book_ask if bid is None: bid = book_bid if ask_sz is None: ask_sz = book_ask_sz if bid_sz is None: bid_sz = book_bid_sz mark = _safe_float(t.get("markPx")) tick_sz = meta.get("tickSz") # 买一缺失时仍可用标记价补展示(平仓路径读 bid);开仓 ask 不顶包 if bid is None and mark is not None: bid = round_option_px(mark, tick_sz, "sell") ref_ask = None if ask is None and mark is not None and mark > 0: ref_ask = round_option_px(mark, tick_sz, "buy") can_open, open_block_msg = option_buy_liquidity_ok(ask, ask_sz) book_ask = ask book_ask_sz = ask_sz uly = str(meta.get("uly") or "") idx = fetch_index_price(ex, uly) opt_type = meta.get("optType") strike = _safe_float(meta.get("stk")) expiry_be = expiry_breakeven_from_ask( opt_type=str(opt_type or ""), strike=strike, ask_px=book_ask if can_open else None, mark_px=mark, inst_id=inst_id, ) return { "ok": True, "inst_id": inst_id, "meta": meta, "ask": book_ask if can_open else None, "bid": bid, "ask_sz": book_ask_sz if can_open else None, "bid_sz": bid_sz, "mark": mark, "ref_ask": ref_ask, "book_ask": book_ask, "book_ask_sz": book_ask_sz, "can_open": can_open, "ask_source": "book" if can_open else "none", "open_block_msg": "" if can_open else open_block_msg, "index_px": idx, "expiry_be_px": expiry_be, "dist_expiry_be": strike_distance_to_be(strike, expiry_be, opt_type=str(opt_type or "")), "ct_mult": _safe_float(meta.get("ctMult")) or 0.01, "min_sz": int(_safe_float(meta.get("minSz")) or 1), "tick_sz": tick_sz, "strike": strike, "opt_type": opt_type, "exp_time": meta.get("expTime"), } except Exception as e: return {"ok": False, "msg": str(e)} def fetch_option_pending_orders(ex: ccxt.okx, inst_id: str | None = None) -> list[dict[str, Any]]: """未成交期权委托(限价挂单).""" params: dict[str, Any] = {"instType": "OPTION"} inst = (inst_id or "").strip() if inst: params["instId"] = inst try: rows = ex.private_get_trade_orders_pending(params).get("data") or [] except Exception: return [] out: list[dict[str, Any]] = [] for o in rows: if not isinstance(o, dict): continue oid = str(o.get("ordId") or "").strip() iid = str(o.get("instId") or "").strip() if not oid or not iid: continue side = str(o.get("side") or "").lower() px = _safe_float(o.get("px")) sz = _safe_float(o.get("sz")) fill_sz = _safe_float(o.get("fillSz")) or 0.0 acc_fill = _safe_float(o.get("accFillSz")) if acc_fill is not None: fill_sz = acc_fill out.append( { "ord_id": oid, "inst_id": iid, "side": side, "side_label": "买入" if side == "buy" else ("卖出" if side == "sell" else side or "—"), "px": px, "sz": int(sz) if sz is not None else None, "fill_sz": int(fill_sz) if fill_sz is not None else 0, "state": str(o.get("state") or ""), "ord_type": str(o.get("ordType") or ""), "c_time": o.get("cTime"), "u_time": o.get("uTime"), "reduce_only": str(o.get("reduceOnly") or "").lower() in ("true", "1", "yes"), } ) out.sort(key=lambda x: int(float(x.get("c_time") or 0)), reverse=True) return out def cancel_option_order(ex: ccxt.okx, *, inst_id: str, ord_id: str) -> dict[str, Any]: inst_id = (inst_id or "").strip() ord_id = (ord_id or "").strip() if not inst_id or not ord_id: return {"ok": False, "msg": "缺少 inst_id 或 ord_id"} try: resp = ex.private_post_trade_cancel_order({"instId": inst_id, "ordId": ord_id}) data = (resp or {}).get("data") or [] if data and str(data[0].get("sCode")) == "0": return {"ok": True, "data": data[0], "raw": resp} return {"ok": False, "msg": _okx_trade_error_message(resp=resp), "raw": resp} except Exception as e: return {"ok": False, "msg": _okx_trade_error_message(e)} def fetch_option_order(ex: ccxt.okx, *, inst_id: str, ord_id: str) -> dict[str, Any]: """查询单笔期权订单状态.""" inst_id = (inst_id or "").strip() ord_id = (ord_id or "").strip() if not inst_id or not ord_id: return {"ok": False, "msg": "缺少 inst_id 或 ord_id"} try: resp = ex.private_get_trade_order({"instId": inst_id, "ordId": ord_id}) data = (resp or {}).get("data") or [] if not data or not isinstance(data[0], dict): return {"ok": False, "msg": "订单不存在或暂不可查", "raw": resp} o = data[0] sz = _safe_float(o.get("sz")) acc = _safe_float(o.get("accFillSz")) if acc is None: acc = _safe_float(o.get("fillSz")) or 0.0 avg = _safe_float(o.get("avgPx")) fill_px = _safe_float(o.get("fillPx")) if avg is None or avg <= 0: avg = fill_px state = str(o.get("state") or "").strip().lower() return { "ok": True, "ord_id": str(o.get("ordId") or ord_id), "inst_id": str(o.get("instId") or inst_id), "state": state, "sz": int(sz) if sz is not None else None, "acc_fill_sz": float(acc or 0), "avg_px": avg, "side": str(o.get("side") or "").lower(), "ord_type": str(o.get("ordType") or ""), "raw": o, } except Exception as e: return {"ok": False, "msg": _okx_trade_error_message(e)} def wait_option_order_full_fill( ex: ccxt.okx, *, inst_id: str, ord_id: str, need_sheets: int, timeout_sec: float = 12.0, poll_sec: float = 0.35, cancel_on_timeout: bool = True, ) -> dict[str, Any]: """轮询至完全成交;超时则撤单.未完全成交返回 ok=False.""" need = max(1, int(need_sheets)) deadline = time.time() + max(0.5, float(timeout_sec)) last: dict[str, Any] = {} while time.time() < deadline: last = fetch_option_order(ex, inst_id=inst_id, ord_id=ord_id) if not last.get("ok"): time.sleep(max(0.15, float(poll_sec))) continue acc = float(last.get("acc_fill_sz") or 0) state = str(last.get("state") or "") if acc + 1e-9 >= need or state == "filled": if acc + 1e-9 < need: return { "ok": False, "msg": f"订单已结束但成交不足 {need} 张(已成 {acc:g})", "filled_sheets": acc, "order": last, } return { "ok": True, "filled_sheets": int(round(acc)), "avg_px": last.get("avg_px"), "state": state, "order": last, } if state in ("canceled", "cancelled", "mmp_canceled"): if acc + 1e-9 >= need: return { "ok": True, "filled_sheets": int(round(acc)), "avg_px": last.get("avg_px"), "state": state, "order": last, } return { "ok": False, "msg": f"订单已撤销且未完全成交(已成 {acc:g}/{need})", "filled_sheets": acc, "order": last, } time.sleep(max(0.15, float(poll_sec))) if cancel_on_timeout: cancel_option_order(ex, inst_id=inst_id, ord_id=ord_id) time.sleep(0.25) last = fetch_option_order(ex, inst_id=inst_id, ord_id=ord_id) acc = float((last or {}).get("acc_fill_sz") or 0) if (last or {}).get("ok") else 0.0 if acc + 1e-9 >= need: return { "ok": True, "filled_sheets": int(round(acc)), "avg_px": (last or {}).get("avg_px"), "state": (last or {}).get("state"), "order": last, "timed_out": True, } return { "ok": False, "msg": f"等待成交超时({float(timeout_sec):g}s),已撤未成交部分;已成 {acc:g}/{need}", "filled_sheets": acc, "order": last, "timed_out": True, } def place_option_limit_order( ex: ccxt.okx, *, inst_id: str, side: str, sheets: int, price: float, td_mode: str = "isolated", tick_sz: Any = None, reduce_only: bool = False, pos_side: str | None = None, ord_type: str = "limit", ) -> dict[str, Any]: side_l = (side or "").lower() if side_l not in ("buy", "sell"): return {"ok": False, "msg": "side 必须为 buy 或 sell"} if sheets < 1: return {"ok": False, "msg": "张数至少为 1"} ot = (ord_type or "limit").strip().lower() if ot not in ("limit", "ioc", "fok", "post_only"): return {"ok": False, "msg": f"不支持的 ordType: {ord_type}"} px = round_option_px(float(price), tick_sz, side_l) if px <= 0: return {"ok": False, "msg": "价格无效"} body: dict[str, Any] = { "instId": inst_id, "tdMode": td_mode, "side": side_l, "ordType": ot, "px": format_option_px(px, tick_sz), "sz": str(int(sheets)), } if pos_side: body["posSide"] = pos_side if reduce_only: body["reduceOnly"] = "true" try: resp = ex.private_post_trade_order(body) data = (resp or {}).get("data") or [] if data and str(data[0].get("sCode")) == "0": return {"ok": True, "data": data[0], "raw": resp, "px": px, "ord_type": ot} return {"ok": False, "msg": _okx_trade_error_message(resp=resp), "raw": resp, "px": px} except Exception as e: return {"ok": False, "msg": _okx_trade_error_message(e), "px": px} def place_option_market_order( ex: ccxt.okx, *, inst_id: str, side: str, sheets: int, td_mode: str = "isolated", reduce_only: bool = False, pos_side: str | None = None, ) -> dict[str, Any]: side_l = (side or "").lower() if side_l not in ("buy", "sell"): return {"ok": False, "msg": "side 必须为 buy 或 sell"} if sheets < 1: return {"ok": False, "msg": "张数至少为 1"} body: dict[str, Any] = { "instId": inst_id, "tdMode": td_mode, "side": side_l, "ordType": "market", "sz": str(int(sheets)), } if pos_side: body["posSide"] = pos_side if reduce_only: body["reduceOnly"] = "true" try: resp = ex.private_post_trade_order(body) data = (resp or {}).get("data") or [] if data and str(data[0].get("sCode")) == "0": return {"ok": True, "data": data[0], "raw": resp} return {"ok": False, "msg": _okx_trade_error_message(resp=resp), "raw": resp} except Exception as e: return {"ok": False, "msg": _okx_trade_error_message(e)} _OPTION_POSITIONS_CACHE: dict[str, Any] = {"updated_at": 0.0, "rows": None, "failed": False} _OPTION_POSITIONS_CACHE_LOCK = threading.Lock() _OPTION_POSITIONS_CACHE_TTL = 4.0 _OPTION_POSITIONS_STALE_OK_SEC = 30.0 def invalidate_option_positions_cache() -> None: with _OPTION_POSITIONS_CACHE_LOCK: _OPTION_POSITIONS_CACHE["updated_at"] = 0.0 _OPTION_POSITIONS_CACHE["failed"] = False def fetch_option_positions(ex: ccxt.okx) -> list[dict[str, Any]] | None: """期权持仓:有仓返回列表,无仓返回 [],API 失败返回 None(短时回退缓存).""" now = time.time() with _OPTION_POSITIONS_CACHE_LOCK: age = now - float(_OPTION_POSITIONS_CACHE["updated_at"] or 0.0) cached = _OPTION_POSITIONS_CACHE["rows"] if age < _OPTION_POSITIONS_CACHE_TTL and cached is not None and not _OPTION_POSITIONS_CACHE["failed"]: return list(cached) try: rows = ex.private_get_account_positions({"instType": "OPTION"}).get("data") or [] out = [] for r in rows: if not isinstance(r, dict): continue pos = _safe_float(r.get("pos")) if pos is None or abs(pos) < 1e-12: continue out.append(r) with _OPTION_POSITIONS_CACHE_LOCK: _OPTION_POSITIONS_CACHE["updated_at"] = now _OPTION_POSITIONS_CACHE["rows"] = out _OPTION_POSITIONS_CACHE["failed"] = False return out except Exception: with _OPTION_POSITIONS_CACHE_LOCK: cached = _OPTION_POSITIONS_CACHE["rows"] age = now - float(_OPTION_POSITIONS_CACHE["updated_at"] or 0.0) if cached is not None and age < _OPTION_POSITIONS_STALE_OK_SEC: return list(cached) _OPTION_POSITIONS_CACHE["updated_at"] = now _OPTION_POSITIONS_CACHE["rows"] = None _OPTION_POSITIONS_CACHE["failed"] = True return None def fetch_option_position_history( ex: ccxt.okx, inst_id: str, *, limit: int = 20, ) -> list[dict[str, Any]]: """OKX 期权历史仓位(含到期结算/平仓).""" inst_id = (inst_id or "").strip() if not inst_id: return [] try: resp = ex.private_get_account_positions_history( { "instType": "OPTION", "instId": inst_id, "limit": str(max(1, min(int(limit), 100))), } ) rows = (resp or {}).get("data") or [] return [r for r in rows if isinstance(r, dict)] except Exception: return [] def fetch_all_option_positions_history( ex: ccxt.okx, *, limit: int = 200, ) -> list[dict[str, Any]]: """拉取 OKX 期权全部历史仓位(分页,按平仓时间倒序).""" cap = max(1, min(int(limit), 500)) out: list[dict[str, Any]] = [] after: str | None = None while len(out) < cap: page_limit = min(100, cap - len(out)) params: dict[str, Any] = { "instType": "OPTION", "limit": str(page_limit), } if after is not None: params["after"] = after try: resp = ex.private_get_account_positions_history(params) except Exception: break rows = (resp or {}).get("data") or [] batch = [r for r in rows if isinstance(r, dict)] if not batch: break out.extend(batch) if len(batch) < page_limit: break utimes = [_safe_float(r.get("uTime")) for r in batch] utimes = [int(u) for u in utimes if u is not None and u > 0] if not utimes: break oldest = min(utimes) if after is not None and str(oldest) == after: break after = str(oldest) out = [r for r in out if is_option_full_close_history(r)] out.sort(key=lambda r: int(_safe_float(r.get("uTime")) or 0), reverse=True) return out[:cap] def format_option_history_row( raw: dict[str, Any], *, tick_sz: Any = None, ct_mult: float = 0.01, ) -> dict[str, Any]: """标准化 OKX positions-history 单条记录供前端展示.""" from lib.options.options_pricing_lib import total_premium inst_id = str(raw.get("instId") or "").strip() open_avg = _safe_float(raw.get("openAvgPx")) close_avg = _safe_float(raw.get("closeAvgPx")) sheets = _safe_float(raw.get("closeTotalPos")) if sheets is None or sheets <= 0: sheets = _safe_float(raw.get("openMaxPos")) sheets_i = int(abs(sheets or 0)) eth_amount = round(abs(sheets or 0) * ct_mult, 8) if sheets else 0.0 premium_paid = ( round(total_premium(open_avg, eth_amount), 8) if open_avg is not None and eth_amount > 0 else None ) realized = _safe_float(raw.get("realizedPnl")) if realized is None: realized = _safe_float(raw.get("pnl")) pnl_ratio = _safe_float(raw.get("pnlRatio")) close_type = str(raw.get("type") or "").strip() utime = _safe_float(raw.get("uTime")) ctime = _safe_float(raw.get("cTime")) opt_type, strike = option_fields_from_inst_id(inst_id) uly = str(raw.get("uly") or inst_id.split("-")[0] or "").replace("-USD_UM", "").replace("-USD", "") try: from lib.options.options_margin_mode_lib import margin_mode_from_inst_id, premium_ccy_for_mode row_mode = margin_mode_from_inst_id(inst_id) if inst_id else "usdc" premium_ccy = premium_ccy_for_mode(row_mode, uly or "ETH") except Exception: row_mode = "usdc" premium_ccy = "USDC" idx_px = _safe_float(raw.get("idxPx") or raw.get("idx_px")) if close_type in ("3", "4"): status_label = "强平" else: status_label = "已平" pos_id = str(raw.get("posId") or "").strip() or None close_ms = int(utime) if utime is not None else None return { "source": "exchange", "history_key": option_history_row_key( source="exchange", inst_id=inst_id, pos_id=pos_id, close_ms=close_ms, ), "pos_id": pos_id, "inst_id": inst_id, "underlying": uly, "opt_type": opt_type, "strike": strike, "sheets": sheets_i, "eth_amount": eth_amount, "ct_mult": ct_mult, "open_avg_px": open_avg, "open_avg_px_fmt": format_option_px(open_avg, tick_sz) if open_avg is not None else None, "close_avg_px": close_avg, "close_avg_px_fmt": format_option_px(close_avg, tick_sz) if close_avg is not None else None, "premium_paid": premium_paid, "premium_paid_fmt": format_premium_amount(premium_paid, ccy=premium_ccy), "premium_ccy": premium_ccy, "margin_mode": row_mode, "margin_mode_label": "币本位" if row_mode == "coin" else "USDC", "idx_px": idx_px, "realized_pnl": realized, "pnl_ratio_pct": round(pnl_ratio * 100, 2) if pnl_ratio is not None else None, "status": "closed", "status_label": status_label, "close_type": close_type, "created_at": _ms_to_iso(ctime), "closed_at": _ms_to_iso(utime), "close_ms": close_ms, "tick_sz": tick_sz, "raw": raw, } def format_live_option_history_row( row: dict[str, Any], *, open_ms: int | None = None, ) -> dict[str, Any]: """将当前持仓格式化为历史列表中的「持仓中」行.""" inst_id = str(row.get("inst_id") or "").strip() pos_id = str((row.get("raw") or {}).get("posId") or "").strip() or None close_ms = open_ms premium_ccy = str(row.get("premium_ccy") or "USDC").strip().upper() or "USDC" return { "source": "live", "history_key": option_history_row_key( source="live", inst_id=inst_id, pos_id=pos_id, close_ms=close_ms, ), "pos_id": pos_id, "inst_id": inst_id, "underlying": str(row.get("underlying") or inst_id.split("-")[0] or ""), "opt_type": row.get("opt_type"), "strike": row.get("strike"), "sheets": int(abs(_safe_float(row.get("pos")) or 0)), "eth_amount": row.get("eth_amount"), "open_avg_px": row.get("avg_px"), "open_avg_px_fmt": row.get("avg_px_fmt"), "close_avg_px": None, "close_avg_px_fmt": None, "premium_paid": row.get("premium_paid"), "premium_paid_fmt": row.get("premium_paid_fmt"), "premium_ccy": premium_ccy, "margin_mode": row.get("margin_mode"), "margin_mode_label": row.get("margin_mode_label"), "idx_px": row.get("idx_px"), "realized_pnl": row.get("upl"), "pnl_ratio_pct": row.get("upl_ratio_pct"), "status": "open", "status_label": "持仓中", "close_type": None, "created_at": _ms_to_iso(open_ms), "closed_at": None, "close_ms": open_ms, "tick_sz": row.get("tick_sz"), "raw": row.get("raw"), } def resolve_option_close_from_history( hist_rows: list[dict[str, Any]], *, open_ms: int | None = None, close_ms: int | None = None, sheets: float | int | None = None, ) -> dict[str, Any] | None: """从 positions-history 选取匹配的平仓记录. 同合约多次开平时,优先按开仓时间(cTime≈open_ms)对齐,再按平仓时间/张数; 无锚点时取开仓后最晚一条(供刚平掉的持仓同步)。 """ candidates: list[tuple[int, dict[str, Any]]] = [] for row in hist_rows: u_ms = _safe_float(row.get("uTime")) if u_ms is None or u_ms <= 0: continue u_i = int(u_ms) # 本地时间偶发与交易所差整时区时,放宽到 12h,主要靠 cTime/张数精配 if open_ms is not None and u_i < int(open_ms) - 12 * 3600_000: continue candidates.append((u_i, row)) if not candidates: return None has_ctime = any(_safe_float(row.get("cTime")) is not None for _, row in candidates) want_sheets = _safe_float(sheets) def _score(item: tuple[int, dict[str, Any]]) -> tuple: u_i, row = item c_ms = _safe_float(row.get("cTime")) parts: list[float] = [] # 张数优先:同合约多笔时最稳,且不受本地/交易所时区偏差影响 if want_sheets is not None: hist_sheets = _safe_float(row.get("closeTotalPos")) if hist_sheets is None: hist_sheets = _safe_float(row.get("openMaxPos")) parts.append( abs(float(hist_sheets) - float(want_sheets)) if hist_sheets is not None else 1e12 ) if open_ms is not None and c_ms is not None: parts.append(float(abs(int(c_ms) - int(open_ms)))) if close_ms is not None: parts.append(float(abs(u_i - int(close_ms)))) if not parts: parts.append(float(-u_i)) # 同距时偏向更晚平仓 parts.append(float(-u_i)) return tuple(parts) if open_ms is None and close_ms is None and want_sheets is None: u_i, best = max(candidates, key=lambda item: item[0]) elif open_ms is not None and close_ms is None and want_sheets is None and not has_ctime: # 兼容旧调用:只有 open_ms 时仍取最晚一条 u_i, best = max(candidates, key=lambda item: item[0]) else: u_i, best = min(candidates, key=_score) realized = _safe_float(best.get("realizedPnl")) if realized is None: realized = _safe_float(best.get("pnl")) return { "close_quote": _safe_float(best.get("closeAvgPx")), "realized_pnl": realized, "close_ms": u_i, "pos_id": str(best.get("posId") or "").strip() or None, } def fetch_options_unrealized_pnl_usdc(ex: ccxt.okx) -> float | None: """ 期权浮盈合计(USDC≈U). 优先返回交易所标记价 upl;实例顶栏应改用 `options_positions_lib.sum_options_net_pnl_usdc`(买一净盈亏)以与持仓卡一致. """ positions = fetch_option_positions(ex) if positions is None: return None total = 0.0 found = False for pos in positions: upl = _safe_float(pos.get("upl")) if upl is None: continue found = True total += upl return round(total, 4) if found else None def estimate_usdt_to_usdc(ex: ccxt.okx, usdt_amount: float) -> dict[str, Any]: if usdt_amount <= 0: return {"ok": False, "msg": "兑换数量须大于 0"} try: resp = ex.private_post_asset_convert_estimate_quote( { "baseCcy": "USDC", "quoteCcy": "USDT", "side": "buy", "rfqSz": str(usdt_amount), "rfqSzCcy": "USDT", } ) data = (resp or {}).get("data") or [] if not data: return {"ok": False, "msg": "询价失败", "raw": resp} row = data[0] return { "ok": True, "quote_id": row.get("quoteId"), "base_ccy": row.get("baseCcy"), "quote_ccy": row.get("quoteCcy"), "cnvt_px": _safe_float(row.get("cnvtPx")), "base_sz": _safe_float(row.get("baseSz")), "quote_sz": _safe_float(row.get("quoteSz")), "rfq_sz": usdt_amount, "raw": row, } except Exception as e: return {"ok": False, "msg": str(e)} def execute_convert(ex: ccxt.okx, quote_id: str) -> dict[str, Any]: if not quote_id: return {"ok": False, "msg": "缺少 quoteId"} try: resp = ex.private_post_asset_convert_trade({"quoteId": str(quote_id)}) data = (resp or {}).get("data") or [] if data and str(data[0].get("sCode", "0")) == "0": return {"ok": True, "data": data[0], "raw": resp} return {"ok": False, "msg": _okx_trade_error_message(resp=resp), "raw": resp} except Exception as e: return {"ok": False, "msg": _okx_trade_error_message(e)} def transfer_ccy( ex: ccxt.okx, ccy: str, amount: float, from_account: str, to_account: str, ) -> dict[str, Any]: if amount <= 0: return {"ok": False, "msg": "划转金额须大于 0"} try: resp = ex.transfer(str(ccy).upper(), float(amount), from_account, to_account) return {"ok": True, "data": resp} except Exception as e: return {"ok": False, "msg": _okx_trade_error_message(e)} _OKX_ACCT_CODE = {"funding": "6", "trading": "18", "spot": "18"} def fetch_options_trading_usdc(ex: ccxt.okx, *, force: bool = False) -> float | None: bal = fetch_options_balances(ex, force=force) v = bal.get("trading_usdc") if v is None: return None return round(float(v), 2) def fetch_options_funding_usdc(ex: ccxt.okx, *, force: bool = False) -> float | None: bal = fetch_options_balances(ex, force=force) v = bal.get("funding_usdc") if v is None: return None return round(float(v), 2) def fetch_options_funding_usdt(ex: ccxt.okx, *, force: bool = False) -> float | None: bal = fetch_options_balances(ex, force=force) v = bal.get("funding_usdt") if v is None: return None return round(float(v), 2) def spot_market_swap_usdt_usdc( ex: ccxt.okx, *, direction: str, amount: float, ) -> dict[str, Any]: """现货市价兑换 USDC-USDT.direction: usdt_to_usdc | usdc_to_usdt.""" if amount <= 0: return {"ok": False, "msg": "数量须大于 0"} d = (direction or "").lower() inst_id = "USDC-USDT" try: if d == "usdt_to_usdc": body = { "instId": inst_id, "tdMode": "cash", "side": "buy", "ordType": "market", "sz": str(amount), "tgtCcy": "quote_ccy", } elif d == "usdc_to_usdt": body = { "instId": inst_id, "tdMode": "cash", "side": "sell", "ordType": "market", "sz": str(amount), "tgtCcy": "base_ccy", } else: return {"ok": False, "msg": "direction 须为 usdt_to_usdc 或 usdc_to_usdt"} resp = ex.private_post_trade_order(body) data = (resp or {}).get("data") or [] if data and str(data[0].get("sCode")) == "0": return {"ok": True, "data": data[0], "raw": resp} return {"ok": False, "msg": _okx_trade_error_message(resp=resp), "raw": resp} except Exception as e: return {"ok": False, "msg": _okx_trade_error_message(e)} def format_position_row( pos: dict[str, Any], ct_mult: float = 0.01, *, tick_sz: Any = None, ) -> dict[str, Any]: from lib.options.options_pricing_lib import ( close_breakeven_idx, expiry_breakeven_px, idx_distance_to_be, strike_distance_to_be, total_premium, ) sheets = _safe_float(pos.get("pos")) or 0.0 avg = _safe_float(pos.get("avgPx")) mark = _safe_float(pos.get("markPx")) upl = _safe_float(pos.get("upl")) upl_ratio = _safe_float(pos.get("uplRatio")) idx_px = _safe_float(pos.get("idxPx")) inst_id = str(pos.get("instId") or "") opt_type = pos.get("optType") strike = _safe_float(pos.get("stk")) parsed_type, parsed_strike = option_fields_from_inst_id(inst_id) if not opt_type: opt_type = parsed_type if strike is None: strike = parsed_strike try: from lib.options.options_margin_mode_lib import margin_mode_from_inst_id, premium_ccy_for_mode row_mode = margin_mode_from_inst_id(inst_id) if inst_id else "usdc" underly = (inst_id.split("-")[0] if inst_id else "ETH") or "ETH" premium_ccy = premium_ccy_for_mode(row_mode, underly) except Exception: row_mode = "usdc" underly = (inst_id.split("-")[0] if inst_id else "ETH") or "ETH" premium_ccy = "USDC" eth_amount = round(abs(sheets) * ct_mult, 8) premium_paid = ( round(total_premium(avg, eth_amount), 8) if avg is not None and eth_amount > 0 else None ) delta_pa = _safe_float(pos.get("deltaPA")) expiry_be = expiry_breakeven_px( opt_type=str(opt_type or ""), strike=strike, avg_px=avg, be_px_api=_safe_float(pos.get("bePx")), inst_id=inst_id, margin_mode=row_mode, ) close_be = close_breakeven_idx( opt_type=str(opt_type or ""), idx_px=idx_px, mark_px=mark, avg_px=avg, delta_pa=delta_pa, pos=sheets, ct_mult=ct_mult, ) exp_time_ms = normalize_option_exp_ms(pos.get("expTime"), inst_id) return { "inst_id": inst_id or pos.get("instId"), "pos": sheets, "eth_amount": eth_amount, "avg_px": avg, "mark_px": mark, "avg_px_fmt": format_option_px(avg, tick_sz) if avg is not None else None, "mark_px_fmt": format_option_px(mark, tick_sz) if mark is not None else None, "premium_paid_fmt": format_premium_amount(premium_paid, ccy=premium_ccy), "tick_sz": tick_sz, "ct_mult": ct_mult, "idx_px": idx_px, "premium_paid": premium_paid, "margin_mode": row_mode, "premium_ccy": premium_ccy, "underlying": underly, "upl": upl, "upl_ratio_pct": round(upl_ratio * 100, 2) if upl_ratio is not None else None, "exp_time": exp_time_ms, "exp_time_ms": exp_time_ms, "opt_type": opt_type, "strike": strike, "avail_pos": _safe_float(pos.get("availPos")), "expiry_be_px": expiry_be, "close_be_px": close_be, "dist_expiry_be": strike_distance_to_be(strike, expiry_be, opt_type=str(opt_type or "")), "dist_close_be": idx_distance_to_be(idx_px, close_be), "raw": pos, } def options_api_ready(ex: ccxt.okx | None) -> tuple[bool, str]: if ex is None: return False, "期权 API 未配置" if not ex.apiKey or not ex.secret or not ex.password: return False, "期权 API Key 不完整" return True, ""