Update hub_dashboard_service.py

hub_dashboard_service.py

@@ -1,17 +1,18 @@
 #!/usr/bin/env python3
 """
 ╔══════════════════════════════════════════════════════════════════════════════════════╗
-║  K1RL QUASAR — HUB DASHBOARD SERVICE (FIXED v2.5)                                   ║
+║  K1RL QUASAR — HUB DASHBOARD SERVICE (with Trade Log Parser) — FIXED v2.4           ║
 ║  ──────────────────────────────────────────────────────────────────────────────────  ║
 ║  Architecture role: READ-ONLY subscriber → serves dashboard UI                      ║
-║  VERSION: v2.5 (FIXED Training Logs 404 + Category Filter)  |  2026-04-04          ║
+║  VERSION: v2.4 (SELF-CONTAINED — all /api/ranker/logs/* routes inline)              ║
 ║                                                                                      ║
-║  CRITICAL FIXES IN v2.5:                                                            ║
-║  ✅ FIX #5.1: Ensure log directory exists before _log_adapter initialization       ║
-║  ✅ FIX #5.2: Fix category filtering to use field match instead of substring       ║
-║  ✅ FIX #5.3: HTML now requests ?category=TRAINING for training logs               ║
-║  ✅ FIX #5.4: Validate Flask routes are registered before app.run()                ║
-║  ✅ FIX #5.5: Add explicit error logging for 404 debugging                         ║
+║  FIXES APPLIED:                                                                      ║
+║  ✅ FIX v2.4: All /api/ranker/logs/* routes moved inline (Blueprint removed)        ║
+║  ✅ FIX v2.4: Category filter uses exact field match (not substring)                ║
+║  ✅ FIX v2.4: Log directory auto-created on startup if missing                      ║
+║  ✅ FIX #1: Include rotated log files (*.log, *.log.1, *.log.2, etc.)              ║
+║  ✅ FIX #2: Improved regex to catch all trade close formats                        ║
+║  ✅ FIX #3: Added unrealized P&L tracking for open positions                       ║
 ╚══════════════════════════════════════════════════════════════════════════════════════╝
 """
 
