"""永期对冲 · 当前持仓:期权+永续实时卡(组合净利扣开平双边手续费).""" from __future__ import annotations from typing import Any, Optional from lib.hedge_plan.hedge_plan_option_primary_lib import ( estimate_combo_net_pnl, fee_rate, opt_type_for_view, perp_direction_for_view, ) def _sf(v: Any) -> Optional[float]: try: if v is None or v == "": return None return float(v) except (TypeError, ValueError): return None def _pick_option_row(rows: list[dict[str, Any]] | None, inst_id: str) -> Optional[dict[str, Any]]: want = str(inst_id or "").strip() if not want or not rows: return None for r in rows: if not isinstance(r, dict): continue rid = str(r.get("instId") or r.get("inst_id") or "").strip() if rid == want: return r return None def _option_bid(ex: Any, inst_id: str, fallback: Optional[float]) -> Optional[float]: try: from lib.exchange.okx_options_lib import fetch_option_book_depth book = fetch_option_book_depth(ex, inst_id) or {} bids = book.get("bids") or [] if bids and isinstance(bids[0], dict): bid = _sf(bids[0].get("px") or bids[0].get("price")) if bid is not None and bid > 0: return bid except Exception: pass return fallback def build_live_hedge_cards(cfg: dict[str, Any], conn: Any) -> list[dict[str, Any]]: """进行中的永期计划 → 期权/永续腿实时卡 + 扣费组合净利.""" from lib.hedge_plan.hedge_plan_db import attach_legs_to_plans, list_plans rows: list[dict[str, Any]] = [] for status in ("opening", "active", "partial"): rows.extend(list_plans(conn, status=status, limit=40)) rows = [r for r in rows if str(r.get("plan_type") or "") == "perp_options"] rows.sort(key=lambda row: int(row.get("id") or 0), reverse=True) if not rows: return [] plans = attach_legs_to_plans(conn, rows) fetch_opt = cfg.get("fetch_option_positions") ex_opt = None try: get_ex = cfg.get("exchange_options") ex_opt = get_ex() if callable(get_ex) else get_ex except Exception: ex_opt = None opt_rows: list[dict[str, Any]] = [] if callable(fetch_opt) and ex_opt is not None: try: opt_rows = fetch_opt(ex_opt) or [] except Exception: opt_rows = [] get_perp_metrics = cfg.get("get_live_position_exchange_metrics") get_cs = cfg.get("get_contract_size") fetch_idx = cfg.get("fetch_index_price") fr = fee_rate() out: list[dict[str, Any]] = [] for plan in plans: legs = plan.get("legs") or [] opt_leg = next((x for x in legs if str(x.get("leg_role") or "") in ("option_hedge", "option_a")), None) perp_leg = next((x for x in legs if str(x.get("leg_role") or "") == "perp"), None) if not opt_leg and not perp_leg: continue view = str(plan.get("direction") or "long").lower() uly = str(plan.get("underlying") or "ETH").upper() inst_id = str((opt_leg or {}).get("inst_id") or "") opt_raw = _pick_option_row(opt_rows, inst_id) if inst_id else None sheets = _sf((opt_leg or {}).get("size")) or _sf((opt_raw or {}).get("pos")) or 0.0 ct = _sf((opt_leg or {}).get("ct_mult")) or 0.01 avg_open = _sf((opt_leg or {}).get("avg_open")) or _sf((opt_raw or {}).get("avgPx")) or 0.0 mark_opt = _sf((opt_raw or {}).get("markPx")) or _sf((opt_raw or {}).get("last")) or avg_open bid = _option_bid(ex_opt, inst_id, mark_opt) if inst_id else mark_opt strike = _sf((opt_leg or {}).get("strike")) or 0.0 opt_type = str((opt_leg or {}).get("opt_type") or opt_type_for_view(view)).upper() sym = str((perp_leg or {}).get("symbol") or plan.get("exchange_symbol") or "") perp_dir = str((perp_leg or {}).get("side") or plan.get("perp_direction") or perp_direction_for_view(view)).lower() contracts = _sf((perp_leg or {}).get("size")) or _sf(plan.get("perp_size")) or 0.0 perp_entry = _sf((perp_leg or {}).get("avg_open")) or _sf(plan.get("entry_mark")) or 0.0 cs = 0.01 if callable(get_cs) and sym: try: cs = float(get_cs(sym) or 0.01) except Exception: cs = 0.01 perp_mark = None perp_upl = None live_contracts = None if callable(get_perp_metrics) and sym: try: m = get_perp_metrics(sym, perp_dir, order_leverage=plan.get("leverage")) if isinstance(m, dict): perp_mark = _sf(m.get("mark_price") or m.get("mark")) perp_upl = _sf(m.get("unrealized_pnl")) live_contracts = _sf(m.get("contracts") or m.get("size")) except Exception: pass if perp_mark is None or perp_mark <= 0: perp_mark = perp_entry if live_contracts and live_contracts > 0: contracts = live_contracts idx = None if callable(fetch_idx): try: idx = _sf(fetch_idx(uly)) except Exception: idx = None if idx is None or idx <= 0: idx = perp_mark or strike or 0.0 net = estimate_combo_net_pnl( view_side=view, strike=float(strike or 0), index_px=float(idx or 0), ask_open=float(avg_open or 0), bid=float(bid or 0), sheets=float(sheets or 0), ct_mult=float(ct or 0.01), perp_direction=perp_dir, perp_entry=float(perp_entry or 0), perp_mark=float(perp_mark or 0), contracts=float(contracts or 0), contract_size=float(cs or 0.01), fee=fr, ) # 币本位期权净利(币) → USDT;永续净利已是 U opt_net_coin = float(net.get("opt_net") or 0) opt_net_usdt = round(opt_net_coin * float(idx or 0), 4) if idx else None perp_net = float(net.get("perp_net") or 0) combo_usdt = round((opt_net_usdt or 0) + perp_net, 4) if opt_net_usdt is not None else round(perp_net, 4) out.append( { "plan_id": plan.get("id"), "status": plan.get("status"), "underlying": uly, "direction": view, "opened_at": plan.get("opened_at"), "index_px": idx, "fee_rate": fr, "option": { "inst_id": inst_id or "—", "opt_type": opt_type, "strike": strike, "sheets": sheets, "ct_mult": ct, "avg_open": avg_open, "mark_px": mark_opt, "bid": bid, "net_coin": opt_net_coin, "net_usdt": opt_net_usdt, "live": bool(opt_raw), }, "perp": { "symbol": sym or "—", "side": perp_dir, "contracts": contracts, "contract_size": cs, "avg_open": perp_entry, "mark_px": perp_mark, "upl": perp_upl, "net_usdt": perp_net, "live": perp_upl is not None or (perp_mark is not None and perp_mark > 0), }, "combo": { "net_usdt": combo_usdt, "opt_net_usdt": opt_net_usdt, "perp_net_usdt": perp_net, "note": "扣开平双边手续费后估", }, } ) return out