websocket_hub.py

#!/usr/bin/env python3
"""
╔══════════════════════════════════════════════════════════════════════════════════════╗
║  K1RL QUASAR — CENTRAL WEBSOCKET HUB  v2.3-realtime-signals                          ║
║  ──────────────────────────────────────────────────────────────────────────────────  ║
║                                                                                      ║
║  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)                                 ║
║                                                                                      ║
║  RANKER LOGS API (FIX v2.2 — moved here from hub_dashboard_service port 8052):      ║
║    GET /api/ranker/logs/recent   → recent log entries  (?limit=N&category=X)        ║
║    GET /api/ranker/logs/stats    → log statistics                                   ║
║    GET /api/ranker/logs/asset/X  → logs for asset X                                 ║
║    GET /api/ranker/logs/level/X  → logs by level                                    ║
║    GET /api/ranker/logs/export   → download JSON                                    ║
║    GET /api/ranker/logs/debug    → file discovery diagnostics                       ║
║                                                                                      ║
║  TRADE API (served natively — no patch script needed):                               ║
║    GET /api/trades          → full open + closed state + stats                       ║
║    GET /api/trades/open     → open trades only                                       ║
║    GET /api/trades/closed   → recent closed trades + stats  (?limit=N)              ║
║    GET /api/health          → service health including trade counts                  ║
║                                                                                      ║
║  REALTIME SIGNAL CHANNEL (NEW v2.3 — per-tick AVN inferenced signals):              ║
║    WS  /ws/signals          → fan-out of {asset, action, price, ts, seq}            ║
║                               derived from buy/sell count deltas. No engine         ║
║                               changes required. Consumed by the AXRVI ranker.       ║
║                                                                                      ║
║  VERSION: v2.3-realtime-signals  |  2026-04-26                                       ║
╚══════════════════════════════════════════════════════════════════════════════════════╝
"""

import asyncio
import copy
import glob
import json
import logging
import os
import re
import threading
import time
from collections import deque
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Set

import uvicorn
from fastapi import FastAPI, Request, 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
# ══════════════════════════════════════════════════════════════════════════════════════

_ALLOWED_TRAINING_FIELDS: frozenset = frozenset({
    "training_steps",
    "actor_loss",
    "critic_loss",
    "avn_loss",
    "avn_accuracy",
})

_ALLOWED_VOTING_FIELDS: frozenset = frozenset({
    # Canonical (post-v2 schema) ────────────────────────────────────────────
    "flip_direction",
    "flip_action",
    "last_price",
    "signal_source",
    # Legacy aliases accepted on the wire — translated inside the hub ───────
    # Executor spaces still emit dominant_signal / buy_count / sell_count
    # because their publisher code pre-dates the flip_* rename. We accept
    # them at ingestion and map them to flip_direction so downstream code
    # (ranker, dashboard) can stay on the unified flip_* schema.
    "dominant_signal",
    "buy_count",
    "sell_count",
})


def _empty_snapshot(space_name: str) -> dict:
    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": {
            "flip_direction": "NONE",
            "flip_action":    "HOLD",
            "last_price":     0.0,
            "signal_source":  "LOG",
        },
    }


def _validate_and_normalize(space_name: str, raw: dict) -> Optional[dict]:
    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

    def _float(v, default: float = 0.0) -> float:
        try:    return float(v)
        except: return default

    def _int(v, default: int = 0) -> int:
        try:    return int(v)
        except: 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: dict = {}
    if voting_raw:
        # ── Canonical fields (new schema) ──────────────────────────────────
        raw_direction = voting_raw.get("flip_direction", None)
        raw_action    = voting_raw.get("flip_action",    None)
        raw_source    = voting_raw.get("signal_source", "LOG")
        if not isinstance(raw_source, str):
            raw_source = "LOG"

        # ── Legacy fallback: translate dominant_signal → flip_direction ────
        # The asset/executor spaces still publish the old schema
        # ({dominant_signal, buy_count, sell_count}) because their publisher
        # code was never rewritten. We map it to the new flip_* schema here
        # so everything downstream stays on one naming convention.
        #
        #   dominant_signal=BUY     → flip_direction=BUY,  flip_action=ENTRY
        #   dominant_signal=SELL    → flip_direction=SELL, flip_action=ENTRY
        #   dominant_signal=NEUTRAL → flip_direction=NONE, flip_action=HOLD
        #
        # If a payload carries BOTH flip_direction and dominant_signal,
        # flip_direction wins (forward-compat with newer publishers).
        if not isinstance(raw_direction, str) or raw_direction.upper() not in {"BUY", "SELL", "NONE"}:
            legacy_signal = voting_raw.get("dominant_signal", None)
            if isinstance(legacy_signal, str):
                legacy_upper = legacy_signal.upper()
                if legacy_upper in {"BUY", "SELL"}:
                    raw_direction = legacy_upper
                    if not isinstance(raw_action, str) or raw_action.upper() == "HOLD":
                        raw_action = "ENTRY"
                elif legacy_upper == "NEUTRAL":
                    raw_direction = "NONE"
                    raw_action    = "HOLD"

        if not isinstance(raw_direction, str):
            raw_direction = "NONE"
        if not isinstance(raw_action, str):
            raw_action = "HOLD"

        _clean_direction = raw_direction.upper() if raw_direction.upper() in {"BUY", "SELL", "NONE"} else "NONE"
        _clean_action    = raw_action.upper()    if raw_action.upper()    in {"ENTRY", "EXIT", "HOLD", "REBALANCE", "REDUCE", "SKIP"} else "HOLD"

        voting = {
            "flip_direction": _clean_direction,
            "flip_action":    _clean_action,
            "last_price":     _float(voting_raw.get("last_price", 0.0)),
            "signal_source":  raw_source,
            # Preserve vote counts (legacy format) so the ranker can compute
            # vote-ratio confidence when the publisher emits them.
            "buy_count":      _int(voting_raw.get("buy_count",  0)),
            "sell_count":     _int(voting_raw.get("sell_count", 0)),
        }

    return {
        "space_name": space_name,
        "training":   training,
        "voting":     voting,
    }


_METRIC_HISTORY_LEN: int = int(os.environ.get("QUASAR_METRIC_HISTORY", "200"))


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 2 — CONNECTION MANAGER
# ══════════════════════════════════════════════════════════════════════════════════════

