Upload 2 files

Quasar_axrvi_ranker.py

@@ -795,13 +795,43 @@ class AssetSnapshot:
     avn_accuracy:    float = 0.0
     training_steps:  int   = 0
 
+    # `dominant_signal` is the public action field that all downstream code
+    # (Gate A, _ensure_minimum_trades, ranking export, monitoring) reads from.
+    # In v2.3+ it is populated SOLELY by realtime per-tick signals delivered
+    # via /ws/signals → HubSubscriber.inject_signal(). The hub-snapshot path
+    # (apply_update) NO LONGER writes to it — voting aggregates on /ws/subscribe
+    # are intentionally ignored as a source of truth for the action, because
+    # they carry the cumulative dominant of an EMA-style aggregation rather
+    # than the raw per-tick AVN inference.
     dominant_signal: str = "NEUTRAL"
     buy_count:       int = 0
     sell_count:      int = 0
 
+    # ── Realtime signal bookkeeping (populated by inject_signal) ──────────
+    # latest_action mirrors the per-tick action that arrived on /ws/signals.
+    # Domain: BUY | SELL | HOLD. dominant_signal is derived from it
+    # (HOLD → NEUTRAL) so that the rest of the codebase keeps reading the
+    # same field name with the same {BUY, SELL, NEUTRAL} domain it always had.
+    latest_action:        str   = "HOLD"
+    latest_action_price:  float = 0.0
+    latest_action_ts:     float = 0.0
+    latest_action_seq:    int   = 0
+
     last_updated:    float = 0.0
 
     def apply_update(self, snapshot: dict) -> None:
+        """
+        Merge a hub snapshot (from /ws/subscribe) into this AssetSnapshot.
+
+        v2.3+ scope:
+          • Training metrics (actor_loss, critic_loss, avn_loss/accuracy, steps)
+          • Voting counters (buy_count, sell_count) — kept ONLY for the legacy
+            vote-ratio confidence fallback. NOT used to set dominant_signal.
+
+        The per-tick action is delivered out-of-band on /ws/signals and lands
+        via inject_signal() / apply_signal(). Hub-snapshot voting aggregates
+        are deliberately ignored here.
+        """
         training = snapshot.get("training", {})
         voting   = snapshot.get("voting",   {})
 
@@ -815,24 +845,51 @@ class AssetSnapshot:
             self.training_steps = int(training.get("training_steps", self.training_steps))
 
         if voting:
-            # ── PREFER flip_direction (direct signal from asset engine) ──
-            flip_dir = voting.get("flip_direction", "")
-            if flip_dir.upper() in {"BUY", "SELL"}:
-                self.dominant_signal = flip_dir.upper()
-                self._latest_signal_confidence = 1.0   # direct signal, not a vote ratio
-            else:
-                # fallback to dominant_signal (legacy vote‑based aggregation)
-                raw_dominant = voting.get("dominant_signal", self.dominant_signal)
-                if raw_dominant.upper() in {"BUY", "SELL", "NEUTRAL"}:
-                    self.dominant_signal = raw_dominant.upper()
-                # If we used dominant_signal, confidence is vote ratio
-                self._latest_signal_confidence = None
-
+            # Counts are still useful for the vote-ratio fallback in
+            # signal_confidence when no realtime signal has arrived yet.
             self.buy_count  = int(voting.get("buy_count",  self.buy_count))
             self.sell_count = int(voting.get("sell_count", self.sell_count))
 
         self.last_updated = snapshot.get("last_updated", time.time())
