Upload 13 files

DEPLOYMENT_GUIDE.md

@@ -0,0 +1,296 @@
+# QUASAR System — Deployment Guide
+## v2.0 Architecture-Strict | 2026-03-25
+
+---
+
+## Overview
+
+The refactored system enforces a **strict one-way data pipeline**:
+
+```
+Asset Spaces (V75, V100_1s, Crash500)
+        │  WebSocket PUBLISH  (send-only)
+        ▼
+Central WebSocket Hub   ←── ingest, normalize, broadcast
+        │  WebSocket SUBSCRIBE  (read-only)
+        ▼
+Ranker Space (Quasar_axrvi_ranker.py)
+        │  REST / Dashboard outputs
+        ▼
+  Rankings  •  REST API  •  Dashboard
+```
+
+**No feedback loop exists.** The Ranker never writes back to the Hub or Asset Spaces.
+
+---
+
+## File Reference
+
+| File | Role | Deploy As |
+|------|------|-----------|
+| `websocket_hub.py` | Central Hub — ingest, normalize, broadcast | Standalone FastAPI service |
+| `websocket_client.py` | Publisher — Asset Space send-only client | Imported in each Asset Space |
+| `Quasar_axrvi_ranker.py` | Ranker Space — subscriber + neural ranker + trading | Standalone process |
+
+---
+
+## 1. Requirements
+
+```bash
+pip install fastapi uvicorn websockets websocket-client torch numpy pydantic
+```
+
+For Hugging Face Spaces, add to `requirements.txt`:
+```
+fastapi
+uvicorn[standard]
+websockets
+websocket-client
+torch
+numpy
+pydantic
+nest_asyncio
+```
+
+---
+
+## 2. Deploy the Central Hub
+
+```bash
+# Local
+python websocket_hub.py
+
+# With explicit port
+PORT=7860 python websocket_hub.py
+
+# Production (Hugging Face Space)
+# Set Space SDK to "Docker" or use app.py entry with:
+uvicorn websocket_hub:app --host 0.0.0.0 --port 7860
+```
+
+**Hub endpoints:**
+
+| Endpoint | Protocol | Role |
+|----------|----------|------|
+| `/ws/publish/{space_name}` | WebSocket | Publisher (Asset Spaces connect here) |
+| `/ws/subscribe` | WebSocket | Subscriber (Ranker connects here) |
+| `/rankings` | GET | Latest snapshots for all assets |
+| `/metrics/{space_name}` | GET | Single-asset snapshot |
+| `/health` | GET | Hub health and connection stats |
+
+---
+
+## 3. Integrate the Publisher into an Asset Space
+
+Each asset space (V75, V100_1s, Crash500) imports `AssetSpacePublisher`:
+
+```python
+from websocket_client import AssetSpacePublisher, TrainingMetrics, VotingMetrics
+
+# ── Create and start (once, at startup) ──────────────────────────────────────
+publisher = AssetSpacePublisher(
+    space_name = "V75",
+    hub_url    = "ws://your-hub-host:7860/ws/publish/V75",
+)
+publisher.start()
+
+# ── In your training loop ──────────────────────────────────────────────────
+# After each training update:
+publisher.publish_training(TrainingMetrics(
+    training_steps = step,
+    actor_loss     = actor_loss,
+    critic_loss    = critic_loss,
+    avn_loss       = avn_loss,
+    avn_accuracy   = avn_accuracy,
+))
+
+# After each agent vote:
+publisher.publish_voting(VotingMetrics(
+    dominant_signal = "BUY",   # "BUY" | "SELL" | "NEUTRAL"
+    buy_count       = 7,
+    sell_count      = 3,
+))
+
+# Or publish both together (preferred — fewer messages):
+publisher.publish_combined(
+    training = TrainingMetrics(...),
+    voting   = VotingMetrics(...),
+)
+```
+
+**Reminder:** The publisher is send-only. Any unexpected message from the hub
+is logged as a warning and discarded. No callbacks are invoked.
+
+---
+
+## 4. Deploy the Ranker Space
+
+```bash
+# Point at your hub's subscribe endpoint
+python Quasar_axrvi_ranker.py \
+    --hub   ws://your-hub-host:7860/ws/subscribe \
+    --assets V75 V100_1s CRASH1000 \
+    --bandit ucb \
+    --reward simple \
+    --model  deriv_axrvi_model.pt
+
+# Sync/thread mode (e.g., inside a Jupyter notebook or larger process)
+python Quasar_axrvi_ranker.py --sync --hub ws://...
+
+# Component tests (no network required)
+python Quasar_axrvi_ranker.py --test
+```
+
+---
+
+## 5. Ranking Formula
+
+```
+signal_confidence = max(buy_count, sell_count) / (buy_count + sell_count)
+score             = signal_confidence - avn_accuracy
+```
+
+| Scenario | Result |
+|----------|--------|
+| High confidence (0.9) + high accuracy (0.8) | score = +0.10 → good |
+| High confidence (0.9) + low accuracy (0.3)  | score = +0.60 → penalized (large gap) |
+| Low confidence (0.5)  + any accuracy        | score ≤ 0.0  → weak |
+
+Assets are sorted by score in **ascending** order (smallest = most balanced = best).
+
+---
+
+## 6. Strict Data Schema
+
+The hub enforces and broadcasts **only** these fields:
+
+```
+training:
+  training_steps  (int)
+  actor_loss      (float)
+  critic_loss     (float)
+  avn_loss        (float)
+  avn_accuracy    (float, clamped [0,1])
+
+voting:
+  dominant_signal  ("BUY" | "SELL" | "NEUTRAL")
+  buy_count        (int)
+  sell_count       (int)
+```
+
+The following fields are **explicitly stripped** at the hub ingestion layer
+and will never reach the Ranker:
+
+- ❌ rewards (matched, unmatched, duplicates, match_rate)
+- ❌ resource metrics (cpu_percent, memory_percent, memory_used_gb, quasar_memory_gb)
+- ❌ agent-level metrics (q_buy, q_sell, entropy, per-agent data)
+- ❌ buffer_size
+- ❌ any q-values or internal model outputs
+
+---
+
+## 7. Hugging Face Spaces Deployment
+
+### Hub Space (`quasar-hub`)
+
+`app.py`:
+```python
+from websocket_hub import app  # FastAPI app, ready for uvicorn
+```
+
+`README.md` front-matter:
+```yaml
+---
+sdk: docker
+app_port: 7860
+---
+```
+
+### Asset Space (e.g., `quasar-v75`)
+
+In your existing Space's training entry point:
+```python
+from websocket_client import AssetSpacePublisher, TrainingMetrics, VotingMetrics
+import os
+
+HUB_URL = os.environ.get("HUB_WS_URL", "wss://your-hub-space.hf.space/ws/publish/V75")
+publisher = AssetSpacePublisher("V75", HUB_URL)
+publisher.start()
+```
+
+Set the environment variable `HUB_WS_URL` in each Space's settings.
+
+### Ranker Space (`quasar-ranker`)
+
+```python
+# main.py (entry point)
+import asyncio, os
+from Quasar_axrvi_ranker import run_live_trading_system
+
+HUB_SUB = os.environ.get("HUB_SUB_URL", "wss://your-hub-space.hf.space/ws/subscribe")
+
+asyncio.run(run_live_trading_system(
+    asset_symbols   = ["V75", "V100_1s", "CRASH1000"],
+    hub_ws_url      = HUB_SUB,
+))
+```
+
+---
+
+## 8. Architecture Constraints (enforced in code)
+
+| Constraint | Where enforced |
+|------------|----------------|
+| Publishers cannot receive data | `_on_message` in `AssetSpacePublisher` discards all inbound messages |
+| Hub never writes to publishers | Publisher WebSocket endpoint is receive-only; no sends |
+| Subscribers are read-only | Subscriber endpoint drains inbound messages without processing |
+| No feedback loop | `HubSubscriber` has no send methods |
+| Minimal schema | Hub `_validate_and_normalize()` strips all non-permitted fields |
+| Thread-safe | All shared state protected by `asyncio.Lock` (hub) / `threading.Lock` (client, ranker) |
+
+---
+
+## 9. Environment Variables
+
+| Variable | Used By | Description |
+|----------|---------|-------------|
+| `DERIV_API_KEY` | Ranker | Deriv API key for live trading |
+| `PORT` | Hub | FastAPI server port (default 7860) |
+| `HUB_WS_URL` | Asset Spaces | Publisher WebSocket URL |
+| `HUB_SUB_URL` | Ranker | Subscriber WebSocket URL |
+
+---
+
+## 10. Quick Smoke Test
+
+```bash
+# Terminal 1 — start hub
+python websocket_hub.py
+
+# Terminal 2 — ranker tests (no network)
+python Quasar_axrvi_ranker.py --test
+
+# Terminal 3 — simulate a publisher
+python - <<'EOF'
+from websocket_client import AssetSpacePublisher, TrainingMetrics, VotingMetrics
+import time
+
+pub = AssetSpacePublisher("V75_TEST", "ws://localhost:7860/ws/publish/V75_TEST")
+pub.start()
+time.sleep(1)
+for step in range(10):
+    pub.publish_combined(
+        TrainingMetrics(training_steps=step*100, avn_accuracy=0.5+step*0.04),
+        VotingMetrics(dominant_signal="BUY", buy_count=7, sell_count=3),
+    )
+    time.sleep(0.5)
+print("Done. Check /rankings endpoint.")
+EOF
+
+# Terminal 4 — verify hub received data
+curl http://localhost:7860/rankings | python -m json.tool
+```
+
+---
+
+*End of deployment guide.*

