""" Crypto Data Fetcher avec Cache SQLite ====================================== Module pour récupérer et mettre en cache les données crypto depuis Binance Utilise SQLite pour stocker localement et éviter de surcharger l'API Binance """ import sqlite3 import json import time import requests from datetime import datetime, timedelta from pathlib import Path class CryptoDataFetcher: """Gestionnaire de cache SQLite pour les données crypto Binance""" def __init__(self, db_path='crypto_cache.db', cache_duration=300): """ Args: db_path: Chemin vers la base SQLite cache_duration: Durée du cache en secondes (défaut: 5 minutes) """ self.db_path = db_path self.cache_duration = cache_duration self.cache = {} self._init_database() def _init_database(self): """Initialise la base de données SQLite""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Table pour les données crypto cursor.execute(''' CREATE TABLE IF NOT EXISTS crypto_data ( symbol TEXT PRIMARY KEY, price REAL, change_24h REAL, volume_24h REAL, high_24h REAL, low_24h REAL, last_update INTEGER ) ''') # Table pour les métadonnées du cache cursor.execute(''' CREATE TABLE IF NOT EXISTS cache_meta ( key TEXT PRIMARY KEY, value TEXT, timestamp INTEGER ) ''') conn.commit() conn.close() print(f"[CACHE] Base de données initialisée: {self.db_path}") def is_cache_valid(self): """Vérifie si le cache est encore valide""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT timestamp FROM cache_meta WHERE key = ?', ('last_update',)) row = cursor.fetchone() conn.close() if not row: return False last_update = row[0] age = time.time() - last_update return age < self.cache_duration except Exception as e: print(f"[CACHE] Erreur vérification cache: {e}") return False def get_cached_data(self): """Récupère les données depuis le cache SQLite""" if not self.is_cache_valid(): print("[CACHE] Cache expiré ou vide") return None try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('SELECT * FROM crypto_data') rows = cursor.fetchall() conn.close() if not rows: return None # Convertir en dictionnaire symbols = {} for row in rows: symbol = row[0] symbols[symbol] = { 'symbol': symbol, 'price': row[1], 'change_24h': row[2], 'volume_24h': row[3], 'high_24h': row[4], 'low_24h': row[5], 'last_update': row[6] } self.cache = { 'symbols': symbols, 'timestamp': time.time() } print(f"[CACHE] ✅ Données chargées: {len(symbols)} cryptos") return self.cache except Exception as e: print(f"[CACHE] ❌ Erreur chargement cache: {e}") return None def update_from_binance(self, symbols_list): """Met à jour le cache depuis l'API Binance""" print(f"[BINANCE] 🔄 Récupération de {len(symbols_list)} cryptos...") try: # Récupérer les données 24h de tous les symboles url = 'https://api.binance.com/api/v3/ticker/24hr' response = requests.get(url, timeout=10) response.raise_for_status() all_tickers = response.json() # Filtrer les symboles demandés symbols_data = {} for ticker in all_tickers: symbol = ticker['symbol'] if symbol in symbols_list: symbols_data[symbol] = { 'symbol': symbol, 'price': float(ticker['lastPrice']), 'change_24h': float(ticker['priceChangePercent']), 'volume_24h': float(ticker['volume']), 'high_24h': float(ticker['highPrice']), 'low_24h': float(ticker['lowPrice']), 'last_update': int(time.time()) } # Sauvegarder dans SQLite self._save_to_database(symbols_data) self.cache = { 'symbols': symbols_data, 'timestamp': time.time() } print(f"[BINANCE] ✅ {len(symbols_data)} cryptos mises à jour") return symbols_data except requests.exceptions.RequestException as e: print(f"[BINANCE] ❌ Erreur API: {e}") return None except Exception as e: print(f"[BINANCE] ❌ Erreur: {e}") return None def _save_to_database(self, symbols_data): """Sauvegarde les données dans SQLite""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() # Insérer ou mettre à jour les données crypto for symbol, data in symbols_data.items(): cursor.execute(''' INSERT OR REPLACE INTO crypto_data (symbol, price, change_24h, volume_24h, high_24h, low_24h, last_update) VALUES (?, ?, ?, ?, ?, ?, ?) ''', ( symbol, data['price'], data['change_24h'], data['volume_24h'], data['high_24h'], data['low_24h'], data['last_update'] )) # Mettre à jour le timestamp du cache cursor.execute(''' INSERT OR REPLACE INTO cache_meta (key, value, timestamp) VALUES ('last_update', ?, ?) ''', (datetime.now().isoformat(), int(time.time()))) conn.commit() conn.close() print(f"[CACHE] 💾 {len(symbols_data)} cryptos sauvegardées") except Exception as e: print(f"[CACHE] ❌ Erreur sauvegarde: {e}") def get_symbol_data(self, symbol): """Récupère les données d'un symbole spécifique""" data = self.get_cached_data() if data and 'symbols' in data: return data['symbols'].get(symbol) return None def clear_cache(self): """Vide complètement le cache""" try: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute('DELETE FROM crypto_data') cursor.execute('DELETE FROM cache_meta') conn.commit() conn.close() self.cache = {} print("[CACHE] 🗑️ Cache vidé") except Exception as e: print(f"[CACHE] ❌ Erreur vidage cache: {e}") # Instance globale du fetcher _fetcher_instance = None def get_fetcher(db_path='crypto_cache.db', cache_duration=300): """Retourne l'instance globale du fetcher (singleton)""" global _fetcher_instance if _fetcher_instance is None: _fetcher_instance = CryptoDataFetcher(db_path, cache_duration) return _fetcher_instance if __name__ == '__main__': # Test du module print("=== Test du Crypto Data Fetcher ===\n") fetcher = get_fetcher() # Symboles de test test_symbols = ['BTCUSDT', 'ETHUSDT', 'BNBUSDT'] print("1. Mise à jour depuis Binance...") data = fetcher.update_from_binance(test_symbols) if data: print("\n2. Données récupérées:") for symbol, info in data.items(): print(f" {symbol}: ${info['price']:,.2f} ({info['change_24h']:+.2f}%)") print("\n3. Vérification du cache...") cached = fetcher.get_cached_data() if cached: print(f" ✅ Cache valide: {len(cached['symbols'])} symboles") print("\n4. Cache SQLite créé avec succès!")