-      
+
+    def apply_signal(
+        self,
+        action: str,
+        price:  float = 0.0,
+        ts:     Optional[float] = None,
+        seq:    int = 0,
+    ) -> None:
+        """
+        Apply a realtime per-tick signal from /ws/signals.
+
+        Sets latest_action and mirrors it into dominant_signal so all existing
+        gate / ranking / execution code (which reads dominant_signal) sees the
+        per-tick action with no further changes.
+
+          BUY  → dominant_signal = BUY,     latest_action_confidence = 1.0
+          SELL → dominant_signal = SELL,    latest_action_confidence = 1.0
+          HOLD → dominant_signal = NEUTRAL, confidence falls back to vote ratio
+        """
+        if not isinstance(action, str):
+            return
+        action = action.upper()
+        if action not in {"BUY", "SELL", "HOLD"}:
+            action = "HOLD"
+
+        self.latest_action       = action
+        self.latest_action_price = float(price or 0.0)
+        self.latest_action_ts    = float(ts) if ts is not None else time.time()
+        self.latest_action_seq   = int(seq or 0)
+
+        if action in {"BUY", "SELL"}:
+            self.dominant_signal = action
+            self._latest_signal_confidence = 1.0   # direct AVN inference
+        else:  # HOLD
+            self.dominant_signal = "NEUTRAL"
+            self._latest_signal_confidence = None  # fall back to vote ratio
+
+        self.last_updated = self.latest_action_ts
+
 
     @property
     def total_votes(self) -> int:
@@ -849,17 +906,22 @@ class AssetSnapshot:
 
     def to_dict(self) -> dict:
         return {
-            "space_name":        self.space_name,
-            "actor_loss":        self.actor_loss,
-            "critic_loss":       self.critic_loss,
-            "avn_loss":          self.avn_loss,
-            "avn_accuracy":      self.avn_accuracy,
-            "training_steps":    self.training_steps,
-            "dominant_signal":   self.dominant_signal,
-            "buy_count":         self.buy_count,
-            "sell_count":        self.sell_count,
-            "signal_confidence": self.signal_confidence,
-            "last_updated":      self.last_updated,
+            "space_name":          self.space_name,
+            "actor_loss":          self.actor_loss,
+            "critic_loss":         self.critic_loss,
+            "avn_loss":            self.avn_loss,
+            "avn_accuracy":        self.avn_accuracy,
+            "training_steps":      self.training_steps,
+            "dominant_signal":     self.dominant_signal,
+            "buy_count":           self.buy_count,
+            "sell_count":          self.sell_count,
+            "signal_confidence":   self.signal_confidence,
+            # Realtime per-tick signal bookkeeping (v2.3+)
+            "latest_action":       self.latest_action,
+            "latest_action_price": self.latest_action_price,
+            "latest_action_ts":    self.latest_action_ts,
+            "latest_action_seq":   self.latest_action_seq,
+            "last_updated":        self.last_updated,
         }
 
 
@@ -1023,98 +1085,95 @@ class HubSubscriber:
                 logger.error(f"[HubSubscriber] on_update callback error: {e}")
 
     # ──────────────────────────────────────────────────────────────────────────
-    # Public fast-path injector — called by FlipSubscriber when a flip arrives
-    # on the /ws/flips channel. Updates the SAME AssetSnapshot instance the
-    # ranker reads from, so rank_and_gate sees flips with ~30 ms latency
-    # instead of waiting for the next metrics_update on /ws/subscribe.
+    # Public fast-path injector — called by SignalSubscriber when a per-tick
+    # realtime signal arrives on the /ws/signals channel. Updates the SAME
+    # AssetSnapshot instance the ranker reads from, so rank_and_gate sees the
+    # latest action with ~30 ms latency instead of waiting for the next
+    # metrics_update on /ws/subscribe (which carries cumulative aggregates,
+    # not the per-tick action we actually want).
     # ──────────────────────────────────────────────────────────────────────────
