Show hub fullscreen options positions as cards matching instance data.

Enrich hub options snapshot with close preview and render read-only option position cards in exchange fullscreen view.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-11 09:40:35 +08:00
parent baf94b4feb
commit 843d87422d
8 changed files with 292 additions and 23 deletions
+3 -1
View File
@@ -24,10 +24,12 @@ def build_options_hub_snapshot(cfg: dict[str, Any]) -> dict[str, Any]:
if not ok:
return {"ok": False, "enabled": True, "msg": reason or "期权 API 未配置"}
try:
from lib.options.options_positions_lib import build_display_option_positions
raw = cfg["fetch_option_positions"](ex)
if raw is None:
return {"ok": False, "enabled": True, "msg": "获取期权持仓失败"}
positions = [cfg["format_position_row"](p) for p in raw]
positions = build_display_option_positions(cfg, ex, raw)
upl_total = 0.0
has_upl = False
for p in positions:
+82
View File
@@ -0,0 +1,82 @@
"""期权持仓展示(实例页 / 中控快照共用)."""
from __future__ import annotations
from typing import Any
from lib.options.options_db import init_options_tables
from lib.options.options_history_lib import enrich_position_row_display
from lib.options.options_pricing_lib import estimate_close_by_bids
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 []
row["close_preview"] = estimate_close_by_bids(
row["bid_depth"],
target_sheets,
ct_mult=ct_mult,
premium_paid=paid,
)
return row
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 = None
if inst:
rec = conn.execute(
"""
SELECT premium_paid FROM options_trades
WHERE inst_id = ? AND status = 'open'
ORDER BY id DESC LIMIT 1
""",
(inst,),
).fetchone()
if rec and rec["premium_paid"] is not None:
premium_override = float(rec["premium_paid"])
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
+8 -15
View File
@@ -216,22 +216,15 @@ def _attach_close_preview(
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 []
row["close_preview"] = estimate_close_by_bids(
row["bid_depth"],
target_sheets,
ct_mult=ct_mult,
premium_paid=paid,
from lib.options.options_positions_lib import attach_close_preview
return attach_close_preview(
cfg,
ex,
row,
sheets=sheets,
premium_paid=premium_paid,
)
return row
_OPTIONS_SYNC_LOCK = threading.Lock()