Update websocket_hub.py

websocket_hub.py

@@ -313,158 +313,147 @@ class ConnectionManager:
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
-# SECTION 3 — TRADE LOG PARSER  (native — no patch script needed)
+# SECTION 3 — HUB TRADE STORE  (in-memory, fed by WebSocket messages)
+# ══════════════════════════════════════════════════════════════════════════════════════
+#
+# ROOT CAUSE FIX: The previous TradeLogParser read from *.log files on THIS container
+# (/app/ranker_logs).  Those files NEVER exist on the Executo Hub space — they are
+# written by ranker processes running in the individual executor spaces (V75, V50, …),
+# each in their own separate container with their own filesystem.
+#
+# Fix: replace file-based parsing with an in-memory store that is populated when
+# executor spaces send WebSocket trade events to this hub.
+#
+# Executor spaces must send:
+#   {"type": "trade_opened", "data": {trade_id, asset, direction, entry, qty, opened_at}}
+#   {"type": "trade_closed",  "data": {trade_id, asset, pnl, exit_price, closed_at}}
+#
+# See ranker_logging.py — the on_event callback already fires for every log entry.
+# Wire it in your executor space's ranker like this:
+#
+#   def _trade_ws_hook(entry: dict):
+#       cat = entry.get("category", "")
+#       msg = entry.get("message", "")
+#       if cat != "TRADE":
+#           return
+#       meta = entry.get("metadata") or {}
+#       if "TRADE OPENED" in msg:
+#           asyncio.create_task(ws.send_text(json.dumps({
+#               "type": "trade_opened",
+#               "data": {
+#                   "trade_id":  meta.get("trade_id"),
+#                   "asset":     entry.get("asset"),
+#                   "direction": meta.get("direction", "?"),
+#                   "entry":     meta.get("price"),
+#                   "qty":       meta.get("qty", 0.0),
+#                   "opened_at": entry.get("timestamp", ""),
+#               },
+#           })))
+#       elif "TRADE CLOSED" in msg:
+#           asyncio.create_task(ws.send_text(json.dumps({
+#               "type": "trade_closed",
+#               "data": {
+#                   "trade_id":  meta.get("trade_id"),
+#                   "asset":     entry.get("asset"),
+#                   "pnl":       meta.get("pnl", 0.0),
+#                   "exit_price": meta.get("exit_price"),
+#                   "closed_at": entry.get("timestamp", ""),
+#               },
+#           })))
+#
+#   ranker_logger = RankerLogger(..., on_event=_trade_ws_hook)
 # ══════════════════════════════════════════════════════════════════════════════════════
 
-class TradeLogParser:
+class HubTradeStore:
     """
-    Lightweight trade-log reader that watches ranker_logs/*.log files and
-    exposes open/closed trade state via get_state().
-
-    Imported from hub_dashboard_service if available; otherwise this built-in
-    fallback is used so /api/trades routes always work.
+    In-memory trade store populated by WebSocket trade-event messages from
+    executor spaces.  Thread-safe.  Replaces the broken file-based TradeLogParser.
     """
 
