log_metrics_reader.py

#!/usr/bin/env python3
"""
╔══════════════════════════════════════════════════════════════════════════════════════╗
║  K1RL QUASAR — COMBINED METRICS READER  v3.0                                         ║
║  ──────────────────────────────────────────────────────────────────────────────────  ║
║                                                                                      ║
║  Merges two previously separate metric sources into one file, one shared cache,      ║
║  one publisher, and one combined WebSocket payload to the hub.                        ║
║                                                                                      ║
║  ┌─────────────────────────────────────────────────────────────────────────────┐    ║
║  │  SOURCE A — LogMetricsReader  (v1.1-standalone logic, preserved exactly)    │    ║
║  │    • Tails quasar_engine.log every 5 s (subprocess tail → Python fallback)  │    ║
║  │    • Regex-parses TRAINING metrics:                                          │    ║
║  │        training_steps, actor_loss, critic_loss, avn_loss, avn_accuracy      │    ║
║  │    • Field-level cache (merge_into) — never zeros a field not seen this      │    ║
║  │      poll (agent + AVN train at different cadences)                          │    ║
║  │    • Also parses dominant_signal / buy_count / sell_count from log —         │    ║
║  │      used ONLY as voting fallback when Redis has not yet fired               │    ║
║  └─────────────────────────────────────────────────────────────────────────────┘    ║
║                                                                                      ║
║  ┌─────────────────────────────────────────────────────────────────────────────┐    ║
║  │  SOURCE B — RedisSignalReader  (v2.0-redis-signals logic, preserved exactly) │    ║
║  │    • Subscribes to V75:final_signals Redis channel                    │    ║
║  │    • Event-driven — fires on every BUY/SELL signal (not polled)             │    ║
║  │    • Maintains cumulative rolling session counters: buy_count, sell_count   │    ║
║  │    • Full parse chain transplanted verbatim from Rewards.py v5.2.1          │    ║
║  │    • AUTHORITATIVE source for voting once first signal is received          │    ║
║  │    • Asyncio loop in its own daemon thread (never touches host loop)        │    ║
║  │    • Gracefully disabled if redis_config_v75 / redis_connection_      │    ║
║  │      manager imports fail — log-based voting continues as fallback           │    ║
║  └─────────────────────────────────────────────────────────────────────────────┘    ║
║                                                                                      ║
║  ┌─────────────────────────────────────────────────────────────────────────────┐    ║
║  │  SHARED BRIDGE — MetricsCache  (new in v3.0)                                │    ║
║  │    • Thread-safe — both readers write into it concurrently                  │    ║
║  │    • Training always comes from log                                          │    ║
║  │    • Voting source priority: Redis (once active) > log fallback             │    ║
║  │    • Returns (TrainingMetrics | None, VotingMetrics | None) snapshot after  │    ║
║  │      every update so the caller can immediately dispatch the right publish  │    ║
║  └─────────────────────────────────────────────────────────────────────────────┘    ║
║                                                                                      ║
║  ONE publisher, ONE WebSocket connection, ONE combined payload.                       ║
║                                                                                      ║
║  Integration — identical to v1.1 and v2.0, no app.py changes needed:                ║
║      from log_metrics_reader import start_log_publisher                              ║
║      _publisher, _reader = start_log_publisher()                                     ║
║                                                                                      ║
║  Smoke test (log parsing only, no hub connection needed):                            ║
║      python log_metrics_reader.py                                                    ║
║      python log_metrics_reader.py /path/to/quasar_engine.log                        ║
║                                                                                      ║
║  pip dependencies:  websocket-client>=1.6.0                                         ║
║                     redis, redis_config_v75,                                  ║
║                     redis_connection_manager  ← optional, graceful fallback         ║
║                                                                                      ║
║  VERSION: v3.0.1-combined  |  2026-04-20  | V75                                 ║
╚══════════════════════════════════════════════════════════════════════════════════════╝
"""

import asyncio
import json
import logging
import os
import re
import subprocess
import sys
import threading
import time
from dataclasses import dataclass
from typing import Optional, Tuple

import websocket  # pip: websocket-client>=1.6.0

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