class ConnectionManager:
    # All training field names the hub will accept (including common ranker aliases)
    _TRAINING_KEYS: frozenset = frozenset({
        "actor_loss", "critic_loss", "avn_loss", "avn_accuracy", "training_steps",
        "a_loss", "c_loss", "loss_actor", "loss_critic", "loss_avn",
        "acc", "accuracy", "step", "steps", "n_steps",
    })
    _TRAINING_ALIAS: dict = {
        "a_loss":      "actor_loss",  "loss_actor":  "actor_loss",
        "c_loss":      "critic_loss", "loss_critic": "critic_loss",
        "loss_avn":    "avn_loss",
        "acc":         "avn_accuracy","accuracy":    "avn_accuracy",
        "step":        "training_steps","steps":     "training_steps","n_steps": "training_steps",
    }
    _VOTING_KEYS: frozenset = frozenset({
        # Canonical
        "flip_direction", "flip_action", "last_price", "signal_source",
        # Legacy (translated inside _validate_and_normalize)
        "dominant_signal", "buy_count", "sell_count",
    })

    def __init__(self):
        self._publishers:     Dict[str, WebSocket]       = {}
        self._subscribers:    Set[WebSocket]              = set()
        self._snapshots:      Dict[str, dict]             = {}
        self._history:        Dict[str, deque]            = {}   # rolling per-space history
        self._lock            = asyncio.Lock()
        self._total_ingested: int                         = 0
        self._msg_counts:     Dict[str, Dict[str, int]]  = {}   # {space: {msg_type: count}}

    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}  (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}")

    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)})")

    async def ingest(self, space_name: str, raw_payload: dict) -> None:
        normalized = _validate_and_normalize(space_name, raw_payload)
        if normalized is None:
            logger.debug(f"[{space_name}] Payload dropped (no valid fields)")
            return

        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)

            # ── Rolling metric history (for sparkline charts in dashboard) ────────
            # Only record a point when training fields arrive AND at least one
            # loss/accuracy field is non-zero (avoids flooding history with empty
            # default-value points before training metrics connect).
            training = snap["training"]
            if normalized["training"] and any(
                training.get(k, 0) != 0
                for k in ("actor_loss", "critic_loss", "avn_loss", "avn_accuracy")
            ):
                if space_name not in self._history:
                    self._history[space_name] = deque(maxlen=_METRIC_HISTORY_LEN)
                self._history[space_name].append({
                    "ts":           snap["last_updated"],
                    "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),
                    "training_steps": training.get("training_steps", 0),
                })

        await self._broadcast_update(space_name, snap_copy)

    def _extract_signal_metadata(self, snapshot: dict) -> dict:
        """
        Extract signal and score metadata from snapshot for broadcast enrichment.
        Computes confidence scores and signal strength metrics.
        """
        voting = snapshot.get("voting", {})
        training = snapshot.get("training", {})
        
        flip_direction = voting.get("flip_direction", "NONE")
        flip_action = voting.get("flip_action", "HOLD")
        last_price = voting.get("last_price", 0.0)
        signal_source = voting.get("signal_source", "LOG")
        
        # Vote-based confidence (if available from legacy fields)
        buy_count = voting.get("buy_count", 0)
        sell_count = voting.get("sell_count", 0)
        total_votes = buy_count + sell_count
        vote_confidence = 0.0
        if total_votes > 0:
            max_votes = max(buy_count, sell_count)
            vote_confidence = float(max_votes) / total_votes
        
        # Training-based confidence (accuracy metric)
        train_confidence = training.get("avn_accuracy", 0.0)
        
        # Blended confidence score [0, 1]
        if vote_confidence > 0 and train_confidence > 0:
            blended_confidence = (vote_confidence + train_confidence) / 2.0
        elif vote_confidence > 0:
            blended_confidence = vote_confidence
        else:
            blended_confidence = train_confidence
        
        # Signal strength based on action type and confidence
        signal_strength = 0.0
        if flip_direction in ("BUY", "SELL"):
            signal_strength = blended_confidence
        
        return {
            "signal": {
                "direction": flip_direction,
                "action": flip_action,
                "source": signal_source,
            },
            "scores": {
                "vote_confidence": round(vote_confidence, 4),
                "train_confidence": round(train_confidence, 4),
                "blended_confidence": round(blended_confidence, 4),
                "signal_strength": round(signal_strength, 4),
            },
            "voting_stats": {
                "buy_count": buy_count,
                "sell_count": sell_count,
                "total_votes": total_votes,
            },
            "price": last_price,
        }

    async def _broadcast_update(self, space_name: str, snapshot: dict) -> None:
        if not self._subscribers:
            return
        
        # Extract signal metadata for this asset
        signal_metadata = self._extract_signal_metadata(snapshot)
        
        # Enhanced message format with per-asset signals and scores
        message = json.dumps({
            "type":          "metrics_update",
            "asset": {
                "space_name":    space_name,
                "metadata":      signal_metadata,
                "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:
        async with self._lock:
            snapshots_copy = dict(self._snapshots)
        
        # Enrich each snapshot with signal metadata
        enriched_assets = {}
        for space_name, snapshot in snapshots_copy.items():
            signal_metadata = self._extract_signal_metadata(snapshot)
            enriched_assets[space_name] = {
                "metadata": signal_metadata,
                "snapshot": snapshot,
            }
        
        message = json.dumps({
            "type":          "initial_state",
            "assets":        enriched_assets,
            "hub_timestamp": time.time(),
        })
        await ws.send_text(message)

    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 record_msg(self, space_name: str, msg_type: str) -> None:
        """Increment per-space message type counter (non-blocking, called from publisher loop)."""
        counts = self._msg_counts.setdefault(space_name, {})
        counts[msg_type] = counts.get(msg_type, 0) + 1

    def get_msg_counts(self) -> dict:
        return {s: dict(c) for s, c in self._msg_counts.items()}

    def get_metric_history(self) -> dict:
        """Return a plain dict of {space_name: [point, …]} for all spaces with history."""
        return {name: list(dq) for name, dq in self._history.items()}

    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 — HUB TRADE STORE  (in-memory, fed by WebSocket messages)
# ══════════════════════════════════════════════════════════════════════════════════════
#
# ROOT CAUSE FIX: The previous TradeLogParser read from *.log files on THIS container
# (/app/ranker_logs).  Those files NEVER exist on the Executo Hub space — they are
# written by ranker processes running in the individual executor spaces (V75, V50, …),
# each in their own separate container with their own filesystem.
#
# Fix: replace file-based parsing with an in-memory store that is populated when
# executor spaces send WebSocket trade events to this hub.
#
# Executor spaces must send:
#   {"type": "trade_opened", "data": {trade_id, asset, direction, entry, qty, opened_at}}
#   {"type": "trade_closed",  "data": {trade_id, asset, pnl, exit_price, closed_at}}
#
# See ranker_logging.py — the on_event callback already fires for every log entry.
# Wire it in your executor space's ranker like this:
#
#   def _trade_ws_hook(entry: dict):
#       cat = entry.get("category", "")
#       msg = entry.get("message", "")
#       if cat != "TRADE":
#           return
#       meta = entry.get("metadata") or {}
#       if "TRADE OPENED" in msg:
#           asyncio.create_task(ws.send_text(json.dumps({
#               "type": "trade_opened",
#               "data": {
#                   "trade_id":  meta.get("trade_id"),
#                   "asset":     entry.get("asset"),
#                   "direction": meta.get("direction", "?"),
#                   "entry":     meta.get("price"),
#                   "qty":       meta.get("qty", 0.0),
#                   "opened_at": entry.get("timestamp", ""),
#               },
#           })))
#       elif "TRADE CLOSED" in msg:
#           asyncio.create_task(ws.send_text(json.dumps({
#               "type": "trade_closed",
#               "data": {
#                   "trade_id":  meta.get("trade_id"),
#                   "asset":     entry.get("asset"),
#                   "pnl":       meta.get("pnl", 0.0),
#                   "exit_price": meta.get("exit_price"),
#                   "closed_at": entry.get("timestamp", ""),
#               },
#           })))
#
#   ranker_logger = RankerLogger(..., on_event=_trade_ws_hook)
# ══════════════════════════════════════════════════════════════════════════════════════

class HubTradeStore:
    """
    In-memory trade store populated by WebSocket trade-event messages from
    executor spaces.  Thread-safe.  Replaces the broken file-based TradeLogParser.
    """

    def __init__(self) -> None:
        self._open:   Dict[str, dict] = {}   # trade_id → record
        self._closed: List[dict]      = []   # newest-first, capped at 500
        self._stats = {
            "total_opened": 0,
            "total_closed": 0,
            "total_pnl":    0.0,
            "win_count":    0,
            "loss_count":   0,
        }
        self._lock = threading.Lock()

    def open_trade(self, space_name: str, data: dict) -> None:
        trade_id  = data.get("trade_id") or f"{space_name}_{int(time.time())}"
        direction = str(data.get("direction", "?")).upper()
        entry_px  = data.get("entry") or data.get("price") or 0.0
        with self._lock:
            self._open[trade_id] = {
                "trade_id":  trade_id,
                "asset":     data.get("asset", space_name),
                "direction": direction,
                "entry":     float(entry_px),
                "qty":       float(data.get("qty", 0.0)),
                "opened_at": data.get("opened_at", datetime.utcnow().isoformat()[:19]),
                "status":    "OPEN",
            }
            self._stats["total_opened"] += 1
        logger.info(
            f"[HubTradeStore] OPEN  {trade_id} | {direction} @ {entry_px} "
            f"(from {space_name})"
        )

    def close_trade(self, space_name: str, data: dict) -> None:
        trade_id   = data.get("trade_id")
        pnl        = float(data.get("pnl", 0.0))
        exit_price = data.get("exit_price")
        with self._lock:
            trade = self._open.pop(trade_id, {}) if trade_id else {}
            closed_rec = {
                "trade_id":   trade_id or "UNKNOWN",
                "asset":      data.get("asset") or trade.get("asset", space_name),
                "direction":  str(data.get("direction") or trade.get("direction", "?")).upper(),
                "entry":      data.get("entry")      or trade.get("entry",     0.0),
                "exit_price": float(exit_price) if exit_price is not None else None,
                "qty":        data.get("qty")        or trade.get("qty",       0.0),
                "pnl":        pnl,
                "closed_at":  data.get("closed_at", datetime.utcnow().isoformat()[:19]),
                "status":     "CLOSED",
            }
            self._closed.insert(0, closed_rec)       # newest-first
            if len(self._closed) > 500:
                self._closed = self._closed[:500]
            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.info(
            f"[HubTradeStore] CLOSE {trade_id} | pnl={pnl:+.4f} "
            f"(from {space_name})"
        )

    def get_state(self) -> dict:
        with self._lock:
            open_list   = list(self._open.values())
            closed_list = list(self._closed[:100])   # newest 100 for dashboard
            stats       = dict(self._stats)
        total_closed = stats["total_closed"]
        stats["win_rate"] = (
            round(stats["win_count"] / total_closed * 100, 1)
            if total_closed > 0 else 0.0
        )
        return {
            "open":   open_list,
            "closed": closed_list,
            "stats":  stats,
        }


# ── Bootstrap ─────────────────────────────────────────────────────────────────────────
_LOG_DIR    = os.environ.get("RANKER_LOG_DIR", "/app/ranker_logs")
_hub_trades = HubTradeStore()
logger.info("✅ HubTradeStore initialised — awaiting trade_opened/trade_closed WS messages")

# ── AXRVI live rankings store ─────────────────────────────────────────────────────────
# Populated by POST /api/flip/rankings from the Executo ranker after every
# rank_and_gate() cycle (~every 5s). Falls back to hub-snapshot scoring when stale.
_axrvi_rankings:    List[dict] = []
_axrvi_rankings_ts: float      = 0.0
_AXRVI_RANKINGS_TTL: float     = 60.0   # FIX: extended 30→60s to survive slow ranker POST cycles

# ══════════════════════════════════════════════════════════════════════════════════════
# FLIP FAST-PATH CHANNEL
# ══════════════════════════════════════════════════════════════════════════════════════
# A dedicated, high-priority, per-asset latest-wins cache + broadcaster for BUY/SELL
# flip events. Flips are time-sensitive (change every few hundred ms) and must not
# be delayed by the slower training/snapshot machinery.
#
# Architecture:
#   Engine  ──► {"type":"flip", "data": {"flip_direction":"BUY"|"SELL"|"NONE", ...}}
#               │
#               ▼  (short-circuit handler in publisher_handler — skips _validate_and_normalize)
#         _store_flip(asset, payload)  ── atomic write under _flip_lock, seq++
#               │
#               ▼  sets _flip_dirty event
#         _flip_broadcaster_loop()  ── dedicated asyncio task, runs every 30 ms
#               │                      (or immediately when dirty, whichever is first)
#               ▼
#         /ws/flips subscribers (ranker, executor) ── push JSON delta
#
# Per-asset signal consistency:
#   • Latest-wins: newer flip replaces older one atomically under the lock.
#   • Monotonic seq per asset so consumers can detect replays / out-of-order.
#   • On subscribe, a full replay of the current cache is sent first so clients
#     start with a consistent view.
#
# Backward-compat auto-bridge:
#   Any inbound voting message whose normalized flip_direction ∈ {BUY, SELL} is
#   AUTOMATICALLY promoted into the flip cache as well. This means existing
#   publishers that send {"type":"voting","dominant_signal":"BUY"} or metrics with
#   voting nested still benefit from the fast path without any code change on
#   their end.
# ──────────────────────────────────────────────────────────────────────────────────────

_FLIP_BROADCAST_INTERVAL_SEC: float = 0.030   # 30 ms cadence per sketch
_flip_cache: Dict[str, dict]        = {}      # {asset: {flip_direction, flip_action, last_price, ts, seq, ...}}
_flip_seq:   Dict[str, int]         = {}      # monotonic per-asset sequence
_flip_lock:  asyncio.Lock           = asyncio.Lock()
_flip_dirty: asyncio.Event          = asyncio.Event()
_flip_subscribers: Set[WebSocket]   = set()
_flip_subscribers_lock: asyncio.Lock = asyncio.Lock()


async def _store_flip(asset: str, raw: dict) -> Optional[dict]:
    """
    Write one flip into the per-asset cache with signal consistency.
    Returns the normalized flip dict (or None if the payload was rejected).

    Contract:
      • Accepts either {flip_direction: BUY|SELL|NONE|NEUTRAL} or legacy
        {dominant_signal: BUY|SELL|NEUTRAL}. NEUTRAL is coerced to NONE.
      • Write is atomic under _flip_lock (latest-wins).
      • Sequence number is incremented monotonically per asset.
      • Sets _flip_dirty so the broadcaster picks it up within 30 ms.
    """
    if not isinstance(raw, dict):
        return None

    # Normalize direction — accept both new and legacy naming
    raw_dir = raw.get("flip_direction")
    if not isinstance(raw_dir, str) or raw_dir.upper() not in {"BUY", "SELL", "NONE", "NEUTRAL"}:
        raw_dir = raw.get("dominant_signal")
    if isinstance(raw_dir, str):
        raw_dir = raw_dir.upper()
        if raw_dir == "NEUTRAL":
            raw_dir = "NONE"
    else:
        raw_dir = "NONE"
    if raw_dir not in {"BUY", "SELL", "NONE"}:
        raw_dir = "NONE"

    # Normalize action
    raw_act = raw.get("flip_action")
    if isinstance(raw_act, str) and raw_act.upper() in {"ENTRY", "EXIT", "HOLD", "REBALANCE", "REDUCE", "SKIP"}:
        raw_act = raw_act.upper()
    else:
        raw_act = "ENTRY" if raw_dir in {"BUY", "SELL"} else "HOLD"

    try:
        price = float(raw.get("last_price", 0.0) or 0.0)
    except Exception:
        price = 0.0
    source = raw.get("signal_source", "engine")
    if not isinstance(source, str):
        source = "engine"

    async with _flip_lock:
        prev = _flip_cache.get(asset)
        # De-dup: if direction AND action AND price are unchanged, don't bump seq.
        # (Signal consistency: the ranker only reacts to real flips.)
        if prev and prev["flip_direction"] == raw_dir and prev["flip_action"] == raw_act and prev["last_price"] == price:
            # Refresh timestamp only — no seq bump, no dirty flag
            prev["ts"] = time.time()
            return prev

        seq = _flip_seq.get(asset, 0) + 1
        _flip_seq[asset] = seq
        flip = {
            "asset":          asset,
            "flip_direction": raw_dir,
            "flip_action":    raw_act,
            "last_price":     price,
            "signal_source":  source,
            "ts":             time.time(),
            "seq":            seq,
        }
        _flip_cache[asset] = flip

    _flip_dirty.set()
    return flip


async def _flip_broadcaster_loop():
    """
    Dedicated task: every 30 ms (or immediately on dirty) push the current
    flip cache to all /ws/flips subscribers. Dead connections are pruned.
    """
    logger.info(f"🎯 Flip broadcaster started | interval={_FLIP_BROADCAST_INTERVAL_SEC*1000:.0f}ms")
    while True:
        try:
            # Wait for either a dirty signal or the 30 ms timeout
            try:
                await asyncio.wait_for(_flip_dirty.wait(), timeout=_FLIP_BROADCAST_INTERVAL_SEC)
            except asyncio.TimeoutError:
                pass
            _flip_dirty.clear()

            async with _flip_subscribers_lock:
                if not _flip_subscribers:
                    continue
                subscribers_snapshot = list(_flip_subscribers)

            async with _flip_lock:
                if not _flip_cache:
                    continue
                flips_snapshot = list(_flip_cache.values())

            msg = {
                "type":          "flip_delta",
                "flips":         flips_snapshot,
                "total_assets":  len(flips_snapshot),
                "hub_timestamp": time.time(),
            }

            dead: List[WebSocket] = []
            for ws in subscribers_snapshot:
                try:
                    await ws.send_json(msg)
                except Exception:
                    dead.append(ws)

            if dead:
                async with _flip_subscribers_lock:
                    for ws in dead:
                        _flip_subscribers.discard(ws)
                logger.info(f"🎯 Pruned {len(dead)} dead flip subscriber(s)")

        except asyncio.CancelledError:
            logger.info("🎯 Flip broadcaster cancelled")
            raise
        except Exception as e:
            logger.error(f"🎯 Flip broadcaster error: {e}")
            await asyncio.sleep(0.1)


# ══════════════════════════════════════════════════════════════════════════════════════
# REALTIME SIGNAL FAST-PATH CHANNEL  (v2.3 — pure per-tick AVN inferenced signals)
# ══════════════════════════════════════════════════════════════════════════════════════
# Purpose
# ───────
# Stream the *per-tick* AVN action (BUY / SELL / HOLD) — i.e. the realtime
# inferenced signal emitted by each asset engine on every Redis tick — to
# downstream consumers (the AXRVI ranker today, executor tomorrow) with no
# flip / dominant-signal translation layer in between.
#
# Why this exists
# ───────────────
# The pre-existing /ws/flips channel rebroadcasts the *cumulative* dominant
# voting signal aggregated by _best_voting_unlocked() in V75 — i.e. an EMA of
# the per-tick action. It loses the per-tick information by design. The ranker
# wants raw per-tick activity, not aggregates.
#
# Wire format (no V75 changes required)
# ─────────────────────────────────────
# The V75 publisher already emits {buy_count, sell_count, dominant_signal} on
# every tick via the voting payload. update_from_redis(action) increments
# either buy_count or sell_count by 1 *before* the publish, so a delta of +1
# in one of those counters between two consecutive frames *is* the per-tick
# action. This is what we extract and rebroadcast as {action, price, ts, seq}.
# Engine-side code is untouched.
#
#   V75 tick (action=BUY)  ──► buy_count: 4 → 5
#                          ──► publish_voting({buy_count:5, sell_count:3, …})
#                              │
#   Hub _validate_and_normalize() runs ──► normalized voting carries the new counts
#                              │
#   _detect_realtime_signal(asset, normalized_voting):
#       prev = _prev_counts.get(asset, {buy:0, sell:0})
#       db = new.buy - prev.buy  →  +1 (BUY tick happened)
#       ds = new.sell - prev.sell → 0
#       _emit_signal(asset, "BUY", price, source="voting_delta")
#       _prev_counts[asset] = new counts
#
# Per-asset signal consistency
# ────────────────────────────
#   • Latest-wins with monotonic seq per asset (consumer can detect replays).
#   • Atomic write under _signal_lock (no torn reads).
#   • On subscribe, full cache replay first → consistent starting view.
#   • If multiple ticks batch into one frame (db>1 or ds>1), we emit the
#     dominant-side action once (matching the dominant of the batch). This is
#     a graceful degradation — at typical ~10 Hz tick rates batching is rare.
#   • Counter resets (buy/sell counts go down or to zero) are detected and
#     reset _prev_counts without emitting a phantom signal.
# ──────────────────────────────────────────────────────────────────────────────────────

_SIGNAL_BROADCAST_INTERVAL_SEC: float = 0.030   # 30 ms, matching flip cadence
_signal_cache: Dict[str, dict]        = {}      # {asset: {action, price, source, ts, seq, ...}}
_signal_seq:   Dict[str, int]         = {}      # monotonic per-asset
_signal_lock:  asyncio.Lock           = asyncio.Lock()
_signal_dirty: asyncio.Event          = asyncio.Event()
_signal_subscribers: Set[WebSocket]   = set()
_signal_subscribers_lock: asyncio.Lock = asyncio.Lock()

# Per-asset previous (buy_count, sell_count) used to compute per-tick deltas.
# Keyed by space_name. Reset to current values whenever counts go DOWN
# (publisher restart / counter wraparound) without emitting a phantom signal.
_prev_counts: Dict[str, Dict[str, int]] = {}
_prev_counts_lock: threading.Lock        = threading.Lock()


async def _emit_signal(
    asset:  str,
    action: str,
    price:  float,
    source: str = "voting_delta",
) -> Optional[dict]:
    """
    Publish one realtime signal event into the per-asset cache.

    Contract:
      • action ∈ {BUY, SELL, HOLD} (anything else coerced to HOLD).
      • Atomic write under _signal_lock (latest-wins).
      • Monotonic seq per asset.
      • Sets _signal_dirty so the broadcaster picks it up within 30 ms.
      • Returns the normalized signal dict, or None if rejected.
    """
    if not isinstance(action, str):
        return None
    action = action.upper()
    if action not in {"BUY", "SELL", "HOLD"}:
        action = "HOLD"

    try:
        price = float(price or 0.0)
    except Exception:
        price = 0.0

    if not isinstance(source, str):
        source = "voting_delta"

    async with _signal_lock:
        seq = _signal_seq.get(asset, 0) + 1
        _signal_seq[asset] = seq
        sig = {
            "asset":  asset,
            "action": action,
            "price":  price,
            "source": source,
            "ts":     time.time(),
            "seq":    seq,
        }
        _signal_cache[asset] = sig

    _signal_dirty.set()
    return sig


def _detect_realtime_signal(asset: str, voting_normalized: dict) -> Optional[str]:
    """
    Compare current (buy_count, sell_count) against the previously-seen pair
    for this asset and infer which per-tick action just occurred.

    Returns the inferred action (BUY / SELL / HOLD) when a delta is detected,
    or None if nothing actionable changed (e.g. first frame, equal counts,
    counter reset).

    NOTE: This is a *synchronous* helper — it only mutates _prev_counts under
    a thin threading.Lock. Emission is done by the caller via _emit_signal()
    inside the asyncio publisher_handler.
    """
    if not isinstance(voting_normalized, dict):
        return None

    try:
        new_buy  = int(voting_normalized.get("buy_count",  0) or 0)
        new_sell = int(voting_normalized.get("sell_count", 0) or 0)
    except Exception:
        return None

    with _prev_counts_lock:
        prev = _prev_counts.get(asset)
        if prev is None:
            # First time we see this asset — record baseline, no emission.
            _prev_counts[asset] = {"buy": new_buy, "sell": new_sell}
            return None

        prev_buy  = prev.get("buy",  0)
        prev_sell = prev.get("sell", 0)

        # Counter regression (publisher restart / wraparound) — re-baseline,
        # no phantom emission.
        if new_buy < prev_buy or new_sell < prev_sell:
            _prev_counts[asset] = {"buy": new_buy, "sell": new_sell}
            return None

        db = new_buy  - prev_buy
        ds = new_sell - prev_sell

        # Always update the baseline before returning.
        _prev_counts[asset] = {"buy": new_buy, "sell": new_sell}

    if db == 0 and ds == 0:
        # No new ticks since last frame. Could still be a HOLD heartbeat;
        # we treat lack-of-delta as silence (no signal).
        return None

    # Both counters incremented in the same frame: very rare batching case
    # (multiple ticks coalesced). Emit the dominant side — if equal, BUY wins
    # (arbitrary but deterministic). This is a graceful degradation, not the
    # common path.
    if db > ds:
        return "BUY"
    if ds > db:
        return "SELL"
    return "BUY"  # tie-break — extremely rare


async def _signal_broadcaster_loop():
    """
    Dedicated task: every 30 ms (or immediately on dirty) push the current
    signal cache to all /ws/signals subscribers. Dead connections are pruned.
    """
    logger.info(
        f"📡 Signal broadcaster started | "
        f"interval={_SIGNAL_BROADCAST_INTERVAL_SEC * 1000:.0f}ms"
    )
    while True:
        try:
            try:
                await asyncio.wait_for(
                    _signal_dirty.wait(),
                    timeout=_SIGNAL_BROADCAST_INTERVAL_SEC,
                )
            except asyncio.TimeoutError:
                pass
            _signal_dirty.clear()

            async with _signal_subscribers_lock:
                if not _signal_subscribers:
                    continue
                subscribers_snapshot = list(_signal_subscribers)

            async with _signal_lock:
                if not _signal_cache:
                    continue
                signals_snapshot = list(_signal_cache.values())

            msg = {
                "type":          "signal_delta",
                "signals":       signals_snapshot,
                "total_assets":  len(signals_snapshot),
                "hub_timestamp": time.time(),
            }

            dead: List[WebSocket] = []
            for ws in subscribers_snapshot:
                try:
                    await ws.send_json(msg)
                except Exception:
                    dead.append(ws)

            if dead:
                async with _signal_subscribers_lock:
                    for ws in dead:
                        _signal_subscribers.discard(ws)
                logger.info(f"📡 Pruned {len(dead)} dead signal subscriber(s)")

        except asyncio.CancelledError:
            logger.info("📡 Signal broadcaster cancelled")
            raise
        except Exception as e:
            logger.error(f"📡 Signal broadcaster error: {e}")
            await asyncio.sleep(0.1)


async def _maybe_emit_realtime_signal(space_name: str, voting_payload: dict) -> None:
    """
    Convenience wrapper called from the publisher_handler ingestion paths.
    Detects a per-tick delta and emits a signal event when one is found.

    `voting_payload` is the *raw* voting dict received on the wire (still
    carries buy_count/sell_count plus last_price). We pass it through
    directly — no need to wait for the full snapshot normalization pass.
    """
    if not isinstance(voting_payload, dict):
        return

    action = _detect_realtime_signal(space_name, voting_payload)
    if action is None:
        return

    try:
        price = float(voting_payload.get("last_price", 0.0) or 0.0)
    except Exception:
        price = 0.0

    src = voting_payload.get("signal_source", "voting_delta")
    if not isinstance(src, str):
        src = "voting_delta"

    await _emit_signal(space_name, action, price, source=src)


# ── Top-3 WebSocket client registry ───────────────────────────────────────────────────
# top3_client.py connects here and receives top3_rankings broadcasts whenever the
# Executo ranker POSTs new rankings via POST /api/flip/rankings.
_top3_clients: Set[WebSocket] = set()
_top3_clients_lock = asyncio.Lock()


async def _broadcast_top3_rankings(rankings: List[dict]) -> None:
    """
    Broadcast a top3_rankings message to all /ws/top3 subscribers.
    Called immediately after /api/flip/rankings receives a fresh ranking list.
    Dead connections are pruned automatically.
    """
    if not _top3_clients:
        return
    msg = json.dumps({
        "type":          "top3_rankings",
        "rankings":      rankings,
        "total_assets":  len(rankings),
        "hub_timestamp": time.time(),
    })
    dead: list = []
    async with _top3_clients_lock:
        clients = list(_top3_clients)
    for ws in clients:
        try:
            await ws.send_text(msg)
        except Exception:
            dead.append(ws)
    if dead:
        async with _top3_clients_lock:
            for ws in dead:
                _top3_clients.discard(ws)
        logger.debug(f"[top3] Pruned {len(dead)} dead client(s)")


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 4 — FASTAPI APPLICATION
# ══════════════════════════════════════════════════════════════════════════════════════

app = FastAPI(
    title="K1RL QUASAR Hub",
    description="Central WebSocket hub — ingest, normalize, broadcast (one-way)",
    version="2.2.0",
)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

manager = ConnectionManager()


@app.on_event("startup")
async def _on_startup():
    """Start background tasks — flip broadcaster runs on the event loop as a
    dedicated task (Python's equivalent of a dedicated thread for asyncio)."""
    logger.info("🚀 HubTradeStore ready (no background scanner needed)")
    asyncio.create_task(_flip_broadcaster_loop())
    # New realtime per-tick signal channel (parallel to the flip channel,
    # consumed by the AXRVI ranker). See REALTIME SIGNAL FAST-PATH CHANNEL
    # comment block above for protocol details.
    asyncio.create_task(_signal_broadcaster_loop())


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 5 — WEBSOCKET ENDPOINTS
# ══════════════════════════════════════════════════════════════════════════════════════

@app.websocket("/ws/publish/{space_name}")
async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
    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", "")

            # ── Track per-space message type counts (for /api/debug/hub) ─────────
            manager.record_msg(space_name, msg_type)

            # ── Route by type ────────────────────────────────────────────────────
            if msg_type == "flip":
                # ★ HIGH-PRIORITY FLIP FAST PATH ★
                # Bypasses the heavier _validate_and_normalize / snapshot path
                # and writes directly into the flip cache. The broadcaster task
                # picks it up within 30 ms and pushes to all /ws/flips subscribers.
                flip_raw = data.get("data") or {
                    k: v for k, v in data.items() if k != "type"
                }
                stored = await _store_flip(space_name, flip_raw)
                if stored and stored["flip_direction"] in {"BUY", "SELL"}:
                    logger.info(
                        f"[{space_name}] ⚡ flip#{stored['seq']} | "
                        f"{stored['flip_direction']}/{stored['flip_action']} @ {stored['last_price']:.5f}"
                    )
                # Also mirror into the main snapshot so clients that only use
                # /ws/subscribe (dashboard, legacy) stay in sync.
                await manager.ingest(space_name, {
                    "training": {},
                    "voting": {
                        "flip_direction": flip_raw.get("flip_direction") or flip_raw.get("dominant_signal", "NONE"),
                        "flip_action":    flip_raw.get("flip_action", "ENTRY"),
                        "last_price":     flip_raw.get("last_price", 0.0),
                        "signal_source":  flip_raw.get("signal_source", "engine"),
                    },
                })

            elif msg_type == "metrics":
                # Combined payload: top-level "training" and "voting" dicts
                voting_payload = data.get("voting", {})
                await manager.ingest(space_name, {
                    "training": data.get("training", {}),
                    "voting":   voting_payload,
                })
                # Auto-bridge: if the voting carries a directional signal,
                # promote it to the flip cache so subscribers on /ws/flips see
                # it immediately even though the publisher used the legacy type.
                if voting_payload:
                    await _store_flip(space_name, voting_payload)
                    # Realtime per-tick signal channel — detect delta in
                    # buy_count/sell_count and emit a {action, price, ts} event
                    # on /ws/signals. Independent of the cumulative flip cache.
                    await _maybe_emit_realtime_signal(space_name, voting_payload)

            elif msg_type == "training":
                # Bug A fix: try "data" wrapper first, then fall back to top-level fields.
                # Some rankers send {"type":"training","data":{...}},
                # others send {"type":"training","actor_loss":..., ...} directly.
                training_raw = data.get("data") or {
                    manager._TRAINING_ALIAS.get(k, k): v
                    for k, v in data.items()
                    if k in manager._TRAINING_KEYS and k != "type"
                }
                if training_raw:
                    logger.info(
                        f"[{space_name}] ⚙  training msg | "
                        f"keys={list(training_raw.keys())} | "
                        f"actor_loss={training_raw.get('actor_loss', training_raw.get('a_loss', '—'))}"
                    )
                await manager.ingest(space_name, {"training": training_raw, "voting": {}})

            elif msg_type == "voting":
                voting_raw = data.get("data") or {
                    k: v for k, v in data.items()
                    if k in manager._VOTING_KEYS and k != "type"
                }
                await manager.ingest(space_name, {"training": {}, "voting": voting_raw})
                # Auto-bridge into fast path (see comment under "metrics").
                if voting_raw:
                    await _store_flip(space_name, voting_raw)
                    # Realtime per-tick signal — see /ws/signals docstring.
                    await _maybe_emit_realtime_signal(space_name, voting_raw)

            elif msg_type in ("heartbeat", "identify", "ping"):
                pass

            elif msg_type == "trade_opened":
                # Executor space opened a trade — add to the hub's in-memory store.
                # data = {trade_id, asset, direction, entry, qty, opened_at}
                _hub_trades.open_trade(space_name, data.get("data", data))

            elif msg_type == "trade_closed":
                # Executor space closed a trade — update the hub's in-memory store.
                # data = {trade_id, asset, pnl, exit_price, closed_at}
                _hub_trades.close_trade(space_name, data.get("data", data))

            else:
                # Bug B fix: don't silently swallow. Try to rescue training/voting
                # fields that live at the top level of an unrecognised message type.
                rescued_training = {
                    manager._TRAINING_ALIAS.get(k, k): v
                    for k, v in data.items()
                    if k in manager._TRAINING_KEYS
                }
                rescued_voting = {
                    k: v for k, v in data.items()
                    if k in manager._VOTING_KEYS
                }
                if rescued_training or rescued_voting:
                    logger.warning(
                        f"[{space_name}] ⚠  Unknown type='{msg_type}' — "
                        f"auto-rescued: training_keys={list(rescued_training.keys())} "
                        f"voting_keys={list(rescued_voting.keys())}"
                    )
                    await manager.ingest(space_name, {
                        "training": rescued_training,
                        "voting":   rescued_voting,
                    })
                    # Auto-bridge rescued voting into fast path too
                    if rescued_voting:
                        await _store_flip(space_name, rescued_voting)
                        # Realtime per-tick signal — see /ws/signals docstring.
                        await _maybe_emit_realtime_signal(space_name, rescued_voting)
                else:
                    logger.warning(
                        f"[{space_name}] ⚠  Unknown type='{msg_type}' with no "
                        f"extractable fields — dropped. Full keys: {list(data.keys())}"
                    )

    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):
    await manager.register_subscriber(websocket)
    await manager.send_initial_state(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        pass
    except Exception as e:
        logger.error(f"Subscriber error: {e}")
    finally:
        await manager.unregister_subscriber(websocket)


@app.websocket("/ws/top3")
async def ws_top3_endpoint(websocket: WebSocket):
    """
    /ws/top3 — consumed by top3_client.py (MT5 bridge).

    Sends a top3_rankings message immediately on connect (replay of the latest
    known ranking so the client does not have to wait for the next ranker cycle),
    then keeps the socket open to receive subsequent broadcasts triggered by
    POST /api/flip/rankings.

    Message format:
      {"type": "top3_rankings", "rankings": [...], "total_assets": N, "hub_timestamp": T}
    """
    await websocket.accept()
    async with _top3_clients_lock:
        _top3_clients.add(websocket)
    logger.info(f"📈 top3 client connected  (total={len(_top3_clients)})")

    # ── Replay latest rankings immediately so client doesn't wait up to 5 s ───
    if _axrvi_rankings:
        try:
            await websocket.send_text(json.dumps({
                "type":          "top3_rankings",
                "rankings":      _axrvi_rankings,
                "total_assets":  len(_axrvi_rankings),
                "hub_timestamp": _axrvi_rankings_ts,
            }))
        except Exception:
            pass

    try:
        while True:
            await websocket.receive_text()   # keep-alive — client sends nothing
    except WebSocketDisconnect:
        pass
    except Exception as e:
        logger.error(f"[top3] Client error: {e}")
    finally:
        async with _top3_clients_lock:
            _top3_clients.discard(websocket)
        logger.info(f"📉 top3 client disconnected  (remaining={len(_top3_clients)})")


@app.websocket("/ws/flips")
async def ws_flips_endpoint(websocket: WebSocket):
    """
    /ws/flips — HIGH-PRIORITY flip-only channel.

    Consumed by:
      • The ranker's FlipSubscriber (fast path into AssetSnapshot, bypassing
        the slower /ws/subscribe snapshot broadcast).
      • The executor (to send flips to MT5 the moment they arrive).

    Protocol:
      • On connect, the hub replays the full current flip cache as
          {"type": "flip_snapshot", "flips": [...], "hub_timestamp": T}
        so the subscriber starts with a consistent per-asset view.
      • Thereafter, every flip update (either from a type="flip" publisher
        message, or auto-bridged from voting/metrics) is pushed as
          {"type": "flip_delta", "flips": [...], "hub_timestamp": T}
        at a coalesced cadence of _FLIP_BROADCAST_INTERVAL_SEC (30 ms).
      • Each flip carries a monotonically increasing `seq` per asset so
        the consumer can detect replays / out-of-order.

    Signal-consistency guarantee:
      • Writes into _flip_cache are serialized through _flip_lock.
      • Readers take a consistent snapshot under the same lock.
      • No torn reads, no lost updates — the ranker always sees exactly one
        authoritative flip per asset.
    """
    await websocket.accept()
    async with _flip_subscribers_lock:
        _flip_subscribers.add(websocket)
    logger.info(f"🎯 Flip subscriber connected  (total={len(_flip_subscribers)})")

    # ── Replay full cache on connect (so the client has state immediately) ──
    async with _flip_lock:
        replay = list(_flip_cache.values())
    if replay:
        try:
            await websocket.send_text(json.dumps({
                "type":          "flip_snapshot",
                "flips":         replay,
                "total_assets":  len(replay),
                "hub_timestamp": time.time(),
            }))
        except Exception:
            pass

    try:
        while True:
            # Keep-alive — the subscriber doesn't send messages, only receives.
            # If it does send something (e.g. ping), we just discard it.
            await websocket.receive_text()
    except WebSocketDisconnect:
        pass
    except Exception as e:
        logger.error(f"[flips] Subscriber error: {e}")
    finally:
        async with _flip_subscribers_lock:
            _flip_subscribers.discard(websocket)
        logger.info(f"🎯 Flip subscriber disconnected  (remaining={len(_flip_subscribers)})")


@app.websocket("/ws/signals")
async def ws_signals_endpoint(websocket: WebSocket):
    """
    /ws/signals — REALTIME PER-TICK signal channel (v2.3+).

    Consumed by:
      • The AXRVI ranker's SignalSubscriber (one source of truth for the
        per-tick AVN action that drives Gate A and asset-buffer features).
      • Any future consumer that wants raw per-tick signals rather than
        aggregated flips.

    Protocol:
      • On connect, the hub replays the full current signal cache as
          {"type": "signal_snapshot", "signals": [...], "hub_timestamp": T}
        so the subscriber starts with a consistent per-asset view.
      • Thereafter, every realtime signal event (derived from a delta in
        buy_count/sell_count between two consecutive voting frames) is
        pushed as
          {"type": "signal_delta", "signals": [...], "hub_timestamp": T}
        at a coalesced cadence of _SIGNAL_BROADCAST_INTERVAL_SEC (30 ms).
      • Each signal carries a monotonic per-asset `seq` so consumers can
        detect replays / out-of-order deliveries.

    Per-signal payload:
        {
          "asset":   "<space_name>",
          "action":  "BUY" | "SELL" | "HOLD",
          "price":   <last_price>,
          "source":  "voting_delta" | <publisher_supplied>,
          "ts":      <unix_seconds>,
          "seq":     <monotonic_int>
        }

    Signal-consistency guarantee:
      • Writes into _signal_cache are serialized through _signal_lock.
      • Readers take a consistent snapshot under the same lock.
      • No torn reads, no lost updates — the ranker always sees exactly one
        authoritative realtime signal per asset.
    """
    await websocket.accept()
    async with _signal_subscribers_lock:
        _signal_subscribers.add(websocket)
    logger.info(f"📡 Signal subscriber connected  (total={len(_signal_subscribers)})")

    # ── Replay full cache on connect (so the client has state immediately) ──
    async with _signal_lock:
        replay = list(_signal_cache.values())
    if replay:
        try:
            await websocket.send_text(json.dumps({
                "type":          "signal_snapshot",
                "signals":       replay,
                "total_assets":  len(replay),
                "hub_timestamp": time.time(),
            }))
        except Exception:
            pass

    try:
        while True:
            # Keep-alive — the subscriber doesn't send messages, only receives.
            # If it does send something (e.g. ping), we just discard it.
            await websocket.receive_text()
    except WebSocketDisconnect:
        pass
    except Exception as e:
        logger.error(f"[signals] Subscriber error: {e}")
    finally:
        async with _signal_subscribers_lock:
            _signal_subscribers.discard(websocket)
        logger.info(
            f"📡 Signal subscriber disconnected  (remaining={len(_signal_subscribers)})"
        )


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 6 — REST API (READ-ONLY)
# ══════════════════════════════════════════════════════════════════════════════════════

@app.get("/rankings")
async def get_rankings():
    return {
        "snapshots": manager.get_all_snapshots(),
        "timestamp": datetime.utcnow().isoformat() + "Z",
    }


@app.get("/metrics/{space_name}")
async def get_space_metrics(space_name: str):
    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(),
    }


@app.get("/api/debug/hub")
async def api_debug_hub():
    """
    Diagnostic endpoint — exposes exactly what the hub has received and stored.

    Returns per-space:
      msg_counts   — how many messages of each type arrived
      snapshot     — current stored training + voting values
      history_len  — number of history points recorded

    Use this to confirm whether training messages are arriving and being stored.
    If msg_counts shows training=0 for a space, the asset space is NOT sending
    training messages. If training > 0 but snapshot.training shows zeros, there
    is a field-name or format mismatch.
    """
    snapshots   = manager.get_all_snapshots()
    msg_counts  = manager.get_msg_counts()
    history_len = {name: len(dq) for name, dq in manager._history.items()}

    spaces = {}
    for name in set(list(snapshots.keys()) + list(msg_counts.keys())):
        snap = snapshots.get(name, {})
        spaces[name] = {
            "msg_counts":   msg_counts.get(name, {}),
            "history_len":  history_len.get(name, 0),
            "training":     snap.get("training", {}),
            "voting":       snap.get("voting",   {}),
            "last_updated": snap.get("last_updated", 0),
            "stale_s":      round(time.time() - snap.get("last_updated", time.time()), 1),
        }

    return JSONResponse({
        "spaces":           spaces,
        "total_ingested":   manager._total_ingested,
        "publisher_count":  len(manager._publishers),
        "subscriber_count": len(manager._subscribers),
        "timestamp":        datetime.utcnow().isoformat() + "Z",
    })


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 7 — TRADE API  (native — replaces patch_websocket_hub.py)
# ══════════════════════════════════════════════════════════════════════════════════════

@app.get("/api/trades")
async def api_trades():
    """Full trade state: open trades, recent closed trades, summary stats."""
    return JSONResponse(_hub_trades.get_state())


@app.get("/api/trades/open")
async def api_trades_open():
    """Open trades only."""
    state = _hub_trades.get_state()
    return JSONResponse({"open": state["open"]})


@app.get("/api/trades/closed")
async def api_trades_closed(limit: int = 50):
    """Recent closed trades (newest first) + cumulative stats."""
    state = _hub_trades.get_state()
    return JSONResponse({
        "closed": state["closed"][:limit],
        "stats":  state["stats"],
    })


@app.get("/api/health")
async def api_health():
    """Service health — includes live trade counts and log-file inventory."""
    state = _hub_trades.get_state()
    return JSONResponse({
        "service":      "websocket_hub",
        "version":      "v2.2-ranker-logs",
        "status":       "running",
        "log_files":    len(glob.glob(os.path.join(_LOG_DIR, "*.log"))),
        "trade_open":   len(state["open"]),
        "trade_closed": len(state["closed"]),
        "uptime_seconds": round(time.time() - _START_TIME, 0),
    })


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 7b — RANKER LOGS API  (FIX: moved here so routes live on port 7860)
# ══════════════════════════════════════════════════════════════════════════════════════
#
# ROOT CAUSE of HTTP 404 on /api/ranker/logs/*:
#   - hub_dashboard_service.py (Flask) runs on port 8052 — NOT publicly accessible on HF Spaces
#   - websocket_hub.py (FastAPI/uvicorn) runs on port 7860 — the ONLY public port
#   - The browser fetches /api/ranker/logs/recent → hits port 7860 → no route → 404
#
# FIX: FileBasedLoggerAdapter + all /api/ranker/logs/* routes added directly here.
# The ranker writes logs to ./ranker_logs (= /app/ranker_logs). This adapter reads
# those files directly — no dependency on hub_dashboard_service or in-memory ranker.

_TRAINING_RE_HUB = re.compile(
    r'step=(\d+)\s*\|\s*loss=([\d.]+)\s*\|\s*lr=([\d.eE+\-]+)\s*\|\s*assets=(\d+)'
)
_JSON_BLOB_RE_HUB = re.compile(r'(\{.*\})\s*$')


def _enrich_training_entry(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_HUB.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_HUB.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


class FileBasedLoggerAdapter:
    """
    Reads ranker log files from disk and exposes the RankerLogger interface
    expected by the /api/ranker/logs/* endpoints.
    No in-memory ranker process required.
    """
    _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):
        self._log_dir = log_dir
        self._lock    = threading.RLock()

    def _find_files(self) -> list:
        candidate_dirs = [
            self._log_dir,
            "/app/ranker_logs",
            str(Path(__file__).parent / "ranker_logs"),
            "./ranker_logs",
            "/home/user/ranker_logs",
            "/tmp/ranker_logs",
        ]
        all_files: list = []
        seen: set = set()
        for d in candidate_dirs:
            for f in sorted(glob.glob(str(Path(d) / "*.log*"))):
                if f not in seen:
                    seen.add(f)
                    all_files.append(f)
        return all_files

    def _read_lines(self, n_tail: int = 500) -> list:
        files = self._find_files()
        raw: list = []
        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
        return {
            "timestamp": ts_m.group(1),
            "level":     level,
            "category":  cat,
            "message":   line.strip(),
            "asset":     asset,
            "data":      None,
        }

    def get_recent(self, n: int = 50, category: Optional[str] = None) -> list:
        entries: list = []
        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: list = []
        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: list = []
        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: int        = 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) -> None:
        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) -> None:
        pass  # file-based — nothing to clear


