Spaces:
Running
Running
""" | |
Sentinel Arbitrage Engine - v14.0 FINAL (Socket.IO Multi-Asset) | |
The definitive version, combining a robust multi-asset detection engine | |
with a high-performance Socket.IO backend for guaranteed, real-time | |
signal delivery and a professional UI. | |
""" | |
import asyncio | |
import os | |
from contextlib import asynccontextmanager | |
import json | |
import time | |
from datetime import datetime, timezone | |
import httpx | |
import socketio | |
from fastapi import FastAPI | |
from fastapi.staticfiles import StaticFiles | |
from .price_fetcher import PriceFetcher | |
from .arbitrage_analyzer import ArbitrageAnalyzer | |
OPPORTUNITY_THRESHOLD = 0.0015 | |
# --- Socket.IO Server Setup --- | |
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*') | |
socket_app = socketio.ASGIApp(sio) | |
async def lifespan(app: FastAPI): | |
print("π Initializing Sentinel Arbitrage Engine v14.0...") | |
async with httpx.AsyncClient() as client: | |
app.state.price_fetcher = PriceFetcher(client) | |
app.state.arbitrage_analyzer = ArbitrageAnalyzer(client) | |
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 | |
print("β³ Shutting down engine...") | |
arbitrage_task.cancel() | |
try: await arbitrage_task | |
except asyncio.CancelledError: print("Engine shut down gracefully.") | |
async def run_arbitrage_detector(price_fetcher, analyzer): | |
last_opportunity_time = 0 | |
while True: | |
try: | |
await price_fetcher.update_prices_async() | |
all_prices = price_fetcher.get_all_prices() | |
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: | |
current_time = time.time() | |
if current_time - last_opportunity_time > 20: # Throttle Gemini calls | |
last_opportunity_time = current_time | |
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 sio.emit('new_signal', signal) | |
print(f"β Signal Emitted for {asset}: {signal['strategy']}") | |
except Exception as e: | |
print(f"β ERROR in engine loop: {e}") | |
await asyncio.sleep(15) | |
# --- FastAPI App & Socket.IO Setup --- | |
app = FastAPI(lifespan=lifespan) | |
async def connect(sid, environ): | |
print(f"β Client connected: {sid}") | |
async def disconnect(sid): | |
print(f"π₯ Client disconnected: {sid}") | |
app.mount('/socket.io', socket_app) | |
app.mount("/", StaticFiles(directory="static", html=True), name="static") |