# ── Redis imports — OPTIONAL.  If these fail, RedisSignalReader is silently  ─────────
# ── disabled and log-based voting continues to serve the hub as a fallback.  ─────────
_REDIS_AVAILABLE = False
ABLY_SIGNAL_CHANNEL: str = ""   # set below if imports succeed
try:
    from redis_config_v75 import (       # type: ignore[import]
        REDIS_URL,
        REDIS_PASSWORD,
        REDIS_DB_FEATURES,
        prefixed_channel,
    )
    from redis_connection_manager import RedisAblyClient, RedisMessage  # type: ignore[import]
    ABLY_SIGNAL_CHANNEL = prefixed_channel("final_signals")   # → "V75:final_signals"
    _REDIS_AVAILABLE    = True
    logger.info("Redis imports OK — RedisSignalReader will be active")
except ImportError as _redis_import_err:
    logger.warning(
        f"Redis imports unavailable — RedisSignalReader disabled.  "
        f"Voting will fall back to log-based extraction.  ({_redis_import_err})"
    )


# ── Config — override via environment variables if needed ─────────────────────────────
_DEFAULT_HUB_HOST     = os.environ.get("QUASAR_HUB_HOST",     "karlquant-quasar-executo.hf.space")
_DEFAULT_LOG_PATH     = os.environ.get("QUASAR_LOG_PATH",     "/home/user/app/logs/quasar_engine.log")
_DEFAULT_POLL_INTERVAL = float(os.environ.get("QUASAR_POLL_INTERVAL", "5.0"))
_DEFAULT_TAIL_LINES    = int(os.environ.get("QUASAR_TAIL_LINES",      "600"))  # v3.0.1: raised from 200 — worst-case AVN gap is 511 lines, Actor Loss gap peaks at 327


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 1 — METRIC CONTAINERS
# (unchanged from v1.1 / v2.0 — hub schema must stay in sync)
# ══════════════════════════════════════════════════════════════════════════════════════

@dataclass
class TrainingMetrics:
    """
    Training fields sent to the hub.
    Source: quasar_engine.log (LogMetricsReader) — ONLY source.
    Redis never produces these fields.
    """
    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:
    """
    Voting fields sent to the hub.
    Source priority: Redis (authoritative) > log (fallback).
    Redis uses cumulative rolling counters; log uses snapshot values from lines.
    """
    dominant_signal: str = "NEUTRAL"   # "BUY" | "SELL" | "NEUTRAL"
    buy_count:       int = 0
    sell_count:      int = 0


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 2 — LOG PARSING
# (from v1.1 — unchanged.  Used for training metrics and log-fallback voting.)
# ══════════════════════════════════════════════════════════════════════════════════════

@dataclass
class _RawMetrics:
    """
    Intermediate container for a single log parse pass.
    All fields start None — only fields that actually matched a regex are set.
    This is what enables merge_into() to preserve last known good values.
    """
    training_steps:  Optional[int]   = None
    actor_loss:      Optional[float] = None
    critic_loss:     Optional[float] = None
    avn_loss:        Optional[float] = None
    avn_accuracy:    Optional[float] = None   # stored as %, e.g. 87.3 — converted on emit
    dominant_signal: Optional[str]   = None
    buy_count:       Optional[int]   = None
    sell_count:      Optional[int]   = None

    def has_training(self) -> bool:
        return any(v is not None for v in (
            self.training_steps, self.actor_loss,
            self.critic_loss, self.avn_loss, self.avn_accuracy,
        ))

    def has_voting(self) -> bool:
        return any(v is not None for v in (
            self.dominant_signal, self.buy_count, self.sell_count,
        ))

    def to_training(self) -> TrainingMetrics:
        return TrainingMetrics(
            training_steps = self.training_steps or 0,
            actor_loss     = self.actor_loss     or 0.0,
            critic_loss    = self.critic_loss    or 0.0,
            avn_loss       = self.avn_loss       or 0.0,
            # Convert percent → 0-1 float expected by hub schema.
            # If value is already ≤ 1.0 it was logged as a fraction — don't divide again.
            avn_accuracy   = (self.avn_accuracy / 100.0
                              if (self.avn_accuracy or 0.0) > 1.0
                              else (self.avn_accuracy or 0.0)),
        )

    def to_voting(self) -> VotingMetrics:
        signal = (self.dominant_signal or "NEUTRAL").upper()
        if signal not in {"BUY", "SELL", "NEUTRAL"}:
            signal = "NEUTRAL"
        return VotingMetrics(
            dominant_signal = signal,
            buy_count       = self.buy_count  or 0,
            sell_count      = self.sell_count or 0,
        )

    def merge_into(self, fresh: "_RawMetrics") -> None:
        """
        Merge a freshly-parsed _RawMetrics INTO this persistent cache.
        Only fields that were actually found this poll (not None) overwrite
        the cached value.  Fields absent from the latest parse are left
        unchanged — preserving the last known good value.

        This prevents AVN metrics (logged at a slower cadence than actor/critic)
        from being zeroed out every time the agent is mid-training-step.
        """
        if fresh.training_steps  is not None: self.training_steps  = fresh.training_steps
        if fresh.actor_loss      is not None: self.actor_loss      = fresh.actor_loss
        if fresh.critic_loss     is not None: self.critic_loss     = fresh.critic_loss
        if fresh.avn_loss        is not None: self.avn_loss        = fresh.avn_loss
        if fresh.avn_accuracy    is not None: self.avn_accuracy    = fresh.avn_accuracy
        if fresh.dominant_signal is not None: self.dominant_signal = fresh.dominant_signal
        if fresh.buy_count       is not None: self.buy_count       = fresh.buy_count
        if fresh.sell_count      is not None: self.sell_count      = fresh.sell_count


