Upload websocket_hub.py

websocket_hub.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 """
 ╔══════════════════════════════════════════════════════════════════════════════════════╗
-║  K1RL QUASAR — CENTRAL WEBSOCKET HUB                                                ║
+║  K1RL QUASAR — CENTRAL WEBSOCKET HUB  v2.1-integrated                               ║
 ║  ──────────────────────────────────────────────────────────────────────────────────  ║
 ║                                                                                      ║
 ║  Architecture role: INGEST → NORMALIZE → BROADCAST                                  ║
@@ -16,12 +16,19 @@
 ║    training: training_steps, actor_loss, critic_loss, avn_loss, avn_accuracy        ║
 ║    voting:   dominant_signal, buy_count, sell_count                                  ║
 ║                                                                                      ║
-║  VERSION: v2.0-arch-strict  |  2026-03-25                                            ║
+║  TRADE API (served natively — no patch script needed):                               ║
+║    GET /api/trades          → full open + closed state + stats                       ║
+║    GET /api/trades/open     → open trades only                                       ║
+║    GET /api/trades/closed   → recent closed trades + stats  (?limit=N)              ║
+║    GET /api/health          → service health including trade counts                  ║
+║                                                                                      ║
+║  VERSION: v2.1-integrated  |  2026-03-31                                             ║
 ╚══════════════════════════════════════════════════════════════════════════════════════╝
 """
 
 import asyncio
 import copy
+import glob
 import json
 import logging
 import os
@@ -48,7 +55,6 @@ logger = logging.getLogger("QuasarHub")
 # SECTION 1 — STRICT DATA MODEL
 # ══════════════════════════════════════════════════════════════════════════════════════
 