@@ -51,39 +52,390 @@ _HTML_PATH = os.environ.get(
 _LOG_DIR = os.environ.get("RANKER_LOG_DIR", "/app/ranker_logs")
 _METRIC_HISTORY_LEN = int(os.environ.get("QUASAR_METRIC_HISTORY", "200"))
 
-# ╔═══════════════════════════════════════════════════════════════════════════════════╗
-# FIX v2.5.1: ENSURE LOG DIRECTORY EXISTS
-# ╚═══════════════════════════════════════════════════════════════════════════════════╝
-try:
-    log_dir_path = Path(_LOG_DIR)
-    log_dir_path.mkdir(parents=True, exist_ok=True)
-    logger.info(f"[FIX v2.5.1] ✅ Log directory ensured: {_LOG_DIR}")
-except Exception as e:
-    logger.error(f"[FIX v2.5.1] ❌ Failed to create log directory: {e}")
-    sys.exit(1)
 
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 1 — TRADE LOG PARSER (FIXED v2.2)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+class TradeLogParser:
+    """
+    Tails ranker log files and maintains open/closed trade state.
+    Runs in a background thread, refreshing every 2 seconds.
+    
+    FIXED v2.2:
+    ✅ FIX #1: Now reads *.log* pattern to include rotated files (.log.1, .log.2, etc.)
+    ✅ FIX #2: Improved regex to catch all trade close formats (normal, fallback, timeout)
+    ✅ FIX #3: Tracks unrealized P&L for open positions
+    
+    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%
+    [2026-03-30 16:20:45] | INFO | TRADE | V75 | Closed V75 (no-cid fallback) | reward=... | pnl=-2.0
+    [2026-03-30 16:20:50] | INFO | TRADE | CRASH500 | TRADE FORCE-CLOSED (timeout) | reward=... | profit=-1.5
+    """
+
+    # Regex patterns matching the actual log format from ranker_logging.py:
+    TRADE_OPEN_RE = re.compile(
+        r'TRADE OPENED \| ID=(\S+) \| Dir=(\w+) \| Entry=([\d.]+) \| Qty=([\d.]+)'
+    )
+    TRADE_OPEN_RE_NOQTY = re.compile(
+        r'TRADE OPENED \| ID=(\S+) \| Dir=(\w+) \| Entry=([\d.]+)'
+    )
+
+    # FIXED v2.2: Improved regex to catch ALL trade close formats
+    # Matches: "TRADE CLOSED | ID=xxx | pnl=X"
+    #          "no-cid fallback) | ... | pnl=X"
+    #          "FORCE-CLOSED (timeout) | ... | pnl=X"
+    #          "profit=X" (alternative field name)
+    TRADE_CLOSE_RE = re.compile(
+        r'(?:TRADE CLOSED|no-cid fallback|FORCE-CLOSED.*?timeout).*?(?:pnl|profit)=([+-]?[\d.]+)'
+    )
+    
+    # FIX v2.3: Dual-format regex for trade close lines that carry exit_price.
+    # Root cause (v2.2 bug): ranker_logging.trade_close() wrote exit_price ONLY
+    # into the trailing JSON metadata blob, e.g.:
+    #   TRADE CLOSED | ID=... | pnl=... | return=...% | {"exit_price": 4364.21}
+    # but this regex looked for it as a pipe-delimited text field:
+    #   TRADE CLOSED | ID=... | pnl=... | exit_price=...   <- never present
+    # so TRADE_CLOSE_RE_WITH_EXIT never matched and exit_price was always None.
+    #
+    # Fix has TWO parts:
+    #   1. ranker_logging.py now writes exit_price into the message text too.
+    #   2. This regex matches BOTH formats so old log files still parse correctly:
+    #      Group 3 - pipe-delimited text field  (new format, post-fix)
+    #      Group 4 - JSON metadata field        (old format, pre-fix)
+    TRADE_CLOSE_RE_WITH_EXIT = re.compile(
+        r'TRADE CLOSED \| ID=(\S+) \| pnl=([+-]?[\d.]+)'
+        r'.*?(?:\| exit_price=([\d.]+)|"exit_price":\s*([\d.]+))'
+    )
+
+    # Fallback for Line 1: pnl + return%, no exit_price
+    TRADE_CLOSE_RE_STRICT = re.compile(
+        r'TRADE CLOSED \| ID=(\S+) \| pnl=([+-]?[\d.]+)'
+    )
+    
+    TRADE_CLOSE_RE_FALLBACK = re.compile(
+        r'no-cid fallback.*?pnl=([+-]?[\d.]+)'
+    )
+    
+    TRADE_CLOSE_RE_TIMEOUT = re.compile(
+        r'FORCE-CLOSED.*?timeout.*?(?:pnl|profit)=([+-]?[\d.]+)'
+    )
+
+    # Safety-net: Rotation close lines appear BEFORE the TRADE CLOSED lines and
+    # contain the underlying asset exit price.  Format (from ranker log):
+    #   [Rotation] 📤 Closing V75 — no longer in top-3 | price=33999.8690 | trade_id=V75_xxx
+    # We store this price on the open-trade record so it's available when the
+    # TRADE CLOSED line arrives.  If Line 2 (exit_price=) also arrives, it takes
+    # precedence (same value, but more authoritative).
+    ROTATION_CLOSE_RE = re.compile(
+        r'\[Rotation\].*?Closing\b.*?\|\s*price=([\d.]+).*?\|\s*trade_id=(\S+)'
+    )
+
+    # Asset sits between the 4th and 5th pipe-separated fields:
+    # "[ts] | LEVEL    | TRADE           | <ASSET> | ..."
+    TRADE_ASSET_RE = re.compile(r'\|\s*TRADE\s*\|\s*(\w+)\s*\|')
+
+    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,
+            "unrealized_pnl": 0.0,  # NEW: Track unrealized P&L from open positions
+        }
+        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:
+        """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)
 
-# [Previous SECTION 1 — TRADE LOG PARSER code remains unchanged...]
-# [Previous SECTION 2a — HUB SUBSCRIBER CLIENT code remains unchanged...]
-# [Previous SECTION 2b — FILE-BASED LOGGER ADAPTER code with FIX below...]
+    def refresh(self) -> None:
+        """
+        Find all log files, read new lines since last position.
+        
+        FIXED v2.2: Now uses *.log* pattern to include rotated files.
+        On first call for each file, always scan from the beginning so trades
+        that were written before the service started are not missed.
+        """
+        # FIX #1: Changed from "*.log" to "*.log*" to include rotated files
+        pattern = str(self.log_dir / "*.log*")
+        files = sorted(glob.glob(pattern))
+        
+        if not files:
+            # Also check for .txt files as fallback
+            pattern = str(self.log_dir / "*.txt")
+            files = sorted(glob.glob(pattern))
+        
+        for fpath in files:
+            self._tail_file(fpath)
+
+    def _tail_file(self, fpath: str) -> None:
+        """Read only new bytes from fpath since last call.
+        First encounter: start from byte 0 (full scan) so pre-existing trades are loaded."""
+        try:
+            size = os.path.getsize(fpath)
+        except OSError:
+            return
+
+        # Use 0 as default so a file seen for the first time is fully scanned
+        last = self._last_pos.get(fpath, 0)
+        if size <= last:
+            return
+
+        try:
+            with open(fpath, "r", encoding="utf-8", errors="replace") as f:
+                f.seek(last)
+                new_lines = f.readlines()
+                self._last_pos[fpath] = f.tell()
+        except OSError:
+            return
+
+        for line in new_lines:
+            self._parse_line(line)
+
+    def _parse_line(self, line: str) -> None:
+        """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:
+            trade_id, direction, entry, qty = m.group(1), m.group(2), float(m.group(3)), float(m.group(4))
+        else:
+            m2 = self.TRADE_OPEN_RE_NOQTY.search(line)
+            if m2:
+                trade_id, direction, entry, qty = m2.group(1), m2.group(2), float(m2.group(3)), 0.0
+            else:
+                m2 = None
+            m = m2  # unify the branch below
+
+        if m:
+            direction = direction.upper()
+            ts = self._parse_timestamp(line)
+
+            with self._lock:
+                self._open[trade_id] = {
+                    "trade_id": trade_id,
+                    "asset": asset or trade_id.split('_')[0],
+                    "direction": direction,
+                    "entry": entry,
+                    "qty": qty,
+                    "opened_at": ts,
+                    "status": "OPEN",
+                }
+                self._stats["total_opened"] += 1
+
+            logger.debug(f"[TradeLogParser] OPEN: {trade_id} | {direction} @ {entry} qty={qty}")
+            return
+
+        # ── TRADE CLOSED ─────────────────────────────────────────────────────────
+        # Try Line-2 format first (has exit_price — dedicated regex, no ambiguity)
+        m2e = self.TRADE_CLOSE_RE_WITH_EXIT.search(line)
+        pnl = None
+        trade_id = None
+        _exit_price = None
+
+        if m2e:
+            trade_id    = m2e.group(1)
+            pnl         = float(m2e.group(2))
+            # Group 3 = pipe-delimited "| exit_price=..." (new format, post-fix)
+            # Group 4 = JSON metadata   '"exit_price": ...' (old format, pre-fix)
+            _exit_price = float(m2e.group(3) or m2e.group(4))
+            logger.debug(
+                f"[TradeLogParser] Matched CLOSE+EXIT: {trade_id} "
+                f"pnl={pnl} exit_price={_exit_price}"
+            )
+        else:
+            # Try Line-1 format (pnl + return%, no exit_price)
+            m = self.TRADE_CLOSE_RE_STRICT.search(line)
+            if m:
+                trade_id = m.group(1)
+                pnl      = float(m.group(2))
+                logger.debug(f"[TradeLogParser] Matched CLOSE(no exit): {trade_id} pnl={pnl}")
+            else:
+                # Try fallback format (no-cid)
+                m = self.TRADE_CLOSE_RE_FALLBACK.search(line)
+                if m:
+                    pnl = float(m.group(1))
+                    logger.debug(f"[TradeLogParser] Matched FALLBACK close: pnl={pnl}")
+                else:
+                    # Try timeout format
+                    m = self.TRADE_CLOSE_RE_TIMEOUT.search(line)
+                    if m:
+                        pnl = float(m.group(1))
+                        logger.debug(f"[TradeLogParser] Matched TIMEOUT close: pnl={pnl}")
+        
+        # If we found a PnL value (any format), log the closed trade
+        # _exit_price is set in every branch above that sets pnl; default None otherwise
+        if pnl is not None:
+            ts = self._parse_timestamp(line)
+            
+            with self._lock:
+                # FIX 4: The bot emits TWO "TRADE CLOSED" lines per close event:
+                #   Line 1 — has pnl + return but NO exit_price  (matched first)
+                #   Line 2 — has pnl + exit_price + status + contract_id
+                # Previously Line 1 created the closed record (exit_price=None) and
+                # Line 2 created a duplicate with exit_price but no direction/entry.
+                # Fix: if a closed record with the same trade_id already exists,
+                # just patch its exit_price in-place and skip re-appending.
+                if trade_id:
+                    existing_idx = next(
+                        (i for i, t in enumerate(self._closed)
+                         if t.get("trade_id") == trade_id),
+                        None
+                    )
+                    if existing_idx is not None:
+                        # Second log line for the same close — update exit_price if we
+                        # now have it, then stop (don't double-count stats).
+                        if _exit_price is not None:
+                            self._closed[existing_idx]["exit_price"] = _exit_price
+                            logger.debug(
+                                f"[TradeLogParser] CLOSE patch exit_price: "
+                                f"{trade_id} exit_price={_exit_price}"
+                            )
+                        return
+
+                # Try to find the matching open trade by trade_id if available
+                if trade_id:
+                    trade = self._open.pop(trade_id, None)
+                else:
+                    # Fallback: unknown trade_id (from fallback/timeout path)
+                    trade = None
+                
+                closed = {
+                    "trade_id": trade_id or "UNKNOWN",
+                    "asset": asset or (trade.get("asset") if trade else "?"),
+                    "pnl": pnl,
+                    "closed_at": ts,
+                    "status": "CLOSED",
+                    "exit_price": _exit_price,
+                }
+                
+                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.debug(f"[TradeLogParser] CLOSE: {trade_id or '?'} | pnl={pnl:+.2f}")
+            return
+
+    @staticmethod
+    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]
+
+    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,
+            }
+
+    def update_unrealized_pnl(self, unrealized_dict: Dict[str, float]) -> None:
+        """
+        FIX #3: Update unrealized P&L for open positions from external source (WebSocket price feed).
+        Call this every tick when you have current market prices.
+        
+        Args:
+            unrealized_dict: {trade_id: unrealized_pnl_value, ...}
+        """
+        with self._lock:
+            total_unrealized = sum(unrealized_dict.values())
+            self._stats["unrealized_pnl"] = total_unrealized
+            
+            # Update individual open trade unrealized values
+            for trade_id, unrealized in unrealized_dict.items():
+                if trade_id in self._open:
+                    self._open[trade_id]["unrealized_pnl"] = unrealized
+
+    def stop(self) -> None:
+        self._running = False
+        if self._thread:
+            self._thread.join(timeout=3)
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 2b — FILE-BASED LOGGER ADAPTER
+# ═══════════════════════════════════════════════════════════════════════════════���══════
+# The ranker_logs_api Blueprint expects a RankerLogger-style object with get_recent(),
+# get_by_asset(), get_by_level(), get_stats(), export_json(), and clear_buffer().
+# This adapter satisfies that interface by reading from the same log FILES that the
+# TradeLogParser uses — no in-memory ranker process required in the dashboard service.
 
 class FileBasedLoggerAdapter:
     """
-    Implements the RankerLogger interface expected by /api/ranker/logs/* routes.
-    Reads from disk log files instead of an in-memory buffer.
-    
-    FIX v2.5.2: Category filtering now uses field match instead of substring match
+    Implements the RankerLogger interface expected by ranker_logs_api.py Blueprint,
+    but reads from disk log files instead of an in-memory buffer.
+    This lets the dashboard service power ALL Blueprint endpoints without needing a
+    live RankerLogger instance.
     """
 
     # ── Shared compiled patterns ───────────────────────────────────────────────
     _CAT_RE      = re.compile(r'\|\s*(INFO|DEBUG|WARNING|ERROR|CRITICAL)\s*\|\s*([A-Z_]+)\s*\|')
