#!/usr/bin/env python3 """ ╔══════════════════════════════════════════════════════════════════════════════════════╗ ║ K1RL QUASAR — HUB DASHBOARD SERVICE (with Trade Log Parser) — FIXED v2.6 ║ ║ ────────────────────────────────────────────────────────────────────────────────── ║ ║ Architecture role: READ-ONLY subscriber → serves dashboard UI ║ ║ VERSION: v2.6 (REALTIME SIGNALS) | 2026-04-26 ║ ║ ║ ║ v2.6 — Wires the dashboard to the hub's actual broadcast contracts: ║ ║ ✅ MetricsSubscriber → /ws/subscribe — properly unwraps msg["asset"]["snapshot"] ║ ║ and reads voting.flip_direction / buy_count / sell_count / last_price plus ║ ║ training.* fields. Previously v2.5 read flat keys that the hub never emits, ║ ║ so AssetSnapshot defaulted to zero on every message. ║ ║ ✅ SignalSubscriber → /ws/signals — NEW. Consumes the per-tick realtime channel ║ ║ (signal_snapshot + signal_delta), drops out-of-order seq, stores the latest ║ ║ {action, price, ts, seq, source} per asset. ║ ║ ✅ DashboardState.get_state() merges both streams. flip_direction reflects the ║ ║ fresh realtime tick (≤1 s old) so the BUY-after-SELL-streak case is visible ║ ║ within ~30 ms of the V75 tick — same latency budget as the ranker. Falls back ║ ║ to the cumulative voting direction when the realtime stream is silent/stale. ║ ║ ✅ Full per-asset payload (buy/sell counts, training_steps, actor/critic/avn ║ ║ loss, avn_accuracy, signal_confidence, score) so the HTML table populates ║ ║ instead of rendering empty cells. ║ ║ ✅ Backward-compat alias: HubSubscriberClient = MetricsSubscriberClient. ║ ║ ║ ║ Carried over from v2.5: ║ ║ ✅ Default port 8051→7860 (HF Spaces only exposes port 7860) ║ ║ ✅ WebSocket ws://host:7860 → wss://host (HF Spaces TLS proxy) ║ ║ ✅ _find_files() searches all likely HF Spaces log paths ║ ║ ✅ /api/debug endpoint for live diagnostics ║ ║ ✅ All /api/ranker/logs/* routes inline — no Blueprint dependency ║ ║ ✅ Training KPI enrichment (_enrich_training) applied on /recent ║ ║ ✅ Rotated log files (*.log, *.log.1, *.log.2, etc.) included ║ ║ ✅ Improved regex to catch all trade close formats ║ ║ ✅ Unrealized P&L tracking for open positions ║ ║ ║ ║ DEPLOYMENT: Just restart the service — routes are already inline in this file. ║ ╚══════════════════════════════════════════════════════════════════════════════════════╝ """ 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, send_file 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", "7860")) _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 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 | | ..." 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) 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 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, # env-configured (default /app/ranker_logs) str(Path(__file__).parent / "ranker_logs"), # alongside this script "./ranker_logs", # cwd "/app/ranker_logs", # HF Spaces default app dir "/home/user/ranker_logs", # HF Spaces user home "/tmp/ranker_logs", # fallback tmp str(Path.home() / "ranker_logs"), # user home str(Path(__file__).parent), # script dir itself (*.log*) ] # Deduplicate while preserving order seen = set() unique_dirs = [] for d in candidate_dirs: if d not in seen: seen.add(d) unique_dirs.append(d) all_files = [] for cdir in unique_dirs: found = sorted(glob.glob(str(Path(cdir) / "*.log*"))) if found: all_files.extend(found) # Also check for any .log* directly in /app and /home/user for root_dir in ["/app", "/home/user", "."]: found = sorted(glob.glob(str(Path(root_dir) / "*.log*"))) all_files.extend(found) # Deduplicate file list seen_files = set() unique_files = [] for f in all_files: if f not in seen_files: seen_files.add(f) unique_files.append(f) return sorted(unique_files) 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) -> Optional[dict]: ts_m = self._TS_RE.search(line) if not ts_m: return None cat_m = self._CAT_RE.search(line) level = cat_m.group(1) if cat_m else "INFO" cat = cat_m.group(2).strip() if cat_m else "" ast_m = self._ASSET_RE.search(line) asset = ast_m.group(1) if ast_m else None # 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: Optional[str] = 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 # ══════════════════════════════════════════════════════════════════════════════════════ # v2.6 — REWRITTEN to match the hub's actual broadcast contracts. # # What was broken in v2.5: # 1. AssetSnapshot read flat `signal` / `confidence` keys that the hub never emits. # The hub emits `voting.flip_direction`, `voting.buy_count`, `voting.sell_count` # inside the snapshot, plus a derived `metadata.scores.blended_confidence`. # 2. _on_message read `msg.get("space_name")` and `msg.get("snapshot")` flat, but # the hub wraps both under `msg["asset"]`. Same for `initial_state`, which uses # the key `assets` (dict of space_name → {metadata, snapshot}), not `snapshots`. # 3. There was no subscriber for /ws/signals at all, so the per-tick realtime # action emitted by the hub's signal broadcaster never reached the dashboard. # # What v2.6 does: # • MetricsSubscriberClient → /ws/subscribe — properly unwraps `msg["asset"]`, # reads `voting.flip_direction` / `voting.buy_count` / `voting.sell_count` / # `voting.last_price`, plus `training.*` for the table columns the HTML reads. # • SignalSubscriberClient → /ws/signals — handles signal_snapshot + signal_delta, # drops out-of-order messages by per-asset `seq`, stores the realtime action # keyed by space_name. # • DashboardState.get_state() merges both streams. For each asset it emits: # - flip_direction: realtime per-tick action when fresh, else cumulative # (HTML's vecOf reads this first → realtime wins display) # - latest_signal: pure realtime action (forward-compat field) # - cumulative_flip_direction: voting.flip_direction (diagnostics) # - signal_confidence: blended_confidence (HTML "Engaged" badge + Certainty) # - buy_count / sell_count / training_steps / actor_loss / critic_loss / # avn_accuracy / score / last_updated (everything the HTML table reads) # ══════════════════════════════════════════════════════════════════════════════════════ # A realtime tick older than this is treated as stale — the cumulative direction # wins instead. 1 s is tight: a tick has to land in the second before the /api/state # poll for the realtime override to fire, otherwise the badge falls back to the # cumulative voting direction. Reflects the design choice that the dashboard should # only paint a realtime BUY/SELL when the asset is *actively* ticking. _REALTIME_SIGNAL_FRESH_SEC = 1.0 @dataclass class AssetSnapshot: """Per-asset cumulative state, derived from /ws/subscribe `metrics_update`.""" space_name: str # Voting (cumulative) flip_direction: str = "NONE" # "BUY" | "SELL" | "NONE" flip_action: str = "HOLD" buy_count: int = 0 sell_count: int = 0 last_price: float = 0.0 # Confidence scores (from hub's signal_metadata) vote_confidence: float = 0.0 train_confidence: float = 0.0 blended_confidence: float = 0.0 # Training/learning fields used by the HTML table & detail panel training_steps: int = 0 actor_loss: float = 0.0 critic_loss: float = 0.0 avn_accuracy: float = 0.0 avn_loss: float = 0.0 # Composite score (HTML reads `r.score` for the bar chart) score: float = 0.0 # Bookkeeping last_updated: float = 0.0 @dataclass class RealtimeSignal: """Per-asset realtime per-tick state, from /ws/signals.""" action: str = "NONE" # "BUY" | "SELL" | "HOLD" | "NONE" price: float = 0.0 seq: int = 0 ts: float = 0.0 source: str = "" class DashboardState: """Centralized state — merges /ws/subscribe metrics with /ws/signals realtime.""" def __init__(self): self._snapshots: Dict[str, AssetSnapshot] = {} self._signals: Dict[str, RealtimeSignal] = {} self._lock = threading.RLock() # ── Writers ──────────────────────────────────────────────────────────────── def update_from_metrics( self, space_name: str, snapshot: dict, metadata: Optional[dict] = None, ) -> None: """ Apply a metrics_update payload from /ws/subscribe. Reads from the snapshot's actual structure (voting.*, training.*) plus the pre-computed `metadata.scores` block the hub attaches. """ if not space_name or not snapshot: return metadata = metadata or {} voting = snapshot.get("voting", {}) or {} training = snapshot.get("training", {}) or {} scores = (metadata.get("scores") or {}) with self._lock: snap = self._snapshots.get(space_name) if snap is None: snap = AssetSnapshot(space_name=space_name) self._snapshots[space_name] = snap # Voting / cumulative direction snap.flip_direction = str(voting.get("flip_direction", "NONE")).upper() snap.flip_action = str(voting.get("flip_action", "HOLD")).upper() snap.buy_count = int(voting.get("buy_count", 0) or 0) snap.sell_count = int(voting.get("sell_count", 0) or 0) try: snap.last_price = float(voting.get("last_price", 0.0) or 0.0) except (TypeError, ValueError): snap.last_price = 0.0 # Confidence scores (prefer hub-derived metadata; fall back to local calc) snap.vote_confidence = float(scores.get("vote_confidence", 0.0) or 0.0) snap.train_confidence = float(scores.get("train_confidence", 0.0) or 0.0) snap.blended_confidence = float(scores.get("blended_confidence", 0.0) or 0.0) if snap.blended_confidence == 0.0 and snap.vote_confidence == 0.0: # Hub didn't attach metadata for some reason — derive locally. total_votes = snap.buy_count + snap.sell_count if total_votes > 0: snap.vote_confidence = max(snap.buy_count, snap.sell_count) / total_votes snap.train_confidence = float(training.get("avn_accuracy", 0.0) or 0.0) if snap.vote_confidence > 0 and snap.train_confidence > 0: snap.blended_confidence = (snap.vote_confidence + snap.train_confidence) / 2.0 else: snap.blended_confidence = snap.vote_confidence or snap.train_confidence # Training fields the HTML table & detail panel read snap.training_steps = int(training.get("training_steps", 0) or 0) snap.actor_loss = float(training.get("actor_loss", 0.0) or 0.0) snap.critic_loss = float(training.get("critic_loss", 0.0) or 0.0) snap.avn_accuracy = float(training.get("avn_accuracy", 0.0) or 0.0) snap.avn_loss = float(training.get("avn_loss", 0.0) or 0.0) # Composite score: prefer an explicit field if the hub ever supplies one, # otherwise fall back to signal_strength (already a [0,1] confidence-weighted # directional score — close enough for the bar chart's relative ranking). score = snapshot.get("score") if score is None: score = scores.get("signal_strength", 0.0) try: snap.score = float(score or 0.0) except (TypeError, ValueError): snap.score = 0.0 snap.last_updated = float( snapshot.get("last_updated") or time.time() ) def update_from_signal(self, signal: dict) -> None: """Apply a per-tick signal from /ws/signals. Drops out-of-order seq.""" asset = signal.get("asset") if not asset: return try: seq = int(signal.get("seq", 0) or 0) except (TypeError, ValueError): seq = 0 with self._lock: cur = self._signals.get(asset) if cur is not None and seq <= cur.seq: # Replay or out-of-order — ignore (matches ranker SignalSubscriber semantics) return try: price = float(signal.get("price", 0.0) or 0.0) except (TypeError, ValueError): price = 0.0 self._signals[asset] = RealtimeSignal( action = str(signal.get("action", "NONE")).upper(), price = price, seq = seq, ts = float(signal.get("ts") or time.time()), source = str(signal.get("source", "") or ""), ) # ── Reader ───────────────────────────────────────────────────────────────── def get_state(self) -> dict: """Merge cumulative + realtime views into the payload `/api/state` returns.""" now = time.time() with self._lock: snaps_out = [] for snap in self._snapshots.values(): rt = self._signals.get(snap.space_name) # Realtime per-tick action (may be empty if /ws/signals never fired) latest_signal = rt.action if rt else "" rt_fresh = ( rt is not None and rt.action in ("BUY", "SELL") and (now - rt.ts) <= _REALTIME_SIGNAL_FRESH_SEC ) # vecOf in the HTML reads flip_direction FIRST. Put the per-tick # action there when it's fresh and directional, so the dashboard # reflects the most recent signal within ~30 ms of the tick. # Otherwise fall back to the cumulative voting direction. display_direction = rt.action if rt_fresh else snap.flip_direction snaps_out.append({ "space_name": snap.space_name, "flip_direction": display_direction, "cumulative_flip_direction": snap.flip_direction, "latest_signal": latest_signal, "flip_action": snap.flip_action, "last_price": round(snap.last_price, 6), "buy_count": snap.buy_count, "sell_count": snap.sell_count, "signal_confidence": round(snap.blended_confidence, 4), "confidence": round(snap.blended_confidence, 4), "vote_confidence": round(snap.vote_confidence, 4), "train_confidence": round(snap.train_confidence, 4), "training_steps": snap.training_steps, "actor_loss": round(snap.actor_loss, 6), "critic_loss": round(snap.critic_loss, 6), "avn_loss": round(snap.avn_loss, 6), "avn_accuracy": round(snap.avn_accuracy, 4), "score": round(snap.score, 4), "last_updated": snap.last_updated, # Realtime diagnostics "realtime_seq": rt.seq if rt else 0, "realtime_ts": rt.ts if rt else 0.0, "realtime_source": rt.source if rt else "", }) return {"snapshots": snaps_out} # ══════════════════════════════════════════════════════════════════════════════════════ # WebSocket subscriber base — DRY shared reconnect/backoff loop # ══════════════════════════════════════════════════════════════════════════════════════ class _BaseSubscriber: """Shared reconnect loop for both metrics and signals subscribers.""" NAME = "Subscriber" MAX_BACKOFF = 30 def __init__(self, url: str): self.url = url self._ws = 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=self.NAME ) self._thread.start() logger.info(f"[{self.NAME}] Starting → {self.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"[{self.NAME}] error: {e}") if not self._running: break backoff = min(self.MAX_BACKOFF, 2 ** min(self._reconnect_count, 4)) logger.info( f"[{self.NAME}] 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.url, on_message = self._on_message, on_open = lambda ws: self._on_open(), on_error = lambda ws, e: logger.warning(f"[{self.NAME}] WS error: {e}"), on_close = lambda ws, c, m: logger.info(f"[{self.NAME}] closed code={c}"), ) self._ws.run_forever(ping_interval=30, ping_timeout=10, reconnect=0) def _on_open(self) -> None: self._reconnect_count = 0 logger.info(f"[{self.NAME}] ✅ Connected") def _on_message(self, ws, raw: str) -> None: # pragma: no cover — overridden raise NotImplementedError class MetricsSubscriberClient(_BaseSubscriber): """Subscribes to /ws/subscribe for cumulative per-asset snapshots.""" NAME = "MetricsSubscriber" def __init__(self, state: DashboardState): super().__init__(f"wss://{_HUB_HOST}/ws/subscribe") self.state = state def _on_message(self, ws, raw: str) -> None: try: msg = json.loads(raw) kind = msg.get("type", "") if kind == "metrics_update": # Hub format: {"type":"metrics_update","asset":{"space_name","metadata","snapshot"},...} asset = msg.get("asset") or {} space_name = asset.get("space_name", "") snapshot = asset.get("snapshot", {}) or {} metadata = asset.get("metadata", {}) or {} if space_name and snapshot: self.state.update_from_metrics(space_name, snapshot, metadata) elif kind == "initial_state": # Hub format: {"type":"initial_state","assets":{ "":{"metadata","snapshot"} },...} # NOTE: legacy key was "snapshots" — check both for safety across versions. assets = msg.get("assets") or msg.get("snapshots") or {} if isinstance(assets, dict): for space_name, payload in assets.items(): if not isinstance(payload, dict): continue # New format wraps under {metadata, snapshot}; old format was the snapshot itself. if "snapshot" in payload: self.state.update_from_metrics( space_name, payload.get("snapshot", {}) or {}, payload.get("metadata", {}) or {}, ) else: self.state.update_from_metrics(space_name, payload, {}) except Exception as e: logger.debug(f"[{self.NAME}] parse error: {e}") class SignalSubscriberClient(_BaseSubscriber): """Subscribes to /ws/signals for realtime per-tick BUY/SELL/HOLD actions.""" NAME = "SignalSubscriber" def __init__(self, state: DashboardState): super().__init__(f"wss://{_HUB_HOST}/ws/signals") self.state = state def _on_message(self, ws, raw: str) -> None: try: msg = json.loads(raw) kind = msg.get("type", "") if kind not in ("signal_snapshot", "signal_delta"): return signals = msg.get("signals") or [] if not isinstance(signals, list): return for sig in signals: if isinstance(sig, dict): self.state.update_from_signal(sig) except Exception as e: logger.debug(f"[{self.NAME}] parse error: {e}") # Backward-compat alias — anything that imported HubSubscriberClient from earlier # revisions of this file keeps working without touching its imports. HubSubscriberClient = MetricsSubscriberClient # ══════════════════════════════════════════════════════════════════════════════════════ # SECTION 3 — FLASK APP # ══════════════════════════════════════════════════════════════════════════════════════ _state = DashboardState() _trade_parser = TradeLogParser(log_dir=_LOG_DIR) _trade_parser.start_background() # Start hub subscribers — _state stays in sync with both cumulative metrics # (/ws/subscribe) and realtime per-tick signals (/ws/signals). Each runs in # its own daemon thread with independent reconnect/backoff. _metrics_subscriber = MetricsSubscriberClient(state=_state) _metrics_subscriber.start() _signal_subscriber = SignalSubscriberClient(state=_state) _signal_subscriber.start() # Backward-compat name in case anything else in the process imports this. _hub_subscriber = _metrics_subscriber 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) # ══════════════════════════════════════════════════════════════════════════════════════ # SECTION 4 — RANKER LOG ROUTES (self-contained — no Blueprint dependency) # ══════════════════════════════════════════════════════════════════════════════════════ # # 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+)' ) _JSON_BLOB_RE_INLINE = re.compile(r'(\{.*\})\s*$') def _enrich_training(entry: dict) -> dict: """Attach parsed `data` dict to TRAINING entries so dashboard KPI cards populate.""" if entry.get("category", "").upper() != "TRAINING": return entry if entry.get("data"): return entry msg = entry.get("message", "") m = _TRAINING_RE_INLINE.search(msg) if m: entry["data"] = { "step": int(m.group(1)), "loss": float(m.group(2)), "lr": float(m.group(3)), "asset_count": int(m.group(4)), } return entry jm = _JSON_BLOB_RE_INLINE.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 @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") entries = _log_adapter.get_recent(n=limit, category=category) entries = [_enrich_training(e) for e in entries] return jsonify({ "logs": entries, "count": len(entries), "stats": _log_adapter.get_stats(), }) except Exception as exc: logger.exception(f"[api_logs_recent] error: {exc}") return jsonify({"logs": [], "count": 0, "error": str(exc)}), 200 @app.route("/api/ranker/logs/stats", methods=["GET"]) def api_logs_stats(): """GET /api/ranker/logs/stats""" try: return jsonify(_log_adapter.get_stats()) 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/ranker/logs/asset/", methods=["GET"]) def api_logs_asset(asset: str): """GET /api/ranker/logs/asset/V75?limit=30""" try: limit = int(request.args.get("limit", 30)) entries = _log_adapter.get_by_asset(asset, n=limit) return jsonify({"asset": asset, "logs": entries, "count": len(entries)}) except Exception as exc: logger.exception(f"[api_logs_asset] error: {exc}") return jsonify({"asset": asset, "logs": [], "count": 0, "error": str(exc)}), 200 @app.route("/api/ranker/logs/level/", methods=["GET"]) def api_logs_level(level: str): """GET /api/ranker/logs/level/ERROR?limit=50""" try: limit = int(request.args.get("limit", 50)) entries = _log_adapter.get_by_level(level, n=limit) return jsonify({"level": level.upper(), "logs": entries, "count": len(entries)}) except Exception as exc: logger.exception(f"[api_logs_level] error: {exc}") return jsonify({"level": level.upper(), "logs": [], "count": 0, "error": str(exc)}), 200 @app.route("/api/ranker/logs/export", methods=["GET"]) def api_logs_export(): """GET /api/ranker/logs/export?limit=500 → download JSON""" try: limit = int(request.args.get("limit", 500)) export_path = Path("/tmp/ranker_logs_export.json") _log_adapter.export_json(str(export_path), n=limit) return send_file( export_path, mimetype="application/json", as_attachment=True, download_name="ranker_logs_export.json", ) except Exception as exc: logger.exception(f"[api_logs_export] error: {exc}") return jsonify({"error": str(exc)}), 500 @app.route("/api/ranker/logs/clear", methods=["POST"]) def api_logs_clear(): """POST /api/ranker/logs/clear — no-op for file-based adapter""" try: _log_adapter.clear_buffer() return jsonify({"status": "cleared"}) except Exception as exc: return jsonify({"error": str(exc)}), 500 @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 ( "

