dekun
94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
"""期权历史列表(交易所 positions-history + 当前持仓)."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from lib.options.options_db import init_options_tables, sum_open_premium_paid
|
|
|
|
|
|
def enrich_position_row_display(
|
|
cfg: dict[str, Any],
|
|
ex: Any,
|
|
raw_pos: dict[str, Any],
|
|
*,
|
|
meta_cache: dict[str, dict[str, Any] | None] | None = None,
|
|
premium_override: float | None = None,
|
|
) -> dict[str, Any]:
|
|
from lib.exchange.okx_options_lib import format_position_row, format_premium_amount, tick_sz_and_ct_mult
|
|
from lib.options.options_margin_mode_lib import margin_mode_from_inst_id, premium_ccy_for_mode
|
|
|
|
inst_id = str(raw_pos.get("instId") or "").strip()
|
|
tick_sz, ct_mult = tick_sz_and_ct_mult(ex, inst_id, meta_cache)
|
|
row = format_position_row(raw_pos, ct_mult=ct_mult, tick_sz=tick_sz)
|
|
row_mode = margin_mode_from_inst_id(inst_id) if inst_id else "usdc"
|
|
underly = str(row.get("underlying") or (inst_id.split("-")[0] if inst_id else "ETH") or "ETH")
|
|
premium_ccy = premium_ccy_for_mode(row_mode, underly)
|
|
row["margin_mode"] = row_mode
|
|
row["premium_ccy"] = premium_ccy
|
|
row["margin_mode_label"] = "币本位" if row_mode == "coin" else "USDC"
|
|
if premium_override is not None:
|
|
row["premium_paid"] = premium_override
|
|
row["premium_paid_fmt"] = format_premium_amount(row.get("premium_paid"), ccy=premium_ccy)
|
|
return row
|
|
|
|
|
|
def load_options_history(ex: Any, cfg: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""与期权历史页相同的数据源:交易所全平记录 + 当前持仓,排除本地隐藏项."""
|
|
from lib.exchange.okx_options_lib import (
|
|
fetch_all_option_positions_history,
|
|
format_live_option_history_row,
|
|
format_option_history_row,
|
|
tick_sz_and_ct_mult,
|
|
)
|
|
|
|
meta_cache: dict[str, dict[str, Any] | None] = {}
|
|
items: list[dict[str, Any]] = []
|
|
|
|
raw_live = cfg["fetch_option_positions"](ex)
|
|
if raw_live is None:
|
|
return []
|
|
|
|
conn = cfg["get_db"]()
|
|
try:
|
|
init_options_tables(conn)
|
|
hidden_keys = {
|
|
str(r["history_key"])
|
|
for r in conn.execute("SELECT history_key FROM options_history_hidden").fetchall()
|
|
}
|
|
for p in raw_live:
|
|
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,
|
|
)
|
|
open_ms = None
|
|
ctime = p.get("cTime") or (row.get("raw") or {}).get("cTime")
|
|
try:
|
|
if ctime is not None and str(ctime).strip():
|
|
open_ms = int(float(ctime))
|
|
except (TypeError, ValueError):
|
|
open_ms = None
|
|
items.append(format_live_option_history_row(row, open_ms=open_ms))
|
|
finally:
|
|
conn.close()
|
|
|
|
hist_raw = fetch_all_option_positions_history(ex, limit=200)
|
|
for raw in hist_raw:
|
|
inst_id = str(raw.get("instId") or "").strip()
|
|
tick_sz, ct_mult = tick_sz_and_ct_mult(ex, inst_id, meta_cache)
|
|
items.append(format_option_history_row(raw, tick_sz=tick_sz, ct_mult=ct_mult))
|
|
|
|
open_rows = [x for x in items if x.get("status") == "open"]
|
|
closed = [x for x in items if x.get("status") != "open"]
|
|
closed.sort(key=lambda x: int(x.get("close_ms") or 0), reverse=True)
|
|
open_rows.sort(key=lambda x: int(x.get("close_ms") or 0), reverse=True)
|
|
return [
|
|
x
|
|
for x in (open_rows + closed)
|
|
if str(x.get("history_key") or "") not in hidden_keys
|
|
]
|