Update websocket_hub.py

websocket_hub.py

@@ -71,10 +71,19 @@ _ALLOWED_TRAINING_FIELDS: frozenset = frozenset({
 })
 
 _ALLOWED_VOTING_FIELDS: frozenset = frozenset({
+    # Canonical (post-v2 schema) ────────────────────────────────────────────
     "flip_direction",
     "flip_action",
     "last_price",
     "signal_source",
+    # Legacy aliases accepted on the wire — translated inside the hub ───────
+    # Executor spaces still emit dominant_signal / buy_count / sell_count
+    # because their publisher code pre-dates the flip_* rename. We accept
+    # them at ingestion and map them to flip_direction so downstream code
+    # (ranker, dashboard) can stay on the unified flip_* schema.
+    "dominant_signal",
+    "buy_count",
+    "sell_count",
 })
 
 
@@ -130,22 +139,54 @@ def _validate_and_normalize(space_name: str, raw: dict) -> Optional[dict]:
 
     voting: dict = {}
     if voting_raw:
-        raw_direction = voting_raw.get("flip_direction", "NONE")
+        # ── Canonical fields (new schema) ──────────────────────────────────
+        raw_direction = voting_raw.get("flip_direction", None)
+        raw_action    = voting_raw.get("flip_action",    None)
+        raw_source    = voting_raw.get("signal_source", "LOG")
+        if not isinstance(raw_source, str):
+            raw_source = "LOG"
+
+        # ── Legacy fallback: translate dominant_signal → flip_direction ────
+        # The asset/executor spaces still publish the old schema
+        # ({dominant_signal, buy_count, sell_count}) because their publisher
+        # code was never rewritten. We map it to the new flip_* schema here
+        # so everything downstream stays on one naming convention.
+        #
+        #   dominant_signal=BUY     → flip_direction=BUY,  flip_action=ENTRY
+        #   dominant_signal=SELL    → flip_direction=SELL, flip_action=ENTRY
+        #   dominant_signal=NEUTRAL → flip_direction=NONE, flip_action=HOLD
+        #
+        # If a payload carries BOTH flip_direction and dominant_signal,
+        # flip_direction wins (forward-compat with newer publishers).
+        if not isinstance(raw_direction, str) or raw_direction.upper() not in {"BUY", "SELL", "NONE"}:
+            legacy_signal = voting_raw.get("dominant_signal", None)
+            if isinstance(legacy_signal, str):
+                legacy_upper = legacy_signal.upper()
+                if legacy_upper in {"BUY", "SELL"}:
+                    raw_direction = legacy_upper
+                    if not isinstance(raw_action, str) or raw_action.upper() == "HOLD":
+                        raw_action = "ENTRY"
+                elif legacy_upper == "NEUTRAL":
+                    raw_direction = "NONE"
+                    raw_action    = "HOLD"
+
         if not isinstance(raw_direction, str):
             raw_direction = "NONE"
-        raw_action = voting_raw.get("flip_action", "HOLD")
         if not isinstance(raw_action, str):
             raw_action = "HOLD"
-        raw_source = voting_raw.get("signal_source", "LOG")
-        if not isinstance(raw_source, str):
-            raw_source = "LOG"
+
         _clean_direction = raw_direction.upper() if raw_direction.upper() in {"BUY", "SELL", "NONE"} else "NONE"
         _clean_action    = raw_action.upper()    if raw_action.upper()    in {"ENTRY", "EXIT", "HOLD", "REBALANCE", "REDUCE", "SKIP"} else "HOLD"
