Update websocket_hub.py

websocket_hub.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 """
 ╔══════════════════════════════════════════════════════════════════════════════════════╗
-║  K1RL QUASAR — CENTRAL WEBSOCKET HUB  v2.1-integrated                               ║
+║  K1RL QUASAR — CENTRAL WEBSOCKET HUB  v2.2-ranker-logs                               ║
 ║  ──────────────────────────────────────────────────────────────────────────────────  ║
 ║                                                                                      ║
 ║  Architecture role: INGEST → NORMALIZE → BROADCAST                                  ║
@@ -12,9 +12,13 @@
 ║  • Hub NEVER writes back to publishers                                               ║
 ║  • Hub stores latest snapshot per asset (NO history)                                 ║
 ║                                                                                      ║
-║  STRICT DATA MODEL (training + voting fields ONLY):                                  ║
-║    training: training_steps, actor_loss, critic_loss, avn_loss, avn_accuracy        ║
-║    voting:   dominant_signal, buy_count, sell_count                                  ║
+║  RANKER LOGS API (FIX v2.2 — moved here from hub_dashboard_service port 8052):      ║
+║    GET /api/ranker/logs/recent   → recent log entries  (?limit=N&category=X)        ║
+║    GET /api/ranker/logs/stats    → log statistics                                   ║
+║    GET /api/ranker/logs/asset/X  → logs for asset X                                 ║
+║    GET /api/ranker/logs/level/X  → logs by level                                    ║
+║    GET /api/ranker/logs/export   → download JSON                                    ║
+║    GET /api/ranker/logs/debug    → file discovery diagnostics                       ║
 ║                                                                                      ║
 ║  TRADE API (served natively — no patch script needed):                               ║
 ║    GET /api/trades          → full open + closed state + stats                       ║
@@ -22,7 +26,7 @@
 ║    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                                             ║
+║  VERSION: v2.2-ranker-logs  |  2026-04-04                                            ║
 ╚══════════════════════════════════════════════════════════════════════════════════════╝
 """
 
@@ -32,6 +36,8 @@ import glob
 import json
 import logging
 import os
+import re
+import threading
 import time
 from collections import deque
 from datetime import datetime