Dockerfile

@@ -0,0 +1,48 @@
+# Use official Python 3.11 slim image
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    git \
+    curl \
+    wget \
+    supervisor \
+    && rm -rf /var/lib/apt/lists/*
+
+# ── Remove ALL default supervisor configs shipped by apt ──────────────────────
+# The default /etc/supervisor/supervisord.conf includes inet_http_server on
+# port 9001, which collides on restart. We replace it entirely with our own.
+RUN rm -f /etc/supervisor/supervisord.conf \
+          /etc/supervisor/conf.d/*.conf
+
+# Copy requirements first (for layer caching)
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir --upgrade pip setuptools wheel && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# ── Place our supervisor config at the root level so it is fully standalone ──
+COPY supervisord.conf /etc/supervisord.conf
+RUN chmod 644 /etc/supervisord.conf
+
+# ── Copy and permission the entrypoint cleanup script ────────────────────────
+COPY entrypoint.sh /entrypoint.sh
+RUN chmod +x /entrypoint.sh
+
+# Logs directory
+RUN mkdir -p /app/logs && chmod 777 /app/logs
+
+# Expose only the ports our app actually uses
+EXPOSE 7860
+
+USER root
+
+# Entrypoint cleans up stale state, then execs supervisord
+CMD ["/entrypoint.sh"]

README.md

@@ -1,10 +1,628 @@
----
-title: Quasar Executo
-emoji: 😻
-colorFrom: indigo
-colorTo: purple
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+#!/usr/bin/env python3
+"""
+╔══════════════════════════════════════════════════════════════════════════════════════╗
+║  K1RL QUASAR — LOG METRICS READER  (standalone)                                      ║
+║  ──────────────────────────────────────────────────────────────────────────────────  ║
+║                                                                                      ║
+║  Drop this SINGLE FILE into any Asset Space.  No other K1RL files needed.           ║
+║                                                                                      ║
+║  What it does:                                                                       ║
+║    • Subscribes to the V75:final_signals Redis channel (no log polling)          ║
+║    • Parses BUY/SELL signals via the same Redis ingestion flow as Rewards.py        ║
+║    • Publishes extracted voting metrics to the Quasar Hub over wss://               ║
+║    • Runs entirely in a background daemon thread with its own asyncio loop           ║
+║    • Auto-reconnects on disconnection                                                ║
+║    • Auto-detects space name from HuggingFace SPACE_ID env var                     ║
+║                                                                                      ║
+║  SIGNAL SOURCE: Redis (V75:final_signals) — logs are NOT consulted for signals  ║
+║                                                                                      ║
+║  Integration — 3 lines in app.py:                                                    ║
+║      from log_metrics_reader import start_log_publisher                              ║
+║      _publisher, _reader = start_log_publisher()                                     ║
+║      # done — signals stream from Redis to hub automatically                        ║
+║                                                                                      ║
+║  pip dependencies:  websocket-client>=1.6.0, redis, redis_config_v75,            ║
+║                     redis_connection_manager                                         ║
+║                                                                                      ║
+║  VERSION: v2.0-redis-signals  |  2026-03-28  |  V75                                          ║
+║                                                                                      ║
+║  CHANGE LOG:                                                                         ║
+║    v2.0  REMOVED log-based signal scraping entirely (_tail_log, _parse_lines,       ║
+║          extract_metrics_from_log, LogMetricsReader, _RawMetrics).                  ║
+║          REPLACED with RedisSignalReader — identical Redis connection setup,         ║
+║          channel, and parse chain as Rewards.py v5.2.1-V75.                      ║
+╚══════════════════════════════════════════════════════════════════════════════════════╝
+"""
+
+import asyncio
+import json
+import logging
+import os
+import sys
+import threading
+import time
+from dataclasses import dataclass
+from typing import Optional, Tuple
+
+import websocket  # pip: websocket-client>=1.6.0
+
+# ── Redis / V75 imports (same as Rewards.py) ──────────────────────────────────────
+from redis_config_v75 import (
+    REDIS_URL,
+    REDIS_PASSWORD,
+    REDIS_DB_FEATURES,
+    prefixed_channel,
+)
+from redis_connection_manager import RedisAblyClient, RedisMessage
+
+# ── Namespaced signal channel (identical to Rewards.py) ──────────────────────────────
+ABLY_SIGNAL_CHANNEL = prefixed_channel("final_signals")   # → "V75:final_signals"
+
+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("LogMetricsReader")
+
+# ── Config — override via environment variables if needed ─────────────────────────────
+_DEFAULT_HUB_HOST      = os.environ.get("QUASAR_HUB_HOST",  "karlquant-quasar-executo.hf.space")
+_DEFAULT_POLL_INTERVAL = float(os.environ.get("QUASAR_POLL_INTERVAL", "5.0"))   # kept for factory compat
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 1 — STRICT METRIC CONTAINERS  (unchanged)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+@dataclass
+class TrainingMetrics:
+    """The only training fields permitted by the hub schema."""
+    training_steps: int   = 0
+    actor_loss:     float = 0.0
+    critic_loss:    float = 0.0
+    avn_loss:       float = 0.0
+    avn_accuracy:   float = 0.0   # 0.0 – 1.0 float (NOT percent)
+
+
+@dataclass
+class VotingMetrics:
+    """The only voting fields permitted by the hub schema."""
+    dominant_signal: str = "NEUTRAL"   # "BUY" | "SELL" | "NEUTRAL"
+    buy_count:       int = 0
+    sell_count:      int = 0
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 2 — WEBSOCKET PUBLISHER  (unchanged — embedded, no external K1RL imports)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+class AssetSpacePublisher:
+    """
+    Send-only WebSocket publisher.  Runs in a background daemon thread.
+    Auto-reconnects with capped exponential back-off.
+
+    Never blocks the training loop or signal reader.
+    """
+
+    _MAX_BACKOFF: int = 30
+
+    def __init__(
+        self,
+        space_name:           str,
+        hub_url:              str,
+        min_publish_interval: float = 0.5,
+    ):
+        self.space_name           = space_name
+        self.hub_url              = hub_url
+        self.min_publish_interval = min_publish_interval
+
+        self._ws:    Optional[websocket.WebSocketApp] = None
+        self._connected  = False
+        self._running    = False
+        self._thread: Optional[threading.Thread] = None
+        self._reconnect_count = 0
+
+        self._cache_lock       = threading.Lock()
+        self._latest_training: Optional[TrainingMetrics] = None
+        self._latest_voting:   Optional[VotingMetrics]   = None
+
+        self._rate_lock    = threading.Lock()
+        self._last_send_ts = {"training": 0.0, "voting": 0.0, "combined": 0.0}
+
+        self._stats = {
+            "messages_sent":   0,
+            "bytes_sent":      0,
+            "reconnect_count": 0,
+            "last_send_time":  0.0,
+            "dropped_rate":    0,
+        }
+
+    # ── Lifecycle ────────────────────────────────────────────────────────────────────
+
+    def start(self) -> None:
+        if self._running:
+            return
+        self._running = True
+        self._thread  = threading.Thread(
+            target=self._run_loop,
+            daemon=True,
+            name=f"Publisher-{self.space_name}",
+        )
+        self._thread.start()
+        logger.info(f"[{self.space_name}] Publisher started → {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)
+        logger.info(f"[{self.space_name}] Publisher stopped")
+
+    @property
+    def is_connected(self) -> bool:
+        return self._connected
+
+    def get_stats(self) -> dict:
+        return {**self._stats, "connected": self._connected, "running": self._running}
+
+    # ── WebSocket loop ───────────────────────────────────────────────────────────────
+
+    def _run_loop(self) -> None:
+        while self._running:
+            try:
+                self._connect_and_run()
+            except Exception as e:
+                logger.error(f"[{self.space_name}] Connection error: {e}")
+            if not self._running:
+                break
+            backoff = min(self._MAX_BACKOFF, 2 ** min(self._reconnect_count, 4))
+            logger.info(
+                f"[{self.space_name}] Reconnecting in {backoff}s… "
+                f"(attempt #{self._reconnect_count + 1})"
+            )
+            time.sleep(backoff)
+            self._reconnect_count += 1
+            self._stats["reconnect_count"] = self._reconnect_count
+
+    def _connect_and_run(self) -> None:
+        self._ws = websocket.WebSocketApp(
+            self.hub_url,
+            on_open    = self._on_open,
+            on_message = self._on_message,
+            on_error   = self._on_error,
+            on_close   = self._on_close,
+        )
+        self._ws.run_forever(
+            ping_interval = 30,
+            ping_timeout  = 10,
+            sslopt        = {"check_hostname": True},
+        )
+
+    # ── Callbacks ────────────────────────────────────────────────────────────────────
+
+    def _on_open(self, ws) -> None:
+        self._connected       = True
+        self._reconnect_count = 0
+        self._stats["reconnect_count"] = 0
+        logger.info(f"[{self.space_name}] ✅ Connected to hub")
+        self._send_raw({"type": "identify", "space": self.space_name})
+        with self._cache_lock:
+            t, v = self._latest_training, self._latest_voting
+        if t:
+            self._send_training_payload(t)
+        if v:
+            self._send_voting_payload(v)
+
+    def _on_message(self, ws, message: str) -> None:
+        logger.warning(
+            f"[{self.space_name}] ⚠️  Unexpected hub message — discarded: {message[:80]}"
+        )
+
+    def _on_error(self, ws, error) -> None:
+        logger.error(f"[{self.space_name}] WebSocket error: {error}")
+        self._connected = False
+
+    def _on_close(self, ws, code, msg) -> None:
+        self._connected = False
+        logger.info(f"[{self.space_name}] Connection closed (code={code})")
+
+    # ── Rate limiter ──────────────────────────────────────────────────────────────────
+
+    def _rate_ok(self, key: str) -> bool:
+        if self.min_publish_interval <= 0:
+            return True
+        now = time.time()
+        with self._rate_lock:
+            if now - self._last_send_ts[key] >= self.min_publish_interval:
+                self._last_send_ts[key] = now
+                return True
+        self._stats["dropped_rate"] += 1
+        return False
+
+    # ── Send primitives ───────────────────────────────────────────────────────────────
+
+    def _send_raw(self, payload: dict) -> bool:
+        if not (self._ws and self._connected):
+            return False
+        try:
+            text = json.dumps(payload)
+            self._ws.send(text)
+            self._stats["messages_sent"] += 1
+            self._stats["bytes_sent"]    += len(text)
+            self._stats["last_send_time"] = time.time()
+            return True
+        except Exception as e:
+            logger.error(f"[{self.space_name}] Send error: {e}")
+            self._connected = False
+            return False
+
+    def _send_training_payload(self, m: TrainingMetrics) -> bool:
+        return self._send_raw({
+            "type": "training",
+            "data": {
+                "training_steps": m.training_steps,
+                "actor_loss":     m.actor_loss,
+                "critic_loss":    m.critic_loss,
+                "avn_loss":       m.avn_loss,
+                "avn_accuracy":   m.avn_accuracy,
+            },
+        })
+
+    def _send_voting_payload(self, m: VotingMetrics) -> bool:
+        return self._send_raw({
+            "type": "voting",
+            "data": {
+                "dominant_signal": m.dominant_signal,
+                "buy_count":       m.buy_count,
+                "sell_count":      m.sell_count,
+            },
+        })
+
+    # ── Public API ────────────────────────────────────────────────────────────────────
+
+    def publish_training(self, metrics: TrainingMetrics) -> bool:
+        with self._cache_lock:
+            self._latest_training = metrics
+        if not self._rate_ok("training"):
+            return False
+        return self._send_training_payload(metrics)
+
+    def publish_voting(self, metrics: VotingMetrics) -> bool:
+        with self._cache_lock:
+            self._latest_voting = metrics
+        if not self._rate_ok("voting"):
+            return False
+        return self._send_voting_payload(metrics)
+
+    def publish_combined(self, training: TrainingMetrics, voting: VotingMetrics) -> bool:
+        with self._cache_lock:
+            self._latest_training = training
+            self._latest_voting   = voting
+        if not self._rate_ok("combined"):
+            return False
+        return self._send_raw({
+            "type": "metrics",
+            "training": {
+                "training_steps": training.training_steps,
+                "actor_loss":     training.actor_loss,
+                "critic_loss":    training.critic_loss,
+                "avn_loss":       training.avn_loss,
+                "avn_accuracy":   training.avn_accuracy,
+            },
+            "voting": {
+                "dominant_signal": voting.dominant_signal,
+                "buy_count":       voting.buy_count,
+                "sell_count":      voting.sell_count,
+            },
+        })
+
+    def publish_heartbeat(self) -> bool:
+        return self._send_raw({"type": "heartbeat"})
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 3 — REDIS SIGNAL READER
+#
+# ✅ REPLACES: LogMetricsReader (log polling), _tail_log, _parse_lines,
+#              extract_metrics_from_log, _RawMetrics — all removed.
+#
+# Signal source is now exclusively the V75:final_signals Redis channel.
+# Connection setup, channel name, and full parse chain are transplanted
+# verbatim from Rewards.py v5.2.1-V75._on_signal / initialize().
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+class RedisSignalReader:
+    """
+    Subscribes to the namespaced V75:final_signals Redis channel and feeds
+    parsed BUY/SELL signals directly into an AssetSpacePublisher as VotingMetrics.
+
+    Redis is the ONLY source of truth for signals — no log file is consulted.
+
+    Parse chain (identical to Rewards.py._on_signal):
+      1. Unwrap RedisMessage → .data attribute
+      2. Unwrap nested envelope  → {"data": {...}}
+      3. Decode JSON strings
+      4. Extract final_action with fallback to action
+      5. Extract signal_keys (normalised to list)
+      6. Extract price
+      7. Validate: action must be BUY|SELL, price must be non-zero
+      8. Silently ignore any malformed payload (no exception raised)
+
+    Threading model:
+      Runs in a daemon thread that owns its own asyncio event loop, mirroring
+      how Rewards.py bridges the blocking RedisAblyClient listener thread into
+      async coroutines without touching the caller's event loop.
+    """
+
+    def __init__(self, publisher: AssetSpacePublisher):
+        self.publisher = publisher
+
+        self._running = False
+        self._thread: Optional[threading.Thread] = None
+        self._loop:   Optional[asyncio.AbstractEventLoop] = None
+
+        # Rolling buy/sell counters → VotingMetrics
+        self._buy_count  = 0
+        self._sell_count = 0
+        self._count_lock = threading.Lock()
+
+        self._stats = {
+            "signals_received": 0,
+            "signals_parsed":   0,
+            "signals_dropped":  0,
+        }
+
+    # ── Lifecycle ────────────────────────────────────────────────────────────────────
+
+    def start(self) -> None:
+        if self._running:
+            return
+        self._running = True
+        self._thread = threading.Thread(
+            target=self._run_loop,
+            daemon=True,
+            name=f"RedisSignalReader-{self.publisher.space_name}",
+        )
+        self._thread.start()
+        logger.info(
+            f"[{self.publisher.space_name}] RedisSignalReader started "
+            f"→ channel={ABLY_SIGNAL_CHANNEL}"
+        )
+
+    def stop(self) -> None:
+        self._running = False
+        if self._loop and self._loop.is_running():
+            self._loop.call_soon_threadsafe(self._loop.stop)
+        if self._thread:
+            self._thread.join(timeout=5)
+        logger.info(f"[{self.publisher.space_name}] RedisSignalReader stopped")
+
+    def get_stats(self) -> dict:
+        return {**self._stats, "running": self._running}
+
+    # ── Asyncio thread entry point ────────────────────────────────────────────────────
+
+    def _run_loop(self) -> None:
+        """
+        Runs in a daemon thread.  Creates a fresh asyncio event loop so this
+        reader never interferes with any event loop the host process may have.
+        """
+        self._loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self._loop)
+        try:
+            self._loop.run_until_complete(self._async_main())
+        except Exception as e:
+            logger.error(
+                f"[{self.publisher.space_name}] RedisSignalReader loop error: {e}"
+            )
+        finally:
+            self._loop.close()
+
+    # ── Core async subscription ───────────────────────────────────────────────────────
+
+    async def _async_main(self) -> None:
+        """
+        Initialise the RedisAblyClient exactly as Rewards.py does, subscribe to
+        ABLY_SIGNAL_CHANNEL, and keep the loop alive until stop() is called.
+
+        Connection parameters mirror Rewards.py.RewardsEngine.initialize():
+          • redis_url    = REDIS_URL      (from redis_config_v75)
+          • password     = REDIS_PASSWORD
+          • use_streams  = True
+          • database     = REDIS_DB_FEATURES   (V75: DB 0)
+        """
+        ably = RedisAblyClient(
+            redis_url=REDIS_URL,
+            password=REDIS_PASSWORD,
+            use_streams=True,
+            database=REDIS_DB_FEATURES,
+        )
+
+        channel = ably.channels.get(ABLY_SIGNAL_CHANNEL)
+
+        # _on_signal is called by the blocking RedisAblyClient listener thread.
+        # Identical callback signature and guard logic to Rewards.py._on_signal.
+        def _on_signal(message) -> None:
+            self._stats["signals_received"] += 1
+
+            parsed = self._parse_signal_message(message)
+            if parsed is None:
+                self._stats["signals_dropped"] += 1
+                return
+
+            self._stats["signals_parsed"] += 1
+            action = parsed["action"]
+
+            with self._count_lock:
+                if action == "BUY":
+                    self._buy_count += 1
+                else:
+                    self._sell_count += 1
+                buy  = self._buy_count
+                sell = self._sell_count
+
+            dominant = "BUY" if buy >= sell else "SELL"
+            vm = VotingMetrics(
+                dominant_signal=dominant,
+                buy_count=buy,
+                sell_count=sell,
+            )
+            self.publisher.publish_voting(vm)
+
+            logger.info(
+                f"[{self.publisher.space_name}] 🔔 Signal {action} "
+                f"@ {parsed['entry_price']:.5f} | "
+                f"keys={len(parsed['signal_keys'])} | "
+                f"buy={buy} sell={sell} dominant={dominant}"
+            )
+
+        await channel.subscribe("message", _on_signal)
+        logger.info(
+            f"[{self.publisher.space_name}] ✅ Subscribed to {ABLY_SIGNAL_CHANNEL} "
+            f"(V75 namespace, DB={REDIS_DB_FEATURES})"
+        )
+
+        # Idle loop — RedisAblyClient delivers messages via its own listener thread.
+        while self._running:
+            await asyncio.sleep(1.0)
+
+        ably.close()
+
+    # ── Signal parser (transplanted verbatim from Rewards.py._on_signal) ─────────────
+
+    def _parse_signal_message(self, message) -> Optional[dict]:
+        """
+        Full parse chain from Rewards.py, extracted into a standalone helper.
+
+        Steps:
+          1. Unwrap RedisMessage → .data
+          2. Unwrap nested {"data": ...} envelope
+          3. Parse JSON strings
+          4. Extract final_action with fallback to action, uppercase
+          5. Extract signal_keys
+          6. Extract price
+          7. Validate action ∈ {BUY, SELL}
+          8. Validate price is non-zero
+          9. Normalise signal_keys to a list (cap at 8, matching Rewards.py)
+         10. Return None silently for any malformed payload — never raises
+
+        Returns a dict with keys: action, signal_keys, entry_price, payload.
+        Returns None if the payload is malformed or the signal should be ignored.
+        """
+        try:
+            # Step 1 — unwrap RedisMessage
+            data = message.data if isinstance(message, RedisMessage) else message
+
+            # Step 2 — unwrap nested data envelope
+            if isinstance(data, dict) and "data" in data:
+                data = data["data"]
+
+            # Step 3 — decode JSON strings
+            if isinstance(data, str):
+                data = json.loads(data)
+
+            # Step 4 — extract action (final_action with fallback to action)
+            action = data.get("final_action", data.get("action", "")).upper()
+
+            # Step 5 — extract signal_keys
+            signal_keys = data.get("signal_keys", [])
+
+            # Step 6 — extract price
+            entry_price = data.get("price", 0.0)
+
+            # Step 7 — validate action
+            if action not in ("BUY", "SELL"):
+                return None
+
+            # Step 8 — validate price
+            if not entry_price or entry_price == 0.0:
+                return None
+
+            # Step 9 — normalise signal_keys to list, cap at 8
+            if not isinstance(signal_keys, list):
+                signal_keys = [str(signal_keys)]
+
+            return {
+                "action":      action,
+                "signal_keys": signal_keys[:8],
+                "entry_price": float(entry_price),
+                "payload":     data,
+            }
+
+        except Exception:
+            # Step 10 — silently discard malformed payloads
+            return None
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 4 — ONE-LINE FACTORY  (signature unchanged — drop-in replacement)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+def start_log_publisher(
+    poll_interval: float = _DEFAULT_POLL_INTERVAL,
+    hub_host:      str   = _DEFAULT_HUB_HOST,
+    space_name:    Optional[str] = None,
+    # Legacy parameters accepted but unused — kept for call-site compatibility.
+    log_path:      Optional[str] = None,
+    tail_lines:    Optional[int] = None,
+) -> Tuple[AssetSpacePublisher, RedisSignalReader]:
+    """
+    One-line drop-in for each asset space.
+
+    Creates the WebSocket publisher AND the Redis signal reader, starts both,
+    and returns them so they stay alive for the process lifetime.
+
+    Signal source:  Redis channel V75:final_signals  (NOT the log file)
+    Auto-detects space name from HuggingFace SPACE_ID env var.
+
+    Usage in app.py
+    ---------------
+        from log_metrics_reader import start_log_publisher
+        _publisher, _reader = start_log_publisher()
+
+    Override defaults
+    -----------------
+        _publisher, _reader = start_log_publisher(
+            poll_interval = 10.0,   # rate-limit floor for WebSocket publishes
+        )
+
+    Environment variable overrides (no code change needed per space)
+    ----------------------------------------------------------------
+        QUASAR_HUB_HOST      = "karlquant-quasar-executo.hf.space"
+        QUASAR_POLL_INTERVAL = "5.0"
+
+    NOTE: log_path and tail_lines parameters are accepted for call-site
+    compatibility but are silently ignored — signal scraping from logs has
+    been removed. Signals are sourced exclusively from Redis.
+    """
+    if log_path is not None or tail_lines is not None:
+        logger.warning(
+            "start_log_publisher: log_path / tail_lines arguments are ignored. "
+            "Signal acquisition from log files has been removed. "
+            "Signals are sourced exclusively from Redis (V75:final_signals)."
+        )
+
+    # Resolve space name — HF injects SPACE_ID="Owner/SpaceName" at runtime
+    if space_name is None:
+        raw        = os.environ.get("SPACE_ID", "")
+        space_name = raw.split("/", 1)[-1] if "/" in raw else (raw or "UnknownSpace")
+
+    # Always wss://, never explicit port — HF proxies through 443
+    hub_url = f"wss://{hub_host}/ws/publish/{space_name}"
+
+    publisher = AssetSpacePublisher(
+        space_name           = space_name,
+        hub_url              = hub_url,
+        min_publish_interval = max(poll_interval * 0.9, 0.5),
+    )
+    publisher.start()
+
+    reader = RedisSignalReader(publisher=publisher)
+    reader.start()
+
+    return publisher, reader

entrypoint.sh

@@ -0,0 +1,33 @@
+#!/bin/bash
+# ════════════════════════════════════════════════════════════════
+# K1RL QUASAR — Entrypoint
+# Cleans up any stale processes/ports before launching supervisord
+# ════════════════════════════════════════════════════════════════
+set -e
+
+echo "=== QUASAR entrypoint: cleaning up stale state ==="
+
+# Kill any lingering supervisord from a previous run
+pkill -f supervisord 2>/dev/null || true
+
+# Kill any lingering hub/ranker Python processes
+pkill -f websocket_hub.py      2>/dev/null || true
+pkill -f Quasar_axrvi_ranker.py 2>/dev/null || true
+
+# Give OS a moment to release sockets
+sleep 1
+
+# Remove stale PID files
+rm -f /tmp/supervisord.pid
+
+# Confirm our app files exist (catches missing-file issues early)
+for f in /app/websocket_hub.py /app/Quasar_axrvi_ranker.py; do
+    if [ ! -f "$f" ]; then
+        echo "ERROR: Required file missing: $f"
+        echo "Make sure it is committed to your HF Space repository."
+        exit 1
+    fi
+done
+
+echo "=== Starting supervisord ==="
+exec /usr/bin/supervisord -c /etc/supervisord.conf

gitattributes

@@ -0,0 +1,35 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

hub_dashboard_service.py

@@ -0,0 +1,388 @@
+#!/usr/bin/env python3
+"""
+╔══════════════════════════════════════════════════════════════════════════════════════╗
+║  K1RL QUASAR — HUB DASHBOARD SERVICE  (WITH RANKER LOGS INTEGRATION)                ║
+║  ──────────────────────────────────────────────────────────────────────────────────  ║
+║  Architecture role: READ-ONLY subscriber → serves dashboard UI                      ║
+║  VERSION: v1.1 (UPDATED)  |  2026-03-26                                              ║
+╚═══════════════════════════════════════════════���══════════════════════════════════════╝
+"""
+
+import json
+import logging
+import os
+import sys
+import threading
+import time
+from collections import deque
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional
+
+import websocket
+from flask import Flask, Response, jsonify, send_from_directory
+from flask_cors import CORS
+
+from ranker_logging import RankerLogger, EventCategory, LogLevel
+from ranker_logs_api import ranker_logs_bp, init_ranker_logs_api
+
+# ── Logging ───────────────────────────────────────────────────────────────────────────
+logging.basicConfig(
+    level=logging.INFO,
+    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"),
+)
+
+_METRIC_HISTORY_LEN = int(os.environ.get("QUASAR_METRIC_HISTORY", "200"))
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 1 — IN-MEMORY STATE STORE
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+class DashboardState:
+    """
+    Thread-safe cache of everything the dashboard needs.
+    """
+
+    def __init__(self, history_len: int = _METRIC_HISTORY_LEN):
+        self._lock    = threading.RLock()
+        self._history_len = history_len
+
+        self.snapshots: Dict[str, dict] = {}
+        self.metric_history: Dict[str, deque] = {}
+
+        self.hub_connected   = False
+        self.start_time      = time.time()
+        self.messages_rx     = 0
+        self.last_update_ts  = 0.0
+        self.reconnect_count = 0
+
+        self._ranked: List[dict] = []
+
+    def ingest_snapshot(self, space_name: str, snapshot: dict) -> None:
+        """Accept a full snapshot dict."""
+        with self._lock:
+            self.snapshots[space_name] = snapshot
+            self.messages_rx += 1
+            self.last_update_ts = time.time()
+            self._append_metric_history(space_name, snapshot)
+            self._recompute_rankings()
+
+    def ingest_bulk(self, snapshots: dict) -> None:
+        """Accept bulk payload."""
+        with self._lock:
+            for name, snap in snapshots.items():
+                self.snapshots[name] = snap
+                self._append_metric_history(name, snap)
+            self.last_update_ts = time.time()
+            self._recompute_rankings()
+
+    def _append_metric_history(self, space_name: str, snap: dict) -> None:
+        """Push latest training numbers onto the per-asset deque."""
+        if space_name not in self.metric_history:
+            self.metric_history[space_name] = deque(maxlen=self._history_len)
+        training = snap.get("training", {})
+        if training:
+            self.metric_history[space_name].append({
+                "ts":            snap.get("last_updated", time.time()),
+                "training_steps": training.get("training_steps", 0),
+                "actor_loss":     training.get("actor_loss",     0.0),
+                "critic_loss":    training.get("critic_loss",    0.0),
+                "avn_loss":       training.get("avn_loss",       0.0),
+                "avn_accuracy":   training.get("avn_accuracy",   0.0),
+            })
+
+    def _recompute_rankings(self) -> None:
+        """Replicate ranker formula: score = signal_confidence - avn_accuracy"""
+        ranked = []
+        for name, snap in self.snapshots.items():
+            training = snap.get("training", {})
+            voting   = snap.get("voting",   {})
+            buy      = voting.get("buy_count",  0)
+            sell     = voting.get("sell_count", 0)
+            total    = buy + sell
+            sig_conf = (max(buy, sell) / total) if total > 0 else 0.0
+            avn_acc  = training.get("avn_accuracy", 0.0)
+            score    = sig_conf - avn_acc
+            ranked.append({
+                "rank":              0,
+                "space_name":        name,
+                "score":             round(score, 4),
+                "signal_confidence": round(sig_conf, 4),
+                "avn_accuracy":      round(avn_acc,  4),
+                "dominant_signal":   voting.get("dominant_signal", "NEUTRAL"),
+                "buy_count":         buy,
+                "sell_count":        sell,
+                "training_steps":    training.get("training_steps", 0),
+                "actor_loss":        training.get("actor_loss",     0.0),
+                "critic_loss":       training.get("critic_loss",    0.0),
+                "avn_loss":          training.get("avn_loss",       0.0),
+                "last_updated":      snap.get("last_updated",       0.0),
+            })
+        ranked.sort(key=lambda r: r["score"], reverse=True)
+        for i, r in enumerate(ranked):
+            r["rank"] = i + 1
+        self._ranked = ranked
+
+    def get_state(self) -> dict:
+        with self._lock:
+            return {
+                "rankings":      list(self._ranked),
+                "metric_history": {
+                    name: list(h)
+                    for name, h in self.metric_history.items()
+                },
+                "health":        self._health_snapshot(),
+                "timestamp":     datetime.utcnow().isoformat() + "Z",
+            }
+
+    def get_rankings(self) -> List[dict]:
+        with self._lock:
+            return list(self._ranked)
+
+    def get_metric_history(self, limit: int = 100) -> dict:
+        with self._lock:
+            return {
+                name: list(h)[-limit:]
+                for name, h in self.metric_history.items()
+            }
+
+    def _health_snapshot(self) -> dict:
+        return {
+            "hub_connected":   self.hub_connected,
+            "spaces_connected": len(self.snapshots),
+            "messages_rx":     self.messages_rx,
+            "last_update_ts":  self.last_update_ts,
+            "last_update_ago": round(time.time() - self.last_update_ts, 1)
+                               if self.last_update_ts else None,
+            "uptime_seconds":  round(time.time() - self.start_time, 0),
+            "reconnect_count": self.reconnect_count,
+        }
+
+    def get_health(self) -> dict:
+        with self._lock:
+            return self._health_snapshot()
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 2 — WEBSOCKET SUBSCRIBER
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+class HubSubscriber:
+    """Connects to hub and maintains per-asset snapshots."""
+
+    _MAX_BACKOFF = 30
+
+    def __init__(
+        self,
+        state: DashboardState,
+        hub_host: str = _HUB_HOST,
+        ranker_logger: Optional[RankerLogger] = None,
+    ):
+        self.state      = state
+        self.hub_url    = f"wss://{hub_host}/ws/subscribe"
+        self.ranker_logger = ranker_logger
+        self._ws:     Optional[websocket.WebSocketApp] = None
+        self._running = False
+        self._thread: Optional[threading.Thread] = None
+        self._reconnect_count = 0
+
+    def start(self) -> None:
+        if self._running:
+            return
+        self._running = True
+        self._thread  = threading.Thread(
+            target=self._run_loop, daemon=True, name="HubSubscriber",
+        )
+        self._thread.start()
+        logger.info(f"HubSubscriber started → {self.hub_url}")
+
+    def stop(self) -> None:
+        self._running = False
+        if self._ws:
+            try:
+                self._ws.close()
+            except Exception:
+                pass
+        if self._thread:
+            self._thread.join(timeout=5)
+
+    def _run_loop(self) -> None:
+        while self._running:
+            try:
+                self._connect()
+            except Exception as e:
+                logger.error(f"[HubSubscriber] Error: {e}")
+            if not self._running:
+                break
+            self.state.hub_connected   = False
+            self.state.reconnect_count = self._reconnect_count
+            backoff = min(self._MAX_BACKOFF, 2 ** min(self._reconnect_count, 4))
+            logger.info(f"[HubSubscriber] Reconnecting in {backoff}s…")
+            time.sleep(backoff)
+            self._reconnect_count += 1
+
+    def _connect(self) -> None:
+        self._ws = websocket.WebSocketApp(
+            self.hub_url,
+            on_open    = self._on_open,
+            on_message = self._on_message,
+            on_error   = self._on_error,
+            on_close   = self._on_close,
+        )
+        self._ws.run_forever(
+            ping_interval = 25,
+            ping_timeout  = 10,
+            sslopt        = {"check_hostname": True},
+        )
+
+    def _on_open(self, ws) -> None:
+        self.state.hub_connected = True
+        self._reconnect_count    = 0
+        if self.ranker_logger:
+            self.ranker_logger.connection_event("Hub WebSocket", "connected")
+        logger.info("[HubSubscriber] ✅ Connected to hub")
+
+    def _on_message(self, ws, raw: str) -> None:
+        try:
+            msg  = json.loads(raw)
+            mtype = msg.get("type", "")
+
+            if mtype == "initial_state":
+                snaps = msg.get("snapshots", {})
+                self.state.ingest_bulk(snaps)
+                logger.info(f"[HubSubscriber] Initial state: {len(snaps)} spaces")
+
+            elif mtype == "metrics_update":
+                space = msg.get("space_name")
+                snap  = msg.get("snapshot", {})
+                if space and snap:
+                    self.state.ingest_snapshot(space, snap)
+                    if self.ranker_logger:
+                        training = snap.get("training", {})
+                        voting = snap.get("voting", {})
+                        self.ranker_logger.hub_update(
+                            asset=space,
+                            avn_accuracy=training.get("avn_accuracy", 0.0),
+                            signal_confidence=(
+                                max(voting.get("buy_count", 0), voting.get("sell_count", 0)) /
+                                (voting.get("buy_count", 0) + voting.get("sell_count", 0) or 1)
+                            )
+                        )
+
+        except json.JSONDecodeError:
+            logger.warning("[HubSubscriber] Malformed JSON")
+        except Exception as e:
+            logger.error(f"[HubSubscriber] Message error: {e}")
+
+    def _on_error(self, ws, error) -> None:
+        logger.error(f"[HubSubscriber] WS error: {error}")
+        self.state.hub_connected = False
+        if self.ranker_logger:
+            self.ranker_logger.connection_event("Hub WebSocket", "error", str(error))
+
+    def _on_close(self, ws, code, msg) -> None:
+        self.state.hub_connected = False
+        logger.info(f"[HubSubscriber] Connection closed (code={code})")
+        if self.ranker_logger:
+            self.ranker_logger.connection_event("Hub WebSocket", "disconnected")
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 3 — FLASK APP
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+_state = DashboardState()
+_ranker_logger = RankerLogger(
+    buffer_size = 1000,
+    log_dir = "./ranker_logs",
+    on_event = None,  # Optional: push to WebSocket here
+)
+
+app = Flask(__name__)
+CORS(app)
+
+# Register ranker logs API
+init_ranker_logs_api(_ranker_logger)
+app.register_blueprint(ranker_logs_bp)
+
+
+@app.route("/")
+def index():
+    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():
+    return jsonify({"rankings": _state.get_rankings()})
+
+
+@app.route("/api/metrics/history")
+def api_metrics_history():
+    limit = int(request.args.get("limit", 100))
+    return jsonify(_state.get_metric_history(limit=limit))
+
+
+@app.route("/api/health")
+def api_health():
+    return jsonify({
+        "service": "hub_dashboard_service",
+        "version": "v1.1",
+        "timestamp": datetime.utcnow().isoformat() + "Z",
+        **_state.get_health(),
+    })
+
+
+def _start_background_services() -> None:
+    sub = HubSubscriber(_state, hub_host=_HUB_HOST, ranker_logger=_ranker_logger)
+    sub.start()
+
+    global _subscriber
+    _subscriber = sub
+
+
+_subscriber = None
+
+_start_background_services()
+
+logger.info("=" * 70)
+logger.info(f"K1RL QUASAR — Hub Dashboard Service  (v1.1 with Ranker Logs)")
+logger.info(f"Hub WebSocket      : wss://{_HUB_HOST}/ws/subscribe")
+logger.info(f"Dashboard HTML     : {_HTML_PATH}")
+logger.info(f"Service port       : {_DASHBOARD_PORT}")
+logger.info(f"Logs API           : /api/ranker/logs/*")
+logger.info(f"Log directory      : ./ranker_logs")
+logger.info("=" * 70)
+
+
+if __name__ == "__main__":
+    app.run(
+        host     = "0.0.0.0",
+        port     = _DASHBOARD_PORT,
+        debug    = False,
+        threaded = True,
+    )

ranker_logging.py

@@ -0,0 +1,56 @@
+import logging
+import sys
+from enum import Enum, auto
+
+class LogLevel(Enum):
+    DEBUG = auto()
+    INFO = auto()
+    WARNING = auto()
+    ERROR = auto()
+    CRITICAL = auto()
+
+class EventCategory(Enum):
+    INITIALIZATION = 'INITIALIZATION'
+    PROCESSING = 'PROCESSING'
+    TERMINATION = 'TERMINATION'
+    ERROR_OCCURRED = 'ERROR_OCCURRED'
+
+class LogEntry:
+    def __init__(self, level: LogLevel, category: EventCategory, message: str):
+        self.level = level
+        self.category = category
+        self.message = message
+        self.timestamp = logging.Formatter.formatTime(logging.LogRecord('', 0, '', 0, message, None, None))
+
+    def __str__(self):
+        return f"[{self.timestamp}] [{self.level.name}] [{self.category.value}] {self.message}"
+
+class RankerLogger:
+    def __init__(self, name: str):
+        self.logger = logging.getLogger(name)
+        self.logger.setLevel(logging.DEBUG)
+        ch = logging.StreamHandler(sys.stdout)
+        ch.setLevel(logging.DEBUG)
+        formatter = logging.Formatter('%(message)s')
+        ch.setFormatter(formatter)
+        self.logger.addHandler(ch)
+
+    def log(self, level: LogLevel, category: EventCategory, message: str):
+        entry = LogEntry(level, category, message)
+        if level == LogLevel.DEBUG:
+            self.logger.debug(str(entry))
+        elif level == LogLevel.INFO:
+            self.logger.info(str(entry))
+        elif level == LogLevel.WARNING:
+            self.logger.warning(str(entry))
+        elif level == LogLevel.ERROR:
+            self.logger.error(str(entry))
+        elif level == LogLevel.CRITICAL:
+            self.logger.critical(str(entry))
+
+class RankerLogBridge:
+    def __init__(self, ranker_logger: RankerLogger):
+        self.ranker_logger = ranker_logger
+
+    def log_event(self, level: LogLevel, category: EventCategory, message: str):
+        self.ranker_logger.log(level, category, message)

ranker_logs_api.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""
+╔══════════════════════════════════════════════════════════════════════════════════════╗
+║  QUASAR RANKER — LOGS REST API                                                      ║
+║  ──────────────────────────────────────────────────────────────────────────────────  ║
+║  Flask blueprint for exposing ranker logs via REST endpoints.                       ║
+║  VERSION: v1.0  |  2026-03-26                                                        ║
+╚══════════════════════════════════════════════════════════════════════════════════════╝
+"""
+
+from flask import Blueprint, jsonify, request, send_file
+from typing import Optional
+from pathlib import Path
+
+ranker_logs_bp = Blueprint("ranker_logs", __name__, url_prefix="/api/ranker/logs")
+
+_logger: Optional[object] = None
+
+
+def init_ranker_logs_api(ranker_logger):
+    """Call this from hub_dashboard_service.py during initialization."""
+    global _logger
+    _logger = ranker_logger
+
+
+@ranker_logs_bp.route("/recent", methods=["GET"])
+def get_recent_logs():
+    """GET /api/ranker/logs/recent?limit=50&category=signal"""
+    if not _logger:
+        return jsonify({"error": "Logger not initialized"}), 500
+    
+    limit    = int(request.args.get("limit", 50))
+    category = request.args.get("category")
+    
+    entries = _logger.get_recent(n=limit, category=category)
+    return jsonify({
+        "logs":  entries,
+        "count": len(entries),
+        "stats": _logger.get_stats(),
+    })
+
+
+@ranker_logs_bp.route("/asset/<asset>", methods=["GET"])
+def get_asset_logs(asset: str):
+    """GET /api/ranker/logs/asset/V75?limit=30"""
+    if not _logger:
+        return jsonify({"error": "Logger not initialized"}), 500
+    
+    limit   = int(request.args.get("limit", 30))
+    entries = _logger.get_by_asset(asset, n=limit)
+    
+    return jsonify({
+        "asset": asset,
+        "logs":  entries,
+        "count": len(entries),
+    })
+
+
+@ranker_logs_bp.route("/level/<level>", methods=["GET"])
+def get_level_logs(level: str):
+    """GET /api/ranker/logs/level/ERROR?limit=50"""
+    if not _logger:
+        return jsonify({"error": "Logger not initialized"}), 500
+    
+    limit   = int(request.args.get("limit", 50))
+    entries = _logger.get_by_level(level, n=limit)
+    
+    return jsonify({
+        "level": level.upper(),
+        "logs":  entries,
+        "count": len(entries),
+    })
+
+
+@ranker_logs_bp.route("/stats", methods=["GET"])
+def get_log_stats():
+    """GET /api/ranker/logs/stats"""
+    if not _logger:
+        return jsonify({"error": "Logger not initialized"}), 500
+    
+    return jsonify(_logger.get_stats())
+
+
+@ranker_logs_bp.route("/export", methods=["GET"])
+def export_logs():
+    """GET /api/ranker/logs/export?limit=500 → download JSON"""
+    if not _logger:
+        return jsonify({"error": "Logger not initialized"}), 500
+    
+    limit = int(request.args.get("limit", 500))
+    export_path = Path("/tmp/ranker_logs_export.json")
+    
+    try:
+        _logger.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 e:
+        return jsonify({"error": str(e)}), 500
+
+
+@ranker_logs_bp.route("/clear", methods=["POST"])
+def clear_logs():
+    """POST /api/ranker/logs/clear — Clear in-memory buffer"""
+    if not _logger:
+        return jsonify({"error": "Logger not initialized"}), 500
+    
+    try:
+        _logger.clear_buffer()
+        return jsonify({"status": "cleared"})
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+
+@ranker_logs_bp.errorhandler(404)
+def not_found(error):
+    return jsonify({"error": "Endpoint not found"}), 404

requirements.txt

@@ -0,0 +1,34 @@
+# ════════════════════════════════════════════════════════════════════════════════
+# K1RL QUASAR — MINIMAL REQUIREMENTS
+# Only packages actually used in the codebase
+# VERSION: 2.2  |  2026-03-29
+# ════════════════════════════════════════════════════════════════════════════════
+
+# Core ML/Neural Network
+torch==2.1.0
+numpy==1.24.3
+
+# Web Framework & WebSocket
+fastapi==0.104.0
+uvicorn==0.24.0
+websockets==14.0
+websocket-client==1.6.0
+
+# Dashboard UI
+flask==3.0.0
+flask-cors==4.0.0
+
+# Process Management
+supervisor==4.2.4
+
+# Async Utilities
+nest_asyncio==1.5.6
+
+# Data Serialization
+orjson==3.9.0
+
+# Environment Management
+python-dotenv==1.0.0
+
+# HuggingFace — persistent checkpoint storage
+huggingface_hub==0.20.3

supervisord.conf

@@ -0,0 +1,41 @@
+[supervisord]
+nodaemon=true
+user=root
+logfile=/dev/null
+logfile_maxbytes=0
+pidfile=/tmp/supervisord.pid
+loglevel=info
+
+[program:app]
+command=python /app/websocket_hub.py
+directory=/app
+autostart=true
+autorestart=true
+startsecs=5
+startretries=10
+stopasgroup=true
+killasgroup=true
+stdout_logfile=/dev/stdout
+stdout_logfile_maxbytes=0
+stderr_logfile=/dev/stderr
+stderr_logfile_maxbytes=0
+environment=PORT=7860,DASHBOARD_HTML=/app/hub_dashboard.html,PYTHONUNBUFFERED=1
+
+[program:worker1]
+command=python /app/Quasar_axrvi_ranker.py --hub ws://localhost:7860/ws/subscribe --sync
+directory=/app
+autostart=true
+autorestart=true
+startsecs=5
+startretries=10
+stopasgroup=true
+killasgroup=true
+stdout_logfile=/dev/stdout
+stdout_logfile_maxbytes=0
+stderr_logfile=/dev/stderr
+stderr_logfile_maxbytes=0
+environment=PYTHONUNBUFFERED=1
+
+[group:space]
+programs=app,worker1
+priority=999

websocket_hub.py

@@ -0,0 +1,553 @@
+#!/usr/bin/env python3
+"""
+╔══════════════════════════════════════════════════════════════════════════════════════╗
+║  K1RL QUASAR — CENTRAL WEBSOCKET HUB                                                ║
+║  ──────────────────────────────────────────────────────────────────────────────────  ║
+║                                                                                      ║
+║  Architecture role: INGEST → NORMALIZE → BROADCAST                                  ║
+║                                                                                      ║
+║  • Accepts publisher connections from Asset Spaces  (/ws/publish/{space_name})      ║
+║  • Accepts subscriber connections from Ranker Space (/ws/subscribe)                 ║
+║  • ONE-WAY:  Publisher → Hub → Subscriber                                           ║
+║  • Hub NEVER writes back to publishers                                               ║
+║  • Hub stores latest snapshot per asset (NO history)                                 ║
+║                                                                                      ║
+║  STRICT DATA MODEL (training + voting fields ONLY):                                  ║
+║    training: training_steps, actor_loss, critic_loss, avn_loss, avn_accuracy        ║
+║    voting:   dominant_signal, buy_count, sell_count                                  ║
+║                                                                                      ║
+║  VERSION: v2.0-arch-strict  |  2026-03-25                                            ║
+╚══════════════════════════════════════════════════════════════════════════════════════╝
+"""
+
+import asyncio
+import copy
+import json
+import logging
+import os
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional, Set
+
+import uvicorn
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import FileResponse, JSONResponse
+
+# ─── Logging ────────────────────────────────────────────────────────────────────────
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger("QuasarHub")
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 1 — STRICT DATA MODEL
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+# Fields that are permitted through the hub.  All others are silently dropped.
+_ALLOWED_TRAINING_FIELDS: frozenset = frozenset({
+    "training_steps",
+    "actor_loss",
+    "critic_loss",
+    "avn_loss",
+    "avn_accuracy",
+})
+
+_ALLOWED_VOTING_FIELDS: frozenset = frozenset({
+    "dominant_signal",
+    "buy_count",
+    "sell_count",
+    "last_price",     # FIX: was silently dropped — needed by subscribers
+    "signal_source",  # FIX: was silently dropped — needed by subscribers
+})
+
+
+def _empty_snapshot(space_name: str) -> dict:
+    """Return a clean, zeroed snapshot for a space."""
+    return {
+        "space_name": space_name,
+        "last_updated": 0.0,
+        "training": {
+            "training_steps": 0,
+            "actor_loss": 0.0,
+            "critic_loss": 0.0,
+            "avn_loss": 0.0,
+            "avn_accuracy": 0.0,
+        },
+        "voting": {
+            "dominant_signal": "NEUTRAL",
+            "buy_count": 0,
+            "sell_count": 0,
+            "last_price": 0.0,       # FIX: added to match allowed fields
+            "signal_source": "LOG",  # FIX: added to match allowed fields
+        },
+    }
+
+
+def _validate_and_normalize(space_name: str, raw: dict) -> Optional[dict]:
+    """
+    Validate incoming payload.  Return a clean normalized dict or None if invalid.
+
+    Strict rules:
+      • Must contain at least one of 'training' or 'voting' keys.
+      • Unknown top-level keys are dropped.
+      • Unknown sub-keys inside training/voting are dropped.
+      • Values are coerced to the correct Python types; malformed values are zeroed.
+    """
+    training_raw = raw.get("training", {})
+    voting_raw = raw.get("voting", {})
+
+    if not isinstance(training_raw, dict):
+        training_raw = {}
+    if not isinstance(voting_raw, dict):
+        voting_raw = {}
+
+    if not training_raw and not voting_raw:
+        return None  # Nothing useful
+
+    # --- training ---
+    def _float(v, default: float = 0.0) -> float:
+        try:
+            return float(v)
+        except (TypeError, ValueError):
+            return default
+
+    def _int(v, default: int = 0) -> int:
+        try:
+            return int(v)
+        except (TypeError, ValueError):
+            return default
+
+    training: dict = {}
+    if training_raw:
+        training = {
+            "training_steps": _int(training_raw.get("training_steps", 0)),
+            "actor_loss":     _float(training_raw.get("actor_loss", 0.0)),
+            "critic_loss":    _float(training_raw.get("critic_loss", 0.0)),
+            "avn_loss":       _float(training_raw.get("avn_loss", 0.0)),
+            "avn_accuracy":   max(0.0, min(1.0, _float(training_raw.get("avn_accuracy", 0.0)))),
+        }
+
+    # --- voting ---
+    voting: dict = {}
+    if voting_raw:
+        raw_signal = voting_raw.get("dominant_signal", "NEUTRAL")
+        if not isinstance(raw_signal, str):
+            raw_signal = "NEUTRAL"
+        raw_source = voting_raw.get("signal_source", "LOG")
+        if not isinstance(raw_source, str):
+            raw_source = "LOG"
+        voting = {
+            "dominant_signal": raw_signal.upper() if raw_signal.upper() in {"BUY", "SELL", "NEUTRAL"} else "NEUTRAL",
+            "buy_count":       _int(voting_raw.get("buy_count", 0)),
+            "sell_count":      _int(voting_raw.get("sell_count", 0)),
+            "last_price":      _float(voting_raw.get("last_price", 0.0)),    # FIX: now passed through
+            "signal_source":   raw_source,                                    # FIX: now passed through
+        }
+
+    return {
+        "space_name": space_name,
+        "training": training,
+        "voting": voting,
+    }
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 2 — CONNECTION MANAGER
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+class ConnectionManager:
+    """
+    Manages publisher (Asset Space) and subscriber (Ranker) WebSocket connections.
+
+    Design:
+      • Publishers  →  write-only lane  (/ws/publish/{space_name})
+      • Subscribers →  read-only lane   (/ws/subscribe)
+      • No cross-talk: subscribers never send to publishers
+    """
+
+    def __init__(self):
+        # Keyed by space_name → WebSocket
+        self._publishers:  Dict[str, WebSocket] = {}
+        # Set of subscriber sockets
+        self._subscribers: Set[WebSocket] = set()
+        # Latest normalized snapshot per space
+        self._snapshots:   Dict[str, dict] = {}
+        # Asyncio lock for thread-safe mutation
+        self._lock = asyncio.Lock()
+        # Message counter (read by /api/state)
+        self._total_ingested: int = 0
+
+    # ── Publisher lifecycle ──────────────────────────────────────────────────────────
+
+    async def register_publisher(self, space_name: str, ws: WebSocket) -> None:
+        await ws.accept()
+        async with self._lock:
+            self._publishers[space_name] = ws
+            if space_name not in self._snapshots:
+                self._snapshots[space_name] = _empty_snapshot(space_name)
+        logger.info(f"📡 Publisher connected: {space_name}  "
+                    f"(total={len(self._publishers)})")
+
+    async def unregister_publisher(self, space_name: str) -> None:
+        async with self._lock:
+            self._publishers.pop(space_name, None)
+        logger.info(f"📡 Publisher disconnected: {space_name}")
+
+    # ── Subscriber lifecycle ─────────────────────────────────────────────────────────
+
+    async def register_subscriber(self, ws: WebSocket) -> None:
+        await ws.accept()
+        async with self._lock:
+            self._subscribers.add(ws)
+        logger.info(f"🔔 Subscriber connected  (total={len(self._subscribers)})")
+
+    async def unregister_subscriber(self, ws: WebSocket) -> None:
+        async with self._lock:
+            self._subscribers.discard(ws)
+        logger.info(f"🔔 Subscriber disconnected  (total={len(self._subscribers)})")
+
+    # ── Ingestion pipeline ───────────────────────────────────────────────────────────
+
+    async def ingest(self, space_name: str, raw_payload: dict) -> None:
+        """
+        Validate → Normalize → Store → Broadcast pipeline.
+        Called for every inbound message from a publisher.
+        """
+        normalized = _validate_and_normalize(space_name, raw_payload)
+        if normalized is None:
+            logger.debug(f"[{space_name}] Payload dropped (no valid fields)")
+            return
+
+        # ✅ FIX #6: deep-copy the snapshot *inside* the lock before releasing.
+        # Without this, snap is a live reference. Another publisher coroutine
+        # can acquire the lock and mutate snap["training"] / snap["voting"]
+        # in-place while json.dumps(snapshot) is still serializing it during
+        # broadcast, producing torn / mixed data sent to subscribers.
+        async with self._lock:
+            snap = self._snapshots.setdefault(space_name, _empty_snapshot(space_name))
+            snap["last_updated"] = time.time()
+
+            if normalized["training"]:
+                snap["training"].update(normalized["training"])
+            if normalized["voting"]:
+                snap["voting"].update(normalized["voting"])
+
+            self._total_ingested += 1
+            snap_copy = copy.deepcopy(snap)  # frozen snapshot for broadcast
+
+        # Broadcast outside the lock using the immutable copy
+        await self._broadcast_update(space_name, snap_copy)
+
+    # ── Broadcast engine ─────────────────────────────────────────────────────────────
+
+    async def _broadcast_update(self, space_name: str, snapshot: dict) -> None:
+        """Fan-out the updated snapshot to all subscribers (event-driven)."""
+        if not self._subscribers:
+            return
+
+        message = json.dumps({
+            "type": "metrics_update",
+            "space_name": space_name,
+            "snapshot": snapshot,
+            "hub_timestamp": time.time(),
+        })
+
+        dead: list = []
+        for ws in list(self._subscribers):
+            try:
+                await ws.send_text(message)
+            except Exception:
+                dead.append(ws)
+
+        if dead:
+            async with self._lock:
+                for ws in dead:
+                    self._subscribers.discard(ws)
+
+    async def send_initial_state(self, ws: WebSocket) -> None:
+        """Send full current state to a newly connected subscriber."""
+        async with self._lock:
+            snapshots_copy = dict(self._snapshots)
+
+        message = json.dumps({
+            "type": "initial_state",
+            "snapshots": snapshots_copy,
+            "hub_timestamp": time.time(),
+        })
+        await ws.send_text(message)
+
+    # ── Read-only accessors (REST API) ───────────────────────────────────────────────
+
+    def get_snapshot(self, space_name: str) -> Optional[dict]:
+        return self._snapshots.get(space_name)
+
+    def get_all_snapshots(self) -> dict:
+        return dict(self._snapshots)
+
+    def get_health(self) -> dict:
+        now = time.time()
+        return {
+            "publishers": {
+                name: {
+                    "last_updated": self._snapshots.get(name, {}).get("last_updated", 0),
+                    "stale_seconds": round(now - self._snapshots.get(name, {}).get("last_updated", now), 1),
+                }
+                for name in self._publishers
+            },
+            "subscriber_count": len(self._subscribers),
+        }
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 3 — FASTAPI APPLICATION
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+app = FastAPI(
+    title="K1RL QUASAR Hub",
+    description="Central WebSocket hub — ingest, normalize, broadcast (one-way)",
+    version="2.0.0",
+)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+manager = ConnectionManager()
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 4 — WEBSOCKET ENDPOINTS
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+@app.websocket("/ws/publish/{space_name}")
+async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
+    """
+    Publisher endpoint — Asset Spaces connect here.
+    SEND ONLY: hub never writes back to this socket.
+
+    Accepted message types:
+      {"type": "metrics", "training": {...}, "voting": {...}}
+      {"type": "training", "data": {...}}
+      {"type": "voting",   "data": {...}}
+      {"type": "heartbeat"}
+    """
+    await manager.register_publisher(space_name, websocket)
+    try:
+        while True:
+            raw_text = await websocket.receive_text()
+            try:
+                data = json.loads(raw_text)
+            except json.JSONDecodeError:
+                logger.warning(f"[{space_name}] Malformed JSON — skipped")
+                continue
+
+            msg_type = data.get("type", "")
+
+            if msg_type == "metrics":
+                # Full combined payload
+                await manager.ingest(space_name, {
+                    "training": data.get("training", {}),
+                    "voting":   data.get("voting",   {}),
+                })
+
+            elif msg_type == "training":
+                await manager.ingest(space_name, {
+                    "training": data.get("data", {}),
+                    "voting":   {},
+                })
+
+            elif msg_type == "voting":
+                await manager.ingest(space_name, {
+                    "training": {},
+                    "voting":   data.get("data", {}),
+                })
+
+            elif msg_type in ("heartbeat", "identify", "ping"):
+                # Silently acknowledged — no reply sent back
+                pass
+
+            else:
+                logger.debug(f"[{space_name}] Unrecognised type '{msg_type}' — dropped")
+
+    except WebSocketDisconnect:
+        pass
+    except Exception as e:
+        logger.error(f"[{space_name}] Publisher error: {e}")
+    finally:
+        await manager.unregister_publisher(space_name)
+
+
+@app.websocket("/ws/subscribe")
+async def ws_subscriber_endpoint(websocket: WebSocket):
+    """
+    Subscriber endpoint — Ranker Space connects here.
+    READ ONLY: subscribers must not send data; any inbound messages are discarded.
+
+    Messages received by subscriber:
+      {"type": "initial_state", "snapshots": {...}, "hub_timestamp": ...}
+      {"type": "metrics_update", "space_name": "...", "snapshot": {...}, "hub_timestamp": ...}
+    """
+    await manager.register_subscriber(websocket)
+    await manager.send_initial_state(websocket)
+    try:
+        while True:
+            # Drain any inbound messages without processing them (read-only contract)
+            await websocket.receive_text()
+    except WebSocketDisconnect:
+        pass
+    except Exception as e:
+        logger.error(f"Subscriber error: {e}")
+    finally:
+        await manager.unregister_subscriber(websocket)
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 5 — REST API (READ-ONLY)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+@app.get("/rankings")
+async def get_rankings():
+    """
+    Return latest snapshot for all assets.
+    Ranker may also poll this endpoint as a fallback.
+    """
+    return {
+        "snapshots": manager.get_all_snapshots(),
+        "timestamp": datetime.utcnow().isoformat() + "Z",
+    }
+
+
+@app.get("/metrics/{space_name}")
+async def get_space_metrics(space_name: str):
+    """Return latest snapshot for a single asset space."""
+    snap = manager.get_snapshot(space_name)
+    if snap is None:
+        return {"error": f"Unknown space: {space_name}"}
+    return snap
+
+
+@app.get("/health")
+async def get_health():
+    return {
+        "status": "ok",
+        "timestamp": datetime.utcnow().isoformat() + "Z",
+        **manager.get_health(),
+    }
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 6 — DASHBOARD UI ROUTES
+# (serves hub_dashboard.html at / and provides /api/state for the polling frontend)
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+_HTML_PATH = Path(os.environ.get(
+    "DASHBOARD_HTML",
+    Path(__file__).parent / "hub_dashboard.html",
+))
+
+
+def _compute_rankings() -> List[dict]:
+    """
+    Derive ranked list from manager snapshots using the AXRVI formula:
+        signal_confidence = max(buy, sell) / (buy + sell)   if total > 0 else 0
+        score             = signal_confidence - avn_accuracy
+    Sorted descending by score.
+    """
+    ranked: List[dict] = []
+    for name, snap in manager.get_all_snapshots().items():
+        training = snap.get("training", {})
+        voting   = snap.get("voting",   {})
+        buy      = voting.get("buy_count",  0)
+        sell     = voting.get("sell_count", 0)
+        total    = buy + sell
+        sig_conf = (max(buy, sell) / total) if total > 0 else 0.0
+        avn_acc  = training.get("avn_accuracy", 0.0)
+        score    = round(sig_conf - avn_acc, 6)
+        ranked.append({
+            "rank":              0,
+            "space_name":        name,
+            "score":             score,
+            "signal_confidence": round(sig_conf, 6),
+            "avn_accuracy":      round(avn_acc,  6),
+            "dominant_signal":   voting.get("dominant_signal", "NEUTRAL"),
+            "buy_count":         buy,
+            "sell_count":        sell,
+            "training_steps":    training.get("training_steps", 0),
+            "actor_loss":        training.get("actor_loss",     0.0),
+            "critic_loss":       training.get("critic_loss",    0.0),
+            "avn_loss":          training.get("avn_loss",       0.0),
+            "last_updated":      snap.get("last_updated",       0.0),
+        })
+    ranked.sort(key=lambda r: r["score"], reverse=True)
+    for i, r in enumerate(ranked):
+        r["rank"] = i + 1
+    return ranked
+
+
+@app.get("/")
+async def serve_dashboard():
+    """Serve the dashboard HTML. Returns a helpful message if the file is missing."""
+    if _HTML_PATH.exists():
+        return FileResponse(str(_HTML_PATH), media_type="text/html")
+    return JSONResponse(
+        status_code=200,
+        content={
+            "service":  "K1RL QUASAR Hub",
+            "status":   "running",
+            "note":     "hub_dashboard.html not found — upload it to the Space",
+            "expected": str(_HTML_PATH),
+            "endpoints": ["/rankings", "/health", "/api/state", "/ws/publish/{space}", "/ws/subscribe"],
+        },
+    )
+
+
+@app.get("/api/state")
+async def api_state():
+    """
+    Full dashboard state polled by hub_dashboard.html every 2 s.
+    Returns rankings + metric history stub + health.
+    """
+    h = manager.get_health()
+    rankings = _compute_rankings()
+    return JSONResponse({
+        "rankings":       rankings,
+        "metric_history": {},          # history kept client-side via snapshot deltas
+        "health": {
+            "hub_connected":    True,  # we ARE the hub
+            "spaces_connected": len(manager.get_all_snapshots()),
+            "messages_rx":      manager._total_ingested,
+            "last_update_ts":   max(
+                (s.get("last_updated", 0) for s in manager.get_all_snapshots().values()),
+                default=0.0,
+            ),
+            "last_update_ago":  round(
+                time.time() - max(
+                    (s.get("last_updated", 0) for s in manager.get_all_snapshots().values()),
+                    default=time.time(),
+                ), 1
+            ),
+            "uptime_seconds":   round(time.time() - _START_TIME, 0),
+            "reconnect_count":  0,
+        },
+        "timestamp": datetime.utcnow().isoformat() + "Z",
+    })
+
+
+_START_TIME = time.time()
+
+
+# ══════════════════════════════════════════════════════════════════════════════════════
+# SECTION 7 — ENTRY POINT
+# ══════════════════════════════════════════════════════════════════════════════════════
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 7860))
+    logger.info(f"🚀 QUASAR Hub starting on port {port}")
+    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")