-    def __init__(self, log_dir: str = "/app/ranker_logs"):
-        self.log_dir  = log_dir
-        self._state   = {"open": {}, "closed": [], "stats": {}}
-        self._lock    = None   # set in start_background()
-        self._running = False
-
-    def start_background(self) -> None:
-        """Schedule background log-scanning on the running event loop."""
-        self._running = True
-        asyncio.get_event_loop().create_task(self._scan_loop())
-        logger.info(f"📂 TradeLogParser started — watching {self.log_dir}")
-
-    async def _scan_loop(self) -> None:
-        while self._running:
-            try:
-                self._refresh()
-            except Exception as e:
-                logger.debug(f"TradeLogParser scan error: {e}")
-            await asyncio.sleep(5)
-
-    def _refresh(self) -> None:
-        """Re-parse all log files and rebuild open/closed state."""
-        open_trades:   Dict[str, dict] = {}
-        closed_trades: List[dict]      = []
-
-        log_files = sorted(glob.glob(os.path.join(self.log_dir, "*.log")))
-        for fpath in log_files:
-            try:
-                with open(fpath, "r", encoding="utf-8", errors="replace") as fh:
-                    for line in fh:
-                        self._parse_line(line, open_trades, closed_trades)
-            except Exception:
-                pass
-
-        # Sort closed newest-first
-        closed_trades.sort(key=lambda t: t.get("closed_ts", 0), reverse=True)
-
-        # Compute stats
-        pnls       = [t.get("pnl", 0.0) for t in closed_trades]
-        wins       = [p for p in pnls if p > 0]
-        total      = len(pnls)
-        stats = {
-            "total_closed":  total,
-            "total_open":    len(open_trades),
-            "win_rate":      round(len(wins) / total, 4) if total else 0.0,
-            "total_pnl":     round(sum(pnls), 6),
-            "avg_pnl":       round(sum(pnls) / total, 6) if total else 0.0,
-            "avg_pnl_trade": round(sum(pnls) / total, 6) if total else 0.0,
+    def __init__(self) -> None:
+        self._open:   Dict[str, dict] = {}   # trade_id → record
+        self._closed: List[dict]      = []   # newest-first, capped at 500
+        self._stats = {
+            "total_opened": 0,
+            "total_closed": 0,
+            "total_pnl":    0.0,
+            "win_count":    0,
+            "loss_count":   0,
         }
-
-        self._state = {
-            "open":   list(open_trades.values()),
-            "closed": closed_trades,
-            "stats":  stats,
-        }
-
-    # ── Log-line parser ──────────────────────────────────────────────────────────────
-    # Matches lines like:
-    #   [2026-03-31 10:02:07] | INFO | TRADE | V75 | TRADE OPENED | ID=V75_... | ...
-    #   [2026-03-31 10:02:07] | INFO | TRADE | V75 | TRADE CLOSED | ID=V75_... | pnl=-32.62 | ...
-
-    def _parse_line(
-        self,
-        line:         str,
-        open_trades:  Dict[str, dict],
-        closed_trades: List[dict],
-    ) -> None:
-        if "| TRADE |" not in line:
-            return
-
-        parts = [p.strip() for p in line.split("|")]
-        # parts[0]=timestamp, [1]=level, [2]="TRADE", [3]=asset, [4]=event, [5..]=fields
-        if len(parts) < 5:
-            return
-
-        try:
-            ts_str = parts[0].strip("[] ")
-            ts     = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").timestamp()
-        except Exception:
-            ts = 0.0
-
-        asset = parts[3].strip()
-        event = parts[4].strip().upper()
-
-        # Build a quick field dict from KEY=VALUE tokens in the rest of the line
-        fields: Dict[str, str] = {}
-        for segment in parts[5:]:
-            for token in segment.split():
-                if "=" in token:
-                    k, _, v = token.partition("=")
-                    fields[k.strip()] = v.strip().rstrip(",")
-
-        trade_id = fields.get("ID", f"{asset}_{int(ts)}")
-
-        def _f(k: str) -> float:
-            try:    return float(fields.get(k, 0))
-            except: return 0.0
-
-        if "TRADE OPENED" in event:
-            open_trades[trade_id] = {
+        self._lock = threading.Lock()
+
+    def open_trade(self, space_name: str, data: dict) -> None:
+        trade_id  = data.get("trade_id") or f"{space_name}_{int(time.time())}"
+        direction = str(data.get("direction", "?")).upper()
+        entry_px  = data.get("entry") or data.get("price") or 0.0
+        with self._lock:
+            self._open[trade_id] = {
                 "trade_id":  trade_id,
-                "asset":     asset,
-                "direction": fields.get("Dir", fields.get("Direction", "?")).capitalize(),
-                "entry":     _f("Entry"),
-                "qty":       _f("Qty"),
-                "open_pnl":  0.0,
-                "opened_ts": ts,
-                "opened":    ts_str,
+                "asset":     data.get("asset", space_name),
+                "direction": direction,
+                "entry":     float(entry_px),
+                "qty":       float(data.get("qty", 0.0)),
+                "opened_at": data.get("opened_at", datetime.utcnow().isoformat()[:19]),
+                "status":    "OPEN",
             }
+            self._stats["total_opened"] += 1
+        logger.info(
+            f"[HubTradeStore] OPEN  {trade_id} | {direction} @ {entry_px} "
+            f"(from {space_name})"
+        )
 
-        elif "TRADE CLOSED" in event:
-            # Remove from open if present
-            trade = open_trades.pop(trade_id, {})
-            pnl   = _f("pnl")
-            ret   = _f("return")
-            closed_trades.append({
-                "trade_id":   trade_id,
-                "asset":      asset,
-                "direction":  trade.get("direction", "?"),
-                "entry":      trade.get("entry", _f("Entry")),
-                "exit":       _f("Exit") or _f("exit"),
-                "qty":        trade.get("qty", _f("Qty")),
+    def close_trade(self, space_name: str, data: dict) -> None:
+        trade_id   = data.get("trade_id")
+        pnl        = float(data.get("pnl", 0.0))
+        exit_price = data.get("exit_price")
+        with self._lock:
+            trade = self._open.pop(trade_id, {}) if trade_id else {}
+            closed_rec = {
+                "trade_id":   trade_id or "UNKNOWN",
+                "asset":      data.get("asset") or trade.get("asset", space_name),
+                "direction":  str(data.get("direction") or trade.get("direction", "?")).upper(),
+                "entry":      data.get("entry")      or trade.get("entry",     0.0),
+                "exit_price": float(exit_price) if exit_price is not None else None,
+                "qty":        data.get("qty")        or trade.get("qty",       0.0),
                 "pnl":        pnl,
-                "return_pct": ret,
-                "closed_ts":  ts,
-                "closed":     ts_str,
-                "duration":   round(ts - trade.get("opened_ts", ts), 0) if trade else 0,
-            })
+                "closed_at":  data.get("closed_at", datetime.utcnow().isoformat()[:19]),
+                "status":     "CLOSED",
+            }
+            self._closed.insert(0, closed_rec)       # newest-first
+            if len(self._closed) > 500:
+                self._closed = self._closed[:500]
+            self._stats["total_closed"] += 1
+            self._stats["total_pnl"]    += pnl
+            if pnl >= 0:
+                self._stats["win_count"]  += 1
+            else:
+                self._stats["loss_count"] += 1
+        logger.info(
+            f"[HubTradeStore] CLOSE {trade_id} | pnl={pnl:+.4f} "
+            f"(from {space_name})"
+        )
 
     def get_state(self) -> dict:
-        return self._state
-
+        with self._lock:
+            open_list   = list(self._open.values())
+            closed_list = list(self._closed[:100])   # newest 100 for dashboard
+            stats       = dict(self._stats)
+        total_closed = stats["total_closed"]
+        stats["win_rate"] = (
+            round(stats["win_count"] / total_closed * 100, 1)
+            if total_closed > 0 else 0.0
+        )
+        return {
+            "open":   open_list,
+            "closed": closed_list,
+            "stats":  stats,
+        }
 
-# ── Bootstrap parser ─────────────────────────────────────────────────────────────────
-# Use built-in TradeLogParser only — importing hub_dashboard_service causes a crash
-# because it instantiates Flask, starts threads, and calls app.run() at module level.
 
-_LOG_DIR = os.environ.get("RANKER_LOG_DIR", "/app/ranker_logs")
-_trade_parser = TradeLogParser(log_dir=_LOG_DIR)
-logger.info("✅ TradeLogParser using built-in")
+# ── Bootstrap ─────────────────────────────────────────────────────────────────────────
+_LOG_DIR    = os.environ.get("RANKER_LOG_DIR", "/app/ranker_logs")
+_hub_trades = HubTradeStore()
+logger.info("✅ HubTradeStore initialised — awaiting trade_opened/trade_closed WS messages")
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -488,9 +477,8 @@ manager = ConnectionManager()
 
 @app.on_event("startup")
 async def _on_startup():
-    """Start background trade-log scanner after the event loop is running."""
-    _trade_parser.start_background()
-    logger.info("🚀 Trade log scanner started")
+    """Nothing to start — HubTradeStore is in-memory, populated by WS messages."""
+    logger.info("🚀 HubTradeStore ready (no background scanner needed)")
 
 
 # ══════════════════════════════════════════════════════════════════════════════════════
@@ -549,6 +537,16 @@ async def ws_publisher_endpoint(websocket: WebSocket, space_name: str):
             elif msg_type in ("heartbeat", "identify", "ping"):
                 pass
 
+            elif msg_type == "trade_opened":
+                # Executor space opened a trade — add to the hub's in-memory store.
+                # data = {trade_id, asset, direction, entry, qty, opened_at}
+                _hub_trades.open_trade(space_name, data.get("data", data))
+
+            elif msg_type == "trade_closed":
+                # Executor space closed a trade — update the hub's in-memory store.
+                # data = {trade_id, asset, pnl, exit_price, closed_at}
+                _hub_trades.close_trade(space_name, data.get("data", data))
+
             else:
                 # Bug B fix: don't silently swallow. Try to rescue training/voting
                 # fields that live at the top level of an unrecognised message type.
@@ -676,20 +674,20 @@ async def api_debug_hub():
 @app.get("/api/trades")
 async def api_trades():
     """Full trade state: open trades, recent closed trades, summary stats."""
-    return JSONResponse(_trade_parser.get_state())
+    return JSONResponse(_hub_trades.get_state())
 
 
 @app.get("/api/trades/open")
 async def api_trades_open():
     """Open trades only."""
-    state = _trade_parser.get_state()
+    state = _hub_trades.get_state()
     return JSONResponse({"open": state["open"]})
 
 
 @app.get("/api/trades/closed")
 async def api_trades_closed(limit: int = 50):
     """Recent closed trades (newest first) + cumulative stats."""
-    state = _trade_parser.get_state()
+    state = _hub_trades.get_state()
     return JSONResponse({
         "closed": state["closed"][:limit],
         "stats":  state["stats"],