@@ -473,7 +479,7 @@ except Exception:
 app = FastAPI(
     title="K1RL QUASAR Hub",
     description="Central WebSocket hub — ingest, normalize, broadcast (one-way)",
-    version="2.1.0",
+    version="2.2.0",
 )
 app.add_middleware(
     CORSMiddleware,
@@ -701,7 +707,7 @@ async def api_health():
     state = _trade_parser.get_state()
     return JSONResponse({
         "service":      "websocket_hub",
-        "version":      "v2.1-integrated",
+        "version":      "v2.2-ranker-logs",
         "status":       "running",
         "log_files":    len(glob.glob(os.path.join(_LOG_DIR, "*.log"))),
         "trade_open":   len(state["open"]),
@@ -710,6 +716,281 @@ async def api_health():
     })
 
 
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 7b — RANKER LOGS API  (FIX: moved here so routes live on port 7860)
+# ══════════════════════════════════════════════════════════════════════════════════════
+#
+# ROOT CAUSE of HTTP 404 on /api/ranker/logs/*:
+#   - hub_dashboard_service.py (Flask) runs on port 8052 — NOT publicly accessible on HF Spaces
+#   - websocket_hub.py (FastAPI/uvicorn) runs on port 7860 — the ONLY public port
+#   - The browser fetches /api/ranker/logs/recent → hits port 7860 → no route → 404
+#
+# FIX: FileBasedLoggerAdapter + all /api/ranker/logs/* routes added directly here.
+# The ranker writes logs to ./ranker_logs (= /app/ranker_logs). This adapter reads
+# those files directly — no dependency on hub_dashboard_service or in-memory ranker.
+
+_TRAINING_RE_HUB = re.compile(
+    r'step=(\d+)\s*\|\s*loss=([\d.]+)\s*\|\s*lr=([\d.eE+\-]+)\s*\|\s*assets=(\d+)'
+)
+_JSON_BLOB_RE_HUB = re.compile(r'(\{.*\})\s*$')
+
+
+def _enrich_training_entry(entry: dict) -> dict:
+    """Attach parsed `data` dict to TRAINING entries so dashboard KPI cards populate."""
+    if entry.get("category", "").upper() != "TRAINING":
+        return entry
+    if entry.get("data"):
+        return entry
+    msg = entry.get("message", "")
+    m = _TRAINING_RE_HUB.search(msg)
+    if m:
+        entry["data"] = {
+            "step":        int(m.group(1)),
+            "loss":        float(m.group(2)),
+            "lr":          float(m.group(3)),
+            "asset_count": int(m.group(4)),
+        }
+        return entry
+    jm = _JSON_BLOB_RE_HUB.search(msg)
+    if jm:
+        try:
+            blob = json.loads(jm.group(1))
+            if "step" in blob:
+                entry["data"] = {
+                    "step":        blob.get("step", 0),
+                    "loss":        blob.get("loss", 0.0),
+                    "lr":          blob.get("lr", 0.0),
+                    "asset_count": blob.get("asset_count", blob.get("assets", 0)),
+                }
+        except (ValueError, KeyError):
+            pass
+    return entry
+
+
+class FileBasedLoggerAdapter:
+    """
+    Reads ranker log files from disk and exposes the RankerLogger interface
+    expected by the /api/ranker/logs/* endpoints.
+    No in-memory ranker process required.
+    """
+    _CAT_RE   = re.compile(r'\|\s*(INFO|DEBUG|WARNING|ERROR|CRITICAL)\s*\|\s*([A-Z_]+)\s*\|')
+    _ASSET_RE = re.compile(r'\|\s*(?:TRADE|SIGNAL)\s*\|\s*(\w+)\s*\|')
+    _TS_RE    = re.compile(r'\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\]')
+
+    def __init__(self, log_dir: str):
+        self._log_dir = log_dir
+        self._lock    = threading.RLock()
+
+    def _find_files(self) -> list:
+        candidate_dirs = [
+            self._log_dir,
+            "/app/ranker_logs",
+            str(Path(__file__).parent / "ranker_logs"),
+            "./ranker_logs",
+            "/home/user/ranker_logs",
+            "/tmp/ranker_logs",
+        ]
+        all_files: list = []
+        seen: set = set()
+        for d in candidate_dirs:
+            for f in sorted(glob.glob(str(Path(d) / "*.log*"))):
+                if f not in seen:
+                    seen.add(f)
+                    all_files.append(f)
+        return all_files
+
+    def _read_lines(self, n_tail: int = 500) -> list:
+        files = self._find_files()
+        raw: list = []
+        for fpath in files[-3:]:
+            try:
+                with open(fpath, "r", encoding="utf-8", errors="replace") as f:
+                    raw.extend(f.readlines()[-n_tail:])
+            except OSError:
+                pass
+        raw.reverse()  # newest first
+        return raw
+
+    def _line_to_entry(self, line: str) -> Optional[dict]:
+        ts_m = self._TS_RE.search(line)
+        if not ts_m:
+            return None
+        cat_m = self._CAT_RE.search(line)
+        level = cat_m.group(1) if cat_m else "INFO"
+        cat   = cat_m.group(2).strip() if cat_m else ""
+        ast_m = self._ASSET_RE.search(line)
+        asset = ast_m.group(1) if ast_m else None
+        return {
+            "timestamp": ts_m.group(1),
+            "level":     level,
+            "category":  cat,
+            "message":   line.strip(),
+            "asset":     asset,
+            "data":      None,
+        }
+
+    def get_recent(self, n: int = 50, category: Optional[str] = None) -> list:
+        entries: list = []
+        for line in self._read_lines(n_tail=max(n * 3, 200)):
+            e = self._line_to_entry(line)
+            if e is None:
+                continue
+            if category and category.upper() not in line.upper():
+                continue
+            entries.append(e)
+            if len(entries) >= n:
+                break
+        return entries
+
+    def get_by_asset(self, asset: str, n: int = 30) -> list:
+        entries: list = []
+        for line in self._read_lines(n_tail=500):
+            if asset.upper() not in line.upper():
+                continue
+            e = self._line_to_entry(line)
+            if e:
+                entries.append(e)
+            if len(entries) >= n:
+                break
+        return entries
+
+    def get_by_level(self, level: str, n: int = 50) -> list:
+        entries: list = []
+        for line in self._read_lines(n_tail=500):
+            e = self._line_to_entry(line)
+            if e and e["level"].upper() == level.upper():
+                entries.append(e)
+            if len(entries) >= n:
+                break
+        return entries
+
+    def get_stats(self) -> dict:
+        by_category: dict = {}
+        by_level: dict    = {}
+        by_asset: dict    = {}
+        errors: dict      = {}
+        total: int        = 0
+        for line in self._read_lines(n_tail=2000):
+            e = self._line_to_entry(line)
+            if not e:
+                continue
+            total += 1
+            by_level[e["level"]]       = by_level.get(e["level"], 0) + 1
+            by_category[e["category"]] = by_category.get(e["category"], 0) + 1
+            if e["asset"]:
+                by_asset[e["asset"]]   = by_asset.get(e["asset"], 0) + 1
+            if e["level"] in ("ERROR", "CRITICAL"):
+                errors[e["category"]]  = errors.get(e["category"], 0) + 1
+        return {
+            "total_events":    total,
+            "by_level":        by_level,
+            "by_category":     by_category,
+            "by_asset":        by_asset,
+            "errors":          errors,
+            "buffer_size":     total,
+            "buffer_capacity": total,
+        }
+
+    def export_json(self, filepath: str, n: int = 500) -> None:
+        entries = self.get_recent(n)
+        with open(filepath, "w") as f:
+            json.dump({
+                "export_time": datetime.utcnow().isoformat(),
+                "count":       len(entries),
+                "logs":        entries,
+            }, f, indent=2)
+
+    def clear_buffer(self) -> None:
+        pass  # file-based — nothing to clear
+
+
+# Singleton adapter — reads from the same /app/ranker_logs the ranker writes to
+_log_adapter = FileBasedLoggerAdapter(log_dir=_LOG_DIR)
+
+
+@app.get("/api/ranker/logs/recent")
+async def api_ranker_logs_recent(limit: int = 50, category: Optional[str] = None):
+    """GET /api/ranker/logs/recent?limit=80&category=TRAINING"""
+    try:
+        entries = _log_adapter.get_recent(n=limit, category=category)
+        entries = [_enrich_training_entry(e) for e in entries]
+        return JSONResponse({
+            "logs":  entries,
+            "count": len(entries),
+            "stats": _log_adapter.get_stats(),
+        })
+    except Exception as exc:
+        logger.exception(f"[api_ranker_logs_recent] {exc}")
+        return JSONResponse({"logs": [], "count": 0, "error": str(exc)}, status_code=200)
+
+
+@app.get("/api/ranker/logs/stats")
+async def api_ranker_logs_stats():
+    """GET /api/ranker/logs/stats"""
+    try:
+        return JSONResponse(_log_adapter.get_stats())
+    except Exception as exc:
+        return JSONResponse({"error": str(exc)}, status_code=500)
+
+
+@app.get("/api/ranker/logs/asset/{asset}")
+async def api_ranker_logs_asset(asset: str, limit: int = 30):
+    """GET /api/ranker/logs/asset/V75?limit=30"""
+    try:
+        entries = _log_adapter.get_by_asset(asset, n=limit)
+        return JSONResponse({"asset": asset, "logs": entries, "count": len(entries)})
+    except Exception as exc:
+        return JSONResponse({"asset": asset, "logs": [], "count": 0, "error": str(exc)})
+
+
+@app.get("/api/ranker/logs/level/{level}")
+async def api_ranker_logs_level(level: str, limit: int = 50):
+    """GET /api/ranker/logs/level/ERROR?limit=50"""
+    try:
+        entries = _log_adapter.get_by_level(level, n=limit)
+        return JSONResponse({"level": level.upper(), "logs": entries, "count": len(entries)})
+    except Exception as exc:
+        return JSONResponse({"level": level.upper(), "logs": [], "count": 0, "error": str(exc)})
+
+
+@app.get("/api/ranker/logs/export")
+async def api_ranker_logs_export(limit: int = 500):
+    """GET /api/ranker/logs/export — download JSON"""
+    from fastapi.responses import FileResponse as _FileResponse
+    try:
+        export_path = "/tmp/ranker_logs_export.json"
+        _log_adapter.export_json(export_path, n=limit)
+        return _FileResponse(
+            export_path,
+            media_type="application/json",
+            filename="ranker_logs_export.json",
+        )
+    except Exception as exc:
+        return JSONResponse({"error": str(exc)}, status_code=500)
+
+
+@app.post("/api/ranker/logs/clear")
+async def api_ranker_logs_clear():
+    """POST /api/ranker/logs/clear"""
+    try:
+        _log_adapter.clear_buffer()
+        return JSONResponse({"status": "cleared"})
+    except Exception as exc:
+        return JSONResponse({"error": str(exc)}, status_code=500)
+
+
+@app.get("/api/ranker/logs/debug")
+async def api_ranker_logs_debug():
+    """GET /api/ranker/logs/debug — show which log files are found"""
+    files = _log_adapter._find_files()
+    return JSONResponse({
+        "log_dir":     _LOG_DIR,
+        "files_found": files,
+        "file_count":  len(files),
+        "stats":       _log_adapter.get_stats(),
+    })
+
+
 # ══════════════════════════════════════════════════════════════════════════════════════
 # SECTION 8 — DASHBOARD UI ROUTES
 # ══════════════════════════════════════════════════════════════════════════════════════