#!/usr/bin/env python3 """ Binance Data Provider - Récupération des données historiques depuis Binance Utilisé pour les signaux uniquement, le trading reste sur Hyperliquid """ import requests import time import logging from typing import Dict, List, Optional class BinanceDataProvider: """Fournisseur de données historiques depuis Binance""" def __init__(self): self.base_url = "https://api.binance.com/api/v3" self.logger = logging.getLogger(__name__) # Cache des données self.cache = {} self.cache_duration = 300 # 5 minutes # Mapping Hyperliquid -> Binance self.symbol_mapping = { 'BTC': 'BTCUSDT', 'ETH': 'ETHUSDT', 'SOL': 'SOLUSDT', 'AVAX': 'AVAXUSDT', 'LINK': 'LINKUSDT', 'DOT': 'DOTUSDT', 'UNI': 'UNIUSDT', 'AAVE': 'AAVEUSDT', 'MATIC': 'MATICUSDT', 'ADA': 'ADAUSDT', 'ATOM': 'ATOMUSDT', 'LTC': 'LTCUSDT', 'BCH': 'BCHUSDT', 'XRP': 'XRPUSDT', 'DOGE': 'DOGEUSDT', 'TRX': 'TRXUSDT', 'ETC': 'ETCUSDT', 'FIL': 'FILUSDT', 'XLM': 'XLMUSDT', 'VET': 'VETUSDT', 'ICP': 'ICPUSDT', 'THETA': 'THETAUSDT', 'FTM': 'FTMUSDT', 'ALGO': 'ALGOUSDT', 'EOS': 'EOSUSDT', 'HBAR': 'HBARUSDT', 'NEAR': 'NEARUSDT', 'FLOW': 'FLOWUSDT', 'MANA': 'MANAUSDT', 'SAND': 'SANDUSDT', 'CRV': 'CRVUSDT', 'GRT': 'GRTUSDT', 'LRC': 'LRCUSDT', 'ENJ': 'ENJUSDT', 'BAT': 'BATUSDT', 'ZRX': '0XUSDT', # Attention: 0x sur Binance 'COMP': 'COMPUSDT', 'MKR': 'MKRUSDT', 'YFI': 'YFIUSDT', 'SUSHI': 'SUSHIUSDT', 'SNX': 'SNXUSDT', 'REN': 'RENUSDT', 'KNC': 'KNCUSDT', 'BAND': 'BANDUSDT', 'STORJ': 'STORJUSDT', 'ZEC': 'ZECUSDT', 'DASH': 'DASHUSDT', 'NEO': 'NEOUSDT', 'QTUM': 'QTUMUSDT', 'ONT': 'ONTUSDT', 'ZIL': 'ZILUSDT', 'IOST': 'IOSTUSDT', 'HOT': 'HOTUSDT', 'DENT': 'DENTUSDT', 'WIN': 'WINUSDT', 'BTT': 'BTTUSDT', 'FUN': 'FUNUSDT', 'ARK': 'ARKUSDT', 'LSK': 'LSKUSDT', 'BNB': 'BNBUSDT', 'APE': 'APEUSDT', 'SHIB': 'SHIBUSDT', 'APT': 'APTUSDT', 'ARB': 'ARBUSDT', 'OP': 'OPUSDT', 'IMX': 'IMXUSDT', 'DYDX': 'DYDXUSDT', 'BLUR': 'BLURUSDT', 'LDO': 'LDOUSDT', 'WLD': 'WLDUSDT', 'SEI': 'SEIUSDT', 'SUI': 'SUIUSDT', 'PEPE': 'PEPEUSDT', 'BONK': 'BONKUSDT', 'JTO': 'JTOUSDT', 'PYTH': 'PYTHUSDT', 'WIF': 'WIFUSDT', 'JUP': 'JUPUSDT', 'TNSR': 'TNSRUSDT', 'W': 'WUSDT', 'ENA': 'ENAUSDT', 'OMNI': 'OMNIUSDT', 'REZ': 'REZUSDT', 'IO': 'IOUSDT', 'ZK': 'ZKUSDT', 'ZRO': 'ZROUSDT', 'G': 'GUSDT', 'DOGS': 'DOGSUSDT', 'TON': 'TONUSDT', 'CATI': 'CATIUSDT', 'HMSTR': 'HMSTRUSDT', 'EIGEN': 'EIGENUSDT', 'NEIRO': 'NEIROUSDT', 'TURBO': 'TURBOUSDT', 'PEOPLE': 'PEOPLEUSDT' } def get_binance_symbol(self, hyperliquid_symbol: str) -> Optional[str]: """Convertit un symbole Hyperliquid vers Binance""" return self.symbol_mapping.get(hyperliquid_symbol) def get_candles(self, symbol: str, interval: str = '1h', limit: int = 100) -> Optional[List[Dict]]: """ Récupère les candles depuis Binance Args: symbol: Symbole Hyperliquid (ex: 'BTC') interval: Intervalle (1h, 4h, 1d) limit: Nombre de candles max 1000 Returns: Liste de candles au format standardisé """ try: # Vérifier le cache cache_key = f"{symbol}_{interval}_{limit}" if cache_key in self.cache: cached_data, timestamp = self.cache[cache_key] if time.time() - timestamp < self.cache_duration: return cached_data # Convertir le symbole binance_symbol = self.get_binance_symbol(symbol) if not binance_symbol: self.logger.warning(f"Symbole {symbol} non mappé vers Binance") return None # Appel API Binance url = f"{self.base_url}/klines" params = { 'symbol': binance_symbol, 'interval': interval, 'limit': min(limit, 1000) # Binance limite à 1000 } response = requests.get(url, params=params, timeout=10) if response.status_code != 200: self.logger.error(f"Erreur Binance {response.status_code} pour {symbol}") return None raw_candles = response.json() # Convertir au format standardisé candles = [] for raw in raw_candles: # Format Binance: [timestamp, open, high, low, close, volume, ...] candles.append({ 'time': int(raw[0]), 'open': float(raw[1]), 'high': float(raw[2]), 'low': float(raw[3]), 'close': float(raw[4]), 'volume': float(raw[5]) }) # Mettre en cache self.cache[cache_key] = (candles, time.time()) self.logger.debug(f"Récupéré {len(candles)} candles {symbol} depuis Binance") return candles except Exception as e: self.logger.error(f"Erreur récupération candles {symbol}: {e}") return None def get_price(self, symbol: str) -> Optional[float]: """Récupère le prix actuel depuis Binance""" try: binance_symbol = self.get_binance_symbol(symbol) if not binance_symbol: return None url = f"{self.base_url}/ticker/price" params = {'symbol': binance_symbol} response = requests.get(url, params=params, timeout=5) if response.status_code == 200: data = response.json() return float(data['price']) return None except Exception as e: self.logger.error(f"Erreur prix {symbol}: {e}") return None def get_24h_volume(self, symbol: str) -> Optional[float]: """Récupère le volume 24h depuis Binance""" try: binance_symbol = self.get_binance_symbol(symbol) if not binance_symbol: return None url = f"{self.base_url}/ticker/24hr" params = {'symbol': binance_symbol} response = requests.get(url, params=params, timeout=5) if response.status_code == 200: data = response.json() return float(data['quoteVolume']) # Volume en USDT return None except Exception as e: self.logger.error(f"Erreur volume {symbol}: {e}") return None def test_connectivity(self) -> bool: """Teste la connectivité avec Binance""" try: response = requests.get(f"{self.base_url}/ping", timeout=5) return response.status_code == 200 except: return False if __name__ == "__main__": # Tests provider = BinanceDataProvider() print("=== TEST BINANCE DATA PROVIDER ===") # Test connectivité if provider.test_connectivity(): print("✅ Connexion Binance OK") else: print("❌ Problème connexion Binance") # Test candles candles = provider.get_candles('BTC', '1h', 24) if candles: print(f"✅ {len(candles)} candles BTC récupérées") print(f" Dernier prix: ${candles[-1]['close']:,.2f}") else: print("❌ Échec récupération candles") # Test prix price = provider.get_price('ETH') if price: print(f"✅ Prix ETH: ${price:,.2f}") else: print("❌ Échec récupération prix") # Test volume volume = provider.get_24h_volume('BTC') if volume: print(f"✅ Volume BTC 24h: ${volume:,.0f}") else: print("❌ Échec récupération volume")