+
         voting = {
             "flip_direction": _clean_direction,
             "flip_action":    _clean_action,
             "last_price":     _float(voting_raw.get("last_price", 0.0)),
             "signal_source":  raw_source,
+            # Preserve vote counts (legacy format) so the ranker can compute
+            # vote-ratio confidence when the publisher emits them.
+            "buy_count":      _int(voting_raw.get("buy_count",  0)),
+            "sell_count":     _int(voting_raw.get("sell_count", 0)),
         }
 
     return {
@@ -177,7 +218,10 @@ class ConnectionManager:
         "step":        "training_steps","steps":     "training_steps","n_steps": "training_steps",
     }
     _VOTING_KEYS: frozenset = frozenset({
+        # Canonical
         "flip_direction", "flip_action", "last_price", "signal_source",
+        # Legacy (translated inside _validate_and_normalize)
+        "dominant_signal", "buy_count", "sell_count",
     })
 
     def __init__(self):
@@ -463,6 +507,170 @@ _axrvi_rankings:    List[dict] = []
 _axrvi_rankings_ts: float      = 0.0
 _AXRVI_RANKINGS_TTL: float     = 60.0   # FIX: extended 30→60s to survive slow ranker POST cycles
 
+# ══════════════════════════════════════════════════════════════════════════════════════
+# FLIP FAST-PATH CHANNEL
+# ═══════════════════════════════════════���══════════════════════════════════════════════
+# A dedicated, high-priority, per-asset latest-wins cache + broadcaster for BUY/SELL
+# flip events. Flips are time-sensitive (change every few hundred ms) and must not
+# be delayed by the slower training/snapshot machinery.
+#
+# Architecture:
+#   Engine  ──► {"type":"flip", "data": {"flip_direction":"BUY"|"SELL"|"NONE", ...}}
+#               │
+#               ▼  (short-circuit handler in publisher_handler — skips _validate_and_normalize)
+#         _store_flip(asset, payload)  ── atomic write under _flip_lock, seq++
+#               │
+#               ▼  sets _flip_dirty event
+#         _flip_broadcaster_loop()  ── dedicated asyncio task, runs every 30 ms
+#               │                      (or immediately when dirty, whichever is first)
+#               ▼
+#         /ws/flips subscribers (ranker, executor) ── push JSON delta
+#
+# Per-asset signal consistency:
+#   • Latest-wins: newer flip replaces older one atomically under the lock.
+#   • Monotonic seq per asset so consumers can detect replays / out-of-order.
+#   • On subscribe, a full replay of the current cache is sent first so clients
+#     start with a consistent view.
+#
+# Backward-compat auto-bridge:
+#   Any inbound voting message whose normalized flip_direction ∈ {BUY, SELL} is
+#   AUTOMATICALLY promoted into the flip cache as well. This means existing
+#   publishers that send {"type":"voting","dominant_signal":"BUY"} or metrics with
+#   voting nested still benefit from the fast path without any code change on
+#   their end.
+# ──────────────────────────────────────────────────────────────────────────────────────
+
+_FLIP_BROADCAST_INTERVAL_SEC: float = 0.030   # 30 ms cadence per sketch
+_flip_cache: Dict[str, dict]        = {}      # {asset: {flip_direction, flip_action, last_price, ts, seq, ...}}
+_flip_seq:   Dict[str, int]         = {}      # monotonic per-asset sequence
+_flip_lock:  asyncio.Lock           = asyncio.Lock()
+_flip_dirty: asyncio.Event          = asyncio.Event()
+_flip_subscribers: Set[WebSocket]   = set()
+_flip_subscribers_lock: asyncio.Lock = asyncio.Lock()
+
+
+async def _store_flip(asset: str, raw: dict) -> Optional[dict]:
+    """
+    Write one flip into the per-asset cache with signal consistency.
+    Returns the normalized flip dict (or None if the payload was rejected).
+
+    Contract:
+      • Accepts either {flip_direction: BUY|SELL|NONE|NEUTRAL} or legacy
+        {dominant_signal: BUY|SELL|NEUTRAL}. NEUTRAL is coerced to NONE.
+      • Write is atomic under _flip_lock (latest-wins).
+      • Sequence number is incremented monotonically per asset.
+      • Sets _flip_dirty so the broadcaster picks it up within 30 ms.
+    """
+    if not isinstance(raw, dict):
+        return None
+
+    # Normalize direction — accept both new and legacy naming
+    raw_dir = raw.get("flip_direction")
+    if not isinstance(raw_dir, str) or raw_dir.upper() not in {"BUY", "SELL", "NONE", "NEUTRAL"}:
+        raw_dir = raw.get("dominant_signal")
+    if isinstance(raw_dir, str):
+        raw_dir = raw_dir.upper()
+        if raw_dir == "NEUTRAL":
+            raw_dir = "NONE"
+    else:
+        raw_dir = "NONE"
+    if raw_dir not in {"BUY", "SELL", "NONE"}:
+        raw_dir = "NONE"
+
+    # Normalize action
+    raw_act = raw.get("flip_action")
+    if isinstance(raw_act, str) and raw_act.upper() in {"ENTRY", "EXIT", "HOLD", "REBALANCE", "REDUCE", "SKIP"}:
+        raw_act = raw_act.upper()
+    else:
+        raw_act = "ENTRY" if raw_dir in {"BUY", "SELL"} else "HOLD"
+
+    try:
+        price = float(raw.get("last_price", 0.0) or 0.0)
+    except Exception:
+        price = 0.0
+    source = raw.get("signal_source", "engine")
+    if not isinstance(source, str):
+        source = "engine"
+
+    async with _flip_lock:
+        prev = _flip_cache.get(asset)
+        # De-dup: if direction AND action AND price are unchanged, don't bump seq.
+        # (Signal consistency: the ranker only reacts to real flips.)
+        if prev and prev["flip_direction"] == raw_dir and prev["flip_action"] == raw_act and prev["last_price"] == price:
+            # Refresh timestamp only — no seq bump, no dirty flag
+            prev["ts"] = time.time()
+            return prev
+
+        seq = _flip_seq.get(asset, 0) + 1
+        _flip_seq[asset] = seq
+        flip = {
+            "asset":          asset,
+            "flip_direction": raw_dir,
+            "flip_action":    raw_act,
+            "last_price":     price,
+            "signal_source":  source,
+            "ts":             time.time(),
+            "seq":            seq,
+        }
+        _flip_cache[asset] = flip
+
+    _flip_dirty.set()
+    return flip
+
+
+async def _flip_broadcaster_loop():
+    """
+    Dedicated task: every 30 ms (or immediately on dirty) push the current
+    flip cache to all /ws/flips subscribers. Dead connections are pruned.
+    """
+    logger.info(f"🎯 Flip broadcaster started | interval={_FLIP_BROADCAST_INTERVAL_SEC*1000:.0f}ms")
+    while True:
+        try:
+            # Wait for either a dirty signal or the 30 ms timeout
+            try:
+                await asyncio.wait_for(_flip_dirty.wait(), timeout=_FLIP_BROADCAST_INTERVAL_SEC)
+            except asyncio.TimeoutError:
+                pass
+            _flip_dirty.clear()
+
+            async with _flip_subscribers_lock:
+                if not _flip_subscribers:
+                    continue
+                subscribers_snapshot = list(_flip_subscribers)
+
+            async with _flip_lock:
+                if not _flip_cache:
+                    continue
+                flips_snapshot = list(_flip_cache.values())
+
+            msg = {
+                "type":          "flip_delta",
+                "flips":         flips_snapshot,
+                "total_assets":  len(flips_snapshot),
+                "hub_timestamp": time.time(),
+            }
+
+            dead: List[WebSocket] = []
+            for ws in subscribers_snapshot:
+                try:
+                    await ws.send_json(msg)
+                except Exception:
+                    dead.append(ws)
+
+            if dead:
+                async with _flip_subscribers_lock:
+                    for ws in dead:
+                        _flip_subscribers.discard(ws)
+                logger.info(f"🎯 Pruned {len(dead)} dead flip subscriber(s)")
+
+        except asyncio.CancelledError:
+            logger.info("🎯 Flip broadcaster cancelled")
+            raise
+        except Exception as e:
+            logger.error(f"🎯 Flip broadcaster error: {e}")
+            await asyncio.sleep(0.1)
+
+
 # ── 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.
