Upload 3 files

Quasar_axrvi_ranker.py

@@ -4613,7 +4613,10 @@ class QuasarAXRVIBridge:
         self.log_bridge:    Optional[object] = None
         if self.enable_logging:
             try:
-                self.ranker_logger = RankerLogger()
+                # FIX: use RANKER_LOG_DIR env var so the dashboard service can
+                # always locate the log files via the same variable.
+                _rl_log_dir = os.environ.get("RANKER_LOG_DIR", "./ranker_logs")
+                self.ranker_logger = RankerLogger(log_dir=_rl_log_dir)
                 self.log_bridge    = RankerLogBridge(self.ranker_logger)
             except Exception as e:
                 logger.warning(f"Structured logging unavailable: {e} — continuing without")

hub_dashboard_service.py

@@ -30,6 +30,15 @@ import websocket
 from flask import Flask, jsonify, request, send_from_directory
 from flask_cors import CORS
 
+# ── Ranker Logs Blueprint ─────────────────────────────────────────────────────
+# Import lazily so the service starts even if ranker_logs_api.py is absent.
+try:
+    from ranker_logs_api import ranker_logs_bp, init_ranker_logs_api as _init_bp
+    _RANKER_LOGS_BP_AVAILABLE = True
+except ImportError:
+    _RANKER_LOGS_BP_AVAILABLE = False
+    logger_import_warning = "ranker_logs_api.py not found — Blueprint endpoints disabled"
+
 # ── Logging ───────────────────────────────────────────────────────────────────────────
 logging.basicConfig(
     level=logging.INFO,
@@ -405,8 +414,158 @@ class TradeLogParser:
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 2 — DASHBOARD STATE & HUB SUBSCRIBER (unchanged from v2.0)
+# 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 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)\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()
+
+    # ── Internal helpers ───────────────────────────────────────────────────────
+
+    def _find_files(self) -> list:
+        candidate_dirs = [
+            self._log_dir,
+            str(Path(__file__).parent / "ranker_logs"),
+            "./ranker_logs",
+            str(Path.home() / "ranker_logs"),
+        ]
+        for cdir in candidate_dirs:
+            found = sorted(glob.glob(str(Path(cdir) / "*.log*")))
+            if found:
+                return found
+        return []
+
+    def _read_lines(self, n_tail: int = 500) -> list:
+        """Return up to n_tail most-recent lines across the 3 newest log files."""
+        files = self._find_files()
+        raw = []
+        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) -> dict | None:
+        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
+        # Build a minimal dict compatible with what the Blueprint's callers expect.
+        return {
+            "timestamp": ts_m.group(1),
+            "level":     level,
+            "category":  cat,
+            "message":   line.strip(),
+            "asset":     asset,
+            "data":      None,
+        }
+
+    # ── RankerLogger interface ─────────────────────────────────────────────────
+
+    def get_recent(self, n: int = 50, category: str | None = None) -> list:
+        entries = []
+        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 = []
+        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 = []
+        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             = 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):
+        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
+
+
+
 
 from dataclasses import dataclass, field
 
@@ -542,6 +701,15 @@ _hub_subscriber.start()
 app = Flask(__name__)
 CORS(app)
 
+# ── Register ranker logs Blueprint (powers /export, /clear, /asset/<a>, /level/<l>) ──
+if _RANKER_LOGS_BP_AVAILABLE:
+    _file_logger_adapter = FileBasedLoggerAdapter(log_dir=_LOG_DIR)
+    _init_bp(_file_logger_adapter)
+    app.register_blueprint(ranker_logs_bp)
+    logger.info("[Blueprint] ranker_logs_bp registered — all /api/ranker/logs/* endpoints active")
+else:
+    logger.warning("[Blueprint] ranker_logs_api.py not found — inline routes only")
+
 
 @app.route("/")
 def index():
@@ -592,206 +760,10 @@ def api_trades_closed():
     })
 
 