-    _ASSET_RE    = re.compile(r'\|\s*(?:TRADE|SIGNAL|TRAINING)\s*\|\s*(\w+)\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 = _LOG_DIR):
         self._log_dir = log_dir
         self._lock = threading.RLock()
-        logger.info(f"[FileBasedLoggerAdapter] Initialized with log_dir={log_dir}")
+        # FIX v2.4 Bug#3: Auto-create log directory so ranker can start writing without a manual mkdir
+        try:
+            Path(self._log_dir).mkdir(parents=True, exist_ok=True)
+        except OSError as _mkdir_err:
+            logger.warning(f"[FileBasedLoggerAdapter] Could not create log dir {self._log_dir}: {_mkdir_err}")
 
     # ── Internal helpers ───────────────────────────────────────────────────────
 
@@ -97,9 +449,7 @@ class FileBasedLoggerAdapter:
         for cdir in candidate_dirs:
             found = sorted(glob.glob(str(Path(cdir) / "*.log*")))
             if found:
-                logger.debug(f"[_find_files] Found {len(found)} log files in {cdir}")
                 return found
-        logger.warning(f"[_find_files] ⚠️  No log files found in any candidate directory")
         return []
 
     def _read_lines(self, n_tail: int = 500) -> list:
@@ -109,11 +459,8 @@ class FileBasedLoggerAdapter:
         for fpath in files[-3:]:
             try:
                 with open(fpath, "r", encoding="utf-8", errors="replace") as f:
-                    lines = f.readlines()
-                    raw.extend(lines[-n_tail:])
-                    logger.debug(f"[_read_lines] Read {len(lines)} lines from {fpath}")
-            except OSError as e:
-                logger.error(f"[_read_lines] Error reading {fpath}: {e}")
+                    raw.extend(f.readlines()[-n_tail:])
+            except OSError:
                 pass
         raw.reverse()  # newest first
         return raw
@@ -127,9 +474,9 @@ class FileBasedLoggerAdapter:
         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
+        # Build a minimal dict compatible with what the Blueprint's callers expect.
         return {
             "timestamp": ts_m.group(1),
-            "ts": datetime.strptime(ts_m.group(1), "%Y-%m-%d %H:%M:%S").timestamp(),
             "level":     level,
             "category":  cat,
             "message":   line.strip(),
@@ -140,27 +487,17 @@ class FileBasedLoggerAdapter:
     # ── RankerLogger interface ─────────────────────────────────────────────────
 
     def get_recent(self, n: int = 50, category: Optional[str] = None) -> list:
-        """
-        FIX v2.5.2: Changed category filtering from substring match to field match
-        BEFORE: if category and category.upper() not in line.upper():
-        AFTER:  if category and entry.get("category", "").upper() != category.upper():
-        """
         entries = []
         for line in self._read_lines(n_tail=max(n * 3, 200)):
             e = self._line_to_entry(line)
             if e is None:
                 continue
-            
-            # FIX v2.5.2: Use field match instead of substring match
-            if category:
-                if e.get("category", "").upper() != category.upper():
-                    continue
-            
+            # FIX v2.4 Bug#4: Use exact category field match, not substring search on raw line
+            if category and e.get("category", "").upper() != category.upper():
+                continue
             entries.append(e)
             if len(entries) >= n:
                 break
-        
-        logger.debug(f"[get_recent] Returned {len(entries)} entries (category={category})")
         return entries
 
     def get_by_asset(self, asset: str, n: int = 30) -> list:
@@ -171,75 +508,212 @@ class FileBasedLoggerAdapter:
             e = self._line_to_entry(line)
             if e:
                 entries.append(e)
-                if len(entries) >= n:
-                    break
+            if len(entries) >= n:
+                break
         return entries
 
     def get_by_level(self, level: str, n: int = 50) -> list:
         entries = []
         for line in self._read_lines(n_tail=500):
-            if level.upper() not in line.upper():
-                continue
             e = self._line_to_entry(line)
-            if e:
+            if e and e["level"].upper() == level.upper():
                 entries.append(e)
-                if len(entries) >= n:
-                    break
+            if len(entries) >= n:
+                break
         return entries
 
     def get_stats(self) -> dict:
-        """Aggregate statistics from all logs."""
-        stats = {
-            "total_events": 0,
-            "by_level": defaultdict(int),
-            "by_category": defaultdict(int),
-            "by_asset": defaultdict(int),
-            "errors": defaultdict(int),
-        }
-        for line in self._read_lines(n_tail=1000):
+        by_category: dict = {}
+        by_level: dict    = {}
+        by_asset: dict    = {}
+        errors: dict      = {}
+        total             = 0
+        for line in self._read_lines(n_tail=2000):
             e = self._line_to_entry(line)
-            if e:
-                stats["total_events"] += 1
-                stats["by_level"][e["level"]] += 1
-                stats["by_category"][e["category"]] += 1
-                if e["asset"]:
-                    stats["by_asset"][e["asset"]] += 1
-                if e["level"] in ("ERROR", "CRITICAL"):
-                    stats["errors"][e["level"]] += 1
-        return {k: (dict(v) if isinstance(v, defaultdict) else v) for k, v in stats.items()}
-
-    def export_json(self, filepath: str, n: int = 500) -> None:
-        """Export recent logs as JSON."""
-        entries = self.get_recent(n=n)
-        with open(filepath, "w") as f:
-            json.dump(entries, f, indent=2)
-        logger.info(f"[export_json] Exported {len(entries)} entries to {filepath}")
+            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 clear_buffer(self) -> None:
-        """No-op for file-based adapter."""
+    def export_json(self, filepath: str, n: int = 500):
+        import json as _json
+        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):
+        # File-based adapter has no in-memory buffer to clear.
+        # No-op — files are managed by the ranker process itself.
         pass
 
 
-# [Previous SECTION 3 — STATE MANAGEMENT code remains unchanged...]
+
+
+from dataclasses import dataclass, field
+
+@dataclass
+class AssetSnapshot:
+    space_name: str
+    signal: float = 0.0
+    confidence: float = 0.0
+    last_updated: float = 0.0
+
+class DashboardState:
+    """Centralized state for the dashboard — collects snapshots from hub."""
+    
+    def __init__(self):
+        self._snapshots: Dict[str, AssetSnapshot] = {}
+        self._lock = threading.RLock()
+
+    def update_from_snapshot(self, space_name: str, snap_dict: dict) -> None:
+        with self._lock:
+            if space_name not in self._snapshots:
+                self._snapshots[space_name] = AssetSnapshot(space_name=space_name)
+            snap = self._snapshots[space_name]
+            snap.signal = snap_dict.get("signal", 0.0)
+            snap.confidence = snap_dict.get("confidence", 0.0)
+            snap.last_updated = snap_dict.get("last_updated", time.time())
+
+    def get_state(self) -> dict:
+        with self._lock:
+            snaps = [
+                {
+                    "space_name": s.space_name,
+                    "signal": round(s.signal, 4),
+                    "confidence": round(s.confidence, 4),
+                }
+                for s in self._snapshots.values()
+            ]
+            return {"snapshots": snaps}
+
+class HubSubscriberClient:
+    """Subscribes to the central hub for live metric updates."""
+    
+    def __init__(self, state: DashboardState):
+        self.state = state
+        self.hub_url = f"ws://{_HUB_HOST}:7860/ws/subscribe"
+        self._ws = None
+        self._running = False
+        self._thread = None
+        self._reconnect_count = 0
+        self._MAX_BACKOFF = 30
+
+    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 4 — FLASK APP INITIALIZATION (FIX v2.5.4)
+# SECTION 3 — FLASK APP
 # ══════════════════════════════════════════════════════════════════════════════════════
 
+_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)
 
 # ── Instantiate the file-based log adapter (used by all /api/ranker/logs/* routes) ──
 _log_adapter = FileBasedLoggerAdapter(log_dir=_LOG_DIR)