# Singleton adapter — reads from the same /app/ranker_logs the ranker writes to
_log_adapter = FileBasedLoggerAdapter(log_dir=_LOG_DIR)


@app.get("/api/ranker/logs/recent")
async def api_ranker_logs_recent(limit: int = 50, category: Optional[str] = None):
    """GET /api/ranker/logs/recent?limit=80&category=TRAINING"""
    try:
        entries = _log_adapter.get_recent(n=limit, category=category)
        entries = [_enrich_training_entry(e) for e in entries]
        return JSONResponse({
            "logs":  entries,
            "count": len(entries),
            "stats": _log_adapter.get_stats(),
        })
    except Exception as exc:
        logger.exception(f"[api_ranker_logs_recent] {exc}")
        return JSONResponse({"logs": [], "count": 0, "error": str(exc)}, status_code=200)


@app.get("/api/ranker/logs/stats")
async def api_ranker_logs_stats():
    """GET /api/ranker/logs/stats"""
    try:
        return JSONResponse(_log_adapter.get_stats())
    except Exception as exc:
        return JSONResponse({"error": str(exc)}, status_code=500)


@app.get("/api/ranker/logs/asset/{asset}")
async def api_ranker_logs_asset(asset: str, limit: int = 30):
    """GET /api/ranker/logs/asset/V75?limit=30"""
    try:
        entries = _log_adapter.get_by_asset(asset, n=limit)
        return JSONResponse({"asset": asset, "logs": entries, "count": len(entries)})
    except Exception as exc:
        return JSONResponse({"asset": asset, "logs": [], "count": 0, "error": str(exc)})