-    def inject_flip(self, space_name: str, flip: dict) -> None:
+    def inject_signal(self, space_name: str, signal: dict) -> None:
         """
-        Directly merge a flip into the AssetSnapshot under the subscriber's
-        lock. Expects `flip` shaped as:
-            {"flip_direction": "BUY|SELL|NONE", "flip_action": "ENTRY|HOLD|...",
-             "last_price": float, "seq": int, "ts": float, ...}
+        Apply a realtime per-tick signal directly to the AssetSnapshot under
+        the subscriber's lock. Expects `signal` shaped as:
+            {"action": "BUY|SELL|HOLD", "price": float, "seq": int,
+             "ts": float, "source": str, ...}
 
-        Handling of flip_direction:
+        Handling:
           • BUY / SELL  → snap.dominant_signal is set accordingly (confidence=1.0)
-          • NONE        → explicit reset to NEUTRAL (clears any prior BUY/SELL
-                          so Gate A stops firing on stale direction)
+          • HOLD        → snap.dominant_signal reset to NEUTRAL so Gate A
+                          stops firing on stale direction
         """
-        if not isinstance(flip, dict):
+        if not isinstance(signal, dict):
             return
 
-        fd = flip.get("flip_direction", "NONE")
-        fd = fd.upper() if isinstance(fd, str) else "NONE"
-
-        voting: dict = {
-            "flip_direction": fd,
-            "flip_action":    flip.get("flip_action",    "HOLD"),
-            "last_price":     flip.get("last_price",     0.0),
-            "signal_source":  flip.get("signal_source",  "engine"),
-        }
-        # Explicit NEUTRAL reset when a NONE flip arrives — apply_update's
-        # fallback branch only overwrites dominant_signal if the incoming
-        # voting dict provides one, so we must include it here.
-        if fd == "NONE":
-            voting["dominant_signal"] = "NEUTRAL"
+        action = signal.get("action", "HOLD")
+        if not isinstance(action, str):
+            action = "HOLD"
 
         with self._lock:
             if space_name not in self._snapshots:
                 self._snapshots[space_name] = AssetSnapshot(space_name=space_name)
             snap = self._snapshots[space_name]
-            snap.apply_update({
-                "training":     {},
-                "voting":       voting,
-                "last_updated": flip.get("ts", time.time()),
-            })
+            snap.apply_signal(
+                action = action,
+                price  = signal.get("price", 0.0),
+                ts     = signal.get("ts"),
+                seq    = signal.get("seq", 0),
+            )
 
         if self.on_update:
             try:
                 self.on_update(space_name, snap)
             except Exception as e:
-                logger.error(f"[HubSubscriber] on_update (flip-path) callback error: {e}")
+                logger.error(f"[HubSubscriber] on_update (signal-path) callback error: {e}")
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 4b — FLIP SUBSCRIBER (high-priority, low-latency side channel)
+# SECTION 4b — SIGNAL SUBSCRIBER (high-priority, low-latency side channel)
 # ══════════════════════════════════════════════════════════════════════════════════════
 #
-# The FlipSubscriber is a dedicated WebSocket client that consumes only flip
-# events from the hub's /ws/flips endpoint. It runs on its own background
-# thread (matching the sketch: "DEDICATED THREAD, poll every 30ms") and
-# bypasses the slower /ws/subscribe snapshot path for BUY/SELL flips.
+# The SignalSubscriber is a dedicated WebSocket client that consumes only the
+# realtime per-tick AVN signal stream from the hub's /ws/signals endpoint. It
+# runs on its own background thread (matching the original sketch:
+# "DEDICATED THREAD, poll every 30ms") and bypasses the slower /ws/subscribe
+# snapshot path for per-tick action delivery.
 #
 # Why a separate subscriber:
 #   • /ws/subscribe carries the full training-metrics firehose (losses, steps,
