diff --git a/lib/hub/hub_bridge_register.py b/lib/hub/hub_bridge_register.py index 32e6dbc..3feaa6d 100644 --- a/lib/hub/hub_bridge_register.py +++ b/lib/hub/hub_bridge_register.py @@ -284,13 +284,12 @@ def register_hub_bridge_routes(app: Flask, app_module: Any) -> None: cfg = hub_ctx.get("options_cfg") if not isinstance(cfg, dict): return jsonify({"ok": False, "msg": "期权未就绪", "version": 1}), 503 - from lib.options.options_db import init_options_tables from lib.options.options_margin_mode_lib import is_coin_margin_mode from lib.options.options_positions_lib import sum_options_net_pnl_usdc from lib.options.options_stats_lib import ( - apply_exchange_pnl_freeze, compute_options_stats_for_window, history_pnl_to_usdt, + load_live_exchange_stats_history, load_local_options_stats_history, normalize_stats_window, ) @@ -303,29 +302,26 @@ def register_hub_bridge_routes(app: Flask, app_module: Any) -> None: raw_live = cfg["fetch_option_positions"](ex) if raw_live is None: return jsonify({"ok": False, "msg": "获取期权持仓失败", "version": 1}) - # 模拟盘本地结算;实盘与期权历史同一交易所盈亏 - conn = cfg["get_db"]() + use_sim = False try: - init_options_tables(conn) + from lib.sim.mode_lib import is_sim_mode + + use_sim = is_sim_mode(cfg["get_db"]) + except Exception: use_sim = False + if use_sim: + conn = cfg["get_db"]() try: - from lib.sim.mode_lib import is_sim_mode - - use_sim = is_sim_mode(cfg["get_db"]) - except Exception: - use_sim = False - if use_sim: history = load_local_options_stats_history(conn, ex) - else: - from lib.options.options_history_lib import load_options_history - - history = apply_exchange_pnl_freeze(conn, load_options_history(ex, cfg), ex) - try: - conn.commit() - except Exception: - pass - finally: - conn.close() + try: + conn.commit() + except Exception: + pass + finally: + conn.close() + else: + # 实盘:先网后短事务,禁止握着 DB 锁拉行情(否则中控整机超时) + history = load_live_exchange_stats_history(cfg, ex) stats = compute_options_stats_for_window(history_pnl_to_usdt(history, ex), window) open_float = sum_options_net_pnl_usdc(cfg, ex, raw_live) net_realized = stats.get("net_realized_pnl") diff --git a/lib/options/options_register.py b/lib/options/options_register.py index c1f0bc8..310b7ec 100644 --- a/lib/options/options_register.py +++ b/lib/options/options_register.py @@ -1285,8 +1285,11 @@ def register_options_routes(app: Flask, cfg: dict[str, Any]) -> None: raw_live = cfg["fetch_option_positions"](ex) if raw_live is None: return jsonify({"ok": False, "msg": "获取期权持仓失败"}) - # 实盘:期权历史交易所原生盈亏;模拟盘:本地结算 - stats = compute_options_stats_auto(cfg, ex) + try: + # 实盘:期权历史交易所原生盈亏;模拟盘:本地结算 + stats = compute_options_stats_auto(cfg, ex) + except Exception as e: + return jsonify({"ok": False, "msg": f"统计失败: {e}"}) open_float = sum_options_net_pnl_usdc(cfg, ex, raw_live) net_realized = _safe_float(stats.get("net_realized_pnl")) or 0.0 total_pnl = None diff --git a/lib/options/options_stats_lib.py b/lib/options/options_stats_lib.py index 13c9991..8445700 100644 --- a/lib/options/options_stats_lib.py +++ b/lib/options/options_stats_lib.py @@ -612,82 +612,87 @@ def load_local_options_stats_history(conn, ex: Any = None) -> list[dict[str, Any return history +def _load_exchange_pnl_freeze_map(conn) -> dict[str, dict[str, Any]]: + init_options_tables(conn) + out: dict[str, dict[str, Any]] = {} + try: + rows = conn.execute( + """ + SELECT history_key, realized_pnl_usdt, close_index_px, realized_pnl + FROM options_exchange_pnl_freeze + """ + ).fetchall() + except Exception: + return out + for row in rows or []: + key = str(_row_get(row, "history_key") or "").strip() + if not key: + continue + out[key] = { + "realized_pnl_usdt": _row_get(row, "realized_pnl_usdt"), + "close_index_px": _row_get(row, "close_index_px"), + "realized_pnl": _row_get(row, "realized_pnl"), + } + return out + + def apply_exchange_pnl_freeze( conn, history: list[dict[str, Any]], ex: Any = None, ) -> list[dict[str, Any]]: - """实盘:以期权历史里的交易所 realized_pnl 为准,首次折 U 后按 history_key 冻结.""" - from lib.options.options_pnl_settle_lib import settle_exchange_realized_pnl - from lib.options.options_time_lib import options_app_now_str + """写冻结表.行内已有 close_index_px / _freeze_index_hint 时不打网. - init_options_tables(conn) - out: list[dict[str, Any]] = [] + 若仍缺指数且传了 ex,会在持连接时补拉(仅兼容旧单测).生产请用 + load_live_exchange_stats_history(先网后短事务). + """ + freeze_map = _load_exchange_pnl_freeze_map(conn) + enriched: list[dict[str, Any]] = [] + need_idx: dict[str, str] = {} for row in history or []: r = dict(row) if r.get("status") == "open": - out.append(r) + enriched.append(r) continue key = str(r.get("history_key") or "").strip() pnl = _safe_float(r.get("realized_pnl")) - if not key or pnl is None: - out.append(r) - continue - frozen = conn.execute( - """ - SELECT realized_pnl_usdt, close_index_px, realized_pnl - FROM options_exchange_pnl_freeze WHERE history_key = ? - """, - (key,), - ).fetchone() - if frozen is not None and _safe_float(_row_get(frozen, "realized_pnl_usdt")) is not None: - r["realized_pnl_usdt"] = _row_get(frozen, "realized_pnl_usdt") - r["close_index_px"] = _row_get(frozen, "close_index_px") - # 币数量仍以当前交易所行为准(防历史修订);U 用冻结值 - out.append(r) - continue - ccy = str(r.get("premium_ccy") or "").strip().upper() - underly = str(r.get("underlying") or "").strip() - inst = str(r.get("inst_id") or "") - if not underly and inst: - underly = inst.split("-")[0] + if ( + key + and pnl is not None + and not ( + key in freeze_map + and _safe_float(freeze_map[key].get("realized_pnl_usdt")) is not None + ) + and _safe_float(r.get("close_index_px")) is None + and _safe_float(r.get("_freeze_index_hint")) is None + ): + underly = str(r.get("underlying") or "").strip().upper() + inst = str(r.get("inst_id") or "") + if not underly and inst: + underly = inst.split("-")[0].strip().upper() + if underly: + need_idx[underly] = inst + enriched.append(r) + + if need_idx and ex is not None: from lib.options.options_pnl_settle_lib import resolve_close_index_px - idx = resolve_close_index_px(ex, inst_id=inst, underlying=underly) - settled = settle_exchange_realized_pnl( - realized_pnl=pnl, - premium_ccy=ccy or "ETH", - index_px=idx, - ) - usdt = settled.get("realized_pnl_usdt") - close_idx = settled.get("close_index_px") - if usdt is not None: - conn.execute( - """ - INSERT INTO options_exchange_pnl_freeze( - history_key, realized_pnl, realized_pnl_usdt, close_index_px, - premium_ccy, updated_at - ) VALUES (?, ?, ?, ?, ?, ?) - ON CONFLICT(history_key) DO UPDATE SET - realized_pnl = excluded.realized_pnl, - realized_pnl_usdt = excluded.realized_pnl_usdt, - close_index_px = excluded.close_index_px, - premium_ccy = excluded.premium_ccy, - updated_at = excluded.updated_at - """, - ( - key, - float(pnl), - float(usdt), - close_idx, - ccy or "", - options_app_now_str(), - ), + idx_cache: dict[str, float | None] = {} + for underly, inst in need_idx.items(): + idx_cache[underly] = resolve_close_index_px( + ex, inst_id=inst, underlying=underly ) - r["realized_pnl_usdt"] = float(usdt) - r["close_index_px"] = close_idx - out.append(r) - return out + for r in enriched: + if r.get("status") == "open": + continue + underly = str(r.get("underlying") or "").strip().upper() + inst = str(r.get("inst_id") or "") + if not underly and inst: + underly = inst.split("-")[0].strip().upper() + if underly in idx_cache and _safe_float(r.get("close_index_px")) is None: + r["_freeze_index_hint"] = idx_cache.get(underly) + + return _apply_exchange_pnl_freeze_with_hints(conn, enriched, freeze_map) def compute_options_stats(get_db, ex: Any = None) -> dict[str, Any]: @@ -704,23 +709,151 @@ def compute_options_stats(get_db, ex: Any = None) -> dict[str, Any]: conn.close() -def compute_options_stats_live_exchange(cfg: dict[str, Any], ex: Any) -> dict[str, Any]: - """实盘:与「期权历史」同一数据源(交易所 positions-history),盈亏以交易所为准.""" - from lib.options.options_history_lib import load_options_history +def load_live_exchange_stats_history(cfg: dict[str, Any], ex: Any) -> list[dict[str, Any]]: + """实盘统计用历史:先拉交易所,再无连接时取指数,最后短事务写冻结. - history = load_options_history(ex, cfg) - conn = cfg["get_db"]() + 禁止在持有 DB 连接时打行情/历史网,否则会拖死持仓与中控轮询. + """ + from lib.options.options_history_lib import load_options_history + from lib.options.options_pnl_settle_lib import resolve_close_index_px + + history = load_options_history(ex, cfg) or [] + get_db = cfg["get_db"] + conn = get_db() try: - history = apply_exchange_pnl_freeze(conn, history, ex) + freeze_map = _load_exchange_pnl_freeze_map(conn) try: conn.commit() except Exception: pass finally: conn.close() + + need_idx: dict[str, str] = {} + for row in history: + if row.get("status") == "open": + continue + key = str(row.get("history_key") or "").strip() + if not key or _safe_float(row.get("realized_pnl")) is None: + continue + if key in freeze_map and _safe_float(freeze_map[key].get("realized_pnl_usdt")) is not None: + continue + underly = str(row.get("underlying") or "").strip().upper() + inst = str(row.get("inst_id") or "") + if not underly and inst: + underly = inst.split("-")[0].strip().upper() + if underly and _safe_float(row.get("close_index_px")) is None: + need_idx[underly] = inst + + idx_cache: dict[str, float | None] = {} + for underly, inst in need_idx.items(): + idx_cache[underly] = resolve_close_index_px(ex, inst_id=inst, underlying=underly) + + enriched: list[dict[str, Any]] = [] + for row in history: + r = dict(row) + if r.get("status") != "open": + underly = str(r.get("underlying") or "").strip().upper() + inst = str(r.get("inst_id") or "") + if not underly and inst: + underly = inst.split("-")[0].strip().upper() + key = str(r.get("history_key") or "").strip() + if underly in idx_cache and _safe_float(r.get("close_index_px")) is None: + if not ( + key in freeze_map + and _safe_float(freeze_map[key].get("realized_pnl_usdt")) is not None + ): + r["_freeze_index_hint"] = idx_cache.get(underly) + enriched.append(r) + + conn = get_db() + try: + out = _apply_exchange_pnl_freeze_with_hints(conn, enriched, freeze_map) + try: + conn.commit() + except Exception: + pass + return out + finally: + conn.close() + + +def compute_options_stats_live_exchange(cfg: dict[str, Any], ex: Any) -> dict[str, Any]: + """实盘:与「期权历史」同一数据源(交易所 positions-history),盈亏以交易所为准.""" + history = load_live_exchange_stats_history(cfg, ex) return compute_options_stats_from_history(history_pnl_to_usdt(history, ex)) +def _apply_exchange_pnl_freeze_with_hints( + conn, + history: list[dict[str, Any]], + freeze_map: dict[str, dict[str, Any]] | None = None, +) -> list[dict[str, Any]]: + """短事务写冻结;指数只用来自行内 _freeze_index_hint / 已有 close_index_px.""" + from lib.options.options_pnl_settle_lib import settle_exchange_realized_pnl + from lib.options.options_time_lib import options_app_now_str + + init_options_tables(conn) + fmap = freeze_map if freeze_map is not None else _load_exchange_pnl_freeze_map(conn) + out: list[dict[str, Any]] = [] + for row in history or []: + r = dict(row) + hint = r.pop("_freeze_index_hint", None) + if r.get("status") == "open": + out.append(r) + continue + key = str(r.get("history_key") or "").strip() + pnl = _safe_float(r.get("realized_pnl")) + if not key or pnl is None: + out.append(r) + continue + frozen = fmap.get(key) + if frozen is not None and _safe_float(frozen.get("realized_pnl_usdt")) is not None: + r["realized_pnl_usdt"] = frozen.get("realized_pnl_usdt") + r["close_index_px"] = frozen.get("close_index_px") + out.append(r) + continue + ccy = str(r.get("premium_ccy") or "").strip().upper() + idx = _safe_float(r.get("close_index_px")) or _safe_float(hint) + settled = settle_exchange_realized_pnl( + realized_pnl=pnl, + premium_ccy=ccy or "ETH", + index_px=idx, + ) + usdt = settled.get("realized_pnl_usdt") + close_idx = settled.get("close_index_px") + if usdt is not None: + try: + conn.execute( + """ + INSERT INTO options_exchange_pnl_freeze( + history_key, realized_pnl, realized_pnl_usdt, close_index_px, + premium_ccy, updated_at + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(history_key) DO UPDATE SET + realized_pnl = excluded.realized_pnl, + realized_pnl_usdt = excluded.realized_pnl_usdt, + close_index_px = excluded.close_index_px, + premium_ccy = excluded.premium_ccy, + updated_at = excluded.updated_at + """, + ( + key, + float(pnl), + float(usdt), + close_idx, + ccy or "", + options_app_now_str(), + ), + ) + except Exception: + pass + r["realized_pnl_usdt"] = float(usdt) + r["close_index_px"] = close_idx + out.append(r) + return out + + def compute_options_stats_auto(cfg: dict[str, Any], ex: Any = None) -> dict[str, Any]: """模拟盘走本地结算;实盘走交易所期权历史原生盈亏.""" get_db = cfg.get("get_db")