-logger.info(f"[APP_INIT] Flask app created with CORS enabled")
-logger.info(f"[APP_INIT] FileBasedLoggerAdapter initialized")
+
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 5 — RANKER LOG ROUTES (INLINE — NO BLUEPRINT DEPENDENCY)
+# SECTION 4 — RANKER LOG ROUTES  (self-contained — no Blueprint dependency)
 # ══════════════════════════════════════════════════════════════════════════════════════
 #
-# FIX v2.5: Routes are defined inline so service is fully self-contained.
-# All routes have explicit error logging for debugging (FIX v2.5.5).
+# FIX v2.4: These routes were previously delegated to ranker_logs_api.py Blueprint.
+# That Blueprint was never registered, so every /api/ranker/logs/* call returned 404.
+# Routes are now defined inline so hub_dashboard_service.py is fully self-contained.
+# FileBasedLoggerAdapter (above) satisfies the full RankerLogger interface by reading
+# the ranker's disk log files — no in-process ranker instance required.
 
 _TRAINING_RE_INLINE = re.compile(
     r'step=(\d+)\s*\|\s*loss=([\d.]+)\s*\|\s*lr=([\d.eE+\-]+)\s*\|\s*assets=(\d+)'
@@ -279,30 +753,21 @@ def _enrich_training(entry: dict) -> dict:
     return entry
 
 
-# ╔═══════════════════════════════════════════════════════════════════════════════════╗
-# FIX v2.5.5: ALL ROUTES HAVE EXPLICIT ERROR LOGGING FOR DEBUGGING
-# ╚═══════════════════════════════════════════════════════════════════════════════════╝
-
 @app.route("/api/ranker/logs/recent", methods=["GET"])
 def api_logs_recent():
     """GET /api/ranker/logs/recent?limit=50&category=TRAINING"""
     try:
         limit    = int(request.args.get("limit", 50))
         category = request.args.get("category")
-        logger.debug(f"[api_logs_recent] limit={limit}, category={category}")
-        
         entries  = _log_adapter.get_recent(n=limit, category=category)
         entries  = [_enrich_training(e) for e in entries]
-        
-        response = {
+        return jsonify({
             "logs":   entries,
             "count":  len(entries),
             "stats":  _log_adapter.get_stats(),
-        }
-        logger.debug(f"[api_logs_recent] ✅ Returned {len(entries)} logs")
-        return jsonify(response)
+        })
     except Exception as exc:
-        logger.exception(f"[api_logs_recent] ❌ ERROR: {exc}")
+        logger.exception(f"[api_logs_recent] error: {exc}")
         return jsonify({"logs": [], "count": 0, "error": str(exc)}), 200
 
 
@@ -310,11 +775,9 @@ def api_logs_recent():
 def api_logs_stats():
     """GET /api/ranker/logs/stats"""
     try:
-        stats = _log_adapter.get_stats()
-        logger.debug(f"[api_logs_stats] ✅ Returned stats")
-        return jsonify(stats)
+        return jsonify(_log_adapter.get_stats())
     except Exception as exc:
-        logger.exception(f"[api_logs_stats] ❌ ERROR: {exc}")
+        logger.exception(f"[api_logs_stats] error: {exc}")
         return jsonify({"total_events": 0, "by_level": {}, "by_category": {},
                         "by_asset": {}, "errors": {}, "error": str(exc)}), 200
 
@@ -325,10 +788,9 @@ def api_logs_asset(asset: str):
     try:
         limit   = int(request.args.get("limit", 30))
         entries = _log_adapter.get_by_asset(asset, n=limit)
-        logger.debug(f"[api_logs_asset] asset={asset}, limit={limit}, returned {len(entries)}")
         return jsonify({"asset": asset, "logs": entries, "count": len(entries)})
     except Exception as exc:
-        logger.exception(f"[api_logs_asset] ❌ ERROR: {exc}")
+        logger.exception(f"[api_logs_asset] error: {exc}")
         return jsonify({"asset": asset, "logs": [], "count": 0, "error": str(exc)}), 200
 
 
@@ -338,10 +800,9 @@ def api_logs_level(level: str):
     try:
         limit   = int(request.args.get("limit", 50))
         entries = _log_adapter.get_by_level(level, n=limit)
-        logger.debug(f"[api_logs_level] level={level}, limit={limit}, returned {len(entries)}")
         return jsonify({"level": level.upper(), "logs": entries, "count": len(entries)})
     except Exception as exc:
-        logger.exception(f"[api_logs_level] ❌ ERROR: {exc}")
+        logger.exception(f"[api_logs_level] error: {exc}")
         return jsonify({"level": level.upper(), "logs": [], "count": 0, "error": str(exc)}), 200
 
 
@@ -352,7 +813,6 @@ def api_logs_export():
         limit       = int(request.args.get("limit", 500))
         export_path = Path("/tmp/ranker_logs_export.json")
         _log_adapter.export_json(str(export_path), n=limit)
-        logger.info(f"[api_logs_export] ✅ Exported logs to {export_path}")
         return send_file(
             export_path,
             mimetype="application/json",
@@ -360,7 +820,7 @@ def api_logs_export():
             download_name="ranker_logs_export.json",
         )
     except Exception as exc:
-        logger.exception(f"[api_logs_export] ❌ ERROR: {exc}")
+        logger.exception(f"[api_logs_export] error: {exc}")
         return jsonify({"error": str(exc)}), 500
 
 
@@ -369,54 +829,11 @@ def api_logs_clear():
     """POST /api/ranker/logs/clear — no-op for file-based adapter"""
     try:
         _log_adapter.clear_buffer()
-        logger.info(f"[api_logs_clear] ✅ Cleared (no-op for file adapter)")
         return jsonify({"status": "cleared"})
     except Exception as exc:
-        logger.exception(f"[api_logs_clear] ❌ ERROR: {exc}")
         return jsonify({"error": str(exc)}), 500
 
 
-# ╔═══════════════════════════════════════════════════════════════════════════════════╗
-# FIX v2.5.4: EXPLICIT ROUTE REGISTRATION VALIDATION
-# ╚═══════════════════════════════════════════════════════════════════════════════════╝
-
-def _validate_routes():
-    """
-    Verify all /api/ranker/logs/* routes are properly registered.
-    Call this before app.run() to catch registration issues early.
-    """
-    required_routes = {
-        "/api/ranker/logs/recent": "GET",
-        "/api/ranker/logs/stats": "GET",
-        "/api/ranker/logs/asset/<asset>": "GET",
-        "/api/ranker/logs/level/<level>": "GET",
-        "/api/ranker/logs/export": "GET",
-        "/api/ranker/logs/clear": "POST",
-    }
-    
-    registered_routes = {}
-    for rule in app.url_map.iter_rules():
-        registered_routes[rule.rule] = list(rule.methods - {"HEAD", "OPTIONS"})
-    
-    logger.info("[_validate_routes] Checking route registration...")
-    all_registered = True
-    for route, expected_method in required_routes.items():
-        if route in registered_routes:
-            methods = registered_routes[route]
-            if expected_method in methods:
-                logger.info(f"  ✅ {route} [{expected_method}] is registered")
-            else:
-                logger.error(f"  ❌ {route} exists but {expected_method} is NOT registered (has {methods})")
-                all_registered = False
-        else:
-            logger.error(f"  ❌ {route} is NOT registered at all!")
-            all_registered = False
-    
-    if all_registered:
-        logger.info("[_validate_routes] ✅ All routes validated successfully")
-    else:
-        logger.error("[_validate_routes] ❌ Some routes are missing or misconfigured!")
-        sys.exit(1)
 
 
 @app.route("/")
@@ -432,32 +849,57 @@ def index():
     )
 
 
