"""实例数据看板用的轻量期权持仓(无盘口/余额/历史).""" from __future__ import annotations from typing import Any def fetch_light_option_positions_for_dashboard(cfg: dict[str, Any]) -> list[dict[str, Any]]: """ 仅拉期权持仓 + 本地目标/对冲标注,供看板后台聚合. 不走 build_options_hub_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 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 rows.append(enrich_position_row_display(cfg, ex, p, meta_cache=meta_cache)) 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, ) 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