Update websocket_hub.py

websocket_hub.py

@@ -71,9 +71,8 @@ _ALLOWED_TRAINING_FIELDS: frozenset = frozenset({
 })
 
 _ALLOWED_VOTING_FIELDS: frozenset = frozenset({
-    "dominant_signal",
-    "buy_count",
-    "sell_count",
+    "flip_direction",
+    "flip_action",
     "last_price",
     "signal_source",
 })
@@ -91,12 +90,10 @@ def _empty_snapshot(space_name: str) -> dict:
             "avn_accuracy":   0.0,
         },
         "voting": {
-            "dominant_signal": "NEUTRAL",
-            "latest_signal":   "NEUTRAL",   # FIX: most recent signal from inference (not vote majority)
-            "buy_count":       0,
-            "sell_count":      0,
-            "last_price":      0.0,
-            "signal_source":   "LOG",
+            "flip_direction": "NONE",
+            "flip_action":    "HOLD",
+            "last_price":     0.0,
+            "signal_source":  "LOG",
         },
     }
 
@@ -133,20 +130,22 @@ def _validate_and_normalize(space_name: str, raw: dict) -> Optional[dict]:
 
     voting: dict = {}
     if voting_raw:
-        raw_signal = voting_raw.get("dominant_signal", "NEUTRAL")
-        if not isinstance(raw_signal, str):
-            raw_signal = "NEUTRAL"
+        raw_direction = voting_raw.get("flip_direction", "NONE")
+        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_signal = raw_signal.upper() if raw_signal.upper() in {"BUY", "SELL", "NEUTRAL"} else "NEUTRAL"
+        _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 = {
-            "dominant_signal": _clean_signal,
-            "latest_signal":   _clean_signal,   # FIX: capture per-update for freshness
-            "buy_count":       _int(voting_raw.get("buy_count",   0)),
-            "sell_count":      _int(voting_raw.get("sell_count",  0)),
-            "last_price":      _float(voting_raw.get("last_price", 0.0)),
-            "signal_source":   raw_source,
+            "flip_direction": _clean_direction,
+            "flip_action":    _clean_action,
+            "last_price":     _float(voting_raw.get("last_price", 0.0)),
+            "signal_source":  raw_source,
         }
 
     return {
@@ -178,8 +177,7 @@ class ConnectionManager:
         "step":        "training_steps","steps":     "training_steps","n_steps": "training_steps",
     }
     _VOTING_KEYS: frozenset = frozenset({
-        "dominant_signal", "latest_signal", "buy_count", "sell_count", "last_price", "signal_source",
-        "signal", "buy", "sell",
+        "flip_direction", "flip_action", "last_price", "signal_source",
     })
 
     def __init__(self):
@@ -459,7 +457,7 @@ _hub_trades = HubTradeStore()
 logger.info("✅ HubTradeStore initialised — awaiting trade_opened/trade_closed WS messages")
 
 # ── AXRVI live rankings store ─────────────────────────────────────────────────────────
-# Populated by POST /api/axrvi/rankings from the Executo ranker after every
+# Populated by POST /api/flip/rankings from the Executo ranker after every
 # rank_and_gate() cycle (~every 5s). Falls back to hub-snapshot scoring when stale.
 _axrvi_rankings:    List[dict] = []
 _axrvi_rankings_ts: float      = 0.0
@@ -467,7 +465,7 @@ _AXRVI_RANKINGS_TTL: float     = 60.0   # FIX: extended 30→60s to survive slow
 
 # ── 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/axrvi/rankings.
+# Executo ranker POSTs new rankings via POST /api/flip/rankings.
 _top3_clients: Set[WebSocket] = set()
 _top3_clients_lock = asyncio.Lock()
 
@@ -475,7 +473,7 @@ _top3_clients_lock = asyncio.Lock()
 async def _broadcast_top3_rankings(rankings: List[dict]) -> None:
     """
     Broadcast a top3_rankings message to all /ws/top3 subscribers.
-    Called immediately after /api/axrvi/rankings receives a fresh ranking list.
+    Called immediately after /api/flip/rankings receives a fresh ranking list.
     Dead connections are pruned automatically.
     """
     if not _top3_clients:
@@ -651,7 +649,7 @@ async def ws_top3_endpoint(websocket: WebSocket):
     Sends a top3_rankings message immediately on connect (replay of the latest
     known ranking so the client does not have to wait for the next ranker cycle),
     then keeps the socket open to receive subsequent broadcasts triggered by
-    POST /api/axrvi/rankings.
+    POST /api/flip/rankings.
 
     Message format:
       {"type": "top3_rankings", "rankings": [...], "total_assets": N, "hub_timestamp": T}
@@ -1086,12 +1084,19 @@ def _compute_rankings() -> List[dict]:
     """
     Build the rankings list served by /api/state.
 
-    Priority order:
-      1. Live AXRVI rankings pushed by the Executo ranker via
-         POST /api/axrvi/rankings (within the last 30 s).
-         These contain real softmax-Shreve priorities from AXRVINet.
-      2. Fallback: hub-snapshot vote-ratio scoring used before the
-         ranker connects or if the push is stale (e.g. ranker restart).
+    Path 1 — live ranker push (preferred):
+      Rankings received via POST /api/flip/rankings within the last
+      _AXRVI_RANKINGS_TTL seconds. Contains real Shreve priorities,
+      signal_confidence, and epistemic_std from the AXRVI ranker.
+      Hub snapshot fields (training, last_price, flip_direction) are
+      merged in on top to give the freshest available state.
+
+    Path 2 — hub-snapshot pass-through (ranker absent or stale):
+      Emits only what the hub snapshot actually contains: flip_direction,
+      flip_action, last_price, signal_source, and training metrics.
+      score / final_priority / signal_confidence / epistemic_std are
+      set to None — they are NOT synthesised or approximated.
+      Assets are ordered by last_updated (most recently active first).
     """
     global _axrvi_rankings, _axrvi_rankings_ts
 
@@ -1104,17 +1109,15 @@ def _compute_rankings() -> List[dict]:
             snap     = snapshots.get(name, {})
             training = snap.get("training", {})
             voting   = snap.get("voting",   {})
-            buy      = voting.get("buy_count",  r.get("buy_count",  0))
-            sell     = voting.get("sell_count", r.get("sell_count", 0))
             merged.append({
-                # Core AXRVI fields — these are the live ranker values
+                # Core flip-channel fields — live ranker values
                 "rank":              r.get("rank",             0),
                 "space_name":        name,
                 "score":             r.get("score",            0.0),
                 "final_priority":    r.get("final_priority",   r.get("score", 0.0)),
                 "signal_confidence": r.get("signal_confidence",0.0),
-                "dominant_signal":   r.get("dominant_signal",  "NEUTRAL"),
-                "latest_signal":     snap.get("voting", {}).get("latest_signal", r.get("dominant_signal", "NEUTRAL")),
+                "flip_direction":    voting.get("flip_direction", r.get("flip_direction", "NONE")),
+                "flip_action":       voting.get("flip_action",    r.get("flip_action",    "HOLD")),
                 "avn_accuracy":      r.get("avn_accuracy",     0.0),
                 "epistemic_std":     r.get("epistemic_std",    0.0),
                 "training_steps":    r.get("training_steps",   training.get("training_steps", 0)),
@@ -1122,45 +1125,44 @@ def _compute_rankings() -> List[dict]:
                 "actor_loss":        training.get("actor_loss",  0.0),
                 "critic_loss":       training.get("critic_loss", 0.0),
                 "avn_loss":          training.get("avn_loss",    0.0),
-                "buy_count":         buy,
-                "sell_count":        sell,
                 "last_updated":      snap.get("last_updated",   _axrvi_rankings_ts),
             })
         return merged
 
-    # ── Path 2: fallback hub-snapshot scoring ────────────────────────────────
-    ranked: List[dict] = []
+    # ── Path 2: hub-snapshot pass-through (ranker not yet connected or stale) ──
+    # Only fields that are genuinely present in the hub snapshot are emitted.
+    # No score, signal_confidence, or priority is synthesised — those values
+    # belong to the Shreve/AXRVI ranker and must not be approximated here.
+    # Assets are ordered by last_updated so the most recently active asset
+    # appears first; rank is assigned on that basis alone.
+    raw_snaps: List[dict] = []
     for name, snap in manager.get_all_snapshots().items():