@app.get("/api/ranker/logs/level/{level}")
async def api_ranker_logs_level(level: str, limit: int = 50):
    """GET /api/ranker/logs/level/ERROR?limit=50"""
    try:
        entries = _log_adapter.get_by_level(level, n=limit)
        return JSONResponse({"level": level.upper(), "logs": entries, "count": len(entries)})
    except Exception as exc:
        return JSONResponse({"level": level.upper(), "logs": [], "count": 0, "error": str(exc)})


@app.get("/api/ranker/logs/export")
async def api_ranker_logs_export(limit: int = 500):
    """GET /api/ranker/logs/export — download JSON"""
    from fastapi.responses import FileResponse as _FileResponse
    try:
        export_path = "/tmp/ranker_logs_export.json"
        _log_adapter.export_json(export_path, n=limit)
        return _FileResponse(
            export_path,
            media_type="application/json",
            filename="ranker_logs_export.json",
        )
    except Exception as exc:
        return JSONResponse({"error": str(exc)}, status_code=500)


@app.post("/api/ranker/logs/clear")
async def api_ranker_logs_clear():
    """POST /api/ranker/logs/clear"""
    try:
        _log_adapter.clear_buffer()
        return JSONResponse({"status": "cleared"})
    except Exception as exc:
        return JSONResponse({"error": str(exc)}, status_code=500)


