Update hub_dashboard_service.py

hub_dashboard_service.py

@@ -280,14 +280,75 @@ class DashboardState:
         
         self._rankings: List[dict] = []
         self._snapshots: Dict[str, dict] = {}
+        self._history:   Dict[str, deque] = {}   # rolling per-space metric history
     
+    def update_from_snapshot(self, space_name: str, snapshot: dict) -> None:
+        """Record a metrics_update frame from the hub WebSocket."""
+        with self._lock:
+            self._snapshots[space_name] = snapshot
+            training = snapshot.get("training", {})
+
+            # Only append a history point when training has non-zero loss/accuracy
+            if any(
+                training.get(k, 0) != 0
+                for k in ("actor_loss", "critic_loss", "avn_loss", "avn_accuracy")
+            ):
+                if space_name not in self._history:
+                    self._history[space_name] = deque(maxlen=self._history_len)
+                self._history[space_name].append({
+                    "ts":             snapshot.get("last_updated", time.time()),
+                    "actor_loss":     training.get("actor_loss",    0.0),
+                    "critic_loss":    training.get("critic_loss",   0.0),
+                    "avn_loss":       training.get("avn_loss",      0.0),
+                    "avn_accuracy":   training.get("avn_accuracy",  0.0),
+                    "training_steps": training.get("training_steps", 0),
+                })
+
+            self.messages_rx += 1
+            self.last_update_ts = time.time()
+            self.hub_connected = True
+
+            # Recompute rankings from the latest snapshot collection
+            self._rankings = self._compute_rankings()
+
+    def _compute_rankings(self) -> List[dict]:
+        ranked: List[dict] = []
+        for name, snap in self._snapshots.items():
+            training = snap.get("training", {})
+            voting   = snap.get("voting",   {})
+            buy      = voting.get("buy_count",  0)
+            sell     = voting.get("sell_count", 0)
+            total    = buy + sell
+            sig_conf = (max(buy, sell) / total) if total > 0 else 0.0
+            avn_acc  = training.get("avn_accuracy", 0.0)
+            score    = round(sig_conf - avn_acc, 6)
+            ranked.append({
+                "rank":              0,
+                "space_name":        name,
+                "score":             score,
+                "signal_confidence": round(sig_conf, 6),
+                "avn_accuracy":      round(avn_acc,  6),
+                "dominant_signal":   voting.get("dominant_signal", "NEUTRAL"),
+                "buy_count":         buy,
+                "sell_count":        sell,
+                "training_steps":    training.get("training_steps", 0),
+                "actor_loss":        training.get("actor_loss",     0.0),
+                "critic_loss":       training.get("critic_loss",    0.0),
+                "avn_loss":          training.get("avn_loss",       0.0),
+                "last_updated":      snap.get("last_updated",       0.0),
+            })
+        ranked.sort(key=lambda r: r["score"], reverse=True)
+        for i, r in enumerate(ranked):
+            r["rank"] = i + 1
+        return ranked
+
     def get_state(self) -> dict:
         with self._lock:
             return {
-                "rankings": list(self._rankings),
-                "metric_history": {},
-                "health": self._health_snapshot(),
-                "timestamp": datetime.utcnow().isoformat() + "Z",
+                "rankings":       list(self._rankings),
+                "metric_history": {name: list(dq) for name, dq in self._history.items()},
+                "health":         self._health_snapshot(),
+                "timestamp":      datetime.utcnow().isoformat() + "Z",
             }
     
     def _health_snapshot(self) -> dict:
@@ -302,6 +363,109 @@ class DashboardState:
         }
 
 
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 2B — HUB SUBSCRIBER CLIENT
+#
+# Connects to websocket_hub.py via /ws/subscribe and feeds incoming
+# metrics_update / initial_state frames into DashboardState.
+# Runs in a daemon thread with exponential-back-off reconnection.
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+_HUB_WS_URL = os.environ.get(
+    "QUASAR_HUB_WS",
+    "ws://127.0.0.1:7860/ws/subscribe",   # local default; override via env var
+)
+
+class HubSubscriberClient:
+    """
+    READ-ONLY WebSocket subscriber that keeps DashboardState in sync with
+    whatever the hub knows, including live training metrics.
+
+    Messages handled:
+      • initial_state — {snapshots: {space: snapshot}}
+      • metrics_update — {space_name: str, snapshot: dict}
+    """
+
+    _MAX_BACKOFF = 30
+
+    def __init__(self, state: "DashboardState", hub_url: str = _HUB_WS_URL):
+        self.state    = state
+        self.hub_url  = hub_url
+        self._ws:     Optional[websocket.WebSocketApp] = None
+        self._running = False
+        self._thread: Optional[threading.Thread] = None
+        self._reconnect_count = 0
+
+    def start(self) -> None:
+        if self._running:
+            return
+        self._running = True
+        self._thread  = threading.Thread(
+            target=self._run_loop, daemon=True, name="HubSubscriberClient"
+        )
+        self._thread.start()
+        logger.info(f"[HubSubscriberClient] Starting → {self.hub_url}")
+
+    def stop(self) -> None:
+        self._running = False
+        if self._ws:
+            try:
+                self._ws.close()
+            except Exception:
+                pass
+        if self._thread:
+            self._thread.join(timeout=3)
+
+    def _run_loop(self) -> None:
+        while self._running:
+            try:
+                self._connect_and_run()
+            except Exception as e:
+                logger.error(f"[HubSubscriberClient] error: {e}")
+            if not self._running:
+                break
+            backoff = min(self._MAX_BACKOFF, 2 ** min(self._reconnect_count, 4))
+            logger.info(
+                f"[HubSubscriberClient] reconnect in {backoff}s "
+                f"(attempt #{self._reconnect_count + 1})"
+            )
+            time.sleep(backoff)
+            self._reconnect_count += 1
+
+    def _connect_and_run(self) -> None:
+        self._ws = websocket.WebSocketApp(
+            self.hub_url,
+            on_message = self._on_message,
+            on_open    = lambda ws: logger.info("[HubSubscriberClient] ✅ Connected"),
+            on_error   = lambda ws, e: logger.warning(f"[HubSubscriberClient] WS error: {e}"),
+            on_close   = lambda ws, c, m: logger.info(f"[HubSubscriberClient] closed code={c}"),
+        )
+        self._ws.run_forever(
+            ping_interval = 30,
+            ping_timeout  = 10,
+            reconnect     = 0,
+        )
+
+    def _on_message(self, ws, raw: str) -> None:
+        try:
+            msg  = json.loads(raw)
+            kind = msg.get("type", "")
+
+            if kind == "metrics_update":
+                space    = msg.get("space_name", "")
+                snapshot = msg.get("snapshot",   {})
+                if space and snapshot:
+                    self.state.update_from_snapshot(space, snapshot)
+
+            elif kind == "initial_state":
+                for space, snapshot in msg.get("snapshots", {}).items():
+                    if space and snapshot:
+                        self.state.update_from_snapshot(space, snapshot)
+
+        except Exception as e:
+            logger.debug(f"[HubSubscriberClient] parse error: {e}")
+
+
 # ══════════════════════════════════════════════════════════════════════════════════════
 # SECTION 3 — FLASK APP
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -310,6 +474,10 @@ _state = DashboardState()
 _trade_parser = TradeLogParser(log_dir=_LOG_DIR)
 _trade_parser.start_background()
 
+# Start hub subscriber so _state stays in sync with live metrics
+_hub_subscriber = HubSubscriberClient(state=_state)
+_hub_subscriber.start()
+
 app = Flask(__name__)
 CORS(app)