Update websocket_hub.py

websocket_hub.py

@@ -462,6 +462,41 @@ _axrvi_rankings:    List[dict] = []
 _axrvi_rankings_ts: float      = 0.0
 _AXRVI_RANKINGS_TTL: float     = 30.0   # seconds before falling back to snapshot scoring
 
+# ── Top-3 WebSocket client registry ───────────────────────────────────────────────────
+# top3_client.py connects here and receives top3_rankings broadcasts whenever the
+# Executo ranker POSTs new rankings via POST /api/axrvi/rankings.
+_top3_clients: Set[WebSocket] = set()
+_top3_clients_lock = asyncio.Lock()
+
+
+async def _broadcast_top3_rankings(rankings: List[dict]) -> None:
+    """
+    Broadcast a top3_rankings message to all /ws/top3 subscribers.
+    Called immediately after /api/axrvi/rankings receives a fresh ranking list.
+    Dead connections are pruned automatically.
+    """
+    if not _top3_clients:
+        return
+    msg = json.dumps({
+        "type":          "top3_rankings",
+        "rankings":      rankings,
+        "total_assets":  len(rankings),
+        "hub_timestamp": time.time(),
+    })
+    dead: list = []
+    async with _top3_clients_lock:
+        clients = list(_top3_clients)
+    for ws in clients:
+        try:
+            await ws.send_text(msg)
+        except Exception:
+            dead.append(ws)
+    if dead:
+        async with _top3_clients_lock:
+            for ws in dead:
+                _top3_clients.discard(ws)
+        logger.debug(f"[top3] Pruned {len(dead)} dead client(s)")
+
 
 # ══════════════════════════════════════════════════════════════════════════════════════
 # SECTION 4 — FASTAPI APPLICATION
@@ -605,6 +640,49 @@ async def ws_subscriber_endpoint(websocket: WebSocket):
         await manager.unregister_subscriber(websocket)
 
 
+@app.websocket("/ws/top3")
+async def ws_top3_endpoint(websocket: WebSocket):
+    """
+    /ws/top3 — consumed by top3_client.py (MT5 bridge).
+
+    Sends a top3_rankings message immediately on connect (replay of the latest
+    known ranking so the client does not have to wait for the next ranker cycle),
+    then keeps the socket open to receive subsequent broadcasts triggered by
+    POST /api/axrvi/rankings.
+
+    Message format:
+      {"type": "top3_rankings", "rankings": [...], "total_assets": N, "hub_timestamp": T}
+    """
+    await websocket.accept()
+    async with _top3_clients_lock:
+        _top3_clients.add(websocket)
+    logger.info(f"📈 top3 client connected  (total={len(_top3_clients)})")
+
+    # ── Replay latest rankings immediately so client doesn't wait up to 5 s ───
+    if _axrvi_rankings:
+        try:
+            await websocket.send_text(json.dumps({
+                "type":          "top3_rankings",
+                "rankings":      _axrvi_rankings,
+                "total_assets":  len(_axrvi_rankings),
+                "hub_timestamp": _axrvi_rankings_ts,
+            }))
+        except Exception:
+            pass
+
+    try:
+        while True:
+            await websocket.receive_text()   # keep-alive — client sends nothing
+    except WebSocketDisconnect:
+        pass
+    except Exception as e:
+        logger.error(f"[top3] Client error: {e}")
+    finally:
+        async with _top3_clients_lock:
+            _top3_clients.discard(websocket)
+        logger.info(f"📉 top3 client disconnected  (remaining={len(_top3_clients)})")
+
+
 # ══════════════════════════════════════════════════════════════════════════════════════
 # SECTION 6 — REST API (READ-ONLY)
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -704,7 +782,7 @@ async def api_trades_closed(limit: int = 50):
 @app.get("/api/health")
 async def api_health():
     """Service health — includes live trade counts and log-file inventory."""
-    state = _trade_parser.get_state()
+    state = _hub_trades.get_state()
     return JSONResponse({
         "service":      "websocket_hub",
         "version":      "v2.2-ranker-logs",
@@ -1128,6 +1206,9 @@ async def receive_axrvi_rankings(request: Request):
         f"top={rankings[0].get('space_name','?')} score={rankings[0].get('score',0):.4f}"
         if rankings else "[AXRVI Rankings] Received empty list"
     )
+    # Broadcast to all connected top3_client.py instances immediately
+    if rankings:
+        asyncio.create_task(_broadcast_top3_rankings(rankings))
     return JSONResponse({"ok": True, "count": len(rankings), "ts": _axrvi_rankings_ts})