Update app/app.py

app/app.py

@@ -1,103 +1,113 @@
 """
-Sentinel Arbitrage Engine - v11.0 FINAL (Multi-Asset)
+Sentinel Arbitrage Engine - v12.0 FINAL (Correct Mount)
 
-Detects and analyzes price dislocations for multiple assets across
-decentralized oracles.
+This version uses the robust singleton broker pattern and correctly mounts
+the Socket.IO application to the path expected by the client, ensuring
+a successful real-time connection.
 """
 import asyncio
 import os
 from contextlib import asynccontextmanager
-from datetime import datetime, timezone
 import json
+import time
+from datetime import datetime, timezone
+
 import httpx
-from fastapi import FastAPI, Request
-from fastapi.responses import HTMLResponse, StreamingResponse
-from fastapi.staticfiles import StaticFiles # <-- We use this instead of Jinja2
+import socketio
+from fastapi import FastAPI
+from fastapi.staticfiles import StaticFiles
 
-# Relative imports for package structure
+# --- IMPORTS ---
+# Correctly use relative imports for our local package structure
 from .price_fetcher import PriceFetcher
 from .arbitrage_analyzer import ArbitrageAnalyzer
-from .broker import signal_broker
+from .broker import signal_broker # Using the robust singleton broker
 
-OPPORTUNITY_THRESHOLD = 0.0015  # 0.15% price difference
+OPPORTUNITY_THRESHOLD = 0.0015
 
+# --- SOCKET.IO SERVER SETUP ---
+# This creates the server instance that will handle WebSocket connections
+sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*')
+# This wraps the Socket.IO server in an ASGI application
+socket_app = socketio.ASGIApp(sio)
+
+
+# --- APPLICATION LIFESPAN ---
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """Manages application startup and shutdown events."""
+    """Manages the application's startup and shutdown procedures."""
+    print("🚀 Initializing Sentinel Arbitrage Engine v12.0...")
     async with httpx.AsyncClient() as client:
+        # We can store these on the app state, but the broker handles the critical part
         app.state.price_fetcher = PriceFetcher(client)
         app.state.arbitrage_analyzer = ArbitrageAnalyzer(client)
-        arbitrage_task = asyncio.create_task(run_arbitrage_detector(app))
-        print("🚀 Sentinel Arbitrage Engine v11.0 (Multi-Asset) started.")
-        yield
+        
+        # Launch the engine as a background task
+        arbitrage_task = asyncio.create_task(
+            run_arbitrage_detector(app.state.price_fetcher, app.state.arbitrage_analyzer)
+        )
+        
+        print("✅ Engine is online and hunting for opportunities.")
+        yield # Application runs here
+        
         print("⏳ Shutting down engine...")
         arbitrage_task.cancel()
         try: await arbitrage_task
-        except asyncio.CancelledError: print("Engine shut down.")
+        except asyncio.CancelledError: print("Engine shut down gracefully.")
 
-async def run_arbitrage_detector(app: FastAPI):
-    """The core engine loop. Checks for opportunities and queues them."""
+async def run_arbitrage_detector(price_fetcher, analyzer):
+    """The core engine loop; detects opportunities and puts them in the broker queue."""
     while True:
-        await app.state.price_fetcher.update_prices_async()
-        all_prices = app.state.price_fetcher.get_all_prices()
-        
-        for asset, prices in all_prices.items():
-            pyth_price = prices.get("pyth")
-            chainlink_price = prices.get("chainlink_agg")
+        try:
+            await price_fetcher.update_prices_async()
+            all_prices = price_fetcher.get_all_prices()
             