-@app.route("/api/health")
-def health():
+@app.route("/api/state")
+def api_state():
+    """Full dashboard state — polled by hub_dashboard.html every 2 s."""
+    return jsonify(_state.get_state())
+
+
+@app.route("/api/rankings")
+def api_rankings():
+    """Get current rankings."""
+    return jsonify({"rankings": _state.get_state()["snapshots"]})
+
+
+@app.route("/api/trades")
+def api_trades():
+    """Returns open trades, recent closed trades, and summary stats."""
+    return jsonify(_trade_parser.get_state())
+
+
+@app.route("/api/trades/open")
+def api_trades_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({
-        "status": "ok",
-        "version": "v2.5-fixed",
-        "log_dir": str(_LOG_DIR),
-        "log_files_found": len(_log_adapter._find_files()),
+        "closed": state["closed"][:limit],
+        "stats": state["stats"]
     })
 
 
+
+@app.route("/api/health")
+def health():
+    return jsonify({"status": "ok", "version": "v2.4-fixed"})
+
+
 if __name__ == "__main__":
-    logger.info("╔═════════════════════════════════════════════════════════════════════╗")
-    logger.info("║  K1RL QUASAR HUB DASHBOARD SERVICE v2.5 (FULLY FIXED)              ║")
-    logger.info("╚═════════════════════════════════════════════════════════════════════╝")
+    logger.info("=== K1RL QUASAR HUB DASHBOARD SERVICE v2.4 (SELF-CONTAINED) ===")
     logger.info(f"Dashboard: http://localhost:{_DASHBOARD_PORT}")
     logger.info(f"Log directory: {_LOG_DIR}")