@app.get("/api/ranker/logs/debug")
async def api_ranker_logs_debug():
    """GET /api/ranker/logs/debug — show which log files are found"""
    files = _log_adapter._find_files()
    return JSONResponse({
        "log_dir":     _LOG_DIR,
        "files_found": files,
        "file_count":  len(files),
        "stats":       _log_adapter.get_stats(),
    })


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 8 — DASHBOARD UI ROUTES
# ══════════════════════════════════════════════════════════════════════════════════════

_HTML_PATH = Path(os.environ.get(
    "DASHBOARD_HTML",
    Path(__file__).parent / "hub_dashboard.html",
))


def _compute_rankings() -> List[dict]:
    """
    Build the rankings list served by /api/state.

    Path 1 — live ranker push (preferred):
      Rankings received via POST /api/flip/rankings within the last
      _AXRVI_RANKINGS_TTL seconds. Contains real Shreve priorities,
      signal_confidence, and epistemic_std from the AXRVI ranker.
      Hub snapshot fields (training, last_price, flip_direction) are
      merged in on top to give the freshest available state.

    Path 2 — hub-snapshot pass-through (ranker absent or stale):
      Emits only what the hub snapshot actually contains: flip_direction,
      flip_action, last_price, signal_source, and training metrics.
      score / final_priority / signal_confidence / epistemic_std are
      set to None — they are NOT synthesised or approximated.
      Assets are ordered by last_updated (most recently active first).
    """
    global _axrvi_rankings, _axrvi_rankings_ts

    # ── Path 1: fresh AXRVI rankings ────────────────────────────────────────
    if _axrvi_rankings and (time.time() - _axrvi_rankings_ts) < _AXRVI_RANKINGS_TTL:
        snapshots = manager.get_all_snapshots()
        merged: List[dict] = []
        for r in _axrvi_rankings:
            name     = r.get("space_name", "")
            snap     = snapshots.get(name, {})
            training = snap.get("training", {})
            voting   = snap.get("voting",   {})
            merged.append({
                # Core flip-channel fields — live ranker values
                "rank":              r.get("rank",             0),
                "space_name":        name,
                "score":             r.get("score",            0.0),
                "final_priority":    r.get("final_priority",   r.get("score", 0.0)),
                "signal_confidence": r.get("signal_confidence",0.0),
                "flip_direction":    voting.get("flip_direction", r.get("flip_direction", "NONE")),
                "flip_action":       voting.get("flip_action",    r.get("flip_action",    "HOLD")),
                "avn_accuracy":      r.get("avn_accuracy",     0.0),
                "epistemic_std":     r.get("epistemic_std",    0.0),
                "training_steps":    r.get("training_steps",   training.get("training_steps", 0)),
                # Hub-snapshot fields merged in (latest available)
                "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",   _axrvi_rankings_ts),
            })
        return merged

    # ── Path 2: hub-snapshot pass-through (ranker not yet connected or stale) ──
    # Only fields that are genuinely present in the hub snapshot are emitted.
    # No score, signal_confidence, or priority is synthesised — those values
    # belong to the Shreve/AXRVI ranker and must not be approximated here.
    # Assets are ordered by last_updated so the most recently active asset
    # appears first; rank is assigned on that basis alone.
    raw_snaps: List[dict] = []
    for name, snap in manager.get_all_snapshots().items():
        voting   = snap.get("voting",   {})
        training = snap.get("training", {})
        raw_snaps.append({
            "rank":           0,           # assigned below after sort
            "space_name":     name,
            "score":          None,         # not available without ranker
            "final_priority": None,         # not available without ranker
            "signal_confidence": None,      # not available without ranker
            "epistemic_std":  None,         # not available without ranker
            # ── real flip-channel fields from hub snapshot ──────────────────
            "flip_direction": voting.get("flip_direction", "NONE"),
            "flip_action":    voting.get("flip_action",    "HOLD"),
            "last_price":     voting.get("last_price",     0.0),
            "signal_source":  voting.get("signal_source",  "LOG"),
            # ── real training fields from hub snapshot ───────────────────────
            "avn_accuracy":   training.get("avn_accuracy",   0.0),
            "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),
        })
    raw_snaps.sort(key=lambda r: r["last_updated"], reverse=True)
    for i, r in enumerate(raw_snaps):
        r["rank"] = i + 1
    return raw_snaps


