Upload hub_dashboard_service.py.py

hub_dashboard_service.py.py

@@ -0,0 +1,449 @@
+#!/usr/bin/env python3
+"""
+╔══════════════════════════════════════════════════════════════════════════════════════╗
+║  K1RL QUASAR — HUB DASHBOARD SERVICE (with Trade Log Parser)                        ║
+║  ──────────────────────────────────────────────────────────────────────────────────  ║
+║  Architecture role: READ-ONLY subscriber → serves dashboard UI                      ║
+║  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, defaultdict
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+
+import websocket
+from flask import Flask, jsonify, request, send_from_directory
+from flask_cors import CORS
+
+# ── Logging ───────────────────────────────────────────────────────────────────────────
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+    stream=sys.stdout,
+)
+logger = logging.getLogger("HubDashboardService")
+
+# ── Config ────────────────────────────────────────────────────────────────────────────
+_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(
+    "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 — TRADE LOG PARSER (FIXED for actual log format)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+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%
+    """
+
+    # 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,
+        }
+        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)
+
+    def refresh(self) -> None:
+        """Find all log files, read new lines since last position."""
+        pattern = str(self.log_dir / "*.log")
+        files = sorted(glob.glob(pattern))
+        
+        if not files:
+            # 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 _tail_file(self, fpath: str) -> None:
+        """Read only new bytes from fpath since last call."""
+        try:
+            size = os.path.getsize(fpath)
+        except OSError:
+            return
+
+        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 = 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 or trade_id.split('_')[0],
+                    "direction": direction,
+                    "entry": entry,
+                    "opened_at": ts,
+                    "status": "OPEN",
+                }
+                self._stats["total_opened"] += 1
+            
+            logger.debug(f"[TradeLogParser] OPEN: {trade_id} | {direction} @ {entry}")
+            return
+
+        # ── TRADE CLOSED ─────────────────────────────────────────────────────────
+        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:
+                # Find the matching open trade
+                trade = self._open.pop(trade_id, None)
+                
+                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.debug(f"[TradeLogParser] CLOSE: {trade_id} | 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,
+                "timestamp": datetime.utcnow().isoformat() + "Z",
+            }
+
+    def stop(self) -> None:
+        """Stop the background thread."""
+        self._running = False
+        if self._thread:
+            self._thread.join(timeout=5)
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# 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,
+        }
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 3 — FLASK APP
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+_state = DashboardState()
+_trade_parser = TradeLogParser(log_dir=_LOG_DIR)
+_trade_parser.start_background()
+
+app = Flask(__name__)
+CORS(app)
+
+
+@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)
+    return (
+        "<h1>hub_dashboard.html not found</h1>"
+        f"<p>Expected: <code>{_HTML_PATH}</code></p>",
+        404,
+    )
+
+
+@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()["rankings"]})
+
+
+@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({
+        "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": "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_state()["health"],
+    })
+
+
+@app.errorhandler(404)
+def not_found(error):
+    return jsonify({"error": "Endpoint not found"}), 404
+
+
+@app.errorhandler(500)
+def internal_error(error):
+    return jsonify({"error": "Internal server error"}), 500
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 4 — MAIN
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+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__":
+    main()