-@app.route("/api/ranker/logs/recent")
-def api_logs_recent():
-    """Get recent ranker logs — reads directly from log files.
-    v2.3 FIX: wrapped in try/except so internal errors return JSON (not HTML 500).
-    Also scans multiple candidate directories so relative vs absolute paths both work.
-    """
-    try:
-        limit    = int(request.args.get("limit", 50))
-        category = request.args.get("category")          # optional filter
-
-        logs = []
-
-        # ── Find log files — try several candidate directories ────────────────
-        # The ranker may write to ./ranker_logs/ (relative to its cwd) which
-        # resolves differently from the service's _LOG_DIR.  We try all candidates
-        # and use the first one that contains *.log* files.
-        candidate_dirs = [
-            _LOG_DIR,                                               # /app/ranker_logs
-            str(Path(__file__).parent / "ranker_logs"),             # next to this file
-            "./ranker_logs",                                        # cwd-relative
-            str(Path.home() / "ranker_logs"),                       # home dir
-        ]
-        files = []
-        found_dir = None
-        for cdir in candidate_dirs:
-            pattern = str(Path(cdir) / "*.log*")
-            found = sorted(glob.glob(pattern))
-            if found:
-                files = found
-                found_dir = cdir
-                logger.debug(f"[api_logs_recent] found {len(files)} log file(s) in {cdir}")
-                break
-
-        if not files:
-            searched = ", ".join(candidate_dirs)
-            return jsonify({
-                "logs":  [],
-                "count": 0,
-                "error": f"No *.log* files found. Searched: {searched}",
-            })
-
-        # Read enough lines from the tail of the 3 newest files
-        raw_lines = []
-        for fpath in files[-3:]:
-            try:
-                with open(fpath, "r", encoding="utf-8", errors="replace") as f:
-                    lines = f.readlines()
-                    raw_lines.extend(lines[-200:])
-            except OSError:
-                pass
-
-        # Newest first
-        raw_lines.reverse()
-
-        # Pre-compiled patterns
-        _CAT_RE      = re.compile(r'\|\s*(INFO|DEBUG|WARNING|ERROR|CRITICAL)\s*\|\s*([A-Z_]+)\s*\|')
-        _TRAINING_RE = re.compile(
-            r'step=(\d+)\s*\|\s*loss=([\d.]+)\s*\|\s*lr=([\d.eE+\-]+)\s*\|\s*assets=(\d+)'
-        )
-        _JSON_RE     = re.compile(r'(\{.*\})\s*$')
-        _ASSET_RE    = re.compile(r'\|\s*TRADE\s*\|\s*(\w+)\s*\|')
-
-        for line in raw_lines[:limit]:
-            if not line.strip():
-                continue
-            ts_match = re.search(r'\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\]', line)
-            if not ts_match:
-                continue
-
-            ts = ts_match.group(1)
-
-            # Extract level + category from "[ts] | LEVEL    | CATEGORY        | ..."
-            cat_m          = _CAT_RE.search(line)
-            level          = cat_m.group(1) if cat_m else "INFO"
-            category_field = cat_m.group(2).strip() if cat_m else ""
-
-            if category and category.upper() not in line.upper():
-                continue
-
-            # Try to extract asset name (present for SIGNAL / TRADE lines)
-            asset_m = _ASSET_RE.search(line)
-            asset   = asset_m.group(1) if asset_m else None
-
-            entry = {
-                "timestamp": ts,
-                "level":     level,
-                "category":  category_field,
-                "message":   line.strip(),
-                "asset":     asset,
-                "data":      None,
-            }
-
-            if category_field == "TRAINING":
-                tm = _TRAINING_RE.search(line)
-                if tm:
-                    entry["data"] = {
-                        "step":        int(tm.group(1)),
-                        "loss":        float(tm.group(2)),
-                        "lr":          float(tm.group(3)),
-                        "asset_count": int(tm.group(4)),
-                    }
-                else:
-                    # Fallback: JSON blob at end of line
-                    jm = _JSON_RE.search(line)
-                    if jm:
-                        try:
-                            blob = json.loads(jm.group(1))
-                            # Normalise key names so JS always sees step/loss/lr/asset_count
-                            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
-
-            logs.append(entry)
-
-        return jsonify({
-            "logs":      logs,
-            "count":     len(logs),
-            "log_dir":   found_dir,
-        })
-
-    except Exception as exc:
-        logger.exception(f"[api_logs_recent] unhandled error: {exc}")
-        return jsonify({"logs": [], "count": 0, "error": str(exc)}), 200
-
-
-@app.route("/api/ranker/logs/stats")
-def api_logs_stats():
-    """Return aggregate stats parsed from log files.
-    Previously missing endpoint — dashboard Logs tab called this and got 404.
-    """
-    try:
-        by_category: dict = {}
-        by_level:    dict = {}
-        by_asset:    dict = {}
-        errors:      dict = {}
-        total        = 0
-
-        candidate_dirs = [
-            _LOG_DIR,
-            str(Path(__file__).parent / "ranker_logs"),
-            "./ranker_logs",
-            str(Path.home() / "ranker_logs"),
-        ]
-        files = []
-        for cdir in candidate_dirs:
-            found = sorted(glob.glob(str(Path(cdir) / "*.log*")))
-            if found:
-                files = found
-                break
-
-        _CAT_RE   = re.compile(r'\|\s*(INFO|DEBUG|WARNING|ERROR|CRITICAL)\s*\|\s*([A-Z_]+)\s*\|')
-        _ASSET_RE = re.compile(r'\|\s*TRADE\s*\|\s*(\w+)\s*\|')
-
-        for fpath in files[-3:]:
-            try:
-                with open(fpath, "r", encoding="utf-8", errors="replace") as f:
-                    for line in f:
-                        if not line.strip():
-                            continue
-                        m = _CAT_RE.search(line)
-                        if not m:
-                            continue
-                        level = m.group(1)
-                        cat   = m.group(2).strip()
-                        by_level[level]    = by_level.get(level, 0) + 1
-                        by_category[cat]   = by_category.get(cat, 0) + 1
-                        total += 1
-                        if level in ("ERROR", "CRITICAL"):
-                            errors[cat] = errors.get(cat, 0) + 1
-                        am = _ASSET_RE.search(line)
-                        if am:
-                            a = am.group(1)
-                            by_asset[a] = by_asset.get(a, 0) + 1
-            except OSError:
-                pass
-
-        return jsonify({
-            "total_events": total,
-            "by_level":     by_level,
-            "by_category":  by_category,
-            "by_asset":     by_asset,
-            "errors":       errors,
-            "buffer_size":  total,
-            "buffer_capacity": total,
-        })
-    except Exception as 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
-
 
 @app.route("/api/health")
 def health():