@@ -520,8 +728,10 @@ manager = ConnectionManager()
 
 @app.on_event("startup")
 async def _on_startup():
-    """Nothing to start — HubTradeStore is in-memory, populated by WS messages."""
+    """Start background tasks — flip broadcaster runs on the event loop as a
+    dedicated task (Python's equivalent of a dedicated thread for asyncio)."""
     logger.info("🚀 HubTradeStore ready (no background scanner needed)")
+    asyncio.create_task(_flip_broadcaster_loop())
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -546,12 +756,44 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
             manager.record_msg(space_name, msg_type)
 
             # ── Route by type ────────────────────────────────────────────────────
-            if msg_type == "metrics":
+            if msg_type == "flip":
+                # ★ HIGH-PRIORITY FLIP FAST PATH ★
+                # Bypasses the heavier _validate_and_normalize / snapshot path
+                # and writes directly into the flip cache. The broadcaster task
+                # picks it up within 30 ms and pushes to all /ws/flips subscribers.
+                flip_raw = data.get("data") or {
+                    k: v for k, v in data.items() if k != "type"
+                }
+                stored = await _store_flip(space_name, flip_raw)
+                if stored and stored["flip_direction"] in {"BUY", "SELL"}:
+                    logger.info(
+                        f"[{space_name}] ⚡ flip#{stored['seq']} | "
+                        f"{stored['flip_direction']}/{stored['flip_action']} @ {stored['last_price']:.5f}"
+                    )
+                # Also mirror into the main snapshot so clients that only use
+                # /ws/subscribe (dashboard, legacy) stay in sync.
+                await manager.ingest(space_name, {
+                    "training": {},
+                    "voting": {
+                        "flip_direction": flip_raw.get("flip_direction") or flip_raw.get("dominant_signal", "NONE"),
+                        "flip_action":    flip_raw.get("flip_action", "ENTRY"),
+                        "last_price":     flip_raw.get("last_price", 0.0),
+                        "signal_source":  flip_raw.get("signal_source", "engine"),
+                    },
+                })
+
+            elif msg_type == "metrics":
                 # Combined payload: top-level "training" and "voting" dicts