-#     avn_accuracy) — large payloads at irregular cadence. Flips would queue
-#     behind them.
-#   • /ws/flips carries only {asset, flip_direction, flip_action, last_price,
-#     seq, ts} — tiny payloads, coalesced to 30 ms broadcasts.
+#     avn_accuracy) — large payloads at irregular cadence. Per-tick signals
+#     would queue behind them.
+#   • /ws/subscribe also only carries CUMULATIVE aggregates (the EMA-style
+#     dominant_signal). The per-tick AVN action — what we actually want to
+#     drive Gate A and the asset-buffer features off of — is not in those
+#     payloads, only on /ws/signals.
+#   • /ws/signals carries only {asset, action, price, source, seq, ts} —
+#     tiny payloads, coalesced to 30 ms broadcasts.
 #   • Running on its own thread means a slow rank_and_gate cycle in the main
-#     loop cannot stall flip ingestion.
+#     loop cannot stall realtime-signal ingestion.
 #
 # Per-asset signal consistency:
 #   • The hub assigns a monotonic `seq` per asset. The subscriber tracks
 #     `last_seq` and drops replays / out-of-order deliveries.
 #   • Updates land in the SAME AssetSnapshot dict the ranker reads from, via
-#     HubSubscriber.inject_flip(), so there is exactly one source of truth per
-#     asset at any point in time.
+#     HubSubscriber.inject_signal(), so there is exactly one source of truth
+#     per asset at any point in time.
 #
-class FlipSubscriber:
+class SignalSubscriber:
     """
-    High-priority WS client for /ws/flips. Feeds flips directly into the
-    shared HubSubscriber snapshot store via inject_flip().
+    High-priority WS client for /ws/signals. Feeds per-tick realtime signals
+    directly into the shared HubSubscriber snapshot store via inject_signal().
     """
 
     _MAX_BACKOFF = 30
 
     def __init__(
         self,
-        flip_url:       str,
+        signal_url:     str,
         hub_subscriber: "HubSubscriber",          # snapshots are written through here
         ranker_logger:  Optional[object] = None,
     ):
-        self.flip_url       = flip_url
+        self.signal_url     = signal_url
         self.hub_subscriber = hub_subscriber
         self.ranker_logger  = ranker_logger
 
@@ -1125,11 +1184,11 @@ class FlipSubscriber:
         self._reconnect_count = 0
 
         self.stats = {
-            "flips_received":    0,
-            "flips_applied":     0,
-            "flips_out_of_order": 0,
-            "reconnect_count":   0,
-            "last_flip_time":    0.0,
+            "signals_received":     0,
+            "signals_applied":      0,
+            "signals_out_of_order": 0,
+            "reconnect_count":      0,
+            "last_signal_time":     0.0,
         }
 
     def start(self) -> None:
@@ -1137,21 +1196,21 @@ class FlipSubscriber:
             return
         self._running = True
         self._thread  = threading.Thread(
-            target=self._run_loop, daemon=True, name="FlipSubscriber"
+            target=self._run_loop, daemon=True, name="SignalSubscriber"
         )
         self._thread.start()
-        logger.info(f"[FlipSubscriber] 🎯 Started → {self.flip_url}")
+        logger.info(f"[SignalSubscriber] 📡 Started → {self.signal_url}")
         if self.ranker_logger:
             try:
                 self.ranker_logger.connection_event(
-                    "Flip WebSocket", "connected", "FlipSubscriber started"
+                    "Signal WebSocket", "connected", "SignalSubscriber started"
                 )
             except Exception:
                 pass
 
     def stop(self) -> None:
         self._running = False
-        logger.info("[FlipSubscriber] Stopping…")
+        logger.info("[SignalSubscriber] Stopping…")
 
     def _run_loop(self) -> None:
         while self._running:
@@ -1162,99 +1221,99 @@ class FlipSubscriber:
                 loop.close()
                 self._reconnect_count = 0
             except Exception as e:
-                logger.error(f"[FlipSubscriber] Session error: {e}")
+                logger.error(f"[SignalSubscriber] Session error: {e}")
                 self._reconnect_count += 1
                 self.stats["reconnect_count"] = self._reconnect_count
 
             if not self._running:
                 break
             backoff = min(self._MAX_BACKOFF, 2 ** min(self._reconnect_count, 4))