def _parse_lines(lines: list) -> _RawMetrics:
    """
    Scan log lines in REVERSE (newest-first) and extract the most recent
    value for each metric field.  Stops scanning a field once it is found.
    Patterns kept identical to dashboard_service.py.
    """
    m = _RawMetrics()

    for line in reversed(lines):

        # training_steps ─────────────────────────────────────────────────────────────
        if m.training_steps is None and 'avn_training_steps:' in line:
            hit = re.search(r'avn_training_steps:\s*(\d+)', line)
            if hit:
                m.training_steps = int(hit.group(1))

        # actor_loss — handles both label format and dict format ──────────────────────
        if m.actor_loss is None:
            if 'Actor Loss:' in line:
                hit = re.search(r'Actor Loss:\s*([-\d.]+)', line)
            elif "'actor_loss':" in line or '"actor_loss":' in line:
                hit = re.search(r"""['"]actor_loss['"]\s*:\s*([-\d.]+)""", line)
            else:
                hit = None
            if hit:
                m.actor_loss = float(hit.group(1))

        # critic_loss — both formats; minus sign included ──────────────────────────────
        if m.critic_loss is None:
            if 'Critic Loss:' in line:
                hit = re.search(r'Critic Loss:\s*([-\d.]+)', line)
            elif "'critic_loss':" in line or '"critic_loss":' in line:
                hit = re.search(r"""['"]critic_loss['"]\s*:\s*([-\d.]+)""", line)
            else:
                hit = None
            if hit:
                m.critic_loss = float(hit.group(1))

        # avn_loss — both formats ──────────────────────────────────────────────────────
        if m.avn_loss is None:
            if 'Avg Loss:' in line or '🎯 Avg Loss:' in line:
                hit = re.search(r'Avg Loss:\s*([-\d.]+)', line)
            elif "'avn_loss':" in line or '"avn_loss":' in line:
                hit = re.search(r"""['"]avn_loss['"]\s*:\s*([-\d.]+)""", line)
            else:
                hit = None
            if hit:
                m.avn_loss = float(hit.group(1))

        # avn_accuracy — non-capturing group covers both variants correctly ───────────
        if m.avn_accuracy is None and ('AVN Accuracy:' in line or 'Avg Accuracy:' in line):
            hit = re.search(r'(?:AVN|Avg) Accuracy:\s*([\d.]+)%?', line)
            if hit:
                m.avn_accuracy = float(hit.group(1))   # kept as % here

        # dominant_signal — from log (FALLBACK only when Redis inactive) ─────────────
        if m.dominant_signal is None and re.search(r'[Dd]ominant[\s_][Ss]ignal|Signal:', line):
            hit = re.search(r'(?:[Dd]ominant[\s_][Ss]ignal|Signal):\s*(BUY|SELL|NEUTRAL)', line)
            if hit:
                m.dominant_signal = hit.group(1).upper()

        # buy_count — from log (FALLBACK only when Redis inactive) ────────────────────
        if m.buy_count is None and re.search(
            r'[Bb]uy[\s_][Cc]ount|[Bb]uy[\s_][Vv]otes|buy_count', line
        ):
            hit = re.search(
                r'(?:[Bb]uy[\s_](?:[Cc]ount|[Vv]otes)|buy_count)[:\s=]+(\d+)', line
            )
            if hit:
                m.buy_count = int(hit.group(1))

        # sell_count — from log (FALLBACK only when Redis inactive) ───────────────────
        if m.sell_count is None and re.search(
            r'[Ss]ell[\s_][Cc]ount|[Ss]ell[\s_][Vv]otes|sell_count', line
        ):
            hit = re.search(
                r'(?:[Ss]ell[\s_](?:[Cc]ount|[Vv]otes)|sell_count)[:\s=]+(\d+)', line
            )
            if hit:
                m.sell_count = int(hit.group(1))

        # Early exit once all 8 fields found ─────────────────────────────────────────
        if (m.training_steps is not None and m.actor_loss      is not None and
                m.critic_loss    is not None and m.avn_loss        is not None and
                m.avn_accuracy   is not None and m.dominant_signal is not None and
                m.buy_count      is not None and m.sell_count      is not None):
            break

    return m


