Update hub_dashboard_service.py

hub_dashboard_service.py

@@ -1,40 +1,30 @@
 #!/usr/bin/env python3
 """
 ╔══════════════════════════════════════════════════════════════════════════════════════╗
-║  K1RL QUASAR — HUB DASHBOARD SERVICE  (WITH RANKER LOGS INTEGRATION)                ║
+║  K1RL QUASAR — HUB DASHBOARD SERVICE (with Trade Log Parser)                        ║
 ║  ──────────────────────────────────────────────────────────────────────────────────  ║
 ║  Architecture role: READ-ONLY subscriber → serves dashboard UI                      ║
-║  VERSION: v1.1 (UPDATED)  |  2026-03-26                                              ║
-╚═══════════════════════════════════════════════���══════════════════════════════════════╝
+║  VERSION: v2.0 (FIXED)  |  2026-03-30                                              ║
+╚══════════════════════════════════════════════════════════════════════════════════════╝
 """
 
 import json
 import logging
 import os
 import sys
+import re
+import glob
 import threading
 import time
-from collections import deque
+from collections import deque, defaultdict
 from datetime import datetime
 from pathlib import Path
 from typing import Dict, List, Optional
 
-import re
-import glob
-
 import websocket
-from flask import Flask, Response, jsonify, request, send_from_directory
+from flask import Flask, jsonify, request, send_from_directory
 from flask_cors import CORS
 
