#!/usr/bin/env python3 """ Pyth Network Price Feed - Prix en temps réel depuis l'oracle Pyth Compatible avec l'oracle on-chain utilisé par ApolloX """ import requests import pandas as pd from datetime import datetime import logging from typing import Optional, Dict import time logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class PythPriceFeed: """ Récupère les prix depuis Pyth Network (oracle on-chain) Pyth Network est l'oracle de prix utilisé par ApolloX on-chain. En utilisant Pyth pour nos signaux, on évite les écarts de prix entre notre analyse et l'exécution on-chain. """ # API Hermes (gateway publique Pyth) HERMES_URL = "https://hermes.pyth.network" # Price Feed IDs (Pyth Network) # Source: https://pyth.network/developers/price-feed-ids PRICE_FEED_IDS = { 'BTCUSD': '0xe62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43', 'ETHUSD': '0xff61491a931112ddf1bd8147cd1b641375f79f5825126d665480874634fd0ace', 'BNBUSD': '0x2f95862b045670cd22bee3114c39763a4a08beeb663b145d283c31d7d1101c4f', 'SOLUSD': '0xef0d8b6fda2ceba41da15d4095d1da392a0d2f8ed0c6c7bc0f4cfac8c280b56d', 'ADAUSD': '0x2a01deaec9e51a579277b34b122399984d0bbf57e2458a7e42fecd2829867a0d', 'DOTUSD': '0xca3eed9b267293f6595901c734c7525ce8ef49adafe8284606ceb307afa2ca5b', } def __init__(self): self.cache = {} # Cache pour éviter trop de requêtes self.cache_duration = 1 # 1 seconde de cache # Fallback Binance si Pyth indisponible self.binance_url = "https://api.binance.com/api/v3" def get_price_feed_id(self, symbol: str) -> Optional[str]: """ Récupère le Price Feed ID Pyth pour un symbole Args: symbol: BTCUSD, ETHUSD, BNBUSD, etc. (sans T) Returns: Price Feed ID ou None """ # Normaliser le symbole (enlever le T de USDT/BUSD) symbol = symbol.upper() # Enlever suffixe si présent (à la fin seulement) if symbol.endswith('USDT'): symbol = symbol[:-1] # BTCUSDT → BTCUSD elif symbol.endswith('BUSD'): symbol = symbol[:-1] # BTCBUSD → BTCUSD return self.PRICE_FEED_IDS.get(symbol) def get_current_price(self, symbol: str = "BTCUSDT") -> Optional[float]: """ Récupère le prix actuel depuis Pyth Network Args: symbol: Symbole de la paire (ex: BTCUSDT, BNBUSDT) Returns: Prix en float ou None si erreur """ # Vérifier cache cache_key = f"{symbol}_price" if cache_key in self.cache: cached_time, cached_price = self.cache[cache_key] if time.time() - cached_time < self.cache_duration: return cached_price # Obtenir le price feed ID price_feed_id = self.get_price_feed_id(symbol) if not price_feed_id: logger.warning(f"⚠️ Pas de Price Feed Pyth pour {symbol}, utilisation Binance") return self._get_binance_price(symbol) try: # API Hermes Pyth (enlever le 0x du début) feed_id_clean = price_feed_id.replace('0x', '') url = f"{self.HERMES_URL}/api/latest_price_feeds" params = { 'ids[]': feed_id_clean, 'binary': 'false' } response = requests.get(url, params=params, timeout=5) response.raise_for_status() data = response.json() if not data or len(data) == 0: logger.warning(f"⚠️ Pas de données Pyth pour {symbol}") return self._get_binance_price(symbol) # Parser le prix price_data = data[0]['price'] price = float(price_data['price']) expo = int(price_data['expo']) # Calculer prix réel avec exposant actual_price = price * (10 ** expo) # Mettre en cache self.cache[cache_key] = (time.time(), actual_price) logger.debug(f"📊 Prix Pyth {symbol}: ${actual_price:,.2f}") return actual_price except Exception as e: logger.error(f"❌ Erreur Pyth pour {symbol}: {e}") logger.info(f" → Fallback Binance") return self._get_binance_price(symbol) def _get_binance_price(self, symbol: str) -> Optional[float]: """Fallback: récupère prix depuis Binance""" try: url = f"{self.binance_url}/ticker/price" params = {'symbol': symbol} response = requests.get(url, params=params, timeout=5) response.raise_for_status() data = response.json() return float(data['price']) except Exception as e: logger.error(f"❌ Erreur fallback Binance: {e}") return None def get_klines(self, symbol: str = "BTCUSDT", interval: str = "1m", limit: int = 100) -> Optional[pd.DataFrame]: """ Récupère les données OHLCV historiques Note: Pyth ne fournit pas de données historiques OHLCV, on utilise Binance pour ça (nécessaire pour ATR, RSI, etc.) Mais le prix actuel (close) sera mis à jour avec Pyth. Args: symbol: Paire de trading (ex: BTCUSDT) interval: Timeframe (1m, 5m, 15m, 1h, etc.) limit: Nombre de bougies Returns: DataFrame avec colonnes: timestamp, open, high, low, close, volume """ try: url = f"{self.binance_url}/klines" params = { 'symbol': symbol, 'interval': interval, 'limit': limit } response = requests.get(url, params=params, timeout=10) response.raise_for_status() data = response.json() # Convertir en DataFrame df = pd.DataFrame(data, columns=[ 'timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote', 'ignore' ]) # Garder colonnes importantes df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']] # Convertir types df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') for col in ['open', 'high', 'low', 'close', 'volume']: df[col] = df[col].astype(float) # Mettre à jour le dernier close avec prix Pyth actuel current_pyth_price = self.get_current_price(symbol) if current_pyth_price: df.loc[df.index[-1], 'close'] = current_pyth_price logger.debug(f" ✅ Dernier close mis à jour avec Pyth: ${current_pyth_price:,.2f}") return df except Exception as e: logger.error(f"❌ Erreur récupération klines: {e}") return None def get_multiple_prices(self, symbols: list) -> Dict[str, float]: """ Récupère plusieurs prix en une seule requête Args: symbols: Liste de symboles ['BTCUSDT', 'ETHUSDT', ...] Returns: Dict {symbol: price} """ prices = {} # Obtenir tous les price feed IDs feed_ids = [] symbol_to_feed = {} for symbol in symbols: feed_id = self.get_price_feed_id(symbol) if feed_id: feed_ids.append(feed_id) symbol_to_feed[feed_id] = symbol if not feed_ids: logger.warning(f"⚠️ Aucun Price Feed Pyth trouvé, fallback Binance") for symbol in symbols: prices[symbol] = self._get_binance_price(symbol) return prices try: # Requête groupée (enlever 0x des IDs) feed_ids_clean = [fid.replace('0x', '') for fid in feed_ids] url = f"{self.HERMES_URL}/api/latest_price_feeds" params = {'ids[]': feed_ids_clean, 'binary': 'false'} response = requests.get(url, params=params, timeout=5) response.raise_for_status() data = response.json() for item in data: feed_id = item['id'] # Ajouter 0x pour le matching symbol = symbol_to_feed.get('0x' + feed_id) if symbol: price_data = item['price'] price = float(price_data['price']) expo = int(price_data['expo']) actual_price = price * (10 ** expo) prices[symbol] = actual_price logger.info(f"📊 Récupéré {len(prices)} prix depuis Pyth") except Exception as e: logger.error(f"❌ Erreur récupération multiple Pyth: {e}") # Fallback individuel for symbol in symbols: prices[symbol] = self._get_binance_price(symbol) return prices if __name__ == '__main__': logger.info("="*80) logger.info("🧪 TEST PYTH PRICE FEED") logger.info("="*80) feed = PythPriceFeed() # Test prix actuel BTC logger.info("\n📊 Prix actuel BTC (Pyth Oracle):") btc_price = feed.get_current_price("BTCUSDT") if btc_price: logger.info(f" BTC: ${btc_price:,.2f}") # Test prix actuel BNB logger.info("\n📊 Prix actuel BNB (Pyth Oracle):") bnb_price = feed.get_current_price("BNBUSDT") if bnb_price: logger.info(f" BNB: ${bnb_price:,.2f}") # Test prix multiples logger.info("\n📊 Prix multiples (requête groupée):") prices = feed.get_multiple_prices(['BTCUSDT', 'ETHUSDT', 'BNBUSDT']) for symbol, price in prices.items(): if price: logger.info(f" {symbol}: ${price:,.2f}") # Test klines avec prix Pyth logger.info("\n📊 Klines 1m (Binance + Pyth close):") df = feed.get_klines("BTCUSDT", "1m", 10) if df is not None: logger.info(f" Récupéré: {len(df)} bougies") logger.info(f"\n{df[['timestamp', 'close', 'volume']].tail()}") logger.info(f"\n Dernier close (Pyth): ${df['close'].iloc[-1]:,.2f}") logger.info("\n" + "="*80) logger.info("✅ Test terminé") logger.info("="*80)