-            logger.info(f"[FlipSubscriber] Reconnecting in {backoff}s…")
+            logger.info(f"[SignalSubscriber] Reconnecting in {backoff}s…")
             time.sleep(backoff)
 
     async def _ws_session(self) -> None:
         if websockets is None:
-            logger.error("[FlipSubscriber] websockets library not installed")
+            logger.error("[SignalSubscriber] websockets library not installed")
             await asyncio.sleep(5)
             return
 
         from websockets.exceptions import ConnectionClosed
 
-        async with websockets.connect(self.flip_url) as ws:
+        async with websockets.connect(self.signal_url) as ws:
             self._reconnect_count = 0
-            logger.info("[FlipSubscriber] ✅ Connected to flip channel")
+            logger.info("[SignalSubscriber] ✅ Connected to signal channel")
 
             while self._running:
                 try:
                     raw = await ws.recv()
                     self._handle_message(raw)
                 except ConnectionClosed:
-                    logger.info("[FlipSubscriber] Connection closed by hub")
+                    logger.info("[SignalSubscriber] Connection closed by hub")
                     break
                 except Exception as e:
-                    logger.error(f"[FlipSubscriber] Message error: {e}")
+                    logger.error(f"[SignalSubscriber] Message error: {e}")
 
     def _handle_message(self, raw: str) -> None:
         try:
             data = json.loads(raw)
         except json.JSONDecodeError:
-            logger.warning("[FlipSubscriber] Malformed JSON")
+            logger.warning("[SignalSubscriber] Malformed JSON")
             return
 
         msg_type = data.get("type", "")
-        if msg_type not in ("flip_snapshot", "flip_delta"):
+        if msg_type not in ("signal_snapshot", "signal_delta"):
             return
 
-        flips = data.get("flips", [])
-        if not isinstance(flips, list):
+        signals = data.get("signals", [])
+        if not isinstance(signals, list):
             return
 
-        for flip in flips:
-            self._apply_flip(flip)
+        for sig in signals:
+            self._apply_signal(sig)
 
-    def _apply_flip(self, flip: dict) -> None:
-        asset = flip.get("asset")
-        seq   = flip.get("seq", 0)
+    def _apply_signal(self, signal: dict) -> None:
+        asset = signal.get("asset")
+        seq   = signal.get("seq", 0)
         if not asset:
             return
 
-        self.stats["flips_received"] += 1
+        self.stats["signals_received"] += 1
 
         # Drop replays / out-of-order (per-asset signal consistency)
         with self._lock:
             last = self._last_seq.get(asset, 0)
             if seq <= last:
-                self.stats["flips_out_of_order"] += 1
+                self.stats["signals_out_of_order"] += 1
                 return
             self._last_seq[asset] = seq
 
         # Push directly into the shared AssetSnapshot store
-        self.hub_subscriber.inject_flip(asset, flip)
-        self.stats["flips_applied"] += 1
-        self.stats["last_flip_time"] = time.time()
+        self.hub_subscriber.inject_signal(asset, signal)
+        self.stats["signals_applied"] += 1
+        self.stats["last_signal_time"] = time.time()
 
-        fd = flip.get("flip_direction", "?")
-        if fd in ("BUY", "SELL"):
+        action = signal.get("action", "?")
+        if action in ("BUY", "SELL"):
             logger.info(
-                f"[FlipSubscriber] ⚡ {asset} flip#{seq} → {fd} "
-                f"@ {flip.get('last_price', 0):.5f}"
+                f"[SignalSubscriber] ⚡ {asset} signal#{seq} → {action} "
+                f"@ {signal.get('price', 0):.5f}"
             )
 
 