def _tail_log(log_path: str, n_lines: int) -> list:
    """
    Return last n_lines from log_path as a list of strings.
    Uses subprocess tail (fast).  Falls back to pure-Python on Windows dev boxes.
    """
    if not os.path.exists(log_path):
        return []
    try:
        result = subprocess.run(
            ['tail', f'-{n_lines}', log_path],
            capture_output=True, text=True, timeout=5,
        )
        return result.stdout.split('\n') if result.returncode == 0 else []
    except FileNotFoundError:
        try:
            with open(log_path, 'r', errors='replace') as fh:
                return fh.readlines()[-n_lines:]
        except Exception:
            return []
    except Exception as e:
        logger.warning(f"Log tail error: {e}")
        return []


def extract_metrics_from_log(
    log_path: str = _DEFAULT_LOG_PATH,
    n_lines:  int = _DEFAULT_TAIL_LINES,
) -> Optional[_RawMetrics]:
    """Public helper — parse log and return raw metrics (or None if nothing found)."""
    lines = _tail_log(log_path, n_lines)
    if not lines:
        return None
    raw = _parse_lines(lines)
    return raw if (raw.has_training() or raw.has_voting()) else None


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 3 — METRICS CACHE  (new in v3.0 — the core bridge between both sources)
# ══════════════════════════════════════════════════════════════════════════════════════

