Update websocket_hub.py

websocket_hub.py

@@ -155,6 +155,24 @@ _METRIC_HISTORY_LEN: int = int(os.environ.get("QUASAR_METRIC_HISTORY", "200"))
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 class ConnectionManager:
+    # All training field names the hub will accept (including common ranker aliases)
+    _TRAINING_KEYS: frozenset = frozenset({
+        "actor_loss", "critic_loss", "avn_loss", "avn_accuracy", "training_steps",
+        "a_loss", "c_loss", "loss_actor", "loss_critic", "loss_avn",
+        "acc", "accuracy", "step", "steps", "n_steps",
+    })
+    _TRAINING_ALIAS: dict = {
+        "a_loss":      "actor_loss",  "loss_actor":  "actor_loss",
+        "c_loss":      "critic_loss", "loss_critic": "critic_loss",
+        "loss_avn":    "avn_loss",
+        "acc":         "avn_accuracy","accuracy":    "avn_accuracy",
+        "step":        "training_steps","steps":     "training_steps","n_steps": "training_steps",
+    }
+    _VOTING_KEYS: frozenset = frozenset({
+        "dominant_signal", "buy_count", "sell_count", "last_price", "signal_source",
+        "signal", "buy", "sell",
+    })
+
     def __init__(self):
         self._publishers:     Dict[str, WebSocket]       = {}
         self._subscribers:    Set[WebSocket]              = set()
@@ -162,6 +180,7 @@ class ConnectionManager:
         self._history:        Dict[str, deque]            = {}   # rolling per-space history
         self._lock            = asyncio.Lock()
         self._total_ingested: int                         = 0
+        self._msg_counts:     Dict[str, Dict[str, int]]  = {}   # {space: {msg_type: count}}
 
     async def register_publisher(self, space_name: str, ws: WebSocket) -> None:
         await ws.accept()
@@ -261,6 +280,14 @@ class ConnectionManager:
     def get_all_snapshots(self) -> dict:
         return dict(self._snapshots)
 
+    def record_msg(self, space_name: str, msg_type: str) -> None:
+        """Increment per-space message type counter (non-blocking, called from publisher loop)."""
+        counts = self._msg_counts.setdefault(space_name, {})
+        counts[msg_type] = counts.get(msg_type, 0) + 1
+
+    def get_msg_counts(self) -> dict:
+        return {s: dict(c) for s, c in self._msg_counts.items()}
+
     def get_metric_history(self) -> dict:
         """Return a plain dict of {space_name: [point, …]} for all spaces with history."""
         return {name: list(dq) for name, dq in self._history.items()}
@@ -483,25 +510,71 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
 
             msg_type = data.get("type", "")
 
+            # ── Track per-space message type counts (for /api/debug/hub) ─────────
+            manager.record_msg(space_name, msg_type)
+
+            # ── Route by type ────────────────────────────────────────────────────
             if msg_type == "metrics":
+                # Combined payload: top-level "training" and "voting" dicts
                 await manager.ingest(space_name, {
                     "training": data.get("training", {}),
                     "voting":   data.get("voting",   {}),
                 })
+
             elif msg_type == "training":
-                await manager.ingest(space_name, {
-                    "training": data.get("data", {}),
-                    "voting":   {},
-                })
+                # Bug A fix: try "data" wrapper first, then fall back to top-level fields.
+                # Some rankers send {"type":"training","data":{...}},
+                # others send {"type":"training","actor_loss":..., ...} directly.
+                training_raw = data.get("data") or {
+                    manager._TRAINING_ALIAS.get(k, k): v
+                    for k, v in data.items()
+                    if k in manager._TRAINING_KEYS and k != "type"
+                }
+                if training_raw:
+                    logger.info(
+                        f"[{space_name}] ⚙  training msg | "
+                        f"keys={list(training_raw.keys())} | "
+                        f"actor_loss={training_raw.get('actor_loss', training_raw.get('a_loss', '—'))}"
+                    )
+                await manager.ingest(space_name, {"training": training_raw, "voting": {}})
+
             elif msg_type == "voting":
-                await manager.ingest(space_name, {
-                    "training": {},
-                    "voting":   data.get("data", {}),
-                })
+                voting_raw = data.get("data") or {
+                    k: v for k, v in data.items()
+                    if k in manager._VOTING_KEYS and k != "type"
+                }
+                await manager.ingest(space_name, {"training": {}, "voting": voting_raw})
+
             elif msg_type in ("heartbeat", "identify", "ping"):
                 pass
+
             else:
-                logger.debug(f"[{space_name}] Unrecognised type '{msg_type}' — dropped")
+                # Bug B fix: don't silently swallow. Try to rescue training/voting
+                # fields that live at the top level of an unrecognised message type.
+                rescued_training = {
+                    manager._TRAINING_ALIAS.get(k, k): v
+                    for k, v in data.items()
+                    if k in manager._TRAINING_KEYS
+                }
+                rescued_voting = {
+                    k: v for k, v in data.items()
+                    if k in manager._VOTING_KEYS
+                }
+                if rescued_training or rescued_voting:
+                    logger.warning(
+                        f"[{space_name}] ⚠  Unknown type='{msg_type}' — "
+                        f"auto-rescued: training_keys={list(rescued_training.keys())} "
+                        f"voting_keys={list(rescued_voting.keys())}"
+                    )
+                    await manager.ingest(space_name, {
+                        "training": rescued_training,
+                        "voting":   rescued_voting,
+                    })
+                else:
+                    logger.warning(
+                        f"[{space_name}] ⚠  Unknown type='{msg_type}' with no "
+                        f"extractable fields — dropped. Full keys: {list(data.keys())}"
+                    )
 
     except WebSocketDisconnect:
         pass
@@ -555,6 +628,46 @@ async def get_health():
     }
 
 
+@app.get("/api/debug/hub")
+async def api_debug_hub():
+    """
+    Diagnostic endpoint — exposes exactly what the hub has received and stored.
+
+    Returns per-space:
+      msg_counts   — how many messages of each type arrived
+      snapshot     — current stored training + voting values
+      history_len  — number of history points recorded
+
+    Use this to confirm whether training messages are arriving and being stored.
+    If msg_counts shows training=0 for a space, the asset space is NOT sending
+    training messages. If training > 0 but snapshot.training shows zeros, there
+    is a field-name or format mismatch.
+    """
+    snapshots   = manager.get_all_snapshots()
+    msg_counts  = manager.get_msg_counts()
+    history_len = {name: len(dq) for name, dq in manager._history.items()}
+
+    spaces = {}
+    for name in set(list(snapshots.keys()) + list(msg_counts.keys())):
+        snap = snapshots.get(name, {})
+        spaces[name] = {
+            "msg_counts":   msg_counts.get(name, {}),
+            "history_len":  history_len.get(name, 0),
+            "training":     snap.get("training", {}),
+            "voting":       snap.get("voting",   {}),
+            "last_updated": snap.get("last_updated", 0),
+            "stale_s":      round(time.time() - snap.get("last_updated", time.time()), 1),
+        }
+
+    return JSONResponse({
+        "spaces":           spaces,
+        "total_ingested":   manager._total_ingested,
+        "publisher_count":  len(manager._publishers),
+        "subscriber_count": len(manager._subscribers),
+        "timestamp":        datetime.utcnow().isoformat() + "Z",
+    })
+
+
 # ══════════════════════════════════════════════════════════════════════════════════════
 # SECTION 7 — TRADE API  (native — replaces patch_websocket_hub.py)
 # ══════════════════════════════════════════════════════════════════════════════════════