-    return jsonify({"status": "ok", "version": "v2.3-fixed"})
+    return jsonify({"status": "ok", "version": "v2.4-fixed"})
 
 
 if __name__ == "__main__":

ranker_logs_api.py

@@ -12,10 +12,63 @@ from flask import Blueprint, jsonify, request, send_file
 from typing import Optional
 from pathlib import Path
 
+import re as _re
+import json as _json
+
 ranker_logs_bp = Blueprint("ranker_logs", __name__, url_prefix="/api/ranker/logs")
 
 _logger: Optional[object] = None
 
+# ── Training-metrics extractor ────────────────────────────────────────────────
+# Parses the structured training line emitted by ranker_logging.training_update():
+#   [ts] | DEBUG    | TRAINING        | step=1 | loss=0.1234 | lr=0.000100 | assets=7
+_TRAINING_RE = _re.compile(
+    r'step=(\d+)\s*\|\s*loss=([\d.]+)\s*\|\s*lr=([\d.eE+\-]+)\s*\|\s*assets=(\d+)'
+)
+_JSON_BLOB_RE = _re.compile(r'(\{.*\})\s*$')
+
+
+def _enrich_entry(entry: dict) -> dict:
+    """
+    Attach a parsed `data` dict to TRAINING log entries so the dashboard
+    KPI cards (Loss / LR / Step / Assets) always have structured values.
+    Works from both the pipe-delimited text and any trailing JSON metadata blob.
+    """
+    if entry.get("category", "").upper() != "TRAINING":
+        return entry
+    if entry.get("data"):          # already enriched (e.g. from in-memory logger)
+        return entry
+
+    msg = entry.get("message", "")
+
+    # 1. Try pipe-delimited format
+    m = _TRAINING_RE.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
+
+    # 2. Fallback: trailing JSON blob
+    jm = _JSON_BLOB_RE.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
+
 
 def init_ranker_logs_api(ranker_logger):
     """Call this from hub_dashboard_service.py during initialization."""
@@ -33,6 +86,7 @@ def get_recent_logs():
     category = request.args.get("category")
     
     entries = _logger.get_recent(n=limit, category=category)
+    entries = [_enrich_entry(e) for e in entries]
     return jsonify({
         "logs":  entries,
         "count": len(entries),