-def _derive_flip_url(hub_ws_url: str) -> str:
-    """Derive the /ws/flips URL from whatever hub URL the ranker was given.
-    Accepts ws://host:port/ws/subscribe  →  ws://host:port/ws/flips."""
+def _derive_signal_url(hub_ws_url: str) -> str:
+    """Derive the /ws/signals URL from whatever hub URL the ranker was given.
+    Accepts ws://host:port/ws/subscribe  →  ws://host:port/ws/signals."""
     url = hub_ws_url
     # Replace the path if present, otherwise append
-    for known in ("/ws/subscribe", "/ws/metrics", "/subscribe"):
+    for known in ("/ws/subscribe", "/ws/metrics", "/subscribe", "/ws/flips"):
         if url.endswith(known):
-            return url[: -len(known)] + "/ws/flips"
-    # Fallback: strip any trailing path and append /ws/flips
+            return url[: -len(known)] + "/ws/signals"
+    # Fallback: strip any trailing path and append /ws/signals
     if "://" in url:
         scheme, rest = url.split("://", 1)
         host_and_path = rest.split("/", 1)[0]
-        return f"{scheme}://{host_and_path}/ws/flips"
-    return url.rstrip("/") + "/ws/flips"
+        return f"{scheme}://{host_and_path}/ws/signals"
+    return url.rstrip("/") + "/ws/signals"
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -1817,9 +1876,28 @@ class AssetStateBuffer:
         return (time.time() - self.last_signal_ts) > self.STALE_TIMEOUT
 
     def on_hub_snapshot(self, snap: AssetSnapshot) -> None:
-        """Update from a hub AssetSnapshot — triggers feature extraction + buffer append."""
+        """
+        Update from a hub AssetSnapshot — triggers feature extraction +
+        buffer append.
+
+        v2.3+ action sourcing:
+          The per-tick action is taken from snap.latest_action (populated by
+          SignalSubscriber → HubSubscriber.inject_signal from /ws/signals).
+          snap.dominant_signal is now a mirror of latest_action (BUY/SELL
+          pass-through, HOLD → NEUTRAL) so it stays usable as a fallback if
+          a snapshot arrives via the cumulative path before any realtime
+          signal has landed.
+        """
+        # Prefer the realtime per-tick action from /ws/signals.
+        action = snap.latest_action if snap.latest_action in {"BUY", "SELL", "HOLD"} else "HOLD"
+        # Backward-compat fallback: if no realtime signal has landed yet but
+        # dominant_signal is set (e.g. legacy code path or snapshot replay),
+        # still use it. Domain mapping: NEUTRAL → HOLD.
+        if action == "HOLD" and snap.dominant_signal in {"BUY", "SELL"}:
+            action = snap.dominant_signal
+
         self.latest_signal = {
-            "action":         snap.dominant_signal if snap.dominant_signal != "NEUTRAL" else "HOLD",
+            "action":         action,
             "confidence":     snap.signal_confidence,
             "avn_loss":       snap.avn_loss,
             "avn_accuracy":   snap.avn_accuracy,
@@ -7231,14 +7309,15 @@ class QuasarAXRVIBridge:
             ranker_logger  = self.ranker_logger,
         )
 
