fix(hub): merge mark price from Flask snapshot and fix board refresh

Sync hub positions with instance price_snapshot (order_prices and position_marks).
Fix monitor board UI when hub restarts (version rewind) and queue snapshot fetches.
Expose board aggregate status on /api/ping for diagnostics.

Co-authored-by: Cursor <cursoragent@cursor.com>

manual_trading_hub/hub.py

@@ -16,7 +16,6 @@ if str(_REPO_ROOT) not in sys.path:
 
 from hub_kline_store import format_ohlcv_detail, resolve_chart_bars, retention_days
 from hub_ohlcv_lib import CHART_TIMEFRAME_ORDER, CHART_TIMEFRAMES, bar_limit_for_timeframe
-
 from env_load import load_hub_dotenv
 
 load_hub_dotenv()
@@ -705,6 +704,106 @@ def _merge_flask_position_breakeven(agent_row: dict, snap: dict | None, hub_mon:
             p["sl_breakeven_secured"] = bool(matched["sl_breakeven_secured"])
 
 
+def _agent_position_has_mark(p: dict) -> bool:
+    try:
+        v = float(p.get("mark_price"))
+        return v > 0
+    except (TypeError, ValueError):
+        return False
+
+
+def _apply_agent_mark_price(p: dict, mark_price: object, mark_display: object = None) -> None:
+    try:
+        mpf = float(mark_price)
+    except (TypeError, ValueError):
+        return
+    if mpf <= 0:
+        return
+    p["mark_price"] = mpf
+    disp = mark_display
+    if disp is not None and str(disp).strip() not in ("", "-"):
+        p["mark_price_fmt"] = str(disp)
+
+
+def _find_matched_order_price_op(
+    p: dict,
+    order_prices: list,
+    hub_orders: list,
+    op_by_id: dict,
+) -> dict | None:
+    sym = p.get("symbol") or ""
+    side = (p.get("side") or "").lower()
+    for o in hub_orders:
+        if not isinstance(o, dict):
+            continue
+        o_sym = o.get("exchange_symbol") or o.get("symbol") or ""
+        if not _symbols_match(sym, o_sym):
+            continue
+        if (o.get("direction") or "").lower() != side:
+            continue
+        matched = op_by_id.get(o.get("id"))
+        if isinstance(matched, dict):
+            return matched
+        break
+    for op in order_prices:
+        if not isinstance(op, dict):
+            continue
+        if not _symbols_match(sym, op.get("symbol") or ""):
+            continue
+        return op
+    return None
+
+
+def _merge_flask_position_mark_price(
+    agent_row: dict, snap: dict | None, hub_mon: dict | None
+) -> None:
+    """子代理无标记价时，用实例 price_snapshot 的交易所标记价补全中控持仓展示。"""
+    ag = agent_row.get("agent")
+    if not isinstance(ag, dict) or not isinstance(snap, dict):
+        return
+    positions = ag.get("positions")
+    if not isinstance(positions, list) or not positions:
+        return
+    order_prices = snap.get("order_prices") or []
+    hub_orders = []
+    if isinstance(hub_mon, dict):
+        hub_orders = hub_mon.get("orders") or []
+    op_by_id = {
+        op.get("id"): op
+        for op in order_prices
+        if isinstance(op, dict) and op.get("id") is not None
+    }
+    for p in positions:
+        if not isinstance(p, dict) or _agent_position_has_mark(p):
+            continue
+        matched = _find_matched_order_price_op(p, order_prices, hub_orders, op_by_id)
+        if isinstance(matched, dict):
+            _apply_agent_mark_price(
+                p,
+                matched.get("exchange_mark_price"),
+                matched.get("exchange_mark_price_display"),
+            )
+    position_marks = snap.get("position_marks") or []
+    if not isinstance(position_marks, list):
+        return
+    for p in positions:
+        if not isinstance(p, dict) or _agent_position_has_mark(p):
+            continue
+        sym = p.get("symbol") or ""
+        side = (p.get("side") or "").lower()
+        for pm in position_marks:
+            if not isinstance(pm, dict):
+                continue
+            if not _symbols_match(sym, pm.get("symbol") or ""):
+                continue
+            if (pm.get("side") or "").lower() != side:
+                continue
+            _apply_agent_mark_price(
+                p, pm.get("mark_price"), pm.get("mark_price_display")
+            )
+            break
+
+
 def _merge_flask_exchange_tpsl(agent_row: dict, snap: dict | None, hub_mon: dict | None) -> None:
     """子代理挂单为空时，用实例 Flask 已算好的 exchange_tpsl 补全展示。"""
     ag = agent_row.get("agent")
@@ -764,6 +863,7 @@ async def _assemble_board_row(
         _merge_flask_order_price_fields(hub_mon, snap)
     _merge_flask_exchange_tpsl(agent_row, snap, hub_mon if isinstance(hub_mon, dict) else None)
     _merge_flask_position_breakeven(agent_row, snap, hub_mon if isinstance(hub_mon, dict) else None)
+    _merge_flask_position_mark_price(agent_row, snap, hub_mon if isinstance(hub_mon, dict) else None)
     flask_ok = isinstance(hub_mon, dict) and hub_mon.get("ok") is not False
     raw_review = (ex.get("review_url") or "").strip()
     review_link = browser_url(raw_review) if raw_review else default_review_url(
@@ -1088,6 +1188,11 @@ def api_ping():
         "features": ["monitor", "settings", "auth", "board_sse"],
         "board_poll_interval_sec": HUB_BOARD_POLL_INTERVAL,
         "board_version": board_store.version,
+        "board_aggregating": board_store.aggregating,
+        "board_updated_at": (board_store.payload or {}).get("updated_at")
+        if isinstance(board_store.payload, dict)
+        else None,
+        "board_error": board_store.last_error,
         "password_required": password_required(),
         "env_disabled_ids": sorted(env_force_disabled_ids()),
         "hub_disabled_ids_raw": (os.getenv("HUB_DISABLED_IDS") or ""),