dekun
c025c06fac
Reject target and depth closes when bid is a residual tick, and show invalid-bid UI instead of recycling at junk prices. Co-authored-by: Cursor <cursoragent@cursor.com>
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
"""期权持仓展示(实例页 / 中控快照共用)."""
|
|
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, 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")),
|
|
)
|
|
row["close_preview"] = estimate_close_by_bids(
|
|
row["bid_depth"],
|
|
target_sheets,
|
|
ct_mult=ct_mult,
|
|
premium_paid=paid,
|
|
mark_px=mark_px,
|
|
intrinsic_px=intrinsic,
|
|
)
|
|
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
|