-        # ── Flip subscriber (high-priority side channel) ──────────────────────
-        # Runs on its own thread, consumes /ws/flips at ~30 ms cadence, and
-        # writes flips directly into the shared AssetSnapshot store via
-        # hub_subscriber.inject_flip(). Gives rank_and_gate sub-second
-        # visibility into BUY/SELL flips instead of waiting for the next
-        # metrics_update broadcast on /ws/subscribe.
-        self.flip_subscriber = FlipSubscriber(
-            flip_url       = _derive_flip_url(hub_ws_url),
+        # ── Signal subscriber (high-priority side channel, v2.3+) ─────────────
+        # Runs on its own thread, consumes /ws/signals at ~30 ms cadence, and
+        # writes per-tick realtime signals directly into the shared
+        # AssetSnapshot store via hub_subscriber.inject_signal(). Gives
+        # rank_and_gate sub-second visibility into BUY/SELL actions instead of
+        # waiting for the next metrics_update broadcast on /ws/subscribe
+        # (which only carries cumulative aggregates, not per-tick data).
+        self.signal_subscriber = SignalSubscriber(
+            signal_url     = _derive_signal_url(hub_ws_url),
             hub_subscriber = self.hub_subscriber,
             ranker_logger  = self.ranker_logger,
         )
@@ -7443,7 +7522,7 @@ class QuasarAXRVIBridge:
             self.ranking_engine.asset_buffers = self.asset_buffers
 
             self.hub_subscriber.start()
-            self.flip_subscriber.start()
+            self.signal_subscriber.start()
 
             # ── Full-state checkpoint resume ──────────────────────────────────
             # Must run AFTER all components are initialised so _restore_checkpoint
@@ -9054,7 +9133,7 @@ class QuasarAXRVIBridge:
             self.running = False
             await self.ws_client.close()
             self.hub_subscriber.stop()
-            self.flip_subscriber.stop()
+            self.signal_subscriber.stop()
             # ── Full-state shutdown checkpoint ────────────────────────────────
             # Saves everything: model, replay buffer, bandit, normalizers, etc.
             # This is the "progress matters" save — nothing is lost on restart.
@@ -11004,6 +11083,9 @@ def test_components() -> None:
         "training": {"avn_accuracy": 0.8, "actor_loss": 0.1, "critic_loss": 0.2, "avn_loss": 0.05},
         "voting":   {"buy_count": 7, "sell_count": 3, "dominant_signal": "BUY"},
     })
+    # v2.3+: per-tick action arrives via /ws/signals → apply_signal, not via
+    # the snapshot's voting dict. Simulate that here.
+    snap.apply_signal(action="BUY", price=1500.0)
     buf.on_price(1500.0, 0.03, 0.0)
     buf.on_hub_snapshot(snap)
     seq = buf.get_sequence()
@@ -11038,6 +11120,9 @@ def test_components() -> None:
             "voting":   {"buy_count": buy, "sell_count": sell,
                          "dominant_signal": "BUY" if buy > sell else "NEUTRAL"},
         })
+        # v2.3+: realtime per-tick action lives in latest_action / dominant_signal,
+        # populated by apply_signal (which mirrors the /ws/signals path).
+        s.apply_signal(action="BUY" if buy > sell else "HOLD")
         snapshots[name] = s
 
     sig_map = {"V75": 0.60, "V100_1s": 0.70, "CRASH500": 0.40}

websocket_hub.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 """
 ╔══════════════════════════════════════════════════════════════════════════════════════╗
-║  K1RL QUASAR — CENTRAL WEBSOCKET HUB  v2.2-ranker-logs                               ║
+║  K1RL QUASAR — CENTRAL WEBSOCKET HUB  v2.3-realtime-signals                          ║
 ║  ──────────────────────────────────────────────────────────────────────────────────  ║
 ║                                                                                      ║
 ║  Architecture role: INGEST → NORMALIZE → BROADCAST                                  ║
@@ -26,7 +26,12 @@
 ║    GET /api/trades/closed   → recent closed trades + stats  (?limit=N)              ║
 ║    GET /api/health          → service health including trade counts                  ║
 ║                                                                                      ║
-║  VERSION: v2.2-ranker-logs  |  2026-04-04                                            ║
+║  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                                       ║
 ╚══════════════════════════════════════════════════════════════════════════════════════╝
 """
 
@@ -747,6 +752,264 @@ async def _flip_broadcaster_loop():
             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.
@@ -808,6 +1071,10 @@ async def _on_startup():
     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())
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -870,6 +1137,10 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
                 # 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.
@@ -897,6 +1168,8 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
                 # 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
@@ -936,6 +1209,8 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
                     # 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 "
@@ -1069,6 +1344,81 @@ async def ws_flips_endpoint(websocket: 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)
 # ══════════════════════════════════════════════════════════════════════════════════════