class MetricsCache:
    """
    Thread-safe shared state that merges training data (from log) and voting
    data (from Redis or log fallback) before every publish.

    ╔═══════════════════════════════════════════════════════════════════════╗
    ║  SOURCE RULES                                                         ║
    ║                                                                       ║
    ║  TRAINING  →  log only, always.                                       ║
    ║               Redis never produces training fields.                   ║
    ║               Uses _RawMetrics.merge_into() so individual fields      ║
    ║               (e.g. avn_accuracy) are NOT zeroed between log cadences.║
    ║                                                                       ║
    ║  VOTING    →  Redis (authoritative) once the first signal arrives.    ║
    ║               Uses cumulative rolling counters (_redis_buy/_sell)     ║
    ║               that only increase — never reset to zero mid-session.   ║
    ║                                                                       ║
    ║               Log fallback: active ONLY while _redis_active is False. ║
    ║               Also uses merge_into() so snapshot values persist       ║
    ║               across polls that don't contain voting lines.           ║
    ║               Once Redis fires its first signal, log-based voting     ║
    ║               is permanently suppressed.                              ║
    ╚═══════════════════════════════════════════════════════════════════════╝

    Both update_from_log() and update_from_redis() return the current
    (TrainingMetrics | None, VotingMetrics | None) snapshot so the caller
    can immediately dispatch the correct publish_* method.
    Either value is None when that source has not yet produced any data.
    """

    def __init__(self) -> None:
        self._lock = threading.Lock()

        # Training — field-level persistent cache (log source)
        self._training_raw: _RawMetrics = _RawMetrics()

        # Voting — Redis (authoritative once active)
        self._redis_buy:    int  = 0
        self._redis_sell:   int  = 0
        self._redis_active: bool = False   # flips True on first valid Redis signal, never reverts

        # Voting — log fallback (merge_into semantics, used only while Redis inactive)
        self._log_voting_raw: _RawMetrics = _RawMetrics()

    # ── Update paths ──────────────────────────────────────────────────────────────────

    def update_from_log(
        self, fresh: _RawMetrics
    ) -> Tuple[Optional[TrainingMetrics], Optional[VotingMetrics]]:
        """
        Called by LogMetricsReader after every successful log parse.

        Always merges training fields from fresh into _training_raw.
        Merges voting fields only when Redis has not yet become active.

        Returns the current (TrainingMetrics, VotingMetrics) combined snapshot.
        """
        with self._lock:
            # Training — always accept from log
            self._training_raw.merge_into(fresh)

            # Voting fallback — accept from log only while Redis has not taken over
            if not self._redis_active:
                self._log_voting_raw.merge_into(fresh)

            return self._snapshot_unlocked()

    def update_from_redis(
        self, action: str
    ) -> Tuple[Optional[TrainingMetrics], Optional[VotingMetrics]]:
        """
        Called by RedisSignalReader on every validated BUY or SELL signal.

        Marks Redis as permanently active (suppresses log voting from this point on).
        Increments the rolling session counter for the given action.

        Returns the current (TrainingMetrics, VotingMetrics) combined snapshot.
        TrainingMetrics may be None if the log reader has not yet completed its
        first poll since startup (2 s delay before first poll).
        """
        with self._lock:
            self._redis_active = True
            if action == "BUY":
                self._redis_buy  += 1
            else:
                self._redis_sell += 1

            return self._snapshot_unlocked()

    # ── Internal helpers (must be called with _lock held) ────────────────────────────

    def _snapshot_unlocked(
        self,
    ) -> Tuple[Optional[TrainingMetrics], Optional[VotingMetrics]]:
        tm = self._training_raw.to_training() if self._training_raw.has_training() else None
        vm = self._best_voting_unlocked()
        return tm, vm

    def _best_voting_unlocked(self) -> Optional[VotingMetrics]:
        """
        Returns the best available VotingMetrics from whichever source is active.

        Redis wins once active.  Its buy/sell counters are cumulative session totals
        — very different from the log snapshot values, which reflect whatever was
        last written to the log.

        Returns None only when neither source has produced any data yet.
        """
        if self._redis_active:
            b, s     = self._redis_buy, self._redis_sell
            dominant = "BUY" if b >= s else "SELL"
            return VotingMetrics(dominant_signal=dominant, buy_count=b, sell_count=s)
        elif self._log_voting_raw.has_voting():
            return self._log_voting_raw.to_voting()
        return None

    def get_stats(self) -> dict:
        with self._lock:
            return {
                "redis_active":       self._redis_active,
                "redis_buy":          self._redis_buy,
                "redis_sell":         self._redis_sell,
                "training_ready":     self._training_raw.has_training(),
                "log_voting_ready":   self._log_voting_raw.has_voting(),
                "voting_source":      "redis" if self._redis_active else (
                    "log" if self._log_voting_raw.has_voting() else "none"
                ),
            }


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 4 — WEBSOCKET PUBLISHER
# (identical to v1.1 / v2.0 — single shared connection, unchanged)
# ══════════════════════════════════════════════════════════════════════════════════════

class AssetSpacePublisher:
    """
    Send-only WebSocket publisher.  Runs in a background daemon thread.
    Auto-reconnects with capped exponential back-off.

    Shared between LogMetricsReader and RedisSignalReader — ONE connection,
    ONE rate-limiter, ONE stream to the hub.  Both readers call publish_*
    methods on the same instance; rate limiting prevents flooding.
    """

    _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})
        # Re-send cached state immediately so hub is up-to-date after reconnect
        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 5 — LOG METRICS READER
#
# From v1.1.  Key change from v1.1:
#   Old: called publisher.publish_*() directly from the poll loop.
#   New: pushes fresh parse into MetricsCache.update_from_log(), which returns
#        the best available combined snapshot (training + best voting), then
#        dispatches the correct publish_* call.
#
# This means if Redis is already active, the log poll still triggers a
# publish_combined() that carries the live Redis voting state alongside the
# fresh training metrics — the hub always gets a complete picture.
# ══════════════════════════════════════════════════════════════════════════════════════