-# Fields that are permitted through the hub.  All others are silently dropped.
 _ALLOWED_TRAINING_FIELDS: frozenset = frozenset({
     "training_steps",
     "actor_loss",
@@ -61,45 +67,35 @@ _ALLOWED_VOTING_FIELDS: frozenset = frozenset({
     "dominant_signal",
     "buy_count",
     "sell_count",
-    "last_price",     # FIX: was silently dropped — needed by subscribers
-    "signal_source",  # FIX: was silently dropped — needed by subscribers
+    "last_price",
+    "signal_source",
 })
 
 
 def _empty_snapshot(space_name: str) -> dict:
-    """Return a clean, zeroed snapshot for a space."""
     return {
-        "space_name": space_name,
+        "space_name":   space_name,
         "last_updated": 0.0,
         "training": {
             "training_steps": 0,
-            "actor_loss": 0.0,
-            "critic_loss": 0.0,
-            "avn_loss": 0.0,
-            "avn_accuracy": 0.0,
+            "actor_loss":     0.0,
+            "critic_loss":    0.0,
+            "avn_loss":       0.0,
+            "avn_accuracy":   0.0,
         },
         "voting": {
             "dominant_signal": "NEUTRAL",
-            "buy_count": 0,
-            "sell_count": 0,
-            "last_price": 0.0,       # FIX: added to match allowed fields
-            "signal_source": "LOG",  # FIX: added to match allowed fields
+            "buy_count":       0,
+            "sell_count":      0,
+            "last_price":      0.0,
+            "signal_source":   "LOG",
         },
     }
 
 
 def _validate_and_normalize(space_name: str, raw: dict) -> Optional[dict]:
-    """
-    Validate incoming payload.  Return a clean normalized dict or None if invalid.
-
-    Strict rules:
-      • Must contain at least one of 'training' or 'voting' keys.
-      • Unknown top-level keys are dropped.
-      • Unknown sub-keys inside training/voting are dropped.
-      • Values are coerced to the correct Python types; malformed values are zeroed.
-    """
     training_raw = raw.get("training", {})
-    voting_raw = raw.get("voting", {})
+    voting_raw   = raw.get("voting",   {})
 
     if not isinstance(training_raw, dict):
         training_raw = {}
@@ -107,32 +103,26 @@ def _validate_and_normalize(space_name: str, raw: dict) -> Optional[dict]:
         voting_raw = {}
 
     if not training_raw and not voting_raw:
-        return None  # Nothing useful
+        return None
 
-    # --- training ---
     def _float(v, default: float = 0.0) -> float:
-        try:
-            return float(v)
-        except (TypeError, ValueError):
-            return default
+        try:    return float(v)
+        except: return default
 
     def _int(v, default: int = 0) -> int:
-        try:
-            return int(v)
-        except (TypeError, ValueError):
-            return default
+        try:    return int(v)
+        except: return default
 
     training: dict = {}
     if training_raw:
         training = {
             "training_steps": _int(training_raw.get("training_steps", 0)),
-            "actor_loss":     _float(training_raw.get("actor_loss", 0.0)),
-            "critic_loss":    _float(training_raw.get("critic_loss", 0.0)),
-            "avn_loss":       _float(training_raw.get("avn_loss", 0.0)),
+            "actor_loss":     _float(training_raw.get("actor_loss",   0.0)),
+            "critic_loss":    _float(training_raw.get("critic_loss",  0.0)),
+            "avn_loss":       _float(training_raw.get("avn_loss",     0.0)),
             "avn_accuracy":   max(0.0, min(1.0, _float(training_raw.get("avn_accuracy", 0.0)))),
         }
 
-    # --- voting ---
     voting: dict = {}
     if voting_raw:
         raw_signal = voting_raw.get("dominant_signal", "NEUTRAL")
@@ -143,16 +133,16 @@ def _validate_and_normalize(space_name: str, raw: dict) -> Optional[dict]:
             raw_source = "LOG"
         voting = {
             "dominant_signal": raw_signal.upper() if raw_signal.upper() in {"BUY", "SELL", "NEUTRAL"} else "NEUTRAL",
-            "buy_count":       _int(voting_raw.get("buy_count", 0)),
-            "sell_count":      _int(voting_raw.get("sell_count", 0)),
-            "last_price":      _float(voting_raw.get("last_price", 0.0)),    # FIX: now passed through
-            "signal_source":   raw_source,                                    # FIX: now passed through
+            "buy_count":       _int(voting_raw.get("buy_count",   0)),
+            "sell_count":      _int(voting_raw.get("sell_count",  0)),
+            "last_price":      _float(voting_raw.get("last_price", 0.0)),
+            "signal_source":   raw_source,
         }
 
     return {
         "space_name": space_name,
-        "training": training,
-        "voting": voting,
+        "training":   training,
+        "voting":     voting,
     }
 
 
@@ -161,28 +151,12 @@ def _validate_and_normalize(space_name: str, raw: dict) -> Optional[dict]:
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 class ConnectionManager:
-    """
-    Manages publisher (Asset Space) and subscriber (Ranker) WebSocket connections.
-
-    Design:
-      • Publishers  →  write-only lane  (/ws/publish/{space_name})
-      • Subscribers →  read-only lane   (/ws/subscribe)
-      • No cross-talk: subscribers never send to publishers
-    """
-
     def __init__(self):
-        # Keyed by space_name → WebSocket
-        self._publishers:  Dict[str, WebSocket] = {}
-        # Set of subscriber sockets
-        self._subscribers: Set[WebSocket] = set()
-        # Latest normalized snapshot per space
-        self._snapshots:   Dict[str, dict] = {}
-        # Asyncio lock for thread-safe mutation
-        self._lock = asyncio.Lock()
-        # Message counter (read by /api/state)
-        self._total_ingested: int = 0
-
-    # ── Publisher lifecycle ──────────────────────────────────────────────────────────
+        self._publishers:     Dict[str, WebSocket] = {}
+        self._subscribers:    Set[WebSocket]        = set()
+        self._snapshots:      Dict[str, dict]       = {}
+        self._lock            = asyncio.Lock()
+        self._total_ingested: int                   = 0
 
     async def register_publisher(self, space_name: str, ws: WebSocket) -> None:
         await ws.accept()
@@ -190,16 +164,13 @@ class ConnectionManager:
             self._publishers[space_name] = ws
             if space_name not in self._snapshots:
                 self._snapshots[space_name] = _empty_snapshot(space_name)
-        logger.info(f"📡 Publisher connected: {space_name}  "
-                    f"(total={len(self._publishers)})")
+        logger.info(f"📡 Publisher connected: {space_name}  (total={len(self._publishers)})")
 
     async def unregister_publisher(self, space_name: str) -> None:
         async with self._lock:
             self._publishers.pop(space_name, None)
         logger.info(f"📡 Publisher disconnected: {space_name}")
 
-    # ── Subscriber lifecycle ─────────────────────────────────────────────────────────
-
     async def register_subscriber(self, ws: WebSocket) -> None:
         await ws.accept()
         async with self._lock:
@@ -211,78 +182,54 @@ class ConnectionManager:
             self._subscribers.discard(ws)
         logger.info(f"🔔 Subscriber disconnected  (total={len(self._subscribers)})")
 
-    # ── Ingestion pipeline ───────────────────────────────────────────────────────────
-
     async def ingest(self, space_name: str, raw_payload: dict) -> None:
-        """
-        Validate → Normalize → Store → Broadcast pipeline.
-        Called for every inbound message from a publisher.
-        """
         normalized = _validate_and_normalize(space_name, raw_payload)
         if normalized is None:
             logger.debug(f"[{space_name}] Payload dropped (no valid fields)")
             return
 
-        # ✅ FIX #6: deep-copy the snapshot *inside* the lock before releasing.
-        # Without this, snap is a live reference. Another publisher coroutine
-        # can acquire the lock and mutate snap["training"] / snap["voting"]
-        # in-place while json.dumps(snapshot) is still serializing it during
-        # broadcast, producing torn / mixed data sent to subscribers.
         async with self._lock:
             snap = self._snapshots.setdefault(space_name, _empty_snapshot(space_name))
             snap["last_updated"] = time.time()
-
             if normalized["training"]:
                 snap["training"].update(normalized["training"])
             if normalized["voting"]:
                 snap["voting"].update(normalized["voting"])
-
             self._total_ingested += 1
-            snap_copy = copy.deepcopy(snap)  # frozen snapshot for broadcast
+            snap_copy = copy.deepcopy(snap)
 
-        # Broadcast outside the lock using the immutable copy
         await self._broadcast_update(space_name, snap_copy)
 
-    # ── Broadcast engine ─────────────────────────────────────────────────────────────
-
     async def _broadcast_update(self, space_name: str, snapshot: dict) -> None:
-        """Fan-out the updated snapshot to all subscribers (event-driven)."""
         if not self._subscribers:
             return
-
         message = json.dumps({
-            "type": "metrics_update",
-            "space_name": space_name,
-            "snapshot": snapshot,
+            "type":          "metrics_update",
+            "space_name":    space_name,
+            "snapshot":      snapshot,
             "hub_timestamp": time.time(),
         })
-
         dead: list = []
         for ws in list(self._subscribers):
             try:
                 await ws.send_text(message)
             except Exception:
                 dead.append(ws)
-
         if dead:
             async with self._lock:
                 for ws in dead:
                     self._subscribers.discard(ws)
 
     async def send_initial_state(self, ws: WebSocket) -> None:
-        """Send full current state to a newly connected subscriber."""
         async with self._lock:
             snapshots_copy = dict(self._snapshots)
-
         message = json.dumps({
-            "type": "initial_state",
-            "snapshots": snapshots_copy,
+            "type":          "initial_state",
+            "snapshots":     snapshots_copy,
             "hub_timestamp": time.time(),
         })
         await ws.send_text(message)
 
-    # ── Read-only accessors (REST API) ───────────────────────────────────────────────
-
     def get_snapshot(self, space_name: str) -> Optional[dict]:
         return self._snapshots.get(space_name)
 
@@ -294,7 +241,7 @@ class ConnectionManager:
         return {
             "publishers": {
                 name: {
-                    "last_updated": self._snapshots.get(name, {}).get("last_updated", 0),
+                    "last_updated":  self._snapshots.get(name, {}).get("last_updated", 0),
                     "stale_seconds": round(now - self._snapshots.get(name, {}).get("last_updated", now), 1),
                 }
                 for name in self._publishers
@@ -304,13 +251,173 @@ class ConnectionManager:
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 3 — FASTAPI APPLICATION
+# SECTION 3 — TRADE LOG PARSER  (native — no patch script needed)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+class TradeLogParser:
+    """
+    Lightweight trade-log reader that watches ranker_logs/*.log files and
+    exposes open/closed trade state via get_state().
+
+    Imported from hub_dashboard_service if available; otherwise this built-in
+    fallback is used so /api/trades routes always work.
+    """
+
+    def __init__(self, log_dir: str = "/app/ranker_logs"):
+        self.log_dir  = log_dir
+        self._state   = {"open": {}, "closed": [], "stats": {}}
+        self._lock    = None   # set in start_background()
+        self._running = False
+
+    def start_background(self) -> None:
+        """Schedule background log-scanning on the running event loop."""
+        self._running = True
+        asyncio.get_event_loop().create_task(self._scan_loop())
+        logger.info(f"📂 TradeLogParser started — watching {self.log_dir}")
+
+    async def _scan_loop(self) -> None:
+        while self._running:
+            try:
+                self._refresh()
+            except Exception as e:
+                logger.debug(f"TradeLogParser scan error: {e}")
+            await asyncio.sleep(5)
+
+    def _refresh(self) -> None:
+        """Re-parse all log files and rebuild open/closed state."""
+        open_trades:   Dict[str, dict] = {}
+        closed_trades: List[dict]      = []
+
+        log_files = sorted(glob.glob(os.path.join(self.log_dir, "*.log")))
+        for fpath in log_files:
+            try:
+                with open(fpath, "r", encoding="utf-8", errors="replace") as fh:
+                    for line in fh:
+                        self._parse_line(line, open_trades, closed_trades)
+            except Exception:
+                pass
+
+        # Sort closed newest-first
+        closed_trades.sort(key=lambda t: t.get("closed_ts", 0), reverse=True)
+
+        # Compute stats
+        pnls       = [t.get("pnl", 0.0) for t in closed_trades]
+        wins       = [p for p in pnls if p > 0]
+        total      = len(pnls)
+        stats = {
+            "total_closed":  total,
+            "total_open":    len(open_trades),
+            "win_rate":      round(len(wins) / total, 4) if total else 0.0,
+            "total_pnl":     round(sum(pnls), 6),
+            "avg_pnl":       round(sum(pnls) / total, 6) if total else 0.0,
+            "avg_pnl_trade": round(sum(pnls) / total, 6) if total else 0.0,
+        }
+
+        self._state = {
+            "open":   list(open_trades.values()),
+            "closed": closed_trades,
+            "stats":  stats,
+        }
+
+    # ── Log-line parser ──────────────────────────────────────────────────────────────
+    # Matches lines like:
+    #   [2026-03-31 10:02:07] | INFO | TRADE | V75 | TRADE OPENED | ID=V75_... | ...
+    #   [2026-03-31 10:02:07] | INFO | TRADE | V75 | TRADE CLOSED | ID=V75_... | pnl=-32.62 | ...
+
+    def _parse_line(
+        self,
+        line:         str,
+        open_trades:  Dict[str, dict],
+        closed_trades: List[dict],
+    ) -> None:
+        if "| TRADE |" not in line:
+            return
+
+        parts = [p.strip() for p in line.split("|")]
+        # parts[0]=timestamp, [1]=level, [2]="TRADE", [3]=asset, [4]=event, [5..]=fields
+        if len(parts) < 5:
+            return
+
+        try:
+            ts_str = parts[0].strip("[] ")
+            ts     = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").timestamp()
+        except Exception:
+            ts = 0.0
+
+        asset = parts[3].strip()
+        event = parts[4].strip().upper()
+
+        # Build a quick field dict from KEY=VALUE tokens in the rest of the line
+        fields: Dict[str, str] = {}
+        for segment in parts[5:]:
+            for token in segment.split():
+                if "=" in token:
+                    k, _, v = token.partition("=")
+                    fields[k.strip()] = v.strip().rstrip(",")
+
+        trade_id = fields.get("ID", f"{asset}_{int(ts)}")
+
+        def _f(k: str) -> float:
+            try:    return float(fields.get(k, 0))
+            except: return 0.0
+
+        if "TRADE OPENED" in event:
+            open_trades[trade_id] = {
+                "trade_id":  trade_id,
+                "asset":     asset,
+                "direction": fields.get("Dir", fields.get("Direction", "?")).capitalize(),
+                "entry":     _f("Entry"),
+                "qty":       _f("Qty"),
+                "open_pnl":  0.0,
+                "opened_ts": ts,
+                "opened":    ts_str,
+            }
+
+        elif "TRADE CLOSED" in event:
+            # Remove from open if present
+            trade = open_trades.pop(trade_id, {})
+            pnl   = _f("pnl")
+            ret   = _f("return")
+            closed_trades.append({
+                "trade_id":   trade_id,
+                "asset":      asset,
+                "direction":  trade.get("direction", "?"),
+                "entry":      trade.get("entry", _f("Entry")),
+                "exit":       _f("Exit") or _f("exit"),
+                "qty":        trade.get("qty", _f("Qty")),
+                "pnl":        pnl,
+                "return_pct": ret,
+                "closed_ts":  ts,
+                "closed":     ts_str,
+                "duration":   round(ts - trade.get("opened_ts", ts), 0) if trade else 0,
+            })
+
+    def get_state(self) -> dict:
+        return self._state
+
+
+# ── Bootstrap parser ─────────────────────────────────────────────────────────────────
+# Try the richer version from hub_dashboard_service first; fall back to built-in above.
+
+_LOG_DIR = os.environ.get("RANKER_LOG_DIR", "/app/ranker_logs")
+
+try:
+    from hub_dashboard_service import TradeLogParser as _ExternalParser
+    _trade_parser = _ExternalParser(log_dir=_LOG_DIR)
+    logger.info("✅ TradeLogParser loaded from hub_dashboard_service")
+except Exception:
+    _trade_parser = TradeLogParser(log_dir=_LOG_DIR)
+    logger.info("✅ TradeLogParser using built-in fallback")
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 4 — FASTAPI APPLICATION
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 app = FastAPI(
     title="K1RL QUASAR Hub",
     description="Central WebSocket hub — ingest, normalize, broadcast (one-way)",
-    version="2.0.0",
+    version="2.1.0",
 )
 app.add_middleware(
     CORSMiddleware,
@@ -322,22 +429,19 @@ app.add_middleware(
 manager = ConnectionManager()
 
 
+@app.on_event("startup")
+async def _on_startup():
+    """Start background trade-log scanner after the event loop is running."""
+    _trade_parser.start_background()
+    logger.info("🚀 Trade log scanner started")
+
+
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 4 — WEBSOCKET ENDPOINTS
+# SECTION 5 — WEBSOCKET ENDPOINTS
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 @app.websocket("/ws/publish/{space_name}")
 async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
-    """
-    Publisher endpoint — Asset Spaces connect here.
-    SEND ONLY: hub never writes back to this socket.
-
-    Accepted message types:
-      {"type": "metrics", "training": {...}, "voting": {...}}
-      {"type": "training", "data": {...}}
-      {"type": "voting",   "data": {...}}
-      {"type": "heartbeat"}
-    """
     await manager.register_publisher(space_name, websocket)
     try:
         while True:
@@ -351,28 +455,22 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
             msg_type = data.get("type", "")
 
             if msg_type == "metrics":
-                # Full combined payload
                 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":   {},
                 })
-
             elif msg_type == "voting":
                 await manager.ingest(space_name, {
                     "training": {},
                     "voting":   data.get("data", {}),
                 })
-
             elif msg_type in ("heartbeat", "identify", "ping"):
-                # Silently acknowledged — no reply sent back
                 pass
-
             else:
                 logger.debug(f"[{space_name}] Unrecognised type '{msg_type}' — dropped")
 
@@ -386,19 +484,10 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
 
 @app.websocket("/ws/subscribe")
 async def ws_subscriber_endpoint(websocket: WebSocket):
-    """
-    Subscriber endpoint — Ranker Space connects here.
-    READ ONLY: subscribers must not send data; any inbound messages are discarded.
-
-    Messages received by subscriber:
-      {"type": "initial_state", "snapshots": {...}, "hub_timestamp": ...}
-      {"type": "metrics_update", "space_name": "...", "snapshot": {...}, "hub_timestamp": ...}
-    """
     await manager.register_subscriber(websocket)
     await manager.send_initial_state(websocket)
     try:
         while True:
-            # Drain any inbound messages without processing them (read-only contract)
             await websocket.receive_text()
     except WebSocketDisconnect:
         pass
@@ -409,15 +498,11 @@ async def ws_subscriber_endpoint(websocket: WebSocket):
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 5 — REST API (READ-ONLY)
+# SECTION 6 — REST API (READ-ONLY)
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 @app.get("/rankings")
 async def get_rankings():
-    """
-    Return latest snapshot for all assets.
-    Ranker may also poll this endpoint as a fallback.
-    """
     return {
         "snapshots": manager.get_all_snapshots(),
         "timestamp": datetime.utcnow().isoformat() + "Z",
@@ -426,7 +511,6 @@ async def get_rankings():
 
 @app.get("/metrics/{space_name}")
 async def get_space_metrics(space_name: str):
-    """Return latest snapshot for a single asset space."""
     snap = manager.get_snapshot(space_name)
     if snap is None:
         return {"error": f"Unknown space: {space_name}"}
@@ -436,15 +520,56 @@ async def get_space_metrics(space_name: str):
 @app.get("/health")
 async def get_health():
     return {
-        "status": "ok",
+        "status":    "ok",
         "timestamp": datetime.utcnow().isoformat() + "Z",
         **manager.get_health(),
     }
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 6 — DASHBOARD UI ROUTES
-# (serves hub_dashboard.html at / and provides /api/state for the polling frontend)
+# SECTION 7 — TRADE API  (native — replaces patch_websocket_hub.py)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+@app.get("/api/trades")
+async def api_trades():
+    """Full trade state: open trades, recent closed trades, summary stats."""
+    return JSONResponse(_trade_parser.get_state())
+
+
+@app.get("/api/trades/open")
+async def api_trades_open():
+    """Open trades only."""
+    state = _trade_parser.get_state()
+    return JSONResponse({"open": state["open"]})
+
+
+@app.get("/api/trades/closed")
+async def api_trades_closed(limit: int = 50):
+    """Recent closed trades (newest first) + cumulative stats."""
+    state = _trade_parser.get_state()
+    return JSONResponse({
+        "closed": state["closed"][:limit],
+        "stats":  state["stats"],
+    })
+
+
+@app.get("/api/health")
+async def api_health():
+    """Service health — includes live trade counts and log-file inventory."""
+    state = _trade_parser.get_state()
+    return JSONResponse({
+        "service":      "websocket_hub",
+        "version":      "v2.1-integrated",
+        "status":       "running",
+        "log_files":    len(glob.glob(os.path.join(_LOG_DIR, "*.log"))),
+        "trade_open":   len(state["open"]),
+        "trade_closed": len(state["closed"]),
+        "uptime_seconds": round(time.time() - _START_TIME, 0),
+    })
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 8 — DASHBOARD UI ROUTES
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 _HTML_PATH = Path(os.environ.get(
@@ -454,12 +579,6 @@ _HTML_PATH = Path(os.environ.get(
 
 
 def _compute_rankings() -> List[dict]:
-    """
-    Derive ranked list from manager snapshots using the AXRVI formula:
-        signal_confidence = max(buy, sell) / (buy + sell)   if total > 0 else 0
-        score             = signal_confidence - avn_accuracy
-    Sorted descending by score.
-    """
     ranked: List[dict] = []
     for name, snap in manager.get_all_snapshots().items():
         training = snap.get("training", {})
@@ -493,7 +612,6 @@ def _compute_rankings() -> List[dict]:
 
 @app.get("/")
 async def serve_dashboard():
-    """Serve the dashboard HTML. Returns a helpful message if the file is missing."""
     if _HTML_PATH.exists():
         return FileResponse(str(_HTML_PATH), media_type="text/html")
     return JSONResponse(
@@ -503,24 +621,26 @@ async def serve_dashboard():
             "status":   "running",
             "note":     "hub_dashboard.html not found — upload it to the Space",
             "expected": str(_HTML_PATH),
-            "endpoints": ["/rankings", "/health", "/api/state", "/ws/publish/{space}", "/ws/subscribe"],
+            "endpoints": [
+                "/rankings", "/health",
+                "/api/state", "/api/trades", "/api/trades/open",
+                "/api/trades/closed", "/api/health",
+                "/ws/publish/{space}", "/ws/subscribe",
+            ],
         },
     )
 
 
 @app.get("/api/state")
 async def api_state():
-    """
-    Full dashboard state polled by hub_dashboard.html every 2 s.
-    Returns rankings + metric history stub + health.
-    """
-    h = manager.get_health()
+    """Full dashboard state polled by hub_dashboard.html every 2 s."""
+    h        = manager.get_health()
     rankings = _compute_rankings()
     return JSONResponse({
         "rankings":       rankings,
-        "metric_history": {},          # history kept client-side via snapshot deltas
+        "metric_history": {},
         "health": {
-            "hub_connected":    True,  # we ARE the hub
+            "hub_connected":    True,
             "spaces_connected": len(manager.get_all_snapshots()),
             "messages_rx":      manager._total_ingested,
             "last_update_ts":   max(
@@ -533,8 +653,8 @@ async def api_state():
                     default=time.time(),
                 ), 1
             ),
-            "uptime_seconds":   round(time.time() - _START_TIME, 0),
-            "reconnect_count":  0,
+            "uptime_seconds":  round(time.time() - _START_TIME, 0),
+            "reconnect_count": 0,
         },
         "timestamp": datetime.utcnow().isoformat() + "Z",
     })
@@ -544,10 +664,10 @@ _START_TIME = time.time()
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 7 — ENTRY POINT
+# SECTION 9 — ENTRY POINT
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 if __name__ == "__main__":
     port = int(os.environ.get("PORT", 7860))
     logger.info(f"🚀 QUASAR Hub starting on port {port}")
-    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")
+    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")