"""实例数据看板用的轻量期权持仓(无余额/历史;含 close_preview 供净盈亏).""" from __future__ import annotations from typing import Any def fetch_light_option_positions_for_dashboard(cfg: dict[str, Any]) -> list[dict[str, Any]]: """ 拉期权持仓 + 本地目标/对冲标注 + 买一净盈亏预览,供看板后台聚合. 不走 options dashboard snapshot(避免余额/历史). """ if not cfg.get("enabled"): return [] ex = cfg.get("exchange_options") ready_fn = cfg.get("options_api_ready") if not callable(ready_fn): return [] ok, _reason = ready_fn(ex) if not ok: return [] fetch_fn = cfg.get("fetch_option_positions") if not callable(fetch_fn): return [] raw = fetch_fn(ex) if raw is None: return [] if not raw: return [] 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_positions_lib import attach_close_preview meta_cache: dict[str, dict[str, Any] | None] = {} rows: list[dict[str, Any]] = [] get_db = cfg.get("get_db") if not callable(get_db): for p in raw: if not isinstance(p, dict): continue row = enrich_position_row_display(cfg, ex, p, meta_cache=meta_cache) attach_close_preview(cfg, ex, row) rows.append(row) return rows conn = get_db() try: init_options_tables(conn) try: from lib.hedge_plan.hedge_plan_db import active_options_targets_by_inst from lib.options.options_target_lib import targets_by_inst tgt_map = targets_by_inst(conn) hedge_target_map = active_options_targets_by_inst(conn) except Exception: tgt_map = {} hedge_target_map = {} for p in raw: if not isinstance(p, dict): continue 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=premium_override) mon = tgt_map.get(str(row.get("inst_id") or "")) if mon: row["target_index"] = mon.get("target_index") row["target_monitor_id"] = mon.get("id") row["target_monitor"] = mon hedge_target = hedge_target_map.get(str(row.get("inst_id") or "")) if hedge_target: row["hedge_plan_target"] = hedge_target if not mon: row["target_index"] = hedge_target.get("target_index") rows.append(row) finally: conn.close() return rows