@app.get("/")
async def serve_dashboard():
    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", "/api/trades", "/api/trades/open",
                "/api/trades/closed", "/api/health",
                "/ws/publish/{space}", "/ws/subscribe",
            ],
        },
    )


@app.post("/api/flip/rankings")
async def receive_axrvi_rankings(request: Request):
    """
    Called by the Executo ranker after every rank_and_gate() cycle (~5 s).
    Stores the live AXRVI-scored ranking list so _compute_rankings() can serve
    it from /api/state instead of the hub-snapshot pass-through (Path 2).

    Expected body:
      {"rankings": [{"space_name": "V75", "score": 0.24, "rank": 1, ...}, ...]}
    """
    global _axrvi_rankings, _axrvi_rankings_ts
    try:
        body = await request.json()
    except Exception as e:
        return JSONResponse({"ok": False, "error": f"Bad JSON: {e}"}, status_code=400)

    rankings = body.get("rankings", [])
    if not isinstance(rankings, list):
        return JSONResponse({"ok": False, "error": "rankings must be a list"}, status_code=400)

    _axrvi_rankings    = rankings
    _axrvi_rankings_ts = time.time()
    logger.debug(
        f"[Flip Rankings] Received {len(rankings)} assets | "
        f"top={rankings[0].get('space_name','?')} score={rankings[0].get('score',0):.4f}"
        if rankings else "[Flip Rankings] Received empty list"
    )
    # Broadcast to all connected top3_client.py instances immediately
    if rankings:
        asyncio.create_task(_broadcast_top3_rankings(rankings))
    return JSONResponse({"ok": True, "count": len(rankings), "ts": _axrvi_rankings_ts})


@app.get("/api/flips")
async def api_flips():
    """Current flip cache — snapshot of the latest per-asset flip state.

    Diagnostic + executor fallback: if MT5 executor can't maintain a WS
    connection, it can poll this endpoint. Response is consistent under the
    same lock as /ws/flips, so readers never see a torn write.
    """
    async with _flip_lock:
        flips_now = list(_flip_cache.values())
    return JSONResponse({
        "flips":         flips_now,
        "total_assets":  len(flips_now),
        "hub_timestamp": time.time(),
    })


@app.get("/api/flips/{asset}")
async def api_flips_asset(asset: str):
    """Single-asset flip lookup."""
    async with _flip_lock:
        flip = _flip_cache.get(asset)
    if not flip:
        return JSONResponse({"ok": False, "error": f"No flip for {asset}"}, status_code=404)
    return JSONResponse({"ok": True, "flip": flip, "hub_timestamp": time.time()})


@app.get("/api/state")
async def api_state():
    """Full dashboard state polled by hub_dashboard.html every 2 s."""
    rankings = _compute_rankings()
    return JSONResponse({
        "rankings":       rankings,
        "metric_history": manager.get_metric_history(),
        "health": {
            "hub_connected":    True,
            "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 9 — 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")