hub_dashboard.html not found

" f"

Expected: {_HTML_PATH}

", 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()["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({ "closed": state["closed"][:limit], "stats": state["stats"] }) @app.route("/api/health") def health(): return jsonify({"status": "ok", "version": "v2.6-realtime-signals"}) @app.route("/api/debug") def api_debug(): """GET /api/debug — diagnostics: log files found, paths searched, env vars.""" import glob as _glob candidate_dirs = [ _LOG_DIR, str(Path(__file__).parent / "ranker_logs"), "./ranker_logs", "/app/ranker_logs", "/home/user/ranker_logs", "/tmp/ranker_logs", str(Path.home() / "ranker_logs"), str(Path(__file__).parent), "/app", "/home/user", ".", ] dir_scan = {} for d in candidate_dirs: files = sorted(_glob.glob(str(Path(d) / "*.log*"))) dir_scan[d] = {"exists": Path(d).exists(), "log_files": files} adapter_files = _log_adapter._find_files() return jsonify({ "version": "v2.5-port-fix", "env": { "DASHBOARD_PORT": os.environ.get("DASHBOARD_PORT", "(not set, using 7860)"), "RANKER_LOG_DIR": os.environ.get("RANKER_LOG_DIR", "(not set, using /app/ranker_logs)"), "QUASAR_HUB_HOST": os.environ.get("QUASAR_HUB_HOST", "(not set)"), "cwd": str(Path.cwd()), "script_dir": str(Path(__file__).parent), }, "adapter_files_found": adapter_files, "directory_scan": dir_scan, "adapter_stats": _log_adapter.get_stats(), }) if __name__ == "__main__": logger.info("=== K1RL QUASAR HUB DASHBOARD SERVICE v2.6 (REALTIME SIGNALS) ===") logger.info(f"Dashboard port: {_DASHBOARD_PORT} (HF Spaces public port)") logger.info(f"Log directory: {_LOG_DIR}") logger.info(f"Hub host: {_HUB_HOST}") logger.info("v2.6 fixes:") logger.info(" ✅ MetricsSubscriber → /ws/subscribe (unwraps msg.asset, reads voting.*)") logger.info(" ✅ SignalSubscriber → /ws/signals (per-tick BUY/SELL within ~30 ms)") logger.info(" ✅ get_state() merges both streams; flip_direction reflects realtime tick") logger.info(" ✅ Full snapshot fields exposed (buy/sell counts, training, confidence)") logger.info("Carried over from v2.5:") logger.info(" ✅ port 7860 / wss:// no explicit port / log path scan / /api/debug") logger.info(f" → Visit /api/debug to inspect log file discovery live") app.run(host="0.0.0.0", port=_DASHBOARD_PORT, debug=False, use_reloader=False)