+                voting_payload = data.get("voting", {})
                 await manager.ingest(space_name, {
                     "training": data.get("training", {}),
-                    "voting":   data.get("voting",   {}),
+                    "voting":   voting_payload,
                 })
+                # Auto-bridge: if the voting carries a directional signal,
+                # promote it to the flip cache so subscribers on /ws/flips see
+                # it immediately even though the publisher used the legacy type.
+                if voting_payload:
+                    await _store_flip(space_name, voting_payload)
 
             elif msg_type == "training":
                 # Bug A fix: try "data" wrapper first, then fall back to top-level fields.
@@ -576,6 +818,9 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
                     if k in manager._VOTING_KEYS and k != "type"
                 }
                 await manager.ingest(space_name, {"training": {}, "voting": voting_raw})
+                # Auto-bridge into fast path (see comment under "metrics").
+                if voting_raw:
+                    await _store_flip(space_name, voting_raw)
 
             elif msg_type in ("heartbeat", "identify", "ping"):
                 pass
@@ -612,6 +857,9 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
                         "training": rescued_training,
                         "voting":   rescued_voting,
                     })
+                    # Auto-bridge rescued voting into fast path too
+                    if rescued_voting:
+                        await _store_flip(space_name, rescued_voting)
                 else:
                     logger.warning(
                         f"[{space_name}] ⚠  Unknown type='{msg_type}' with no "
@@ -684,6 +932,67 @@ async def ws_top3_endpoint(websocket: WebSocket):
         logger.info(f"📉 top3 client disconnected  (remaining={len(_top3_clients)})")
 
 
+@app.websocket("/ws/flips")
+async def ws_flips_endpoint(websocket: WebSocket):
+    """
+    /ws/flips — HIGH-PRIORITY flip-only channel.
+
+    Consumed by:
+      • The ranker's FlipSubscriber (fast path into AssetSnapshot, bypassing
+        the slower /ws/subscribe snapshot broadcast).
+      • The executor (to send flips to MT5 the moment they arrive).
+
+    Protocol:
+      • On connect, the hub replays the full current flip cache as
+          {"type": "flip_snapshot", "flips": [...], "hub_timestamp": T}
+        so the subscriber starts with a consistent per-asset view.
+      • Thereafter, every flip update (either from a type="flip" publisher
+        message, or auto-bridged from voting/metrics) is pushed as
+          {"type": "flip_delta", "flips": [...], "hub_timestamp": T}
+        at a coalesced cadence of _FLIP_BROADCAST_INTERVAL_SEC (30 ms).
+      • Each flip carries a monotonically increasing `seq` per asset so
+        the consumer can detect replays / out-of-order.
+
+    Signal-consistency guarantee:
+      • Writes into _flip_cache are serialized through _flip_lock.
+      • Readers take a consistent snapshot under the same lock.
+      • No torn reads, no lost updates — the ranker always sees exactly one
+        authoritative flip per asset.
+    """
+    await websocket.accept()
+    async with _flip_subscribers_lock:
+        _flip_subscribers.add(websocket)
+    logger.info(f"🎯 Flip subscriber connected  (total={len(_flip_subscribers)})")
+
+    # ── Replay full cache on connect (so the client has state immediately) ──
+    async with _flip_lock:
+        replay = list(_flip_cache.values())
+    if replay:
+        try:
+            await websocket.send_text(json.dumps({
+                "type":          "flip_snapshot",
+                "flips":         replay,
+                "total_assets":  len(replay),
+                "hub_timestamp": time.time(),
+            }))
+        except Exception:
+            pass
+
+    try:
+        while True:
+            # Keep-alive — the subscriber doesn't send messages, only receives.
+            # If it does send something (e.g. ping), we just discard it.
+            await websocket.receive_text()
+    except WebSocketDisconnect:
+        pass
+    except Exception as e:
+        logger.error(f"[flips] Subscriber error: {e}")
+    finally:
+        async with _flip_subscribers_lock:
+            _flip_subscribers.discard(websocket)
+        logger.info(f"🎯 Flip subscriber disconnected  (remaining={len(_flip_subscribers)})")
+
+
 # ══════════════════════════════════════════════════════════════════════════════════════
 # SECTION 6 — REST API (READ-ONLY)
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -1219,6 +1528,33 @@ async def receive_axrvi_rankings(request: Request):
     return JSONResponse({"ok": True, "count": len(rankings), "ts": _axrvi_rankings_ts})
 
 
+@app.get("/api/flips")
+async def api_flips():
+    """Current flip cache — snapshot of the latest per-asset flip state.
+
+    Diagnostic + executor fallback: if MT5 executor can't maintain a WS
+    connection, it can poll this endpoint. Response is consistent under the
+    same lock as /ws/flips, so readers never see a torn write.
+    """
+    async with _flip_lock:
+        flips_now = list(_flip_cache.values())
+    return JSONResponse({
+        "flips":         flips_now,
+        "total_assets":  len(flips_now),
+        "hub_timestamp": time.time(),
+    })
+
+
+@app.get("/api/flips/{asset}")
+async def api_flips_asset(asset: str):
+    """Single-asset flip lookup."""
+    async with _flip_lock:
+        flip = _flip_cache.get(asset)
+    if not flip:
+        return JSONResponse({"ok": False, "error": f"No flip for {asset}"}, status_code=404)
+    return JSONResponse({"ok": True, "flip": flip, "hub_timestamp": time.time()})
+
+
 @app.get("/api/state")
 async def api_state():
     """Full dashboard state polled by hub_dashboard.html every 2 s."""