Update app/price_fetcher.py

app/price_fetcher.py

@@ -1,34 +1,41 @@
 """
-A high-throughput, fault-tolerant, multi-source price fetcher engine.
-Designed to bypass rate-limiting and ensure constant data flow.
+A professional-grade, WebSocket-based, high-frequency price engine.
+
+This engine establishes a persistent connection to the Binance real-time data
+stream for BTC/USDT, providing millisecond-latency price updates. It uses
+Pyth as a redundant, cross-checking data source.
 """
 import asyncio
+import json
 import logging
-from typing import Dict, Optional, List
+from typing import Dict, Optional
 import httpx
-import os
-import statistics
+import websockets
 
 logger = logging.getLogger(__name__)
 
 class PriceFetcher:
-    # Pyth for our on-chain data
-    PYTH_URL = "https://hermes.pyth.network/v2/updates/price/latest?ids[]=e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415B43"
-
-    # A list of our off-chain aggregator sources
-    AGGREGATOR_SOURCES = {
-        "coingecko": "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd",
-        "coincap": "https://api.coincap.io/v2/assets/bitcoin",
-    }
+    # Primary data source: Binance WebSocket for real-time trades
+    BINANCE_WS_URL = "wss://stream.binance.com:9443/ws/btcusdt@trade"
     
-    # Use Pro API if a key is provided as a secret
-    COINGECKO_PRO_URL = "https://pro-api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd"
+    # Secondary source for sanity checks
+    PYTH_URL = "https://hermes.pyth.network/v2/updates/price/latest?ids[]=e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415B43"
 
     def __init__(self, client: httpx.AsyncClient):
         self.client = client
-        self._prices: Dict[str, Optional[float]] = {"on_chain_pyth": None, "off_chain_agg": None}
+        self._prices: Dict[str, Optional[float]] = {"realtime_binance": None, "on_chain_pyth": None}
         self._lock = asyncio.Lock()
-        self.coingecko_api_key = os.getenv("COINGECKO_API_KEY")
+        self.last_binance_price: Optional[float] = None
+
+    async def get_current_prices(self) -> Dict[str, Optional[float]]:
+        """Returns a copy of the latest prices."""
+        # Fetch a fresh Pyth price for the most up-to-date comparison
+        pyth_price = await self._fetch_pyth()
+        async with self._lock:
+            self._prices["on_chain_pyth"] = pyth_price
+            # Use the last price received from the continuously running websocket
+            self._prices["realtime_binance"] = self.last_binance_price
+            return self._prices.copy()
 
     async def _fetch_pyth(self) -> Optional[float]:
         try:
@@ -36,60 +43,27 @@ class PriceFetcher:
             resp.raise_for_status()
             price_data = resp.json()['parsed'][0]['price']
             return int(price_data['price']) / (10 ** abs(int(price_data['expo'])))
-        except Exception:
-            return None # Fail silently, we have other sources
-
-    async def _fetch_aggregator(self, name: str, url: str) -> Optional[float]:
-        try:
-            headers = {}
-            if name == "coingecko" and self.coingecko_api_key:
-                url = self.COINGECKO_PRO_URL
-                headers["X-Cg-Pro-Api-Key"] = self.coingecko_api_key
-            
-            resp = await self.client.get(url, timeout=5, headers=headers)
-            if resp.status_code == 429: # Explicitly handle rate limiting
-                logger.warning(f"⚠️ Rate limited by {name}. Skipping.")
-                return None
-            resp.raise_for_status()
-            data = resp.json()
-
-            if name == "coingecko": return float(data['bitcoin']['usd'])
-            if name == "coincap": return float(data['data']['priceUsd'])
-            
         except Exception as e:
-            logger.error(f"❌ Failed to fetch from {name}: {e}")
-        return None
-
-    def _get_sane_average(self, prices: List[float]) -> Optional[float]:
-        """Calculates the average price, discarding outliers."""
-        if not prices: return None
-        if len(prices) < 3: return statistics.mean(prices)
-        
-        # For 3 or more sources, discard any price that is more than 2%
-        # different from the median, then average the rest.
-        median = statistics.median(prices)
-        sane_prices = [p for p in prices if abs(p - median) / median < 0.02]
-        
-        if not sane_prices: return median # Fallback to median if all are outliers
-        return statistics.mean(sane_prices)
-
-    async def update_prices_async(self):
-        # Fetch all sources concurrently
-        pyth_task = self._fetch_pyth()
-        agg_tasks = [self._fetch_aggregator(name, url) for name, url in self.AGGREGATOR_SOURCES.items()]
-        
-        results = await asyncio.gather(pyth_task, *agg_tasks)
-        
-        pyth_price = results[0]
-        agg_prices = [p for p in results[1:] if p is not None]
-        
-        sane_agg_price = self._get_sane_average(agg_prices)
-        
-        async with self._lock:
-            self._prices["on_chain_pyth"] = pyth_price
-            self._prices["off_chain_agg"] = sane_agg_price
-        
-        logger.info(f"✅ Prices updated: On-Chain={pyth_price}, Off-Chain Agg={sane_agg_price} (from {len(agg_prices)} sources)")
+            logger.error(f"❌ Failed to fetch from Pyth: {e}")
+            return None
 
-    def get_current_prices(self) -> Dict[str, Optional[float]]:
-        return self._prices.copy()
+    async def run_websocket_listener(self):
+        """
+        Connects to the Binance WebSocket and continuously updates the latest price.
+        This function runs forever in the background.
+        """
+        while True:
+            try:
+                async with websockets.connect(self.BINANCE_WS_URL) as websocket:
+                    logger.info("✅ Connected to Binance WebSocket stream.")
+                    while True:
+                        message = await websocket.recv()
+                        data = json.loads(message)
+                        # The 'p' key in the trade stream is the price
+                        if 'p' in data:
+                            price = float(data['p'])
+                            async with self._lock:
+                                self.last_binance_price = price
+            except Exception as e:
+                logger.error(f"❌ WebSocket connection error: {e}. Reconnecting in 5 seconds...")
+                await asyncio.sleep(5)