-            if pyth_price and chainlink_price and pyth_price > 0:
-                spread = abs(pyth_price - chainlink_price) / chainlink_price
-                if spread > OPPORTUNITY_THRESHOLD:
-                    opportunity = {
-                        "asset": asset,
-                        "pyth_price": pyth_price, "chainlink_price": chainlink_price,
-                        "spread_pct": spread * 100
-                    }
-                    print(f"⚡️ Dislocation for {asset}: {opportunity['spread_pct']:.3f}%")
-                    briefing = await app.state.arbitrage_analyzer.get_alpha_briefing(asset, opportunity)
-                    if briefing:
-                        signal = {**opportunity, **briefing, "timestamp": datetime.now(timezone.utc).isoformat()}
-                        await signal_broker.queue.put(signal)
+            for asset, prices in all_prices.items():
+                pyth_price = prices.get("pyth")
+                chainlink_price = prices.get("chainlink_agg")
+                
+                if pyth_price and chainlink_price and pyth_price > 0:
+                    spread = abs(pyth_price - chainlink_price) / chainlink_price
+                    if spread > OPPORTUNITY_THRESHOLD:
+                        opportunity = {
+                            "asset": asset, "pyth_price": pyth_price,
+                            "chainlink_price": chainlink_price, "spread_pct": spread * 100
+                        }
+                        print(f"⚡️ Dislocation for {asset}: {opportunity['spread_pct']:.3f}%")
+                        briefing = await analyzer.get_alpha_briefing(asset, opportunity)
+                        if briefing:
+                            signal = {**opportunity, **briefing, "timestamp": datetime.now(timezone.utc).isoformat()}
+                            await signal_broker.queue.put(signal)
+                            print(f"✅ Signal Emitted via Broker: {signal['strategy']}")
+        except Exception as e:
+            print(f"❌ ERROR in engine loop: {e}")
 
         await asyncio.sleep(15)
 
-# --- FastAPI App Initialization ---
-app = FastAPI(title="Sentinel Arbitrage Engine", lifespan=lifespan)
+# --- FASTAPI APP SETUP ---
+app = FastAPI(lifespan=lifespan)
+
+# --- SOCKET.IO EVENT HANDLERS ---
+@sio.event
+async def connect(sid, environ):
+    """Handles new client connections."""
+    print(f"✅ Client connected: {sid}")
+    # We can even send a welcome message here if we want
+    await sio.emit('welcome', {'message': 'Connection to Sentinel Engine established!'}, to=sid)
+
+@sio.event
+async def disconnect(sid):
+    """Handles client disconnections."""
+    print(f"🔥 Client disconnected: {sid}")
 
-# --- HTML Rendering Helper ---
-def render_signal_card(payload: dict) -> str:
-    """Renders a dictionary of analysis into a styled HTML table row."""
-    s = payload
-    time_str = datetime.fromisoformat(s['timestamp']).strftime('%H:%M:%S UTC')
-    pyth_class = "buy" if s['pyth_price'] < s['chainlink_price'] else "sell"
-    chainlink_class = "sell" if s['pyth_price'] < s['chainlink_price'] else "buy"
-    
-    return f"""
-    <tr hx-swap-oob="afterbegin:#opportunities-table">
-        <td><strong>{s['asset']}/USD</strong></td>
-        <td><span class="{pyth_class}">Pyth Network</span><br>${s['pyth_price']:,.2f}</td>
-        <td><span class="{chainlink_class}">Chainlink Agg.</span><br>${s['chainlink_price']:,.2f}</td>
-        <td><strong>{s['spread_pct']:.3f}%</strong></td>
-        <td><span class="risk-{s.get('risk', 'low').lower()}">{s.get('risk', 'N/A')}</span></td>
-        <td>{s.get('rationale', 'N/A')}</td>
-        <td><button class="trade-btn">{s.get('strategy', 'N/A')}</button></td>
-    </tr>
-    <div id="last-update-time" hx-swap-oob="true">{time_str}</div>
-    """
+# --- THE CRITICAL FIX ---
+# Mount the Socket.IO application at the path the client expects ('/socket.io/').
+# This MUST come before the root static file mount.
+app.mount('/socket.io', socket_app)
 
-# --- API Endpoints ---
-@app.get("/api/signals/stream")
-async def signal_stream(request: Request):
-    """SSE stream for the automated Signal Stream."""
-    async def event_generator():
-        while True:
-            payload = await signal_broker.queue.get()
-            html_card = render_signal_card(payload)
-            data_payload = html_card.replace('\n', ' ').strip()
-            yield f"event: message\ndata: {data_payload}\n\n"
-    return StreamingResponse(event_generator(), media_type="text/event-stream")
+# Mount the static directory to serve index.html at the root path "/".
+app.mount("/", StaticFiles(directory="static", html=True), name="static")
 
-# --- Static File Server ---
-# This single mount point serves index.html for the root path "/"
-# and any other files like CSS or JS from the "static" directory.
-app.mount("/", StaticFiles(directory="static", html=True), name="static")
+# We no longer need the old SSE endpoint, as Socket.IO handles everything.
+# The `run_arbitrage_detector` now emits directly to clients,
+# and the `static/index.html`'s JavaScript listens for these events.