-    logger.info("")
-    logger.info("✅ FIXES APPLIED IN v2.5:")
-    logger.info("  ✅ FIX v2.5.1: Log directory ensured before adapter initialization")
-    logger.info("  ✅ FIX v2.5.2: Category filtering fixed (field match vs substring)")
-    logger.info("  ✅ FIX v2.5.3: HTML requests ?category=TRAINING parameter")
-    logger.info("  ✅ FIX v2.5.4: Explicit route registration validation")
-    logger.info("  ✅ FIX v2.5.5: All routes have explicit error logging")
-    logger.info("")
-    
-    # FIX v2.5.4: Validate routes before starting
-    _validate_routes()
-    
+    logger.info("Fixes applied:")
+    logger.info("  ✅ FIX v2.4: All /api/ranker/logs/* routes inline — no Blueprint dependency")
+    logger.info("  ✅ FIX v2.4: Training KPI enrichment (_enrich_training) applied on /recent")
+    logger.info("  ✅ FIX #1: Now reading *.log* (includes rotated files)")
+    logger.info("  ✅ FIX #2: Improved regex to catch all trade close formats")
+    logger.info("  ✅ FIX #4: Dedicated exit_price regex (no optional-group ambiguity)")
+
     app.run(host="0.0.0.0", port=_DASHBOARD_PORT, debug=False, use_reloader=False)