-# ── Trade log regexes ─────────────────────────────────────────────────────────────────
-TRADE_OPEN_RE  = re.compile(r'TRADE OPENED \| ID=(\S+) \| Dir=(\w+) \| Entry=([\d.]+)')
-TRADE_CLOSE_RE = re.compile(r'TRADE CLOSED.*?PnL=([+-]?[\d.]+)')
-TRADE_ID_RE    = re.compile(r'\[(\w+)\] TRADE CLOSED')
-REWARD_RE      = re.compile(r'Closed (\w+) \| reward=([+-]?[\d.]+) \| pnl=([+-]?[\d.]+) \| portfolio_dd=([\d.]+)%')
-
-from ranker_logging import RankerLogger, EventCategory, LogLevel
-from ranker_logs_api import ranker_logs_bp, init_ranker_logs_api
-
 # ── Logging ───────────────────────────────────────────────────────────────────────────
 logging.basicConfig(
     level=logging.INFO,
@@ -45,216 +35,94 @@ logging.basicConfig(
 logger = logging.getLogger("HubDashboardService")
 
 # ── Config ────────────────────────────────────────────────────────────────────────────
-_HUB_HOST       = os.environ.get("QUASAR_HUB_HOST",  "karlquant-quasar-executo.hf.space")
+_HUB_HOST = os.environ.get("QUASAR_HUB_HOST", "karlquant-quasar-executo.hf.space")
 _DASHBOARD_PORT = int(os.environ.get("DASHBOARD_PORT", "8051"))
-_HTML_PATH      = os.environ.get(
+_HTML_PATH = os.environ.get(
     "DASHBOARD_HTML",
     str(Path(__file__).parent / "hub_dashboard.html"),
 )
-
+_LOG_DIR = os.environ.get("RANKER_LOG_DIR", "/app/ranker_logs")
 _METRIC_HISTORY_LEN = int(os.environ.get("QUASAR_METRIC_HISTORY", "200"))
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 1 — IN-MEMORY STATE STORE
+# SECTION 1 — TRADE LOG PARSER (FIXED for actual log format)
 # ══════════════════════════════════════════════════════════════════════════════════════
 
-class DashboardState:
-    """
-    Thread-safe cache of everything the dashboard needs.
-    """
-
-    def __init__(self, history_len: int = _METRIC_HISTORY_LEN):
-        self._lock    = threading.RLock()
-        self._history_len = history_len
-
-        self.snapshots: Dict[str, dict] = {}
-        self.metric_history: Dict[str, deque] = {}
-
-        self.hub_connected   = False
-        self.start_time      = time.time()
-        self.messages_rx     = 0
-        self.last_update_ts  = 0.0
-        self.reconnect_count = 0
-
-        self._ranked: List[dict] = []
-
-    def ingest_snapshot(self, space_name: str, snapshot: dict) -> None:
-        """Accept a full snapshot dict."""
-        with self._lock:
-            self.snapshots[space_name] = snapshot
-            self.messages_rx += 1
-            self.last_update_ts = time.time()
-            self._append_metric_history(space_name, snapshot)
-            self._recompute_rankings()
-
-    def ingest_bulk(self, snapshots: dict) -> None:
-        """Accept bulk payload."""
-        with self._lock:
-            for name, snap in snapshots.items():
-                self.snapshots[name] = snap
-                self._append_metric_history(name, snap)
-            self.last_update_ts = time.time()
-            self._recompute_rankings()
-
-    def _append_metric_history(self, space_name: str, snap: dict) -> None:
-        """Push latest training numbers onto the per-asset deque."""
-        if space_name not in self.metric_history:
-            self.metric_history[space_name] = deque(maxlen=self._history_len)
-        training = snap.get("training", {})
-        if training:
-            self.metric_history[space_name].append({
-                "ts":            snap.get("last_updated", time.time()),
-                "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),
-                "avn_accuracy":   training.get("avn_accuracy",   0.0),
-            })
-
-    def _recompute_rankings(self) -> None:
-        """Replicate ranker formula: score = signal_confidence - avn_accuracy"""
-        ranked = []
-        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    = sig_conf - avn_acc
-            ranked.append({
-                "rank":              0,
-                "space_name":        name,
-                "score":             round(score, 4),
-                "signal_confidence": round(sig_conf, 4),
-                "avn_accuracy":      round(avn_acc,  4),
-                "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
-        self._ranked = ranked
-
-    def get_state(self) -> dict:
-        with self._lock:
-            return {
-                "rankings":      list(self._ranked),
-                "metric_history": {
-                    name: list(h)
-                    for name, h in self.metric_history.items()
-                },
-                "health":        self._health_snapshot(),
-                "timestamp":     datetime.utcnow().isoformat() + "Z",
-            }
-
-    def get_rankings(self) -> List[dict]:
-        with self._lock:
-            return list(self._ranked)
-
-    def get_metric_history(self, limit: int = 100) -> dict:
-        with self._lock:
-            return {
-                name: list(h)[-limit:]
-                for name, h in self.metric_history.items()
-            }
-
-    def _health_snapshot(self) -> dict:
-        return {
-            "hub_connected":   self.hub_connected,
-            "spaces_connected": len(self.snapshots),
-            "messages_rx":     self.messages_rx,
-            "last_update_ts":  self.last_update_ts,
-            "last_update_ago": round(time.time() - self.last_update_ts, 1)
-                               if self.last_update_ts else None,
-            "uptime_seconds":  round(time.time() - self.start_time, 0),
-            "reconnect_count": self.reconnect_count,
-        }
-
-    def get_health(self) -> dict:
-        with self._lock:
-            return self._health_snapshot()
-
-
-
-# ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 1b — TRADE LOG PARSER
-# ══════════════════════════════════════════════════════���═══════════════════════════════
 class TradeLogParser:
     """
     Tails ranker log files and maintains open/closed trade state.
     Runs in a background thread, refreshing every 2 seconds.
+    
+    Expected log format from ranker_logging.py:
+    [2026-03-30 16:20:40] | INFO | TRADE | CRASH500 | TRADE OPENED | ID=CRASH500_123 | Dir=long | Entry=3524.6485 | Qty=0.000284
+    [2026-03-30 16:20:39] | INFO | TRADE | CRASH500 | TRADE CLOSED | ID=CRASH500_456 | pnl=-3.5246 | return=+0.01%
     """
 
-    # Updated regex to match ranker's actual log format
-    TRADE_OPEN_RE  = re.compile(r'\[([^\]]+)\] TRADE OPENED \| ID=(\S+) \| Dir=(\w+) \| Entry=([\d.]+)')
-    TRADE_CLOSE_RE = re.compile(r'\[([^\]]+)\] TRADE CLOSED \| reward=[+-]?[\d.]+ \| pnl=([+-]?[\d.]+)')
-    TRADE_ID_RE    = re.compile(r'\[([^\]]+)\] TRADE CLOSED')
-    REWARD_RE      = re.compile(r'Closed (\w+) \| reward=([+-]?[\d.]+) \| pnl=([+-]?[\d.]+) \| portfolio_dd=([\d.]+)%')
-
-    def __init__(self, log_dir: str = "./ranker_logs"):
-        self._lock      = threading.RLock()
-        self._open:   dict  = {}   # trade_id → trade dict
-        self._closed: list  = []   # list of closed trade dicts (most recent last)
-        self._log_dir = log_dir
-        self._last_pos: dict = {}  # filepath → byte offset (tail mode)
+    # Regex patterns matching the actual log format
+    # Pattern for OPEN: ... | TRADE OPENED | ID=xxx | Dir=xxx | Entry=xxx
+    TRADE_OPEN_RE = re.compile(r'TRADE OPENED \| ID=(\S+) \| Dir=(\w+) \| Entry=([\d.]+)')
+    
+    # Pattern for CLOSE: ... | TRADE CLOSED | ID=xxx | pnl=xxx | return=xxx
+    TRADE_CLOSE_RE = re.compile(r'TRADE CLOSED \| ID=(\S+) \| pnl=([+-]?[\d.]+)')
+    
+    # Pattern to extract asset from the log line (the part after TRADE |)
+    TRADE_ASSET_RE = re.compile(r'TRADE \| (\w+) \|')
+
+    def __init__(self, log_dir: str = _LOG_DIR):
+        self.log_dir = Path(log_dir)
+        self._open: Dict[str, dict] = {}
+        self._closed: List[dict] = []
+        self._last_pos: Dict[str, int] = {}
+        self._lock = threading.RLock()
         self._stats = {
             "total_opened": 0,
             "total_closed": 0,
-            "total_pnl":    0.0,
-            "win_count":    0,
-            "loss_count":   0,
+            "total_pnl": 0.0,
+            "win_count": 0,
+            "loss_count": 0,
         }
-        self._debug_counter = 0  # for debugging
+        self._running = False
+        self._thread: Optional[threading.Thread] = None
+        
+        # Create log directory if it doesn't exist
+        self.log_dir.mkdir(parents=True, exist_ok=True)
+        logger.info(f"[TradeLogParser] Initialized | log_dir={self.log_dir}")
 
     def start_background(self, interval: float = 2.0) -> None:
-        def _loop():
-            while True:
-                try:
-                    self.refresh()
-                except Exception as e:
-                    logger.error(f"[TradeLogParser] refresh error: {e}")
-                time.sleep(interval)
-
-        t = threading.Thread(target=_loop, daemon=True, name="TradeLogParser")
-        t.start()
-        logger.info(f"[TradeLogParser] Started — watching {self._log_dir}")
+        """Launch a daemon thread that calls refresh() every `interval` seconds."""
+        if self._running:
+            return
+        
+        self._running = True
+        self._thread = threading.Thread(target=self._loop, daemon=True, name="TradeLogParser")
+        self._thread.start()
+        logger.info(f"[TradeLogParser] Started — watching {self.log_dir} (interval={interval}s)")
+
+    def _loop(self) -> None:
+        """Background loop."""
+        while self._running:
+            try:
+                self.refresh()
+            except Exception as e:
+                logger.error(f"[TradeLogParser] refresh error: {e}")
+            time.sleep(2.0)
 
     def refresh(self) -> None:
-        pattern = os.path.join(self._log_dir, "*.log")
+        """Find all log files, read new lines since last position."""
+        pattern = str(self.log_dir / "*.log")
         files = sorted(glob.glob(pattern))
+        
         if not files:
-            pattern = os.path.join(self._log_dir, "*.txt")
+            # Also check for .txt files
+            pattern = str(self.log_dir / "*.txt")
             files = sorted(glob.glob(pattern))
-
+        
         for fpath in files:
             self._tail_file(fpath)
 
-    def get_state(self) -> dict:
-        with self._lock:
-            open_trades = list(self._open.values())
-            closed_trades = list(reversed(self._closed[-100:]))
-            stats = dict(self._stats)
-            stats["win_rate"] = (
-                round(stats["win_count"] / stats["total_closed"] * 100, 1)
-                if stats["total_closed"] > 0 else 0.0
-            )
-            return {
-                "open":   open_trades,
-                "closed": closed_trades,
-                "stats":  stats,
-                "timestamp": datetime.utcnow().isoformat() + "Z",
-            }
-
     def _tail_file(self, fpath: str) -> None:
+        """Read only new bytes from fpath since last call."""
         try:
             size = os.path.getsize(fpath)
         except OSError:
@@ -276,202 +144,143 @@ class TradeLogParser:
             self._parse_line(line)
 
     def _parse_line(self, line: str) -> None:
-        # Debug first few lines
-        if self._debug_counter < 10:
-            logger.debug(f"[TradeLogParser] Line: {line.strip()}")
-            self._debug_counter += 1
-
+        """Extract trade events from a single log line."""
+        
+        # Extract asset from the line (if present)
+        asset_match = self.TRADE_ASSET_RE.search(line)
+        asset = asset_match.group(1) if asset_match else None
+        
         # ── TRADE OPENED ─────────────────────────────────────────────────────────
         m = self.TRADE_OPEN_RE.search(line)
         if m:
-            asset, trade_id, direction, entry = m.group(1), m.group(2), m.group(3), float(m.group(4))
-            ts = self._parse_ts(line)
+            trade_id, direction, entry = m.group(1), m.group(2), float(m.group(3))
+            # Normalize direction: long -> LONG, short -> SHORT
+            direction = direction.upper()
+            
+            # Parse timestamp from log line
+            ts = self._parse_timestamp(line)
+            
             with self._lock:
                 self._open[trade_id] = {
-                    "trade_id":   trade_id,
-                    "asset":      asset,
-                    "direction":  direction,
-                    "entry":      entry,
-                    "opened_at":  ts,
-                    "status":     "OPEN",
+                    "trade_id": trade_id,
+                    "asset": asset or trade_id.split('_')[0],
+                    "direction": direction,
+                    "entry": entry,
+                    "opened_at": ts,
+                    "status": "OPEN",
                 }
                 self._stats["total_opened"] += 1
-                logger.info(f"[TradeLogParser] Opened: {trade_id} {asset} {direction}")
+            
+            logger.debug(f"[TradeLogParser] OPEN: {trade_id} | {direction} @ {entry}")
             return
 
         # ── TRADE CLOSED ─────────────────────────────────────────────────────────
-        m_close = self.TRADE_CLOSE_RE.search(line)
-        m_id = self.TRADE_ID_RE.search(line)
-        if m_close and m_id:
-            asset = m_id.group(1)
-            pnl = float(m_close.group(2))
-            ts = self._parse_ts(line)
-
+        m = self.TRADE_CLOSE_RE.search(line)
+        if m:
+            trade_id, pnl = m.group(1), float(m.group(2))
+            ts = self._parse_timestamp(line)
+            
             with self._lock:
-                matched_id = None
-                for tid, trade in self._open.items():
-                    if trade["asset"] == asset:
-                        matched_id = tid
-                        break
-
-                trade = self._open.pop(matched_id, None) if matched_id else None
+                # Find the matching open trade
+                trade = self._open.pop(trade_id, None)
+                
                 closed = {
-                    "trade_id":   matched_id or asset,
-                    "asset":      asset,
-                    "direction":  trade["direction"] if trade else "?",
-                    "entry":      trade["entry"] if trade else 0.0,
-                    "pnl":        pnl,
-                    "closed_at":  ts,
-                    "status":     "CLOSED",
+                    "trade_id": trade_id,
+                    "asset": asset or (trade.get("asset") if trade else trade_id.split('_')[0]),
+                    "pnl": pnl,
+                    "closed_at": ts,
+                    "status": "CLOSED",
                 }
+                
+                if trade:
+                    closed["direction"] = trade.get("direction", "?")
+                    closed["entry"] = trade.get("entry", 0.0)
+                
                 self._closed.append(closed)
                 self._stats["total_closed"] += 1
                 self._stats["total_pnl"] += pnl
+                
                 if pnl >= 0:
                     self._stats["win_count"] += 1
                 else:
                     self._stats["loss_count"] += 1
-                logger.info(f"[TradeLogParser] Closed: {asset} pnl={pnl}")
+            
+            logger.debug(f"[TradeLogParser] CLOSE: {trade_id} | pnl={pnl:+.2f}")
             return
 
-        # ── REWARD / portfolio_dd line ────────────────────────────────────────────
-        m = self.REWARD_RE.search(line)
-        if m and self._closed:
-            reward = float(m.group(2))
-            with self._lock:
-                self._closed[-1]["reward"] = reward
-                self._closed[-1]["portfolio_dd"] = float(m.group(4))
-
     @staticmethod
-    def _parse_ts(line: str) -> str:
-        parts = line.split("|")
-        if parts:
-            ts = parts[0].strip()
-            if len(ts) >= 19:
-                return ts[:19].replace(" ", "T")
+    def _parse_timestamp(line: str) -> str:
+        """Extract ISO timestamp from log line prefix."""
+        # Format: [2026-03-30 16:20:40] | ...
+        match = re.search(r'\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\]', line)
+        if match:
+            return match.group(1).replace(" ", "T")
         return datetime.utcnow().isoformat()[:19]
 
-
-
-
-class HubSubscriber:
-    """Connects to hub and maintains per-asset snapshots."""
-
-    _MAX_BACKOFF = 30
-
-    def __init__(
-        self,
-        state: DashboardState,
-        hub_host: str = _HUB_HOST,
-        ranker_logger: Optional[RankerLogger] = None,
-    ):
-        self.state      = state
-        self.hub_url    = f"wss://{hub_host}/ws/subscribe"
-        self.ranker_logger = ranker_logger
-        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="HubSubscriber",
-        )
-        self._thread.start()
-        logger.info(f"HubSubscriber started → {self.hub_url}")
+    def get_state(self) -> dict:
+        """Return current trade state."""
+        with self._lock:
+            open_trades = list(self._open.values())
+            closed_trades = list(reversed(self._closed[-100:]))  # newest first
+            
+            stats = dict(self._stats)
+            stats["win_rate"] = (
+                round(stats["win_count"] / stats["total_closed"] * 100, 1)
+                if stats["total_closed"] > 0 else 0.0
+            )
+            
+            return {
+                "open": open_trades,
+                "closed": closed_trades,
+                "stats": stats,
+                "timestamp": datetime.utcnow().isoformat() + "Z",
+            }
 
     def stop(self) -> None:
+        """Stop the background thread."""
         self._running = False
-        if self._ws:
-            try:
-                self._ws.close()
-            except Exception:
-                pass
         if self._thread:
             self._thread.join(timeout=5)
 
-    def _run_loop(self) -> None:
-        while self._running:
-            try:
-                self._connect()
-            except Exception as e:
-                logger.error(f"[HubSubscriber] Error: {e}")
-            if not self._running:
-                break
-            self.state.hub_connected   = False
-            self.state.reconnect_count = self._reconnect_count
-            backoff = min(self._MAX_BACKOFF, 2 ** min(self._reconnect_count, 4))
-            logger.info(f"[HubSubscriber] Reconnecting in {backoff}s…")
-            time.sleep(backoff)
-            self._reconnect_count += 1
-
-    def _connect(self) -> None:
-        self._ws = websocket.WebSocketApp(
-            self.hub_url,
-            on_open    = self._on_open,
-            on_message = self._on_message,
-            on_error   = self._on_error,
-            on_close   = self._on_close,
-        )
-        self._ws.run_forever(
-            ping_interval = 25,
-            ping_timeout  = 10,
-            sslopt        = {"check_hostname": True},
-        )
-
-    def _on_open(self, ws) -> None:
-        self.state.hub_connected = True
-        self._reconnect_count    = 0
-        if self.ranker_logger:
-            self.ranker_logger.connection_event("Hub WebSocket", "connected")
-        logger.info("[HubSubscriber] ✅ Connected to hub")
-
-    def _on_message(self, ws, raw: str) -> None:
-        try:
-            msg  = json.loads(raw)
-            mtype = msg.get("type", "")
-
-            if mtype == "initial_state":
-                snaps = msg.get("snapshots", {})
-                self.state.ingest_bulk(snaps)
-                logger.info(f"[HubSubscriber] Initial state: {len(snaps)} spaces")
-
-            elif mtype == "metrics_update":
-                space = msg.get("space_name")
-                snap  = msg.get("snapshot", {})
-                if space and snap:
-                    self.state.ingest_snapshot(space, snap)
-                    if self.ranker_logger:
-                        training = snap.get("training", {})
-                        voting = snap.get("voting", {})
-                        self.ranker_logger.hub_update(
-                            asset=space,
-                            avn_accuracy=training.get("avn_accuracy", 0.0),
-                            signal_confidence=(
-                                max(voting.get("buy_count", 0), voting.get("sell_count", 0)) /
-                                (voting.get("buy_count", 0) + voting.get("sell_count", 0) or 1)
-                            )
-                        )
-
-        except json.JSONDecodeError:
-            logger.warning("[HubSubscriber] Malformed JSON")
-        except Exception as e:
-            logger.error(f"[HubSubscriber] Message error: {e}")
-
-    def _on_error(self, ws, error) -> None:
-        logger.error(f"[HubSubscriber] WS error: {error}")
-        self.state.hub_connected = False
-        if self.ranker_logger:
-            self.ranker_logger.connection_event("Hub WebSocket", "error", str(error))
-
-    def _on_close(self, ws, code, msg) -> None:
-        self.state.hub_connected = False
-        logger.info(f"[HubSubscriber] Connection closed (code={code})")
-        if self.ranker_logger:
-            self.ranker_logger.connection_event("Hub WebSocket", "disconnected")
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 2 — IN-MEMORY STATE STORE (minimal for dashboard)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+class DashboardState:
+    """Thread-safe cache of everything the dashboard needs."""
+    
+    def __init__(self, history_len: int = _METRIC_HISTORY_LEN):
+        self._lock = threading.RLock()
+        self._history_len = history_len
+        
+        self.hub_connected = False
+        self.start_time = time.time()
+        self.messages_rx = 0
+        self.last_update_ts = 0.0
+        
+        self._rankings: List[dict] = []
+        self._snapshots: Dict[str, dict] = {}
+    
+    def get_state(self) -> dict:
+        with self._lock:
+            return {
+                "rankings": list(self._rankings),
+                "metric_history": {},
+                "health": self._health_snapshot(),
+                "timestamp": datetime.utcnow().isoformat() + "Z",
+            }
+    
+    def _health_snapshot(self) -> dict:
+        return {
+            "hub_connected": self.hub_connected,
+            "spaces_connected": len(self._snapshots),
+            "messages_rx": self.messages_rx,
+            "last_update_ts": self.last_update_ts,
+            "last_update_ago": round(time.time() - self.last_update_ts, 1) if self.last_update_ts else None,
+            "uptime_seconds": round(time.time() - self.start_time, 0),
+            "reconnect_count": 0,
+        }
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -479,25 +288,16 @@ class HubSubscriber:
 # ══════════════════════════════════════════════════════════════════════════════════════
 
 _state = DashboardState()
-_ranker_logger = RankerLogger(
-    buffer_size = 1000,
-    log_dir = "./ranker_logs",
-    on_event = None,
-)
-
-_trade_parser = TradeLogParser(log_dir="./ranker_logs")
-_trade_parser.start_background(interval=2.0)
+_trade_parser = TradeLogParser(log_dir=_LOG_DIR)
+_trade_parser.start_background()
 
 app = Flask(__name__)
 CORS(app)
 
-# Register ranker logs API
-init_ranker_logs_api(_ranker_logger)
-app.register_blueprint(ranker_logs_bp)
-
 
 @app.route("/")
 def index():
+    """Serve the dashboard HTML."""
     html_path = Path(_HTML_PATH)
     if html_path.exists():
         return send_from_directory(str(html_path.parent), html_path.name)
@@ -516,71 +316,134 @@ def api_state():
 
 @app.route("/api/rankings")
 def api_rankings():
-    return jsonify({"rankings": _state.get_rankings()})
-
-
-@app.route("/api/metrics/history")
-def api_metrics_history():
-    limit = int(request.args.get("limit", 100))
-    return jsonify(_state.get_metric_history(limit=limit))
+    """Get current rankings."""
+    return jsonify({"rankings": _state.get_state()["rankings"]})
 
 
 @app.route("/api/trades")
 def api_trades():
-    """
-    Returns open trades, recent closed trades, and summary stats.
-    Parsed live from ranker log files in ./ranker_logs/.
-    """
+    """Returns open trades, recent closed trades, and summary stats."""
     return jsonify(_trade_parser.get_state())
 
 
 @app.route("/api/trades/open")
 def api_trades_open():
-    return jsonify({"open": _trade_parser.get_state()["open"]})
+    """Get only open trades."""
+    state = _trade_parser.get_state()
+    return jsonify({"open": state["open"]})
 
 
 @app.route("/api/trades/closed")
 def api_trades_closed():
+    """Get closed trades and stats."""
     limit = int(request.args.get("limit", 50))
     state = _trade_parser.get_state()
-    return jsonify({"closed": state["closed"][:limit], "stats": state["stats"]})
+    return jsonify({
+        "closed": state["closed"][:limit],
+        "stats": state["stats"]
+    })
 
 
+@app.route("/api/ranker/logs/recent")
+def api_logs_recent():
+    """Get recent ranker logs (minimal implementation)."""
+    limit = int(request.args.get("limit", 50))
+    
+    # Read directly from log file
+    logs = []
+    pattern = str(Path(_LOG_DIR) / "*.log")
+    files = sorted(glob.glob(pattern))
+    
+    for fpath in files[-3:]:  # Last 3 log files
+        try:
+            with open(fpath, "r") as f:
+                lines = f.readlines()[-limit:]
+                for line in lines:
+                    # Parse log line into structured format
+                    match = re.match(r'\[([^\]]+)\] \| (\w+) \| (\w+) \| (.*)', line.strip())
+                    if match:
+                        logs.append({
+                            "ts": match.group(1),
+                            "level": match.group(2),
+                            "category": match.group(3),
+                            "message": match.group(4),
+                        })
+        except Exception:
+            pass
+    
+    return jsonify({"logs": logs[-limit:], "count": len(logs)})
+
+
+@app.route("/api/ranker/logs/stats")
+def api_logs_stats():
+    """Get log statistics."""
+    pattern = str(Path(_LOG_DIR) / "*.log")
+    files = glob.glob(pattern)
+    
+    total_lines = 0
+    for fpath in files:
+        try:
+            with open(fpath, "r") as f:
+                total_lines += sum(1 for _ in f)
+        except Exception:
+            pass
+    
+    return jsonify({
+        "total_events": total_lines,
+        "by_level": {"INFO": total_lines},
+        "by_category": {},
+        "by_asset": {},
+        "errors": {}
+    })
 
+
+@app.route("/api/health")
+def api_health():
+    """Service health check."""
     return jsonify({
         "service": "hub_dashboard_service",
-        "version": "v1.1",
+        "version": "v2.0",
+        "status": "running",
+        "log_dir": str(_LOG_DIR),
+        "log_files": len(glob.glob(str(Path(_LOG_DIR) / "*.log"))),
         "timestamp": datetime.utcnow().isoformat() + "Z",
-        **_state.get_health(),
+        **_state.get_state()["health"],
     })
 
 
-def _start_background_services() -> None:
-    sub = HubSubscriber(_state, hub_host=_HUB_HOST, ranker_logger=_ranker_logger)
-    sub.start()
+@app.errorhandler(404)
+def not_found(error):
+    return jsonify({"error": "Endpoint not found"}), 404
 
-    global _subscriber
-    _subscriber = sub
 
+@app.errorhandler(500)
+def internal_error(error):
+    return jsonify({"error": "Internal server error"}), 500
 
-_subscriber = None
 
-_start_background_services()
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 4 — MAIN
+# ══════════════════════════════════════════════════════════════════════════════════════
 
-logger.info("=" * 70)
-logger.info(f"K1RL QUASAR — Hub Dashboard Service  (v1.1 with Ranker Logs)")
-logger.info(f"Hub WebSocket      : wss://{_HUB_HOST}/ws/subscribe")
-logger.info(f"Dashboard HTML     : {_HTML_PATH}")
-logger.info(f"Service port       : {_DASHBOARD_PORT}")
-logger.info(f"Logs API           : /api/ranker/logs/*")
-logger.info(f"Log directory      : ./ranker_logs")
-logger.info("=" * 70)
+def main():
+    """Start the dashboard service."""
+    logger.info("=" * 70)
+    logger.info(f"K1RL QUASAR — Hub Dashboard Service  (v2.0 with Trade Log Parser)")
+    logger.info(f"Dashboard HTML     : {_HTML_PATH}")
+    logger.info(f"Log directory      : {_LOG_DIR}")
+    logger.info(f"Service port       : {_DASHBOARD_PORT}")
+    logger.info("=" * 70)
+    
+    # Ensure log directory exists
+    Path(_LOG_DIR).mkdir(parents=True, exist_ok=True)
+    
+    app.run(
+        host="0.0.0.0",
+        port=_DASHBOARD_PORT,
+        debug=False,
+        threaded=True,
+    )
 
 
 if __name__ == "__main__":
-    app.run(
-        host     = "0.0.0.0",
-        port     = _DASHBOARD_PORT,
-        debug    = False,
-        threaded = True,
-    )
+    main()