class LogMetricsReader:
    """
    Polls quasar_engine.log every poll_interval seconds.

    Authoritative source for: training_steps, actor_loss, critic_loss,
                               avn_loss, avn_accuracy
    Fallback source for:      dominant_signal, buy_count, sell_count
                               (only used while _redis_active is False in cache)
    """

    def __init__(
        self,
        publisher:     AssetSpacePublisher,
        cache:         MetricsCache,
        log_path:      str   = _DEFAULT_LOG_PATH,
        poll_interval: float = _DEFAULT_POLL_INTERVAL,
        tail_lines:    int   = _DEFAULT_TAIL_LINES,
    ):
        self.publisher     = publisher
        self.cache         = cache
        self.log_path      = log_path
        self.poll_interval = poll_interval
        self.tail_lines    = tail_lines

        self._running = False
        self._thread: Optional[threading.Thread] = None

        self._stats = {
            "polls":        0,
            "published":    0,
            "parse_errors": 0,
            "missing_log":  0,
        }

    def start(self) -> None:
        if self._running:
            return
        self._running = True
        self._thread  = threading.Thread(
            target = self._poll_loop,
            daemon = True,
            name   = f"LogReader-{self.publisher.space_name}",
        )
        self._thread.start()
        logger.info(
            f"[{self.publisher.space_name}] LogMetricsReader started "
            f"(log={self.log_path}, interval={self.poll_interval}s)"
        )

    def stop(self) -> None:
        self._running = False
        if self._thread:
            self._thread.join(timeout=self.poll_interval + 2)
        logger.info(f"[{self.publisher.space_name}] LogMetricsReader stopped")

    def get_stats(self) -> dict:
        return {**self._stats, "running": self._running}

    # ── Poll loop ────────────────────────────────────────────────────────────────────

    def _poll_loop(self) -> None:
        time.sleep(2.0)   # let publisher connect before first poll
        while self._running:
            try:
                self._poll_once()
            except Exception as e:
                logger.error(f"[{self.publisher.space_name}] Poll error: {e}")
                self._stats["parse_errors"] += 1
            time.sleep(self.poll_interval)

    def _poll_once(self) -> None:
        self._stats["polls"] += 1

        if not os.path.exists(self.log_path):
            if self._stats["missing_log"] % 12 == 0:   # warn once per minute at 5s interval
                logger.warning(
                    f"[{self.publisher.space_name}] Log not found: {self.log_path}"
                )
            self._stats["missing_log"] += 1
            self.publisher.publish_heartbeat()
            return

        lines = _tail_log(self.log_path, self.tail_lines)
        if not lines:
            return

        fresh = _parse_lines(lines)
        if not fresh.has_training() and not fresh.has_voting():
            logger.debug(
                f"[{self.publisher.space_name}] No metrics matched in last "
                f"{self.tail_lines} lines — heartbeat"
            )
            self.publisher.publish_heartbeat()
            return

        # Push into shared cache.  Cache merges training fields and (if Redis
        # inactive) voting fields, then returns the best combined snapshot.
        tm, vm = self.cache.update_from_log(fresh)

        ok = self._dispatch(tm, vm)
        if ok:
            self._stats["published"] += 1
            logger.debug(
                f"[{self.publisher.space_name}] Log poll → published "
                f"steps={getattr(tm, 'training_steps', '?')}  "
                f"signal={getattr(vm, 'dominant_signal', '?')}  "
                f"voting_src={self.cache.get_stats()['voting_source']}"
            )

    def _dispatch(
        self,
        tm: Optional[TrainingMetrics],
        vm: Optional[VotingMetrics],
    ) -> bool:
        if tm and vm:
            return self.publisher.publish_combined(tm, vm)
        elif tm:
            return self.publisher.publish_training(tm)
        elif vm:
            return self.publisher.publish_voting(vm)
        else:
            self.publisher.publish_heartbeat()
            return False


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 6 — REDIS SIGNAL READER
#
# From v2.0.  Key change from v2.0:
#   Old: called publisher.publish_voting() directly on each signal.
#   New: pushes signal action into MetricsCache.update_from_redis(), which returns
#        the current combined snapshot, then dispatches publish_combined() if
#        training is already available — otherwise falls back to publish_voting().
#
# This means every Redis signal now also carries the latest training metrics
# to the hub — even though Redis itself has no training data.
#
# Everything else is preserved verbatim from v2.0:
#   • Full parse chain from Rewards.py v5.2.1-V75 (all 10 steps)
#   • RedisAblyClient connection (REDIS_URL, REDIS_PASSWORD, REDIS_DB_FEATURES)
#   • Channel: V75:final_signals (via prefixed_channel)
#   • Asyncio loop in own daemon thread
#   • Rolling _buy_count / _sell_count counters (now live inside MetricsCache)
# ══════════════════════════════════════════════════════════════════════════════════════