-        training = snap.get("training", {})
         voting   = snap.get("voting",   {})
-        buy      = voting.get("buy_count",  0)
-        sell     = voting.get("sell_count", 0)
-        total    = buy + sell
-        sig_conf = (max(buy, sell) / total) if total > 0 else 0.0
-        avn_acc  = training.get("avn_accuracy", 0.0)
-        score    = round(sig_conf - avn_acc, 6)
-        ranked.append({
-            "rank":              0,
-            "space_name":        name,
-            "score":             score,
-            "final_priority":    score,
-            "signal_confidence": round(sig_conf, 6),
-            "avn_accuracy":      round(avn_acc,  6),
-            "dominant_signal":   voting.get("dominant_signal", "NEUTRAL"),
-            "latest_signal":     voting.get("latest_signal",   voting.get("dominant_signal", "NEUTRAL")),
-            "buy_count":         buy,
-            "sell_count":        sell,
-            "training_steps":    training.get("training_steps", 0),
-            "actor_loss":        training.get("actor_loss",     0.0),
-            "critic_loss":       training.get("critic_loss",    0.0),
-            "avn_loss":          training.get("avn_loss",       0.0),
-            "last_updated":      snap.get("last_updated",       0.0),
-            "epistemic_std":     0.0,
+        training = snap.get("training", {})
+        raw_snaps.append({
+            "rank":           0,           # assigned below after sort
+            "space_name":     name,
+            "score":          None,         # not available without ranker
+            "final_priority": None,         # not available without ranker
+            "signal_confidence": None,      # not available without ranker
+            "epistemic_std":  None,         # not available without ranker
+            # ── real flip-channel fields from hub snapshot ──────────────────
+            "flip_direction": voting.get("flip_direction", "NONE"),
+            "flip_action":    voting.get("flip_action",    "HOLD"),
+            "last_price":     voting.get("last_price",     0.0),
+            "signal_source":  voting.get("signal_source",  "LOG"),
+            # ── real training fields from hub snapshot ───────────────────────
+            "avn_accuracy":   training.get("avn_accuracy",   0.0),
+            "training_steps": training.get("training_steps", 0),
+            "actor_loss":     training.get("actor_loss",     0.0),
+            "critic_loss":    training.get("critic_loss",    0.0),
+            "avn_loss":       training.get("avn_loss",       0.0),
+            "last_updated":   snap.get("last_updated",       0.0),
         })
-    ranked.sort(key=lambda r: r["score"], reverse=True)
-    for i, r in enumerate(ranked):
+    raw_snaps.sort(key=lambda r: r["last_updated"], reverse=True)
+    for i, r in enumerate(raw_snaps):
         r["rank"] = i + 1
-    return ranked
+    return raw_snaps
 
 
 @app.get("/")
@@ -1184,12 +1186,12 @@ async def serve_dashboard():
     )
 
 
-@app.post("/api/axrvi/rankings")
+@app.post("/api/flip/rankings")
 async def receive_axrvi_rankings(request: Request):
     """
     Called by the Executo ranker after every rank_and_gate() cycle (~5 s).
     Stores the live AXRVI-scored ranking list so _compute_rankings() can serve
-    it from /api/state instead of the stale hub-snapshot vote-ratio fallback.
+    it from /api/state instead of the hub-snapshot pass-through (Path 2).
 
     Expected body:
       {"rankings": [{"space_name": "V75", "score": 0.24, "rank": 1, ...}, ...]}
@@ -1207,9 +1209,9 @@ async def receive_axrvi_rankings(request: Request):
     _axrvi_rankings    = rankings
     _axrvi_rankings_ts = time.time()
     logger.debug(
-        f"[AXRVI Rankings] Received {len(rankings)} assets | "
+        f"[Flip Rankings] Received {len(rankings)} assets | "
         f"top={rankings[0].get('space_name','?')} score={rankings[0].get('score',0):.4f}"
-        if rankings else "[AXRVI Rankings] Received empty list"
+        if rankings else "[Flip Rankings] Received empty list"
     )
     # Broadcast to all connected top3_client.py instances immediately
     if rankings: