"""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, is_shallow_itm, option_moneyness, option_moneyness_label, ) _OKX_OPTION_ERR_ZH: dict[str, str] = { "51018": "期权账户不能持有净空头头寸", "51019": "期权买入须使用逐仓模式(全仓模式下不能持有多头净头寸)", } _OPTIONS_BALANCE_CACHE: dict[str, Any] = {"updated_at": 0.0, "data": None} def invalidate_options_balance_cache() -> None: _OPTIONS_BALANCE_CACHE["updated_at"] = 0.0 _OPTIONS_BALANCE_CACHE["data"] = None 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 "") zh = _OKX_OPTION_ERR_ZH.get(code) if zh: return zh msg = str(row.get("sMsg") or "").strip() 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 "下单失败" 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: ex = ccxt.okx( { "apiKey": api_key, "secret": api_secret, "password": passphrase, "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: return str(px) decimals = max(0, -int(round(math.log10(tick)))) if tick < 1 else 0 if tick >= 1: decimals = len(str(tick).split(".")[-1]) if "." in str(tick) else 0 return f"{px:.{decimals}f}".rstrip("0").rstrip(".") or "0" def format_usdc_amount(v: float | None) -> str | None: """USDC 金额展示(权利金/回收等,固定 2 位小数).""" if v is None: return None return f"{float(v):.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'}" if pos_id: 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 _intrinsic_px_per_unit(opt_type: str, strike: float, index_px: float) -> float | None: o = (opt_type or "").upper() if o == "C" and index_px > strike: return float(index_px) - float(strike) if o == "P" and index_px < strike: return float(strike) - float(index_px) return None def _resolve_chain_quote( *, ticker: dict[str, Any], meta: dict[str, Any], opt_type: str, strike: float, index_px: float, ) -> 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 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) 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) 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 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 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 = ex.public_get_public_instruments( {"instType": "OPTION", "instFamily": family} ).get("data") or [] for r in rows: if isinstance(r, dict) and str(r.get("instId")) == inst_id: return r except Exception: return None 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} avail: dict[str, float | None] = {"USDT": None, "USDC": None, "USDG": 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_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") trading, trading_avail = fetch_account_balances_by_type(ex, "trading") if trading.get("USDC") is None: swap_bal, swap_avail = fetch_account_balances_by_type(ex, "swap") if swap_bal.get("USDC") is not None: trading["USDC"] = swap_bal["USDC"] if trading_avail.get("USDC") is None and swap_avail.get("USDC") is not None: trading_avail["USDC"] = swap_avail["USDC"] result = { "scope": "main", "funding_usdt": funding.get("USDT"), "funding_usdc": funding.get("USDC"), "funding_usdg": funding.get("USDG"), "funding_usdt_avail": funding_avail.get("USDT"), "funding_usdc_avail": funding_avail.get("USDC"), "trading_usdt": trading.get("USDT"), "trading_usdc": trading.get("USDC"), "trading_usdg": trading.get("USDG"), "trading_usdt_avail": trading_avail.get("USDT"), "trading_usdc_avail": trading_avail.get("USDC"), } _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,资金 USDC/USDT(单次拉取 + 缓存).""" bal = fetch_options_balances(ex, force=force) def _round(v: Any) -> float | None: if v is None: return None try: return round(float(v), 2) except (TypeError, ValueError): return None return ( _round(bal.get("trading_usdc")), _round(bal.get("funding_usdc")), _round(bal.get("funding_usdt")), _round(bal.get("trading_usdt")), ) 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, ) -> list[dict[str, Any]]: try: rows = ex.public_get_public_instruments( {"instType": "OPTION", "instFamily": inst_family} ).get("data") or [] return [r for r in rows if isinstance(r, dict) and r.get("state") == "live"] except Exception: return [] 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, ) -> dict[str, Any]: u = (underlying or "ETH").upper() family = f"{u}-USD_UM" uly = f"{u}-USD" idx = index_px if index_px is not None else fetch_index_price(ex, uly) now_ms = time.time() * 1000 max_ms = now_ms + max_dte_days * 86400 * 1000 instruments = fetch_option_instruments(ex, family) tickers = fetch_option_tickers(ex, family) expiries: dict[str, list[dict[str, Any]]] = {} 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 or idx is None: 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, ) 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, ) 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": idx_distance_to_be(idx, expiry_be), "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}) return {"underlying": u, "index_px": idx, "inst_family": family, "expiries": exp_list} 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) if not meta: return {"ok": False, "msg": "合约不存在"} t_rows = ex.public_get_market_ticker({"instId": inst_id}).get("data") or [] t = t_rows[0] if t_rows else {} 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") if ask is None and mark is not None: ask = round_option_px(mark, tick_sz, "buy") if bid is None and mark is not None: bid = round_option_px(mark, tick_sz, "sell") 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=ask, mark_px=mark, ) return { "ok": True, "inst_id": inst_id, "meta": meta, "ask": ask, "bid": bid, "ask_sz": ask_sz, "bid_sz": bid_sz, "mark": mark, "index_px": idx, "expiry_be_px": expiry_be, "dist_expiry_be": idx_distance_to_be(idx, expiry_be), "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 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, ) -> 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"} 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": "limit", "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} 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", "") 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, "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_usdc_amount(premium_paid), "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 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"), "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, ) -> dict[str, Any] | None: """从 positions-history 中选取最近一条有效平仓/结算记录.""" best: dict[str, Any] | None = None best_utime = -1 for row in hist_rows: u_ms = _safe_float(row.get("uTime")) if u_ms is None or u_ms <= 0: continue if open_ms is not None and u_ms < int(open_ms) - 60_000: continue if u_ms > best_utime: best = row best_utime = int(u_ms) if not best: return None 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": best_utime, "pos_id": str(best.get("posId") or "").strip() or None, } def fetch_options_unrealized_pnl_usdc(ex: ccxt.okx) -> float | None: """期权持仓未实现盈亏合计(USDC,统计口径与 USDT 1:1).""" 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} msg = data[0].get("sMsg") if data else str(resp) return {"ok": False, "msg": msg or "兑换失败", "raw": resp} except Exception as e: return {"ok": False, "msg": str(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": str(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} msg = data[0].get("sMsg") if data else str(resp) return {"ok": False, "msg": msg or "现货兑换失败", "raw": resp} except Exception as e: return {"ok": False, "msg": str(e)} def transfer_main_sub_account( ex: ccxt.okx, *, ccy: str, amount: float, sub_acct: str, main_to_sub: bool, from_account: str = "funding", to_account: str = "funding", ) -> dict[str, Any]: """主账户与子账户之间划转(须主账户 API).""" if amount <= 0: return {"ok": False, "msg": "划转金额须大于 0"} sub = (sub_acct or "").strip() if not sub: return {"ok": False, "msg": "未配置子账户名称 OKX_SUB_ACCOUNT_NAME"} from_code = _OKX_ACCT_CODE.get((from_account or "funding").lower(), "6") to_code = _OKX_ACCT_CODE.get((to_account or "funding").lower(), "6") try: resp = ex.private_post_asset_transfer( { "type": "1" if main_to_sub else "2", "ccy": str(ccy).upper(), "amt": str(amount), "from": from_code, "to": to_code, "subAcct": sub, } ) data = (resp or {}).get("data") or [] if data and str(data[0].get("sCode", "0")) == "0": return {"ok": True, "data": data[0], "raw": resp} msg = data[0].get("sMsg") if data else str(resp) return {"ok": False, "msg": msg or "主/子账户划转失败", "raw": resp} except Exception as e: return {"ok": False, "msg": str(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, 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 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")), ) 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_usdc_amount(premium_paid), "tick_sz": tick_sz, "ct_mult": ct_mult, "idx_px": idx_px, "premium_paid": premium_paid, "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": idx_distance_to_be(idx_px, expiry_be), "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, ""