Update app/app.py

app/app.py

@@ -1,8 +1,8 @@
 """
-Sentinel Arbitrage Engine - v8.0 FINAL (Multi-Oracle)
+Sentinel Arbitrage Engine - v9.0 FINAL (Singleton Broker)
 
-Detects and analyzes price dislocations between major decentralized oracles.
-This architecture is robust and immune to CEX geoblocking.
+This version uses a singleton message broker to ensure robust, stateful
+communication between the background engine and the API endpoints.
 """
 import asyncio
 import os
@@ -14,28 +14,32 @@ from fastapi import FastAPI, Request
 from fastapi.responses import HTMLResponse, StreamingResponse
 from fastapi.templating import Jinja2Templates
 
+# Import our local modules AND the new singleton broker
 from .price_fetcher import PriceFetcher
 from .arbitrage_analyzer import ArbitrageAnalyzer
+from .broker import signal_broker
 
-OPPORTUNITY_THRESHOLD = 0.001  # 0.1% price difference
+OPPORTUNITY_THRESHOLD = 0.001
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    async with httpx.AsyncClient() as client:
-        app.state.price_fetcher = PriceFetcher(client=client)
-        app.state.arbitrage_analyzer = ArbitrageAnalyzer(client=client)
-        app.state.signal_queue: asyncio.Queue = asyncio.Queue()
-        arbitrage_task = asyncio.create_task(run_arbitrage_detector(app, 15)) # Check every 15 seconds
-        
-        print("🚀 Sentinel Arbitrage Engine v8.0 (Multi-Oracle) started.")
-        yield
-        
-        print("⏳ Shutting down engine...")
-        arbitrage_task.cancel()
-        try: await arbitrage_task
-        except asyncio.CancelledError: print("Engine shut down.")
+    # We no longer need to manage the queue in the app state.
+    # The broker exists independently.
+    app.state.price_fetcher = PriceFetcher(httpx.AsyncClient())
+    app.state.arbitrage_analyzer = ArbitrageAnalyzer(httpx.AsyncClient())
+    
+    arbitrage_task = asyncio.create_task(run_arbitrage_detector(app))
+    
+    print("🚀 Sentinel Arbitrage Engine v9.0 (Singleton Broker) started.")
+    yield
+    
+    print("⏳ Shutting down engine...")
+    arbitrage_task.cancel()
+    try: await arbitrage_task
+    except asyncio.CancelledError: print("Engine shut down.")
 
-async def run_arbitrage_detector(app: FastAPI, interval_seconds: int):
+async def run_arbitrage_detector(app: FastAPI):
+    # Now this function uses the globally imported signal_broker
     while True:
         await app.state.price_fetcher.update_prices_async()
         prices = app.state.price_fetcher.get_current_prices()
@@ -51,18 +55,20 @@ async def run_arbitrage_detector(app: FastAPI, interval_seconds: int):
                     "pyth_price": pyth_price, "chainlink_price": chainlink_price,
                     "spread_pct": spread * 100
                 }
-                print(f"⚡️ Oracle Dislocation Detected: {opportunity['spread_pct']:.3f}%")
                 briefing = await app.state.arbitrage_analyzer.get_alpha_briefing(opportunity)
                 if briefing:
                     signal = {**opportunity, **briefing, "timestamp": datetime.now(timezone.utc).isoformat()}
-                    await app.state.signal_queue.put(signal)
+                    # Put the signal into the singleton broker's queue
+                    await signal_broker.queue.put(signal)
+                    print(f"✅ Signal Queued: {signal['spread_pct']:.3f}% Discrepancy")
 
-        await asyncio.sleep(interval_seconds)
+        await asyncio.sleep(15) # Check every 15 seconds
 
 app = FastAPI(title="Sentinel Arbitrage Engine", lifespan=lifespan)
 templates = Jinja2Templates(directory="templates")
 
 def render_signal_card(payload: dict) -> str:
+    # This rendering function remains the same
     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"
@@ -87,10 +93,11 @@ async def serve_dashboard(request: Request):
 
 @app.get("/api/signals/stream")
 async def signal_stream(request: Request):
-    queue: asyncio.Queue = request.app.state.signal_queue
+    # This endpoint now also imports and uses the singleton broker
     async def event_generator():
         while True:
-            payload = await queue.get()
+            # Get the signal from the singleton broker's queue
+            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"