dekun
37f787e6a9
Support premium multiplier (coin default 1.05) or net-PnL USDT threshold with index conversion. Co-authored-by: Cursor <cursoragent@cursor.com>
197 lines
7.1 KiB
Python
197 lines
7.1 KiB
Python
"""期权持仓展示(实例页 / 中控快照共用)."""
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from lib.options.options_db import init_options_tables, sum_open_premium_paid
|
||
from lib.options.options_history_lib import enrich_position_row_display
|
||
from lib.options.options_close_gate_lib import clear_close_gate, is_close_gate_passed, update_close_gate
|
||
from lib.options.options_pricing_lib import estimate_close_by_bids, intrinsic_px_per_unit
|
||
|
||
|
||
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 attach_close_preview(
|
||
cfg: dict[str, Any],
|
||
ex: Any,
|
||
row: dict[str, Any],
|
||
*,
|
||
sheets: int | None = None,
|
||
premium_paid: float | None = None,
|
||
) -> dict[str, Any]:
|
||
inst_id = str(row.get("inst_id") or row.get("instId") or "").strip()
|
||
if not inst_id:
|
||
return row
|
||
ct_mult = float(row.get("ct_mult") or 0.01)
|
||
target_sheets = int(sheets) if sheets is not None else int(abs(_safe_float(row.get("pos")) or 0))
|
||
paid = premium_paid if premium_paid is not None else _safe_float(row.get("premium_paid"))
|
||
book = cfg["fetch_option_book_depth"](ex, inst_id, 5)
|
||
row["bid_depth"] = book.get("bids") or []
|
||
row["ask_depth"] = book.get("asks") or []
|
||
mark_px = _safe_float(row.get("mark_px") or row.get("markPx"))
|
||
intrinsic = intrinsic_px_per_unit(
|
||
row.get("opt_type") or row.get("optType"),
|
||
_safe_float(row.get("strike") or row.get("stk")),
|
||
_safe_float(row.get("idx_px") or row.get("idxPx")),
|
||
inst_id=inst_id,
|
||
margin_mode=row.get("margin_mode"),
|
||
)
|
||
# 与实盘一致:只按买一估算本轮可平
|
||
preview = estimate_close_by_bids(
|
||
row["bid_depth"],
|
||
target_sheets,
|
||
ct_mult=ct_mult,
|
||
premium_paid=paid,
|
||
mark_px=mark_px,
|
||
intrinsic_px=intrinsic,
|
||
max_levels=1,
|
||
)
|
||
premium_ccy = str(row.get("premium_ccy") or "USDC").strip().upper() or "USDC"
|
||
index_px = _safe_float(row.get("idx_px") or row.get("idxPx") or row.get("index_px"))
|
||
# 残档时不累计门控;有效买一时刷新计时(仅自动平仓需要)
|
||
if preview.get("bid_invalid") or preview.get("auto_close_blocked"):
|
||
gate = update_close_gate(
|
||
inst_id,
|
||
recycle_usdc=None,
|
||
premium_paid=paid,
|
||
premium_ccy=premium_ccy,
|
||
index_px=index_px,
|
||
)
|
||
preview["close_gate"] = gate
|
||
preview["close_gate_blocked"] = True
|
||
preview["close_gate_msg"] = preview.get("bid_invalid_reason") or gate.get("msg")
|
||
preview["manual_close_blocked"] = True
|
||
preview["liquidity_ok"] = False
|
||
else:
|
||
gate = update_close_gate(
|
||
inst_id,
|
||
recycle_usdc=_safe_float(preview.get("total_received")),
|
||
premium_paid=paid,
|
||
premium_ccy=premium_ccy,
|
||
index_px=index_px,
|
||
)
|
||
passed = bool(gate.get("passed") or is_close_gate_passed(inst_id) or gate.get("ready"))
|
||
preview["close_gate"] = gate
|
||
preview["close_gate_blocked"] = not passed
|
||
preview["close_gate_msg"] = gate.get("msg")
|
||
preview["manual_close_blocked"] = False
|
||
preview["liquidity_ok"] = True
|
||
if not passed:
|
||
preview["auto_close_blocked"] = True
|
||
row["close_preview"] = preview
|
||
return row
|
||
|
||
|
||
def forget_close_gate_for_inst(inst_id: str) -> None:
|
||
clear_close_gate(inst_id)
|
||
|
||
|
||
def net_pnl_from_display_row(row: dict[str, Any]) -> float | None:
|
||
"""与持仓卡「净盈亏」同口径:买一可回收 − 权利金;残档买一则无净值."""
|
||
preview = row.get("close_preview") if isinstance(row.get("close_preview"), dict) else {}
|
||
if preview.get("bid_invalid"):
|
||
return None
|
||
net = preview.get("estimated_pnl")
|
||
if net is not None:
|
||
try:
|
||
return float(net)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
# 仅当实际吃到买盘张数时,才用 total_received − 权利金(避免 bid 无效时 total_received=0 算出 −权利金假亏)
|
||
try:
|
||
covered = float(preview.get("covered_sheets") or 0)
|
||
except (TypeError, ValueError):
|
||
covered = 0.0
|
||
recv = _safe_float(preview.get("total_received"))
|
||
paid = _safe_float(row.get("premium_paid"))
|
||
if covered > 0 and recv is not None and paid is not None:
|
||
return round(recv - paid, 4)
|
||
return None
|
||
|
||
|
||
def display_pnl_from_option_row(row: dict[str, Any]) -> float | None:
|
||
"""展示用盈亏:优先买一净盈亏;残档/无买一时回退交易所标记浮盈 upl."""
|
||
net = net_pnl_from_display_row(row)
|
||
if net is not None:
|
||
return net
|
||
return _safe_float(row.get("upl"))
|
||
|
||
|
||
def display_pnl_usdt_from_option_row(row: dict[str, Any]) -> float | None:
|
||
"""顶栏「实时盈亏」用:币本位净盈亏按指数折 USDT;USDC 仓原样."""
|
||
pnl = display_pnl_from_option_row(row)
|
||
if pnl is None:
|
||
return None
|
||
ccy = str(row.get("premium_ccy") or "USDC").strip().upper() or "USDC"
|
||
if ccy in ("ETH", "BTC"):
|
||
idx = _safe_float(row.get("idx_px") or row.get("idxPx") or row.get("index_px"))
|
||
if idx is None or idx <= 0:
|
||
# 无指数时无法折算,跳过该仓以免把「币数量」当成 U
|
||
return None
|
||
return float(pnl) * float(idx)
|
||
return float(pnl)
|
||
|
||
|
||
def sum_options_net_pnl_usdc(
|
||
cfg: dict[str, Any],
|
||
ex: Any,
|
||
raw_positions: list[dict[str, Any]] | None = None,
|
||
) -> float | None:
|
||
"""
|
||
期权浮盈合计(USDT/USDC 口径),与顶栏实时盈亏对齐:
|
||
各仓买一可回收 − 权利金;币本位再 × 指数折成 U;残档回退该仓 upl(同样折算).
|
||
获取失败返回 None;无持仓返回 0.
|
||
"""
|
||
raw = raw_positions
|
||
if raw is None:
|
||
raw = cfg["fetch_option_positions"](ex)
|
||
if raw is None:
|
||
return None
|
||
if not raw:
|
||
return 0.0
|
||
positions = build_display_option_positions(cfg, ex, raw)
|
||
total = 0.0
|
||
found = False
|
||
for p in positions:
|
||
pnl = display_pnl_usdt_from_option_row(p)
|
||
if pnl is None:
|
||
continue
|
||
found = True
|
||
total += float(pnl)
|
||
return round(total, 4) if found else (0.0 if not positions else None)
|
||
|
||
|
||
def build_display_option_positions(
|
||
cfg: dict[str, Any],
|
||
ex: Any,
|
||
raw_positions: list[dict[str, Any]],
|
||
) -> list[dict[str, Any]]:
|
||
"""与实例 /api/options/positions 相同 enrichment + close_preview."""
|
||
meta_cache: dict[str, dict[str, Any] | None] = {}
|
||
rows: list[dict[str, Any]] = []
|
||
conn = cfg["get_db"]()
|
||
try:
|
||
init_options_tables(conn)
|
||
for p in raw_positions:
|
||
inst = str(p.get("instId") or "").strip()
|
||
premium_override = sum_open_premium_paid(conn, inst) if inst else None
|
||
row = enrich_position_row_display(
|
||
cfg,
|
||
ex,
|
||
p,
|
||
meta_cache=meta_cache,
|
||
premium_override=premium_override,
|
||
)
|
||
attach_close_preview(cfg, ex, row, premium_paid=_safe_float(row.get("premium_paid")))
|
||
rows.append(row)
|
||
finally:
|
||
conn.close()
|
||
return rows
|