class RedisSignalReader:
    """
    Subscribes to V75:final_signals and feeds validated signals into
    MetricsCache, then triggers a combined publish via AssetSpacePublisher.

    Only instantiated when _REDIS_AVAILABLE is True.
    If Redis imports failed, this class is never constructed and log-based
    voting continues to serve the hub uninterrupted.
    """

    def __init__(self, publisher: AssetSpacePublisher, cache: MetricsCache):
        self.publisher = publisher
        self.cache     = cache

        self._running = False
        self._thread: Optional[threading.Thread] = None
        self._loop:   Optional[asyncio.AbstractEventLoop] = None

        self._stats = {
            "signals_received": 0,
            "signals_parsed":   0,
            "signals_dropped":  0,
            "published":        0,
        }

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

        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"]

            # Push into shared cache → get combined snapshot
            tm, vm = self.cache.update_from_redis(action)

            # Publish combined if training is ready, else voting only
            if tm and vm:
                ok = self.publisher.publish_combined(tm, vm)
            elif vm:
                ok = self.publisher.publish_voting(vm)
            else:
                ok = False

            if ok:
                self._stats["published"] += 1
                logger.info(
                    f"[{self.publisher.space_name}] 🔔 Signal {action} "
                    f"@ {parsed['entry_price']:.5f} | "
                    f"keys={len(parsed['signal_keys'])} | "
                    f"buy={vm.buy_count if vm else '?'} "
                    f"sell={vm.sell_count if vm else '?'} "
                    f"dominant={vm.dominant_signal if vm else '?'}"
                )

        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 10-step parse chain from Rewards.py v5.2.1-V75.
        Returns a dict with keys: action, signal_keys, entry_price, payload.
        Returns None silently for any malformed payload — never raises.
        """
        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)]
            # Step 10 — return parsed dict
            return {
                "action":      action,
                "signal_keys": signal_keys[:8],
                "entry_price": float(entry_price),
                "payload":     data,
            }
        except Exception:
            return None


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 7 — COMBINED READER WRAPPER
# ══════════════════════════════════════════════════════════════════════════════════════

class CombinedReader:
    """
    Wraps LogMetricsReader and (optionally) RedisSignalReader behind a single
    object so app.py can use the unchanged two-tuple assignment:

        _publisher, _reader = start_log_publisher()

    and still call _reader.stop() / _reader.get_stats() without caring whether
    Redis is active or not.
    """

    def __init__(
        self,
        log_reader:   LogMetricsReader,
        redis_reader: Optional[RedisSignalReader],
        cache:        MetricsCache,
    ):
        self.log_reader   = log_reader
        self.redis_reader = redis_reader
        self.cache        = cache

    def stop(self) -> None:
        self.log_reader.stop()
        if self.redis_reader:
            self.redis_reader.stop()

    def get_stats(self) -> dict:
        return {
            "log_reader":      self.log_reader.get_stats(),
            "redis_reader":    (self.redis_reader.get_stats()
                                if self.redis_reader else {"enabled": False}),
            "cache":           self.cache.get_stats(),
            "redis_available": _REDIS_AVAILABLE,
        }


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 8 — ONE-LINE FACTORY
# ══════════════════════════════════════════════════════════════════════════════════════

def start_log_publisher(
    log_path:      str   = _DEFAULT_LOG_PATH,
    poll_interval: float = _DEFAULT_POLL_INTERVAL,
    tail_lines:    int   = _DEFAULT_TAIL_LINES,
    hub_host:      str   = _DEFAULT_HUB_HOST,
    space_name:    Optional[str] = None,
    # Legacy keyword accepted for call-site compatibility with v2.0 callers.
    # Silently ignored — log_path / tail_lines are now honoured again.
    **_legacy_kwargs,
) -> Tuple[AssetSpacePublisher, CombinedReader]:
    """
    One-line drop-in.  Signature backward-compatible with v1.1 and v2.0.

    Wires together:
      ONE  AssetSpacePublisher  (single WebSocket connection to hub)
      ONE  MetricsCache         (shared state bridge between log and Redis)
      ONE  LogMetricsReader     (polls log → training + fallback voting)
      ONE  RedisSignalReader    (Redis events → authoritative voting) [if available]
      ONE  CombinedReader       (thin wrapper returned as _reader)

    Usage in app.py — UNCHANGED from v1.1 / v2.0:
        from log_metrics_reader import start_log_publisher
        _publisher, _reader = start_log_publisher()

    Or with explicit space name (as in current app.py):
        _publisher, _reader = start_log_publisher(space_name="V75")
    """
    if space_name is None:
        raw        = os.environ.get("SPACE_ID", "")
        space_name = raw.split("/", 1)[-1] if "/" in raw else (raw or "UnknownSpace")

    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),
    )
    cache = MetricsCache()

    log_reader = LogMetricsReader(
        publisher     = publisher,
        cache         = cache,
        log_path      = log_path,
        poll_interval = poll_interval,
        tail_lines    = tail_lines,
    )

    redis_reader: Optional[RedisSignalReader] = None
    if _REDIS_AVAILABLE:
        redis_reader = RedisSignalReader(publisher=publisher, cache=cache)
    else:
        logger.warning(
            f"[{space_name}] RedisSignalReader not started "
            f"(Redis imports unavailable). "
            f"Voting metrics will be sourced from log only."
        )

    # Start in order: publisher first, then readers
    publisher.start()
    log_reader.start()
    if redis_reader:
        redis_reader.start()

    combined = CombinedReader(
        log_reader   = log_reader,
        redis_reader = redis_reader,
        cache        = cache,
    )

    logger.info(
        f"[{space_name}] ✅ Combined metrics pipeline active — "
        f"log=✅  redis={'✅' if redis_reader else '⚠️ disabled'}  "
        f"hub={hub_url}"
    )

    return publisher, combined


# ══════════════════════════════════════════════════════════════════════════════════════
# SECTION 9 — CLI SMOKE TEST  (log parsing only, no hub connection needed)
# ══════════════════════════════════════════════════════════════════════════════════════

if __name__ == "__main__":
    log = sys.argv[1] if len(sys.argv) > 1 else _DEFAULT_LOG_PATH

    print(f"\n{'=' * 62}")
    print(f"  K1RL QUASAR — Combined Metrics Reader  v3.0  Smoke Test")
    print(f"{'=' * 62}")
    print(f"  Log file    : {log}")
    print(f"  Tail lines  : {_DEFAULT_TAIL_LINES}")
    print(f"  Redis avail : {_REDIS_AVAILABLE}")
    if _REDIS_AVAILABLE:
        print(f"  Redis chan   : {ABLY_SIGNAL_CHANNEL}")
    print(f"{'=' * 62}\n")

    raw = extract_metrics_from_log(log_path=log)

    if raw is None:
        print("❌  No metrics found.")
        print("    Log file missing or no regex patterns matched.")
        print(f"\n    Expected patterns in: {log}")
        print("      avn_training_steps: <int>")
        print("      Actor Loss: <float>  |  'actor_loss': <float>")
        print("      Critic Loss: <float> |  'critic_loss': <float>")
        print("      Avg Loss: <float>    |  'avn_loss': <float>")
        print("      AVN Accuracy: <float>%  |  Avg Accuracy: <float>%")
        print("      Dominant Signal: BUY|SELL|NEUTRAL")
        print("      buy_count: <int>   |  Buy Votes: <int>")
        print("      sell_count: <int>  |  Sell Votes: <int>")
        sys.exit(1)

    print("✅  Raw extracted values (from log):")
    print(json.dumps({
        "training_steps":   raw.training_steps,
        "actor_loss":       raw.actor_loss,
        "critic_loss":      raw.critic_loss,
        "avn_loss":         raw.avn_loss,
        "avn_accuracy_pct": raw.avn_accuracy,
        "dominant_signal":  raw.dominant_signal,
        "buy_count":        raw.buy_count,
        "sell_count":       raw.sell_count,
    }, indent=2))

    if raw.has_training():
        tm = raw.to_training()
        print("\n📡  TrainingMetrics (as sent to hub):")
        print(f"    training_steps = {tm.training_steps}")
        print(f"    actor_loss     = {tm.actor_loss}")
        print(f"    critic_loss    = {tm.critic_loss}")
        print(f"    avn_loss       = {tm.avn_loss}")
        print(f"    avn_accuracy   = {tm.avn_accuracy:.6f}  ← 0-1 float, not %")

    if raw.has_voting():
        vm = raw.to_voting()
        print("\n🗳️   VotingMetrics (log-based — shown as fallback reference):")
        print(f"    dominant_signal = {vm.dominant_signal}")
        print(f"    buy_count       = {vm.buy_count}")
        print(f"    sell_count      = {vm.sell_count}")
        if _REDIS_AVAILABLE:
            print(f"\n  ℹ️  Redis is available — in production, these log-based voting values")
            print(f"     will be REPLACED by live Redis signals from {ABLY_SIGNAL_CHANNEL}")
            print(f"     as soon as the first BUY/SELL signal is received.")
    elif _REDIS_AVAILABLE:
        print(f"\n  ℹ️  No voting data in log.  Redis ({ABLY_SIGNAL_CHANNEL}) will supply")
        print(f"     voting metrics in production.")

    print(f"\n{'=' * 62}")
    print("  Log parsing OK — safe to deploy as log_metrics_reader.py")
    print(f"{'=' * 62}\n")