""" Crypto Data Fetcher - Récupère et met en cache les données de 64 cryptos depuis Binance Ce module permet d'accélérer les requêtes en stockant les données localement. Fonctionnalités: - Télécharge les prix, volumes et données OHLCV depuis Binance - Calcule les indicateurs techniques (RSI, EMA, BB, MACD) - Stocke les données en cache JSON pour un accès rapide - Met à jour automatiquement les données selon un intervalle configurable Auteur: Trading Bot System Date: Décembre 2025 """ import os import json import time import asyncio import aiohttp from datetime import datetime, timedelta from typing import Dict, List, Optional, Any import threading # Import des paramètres de config pour cohérence avec le bot try: from config import ( MIN_AI_SCORE_FOR_BUY, RSI_OVERSOLD, RSI_OVERBOUGHT, MIN_BUY_SIGNALS, BLOCK_BUY_ON_BEARISH ) except ImportError: # Valeurs par défaut si config non disponible MIN_AI_SCORE_FOR_BUY = 70 RSI_OVERSOLD = 30 RSI_OVERBOUGHT = 70 MIN_BUY_SIGNALS = 3 BLOCK_BUY_ON_BEARISH = True # Configuration CACHE_DIR = os.path.join(os.path.dirname(__file__), 'crypto_cache') CACHE_FILE = os.path.join(CACHE_DIR, 'crypto_data.json') METADATA_FILE = os.path.join(CACHE_DIR, 'cache_metadata.json') WATCHLIST_FILE = os.path.join(os.path.dirname(__file__), 'watchlist.json') # API Binance BINANCE_API_URL = "https://api.binance.com/api/v3" BINANCE_TESTNET_URL = "https://testnet.binance.vision/api/v3" # Paramètres de cache CACHE_EXPIRY_MINUTES = 3 # Durée de validité du cache (< 120s auto-update) UPDATE_INTERVAL_SECONDS = 60 # Intervalle de mise à jour automatique class CryptoDataFetcher: """Gestionnaire de cache pour les données crypto Binance""" def __init__(self, use_testnet: bool = True): self.use_testnet = use_testnet self.base_url = BINANCE_TESTNET_URL if use_testnet else BINANCE_API_URL self.cache: Dict[str, Any] = {} self.last_update: Optional[datetime] = None self.is_updating = False self._update_thread: Optional[threading.Thread] = None self._stop_event = threading.Event() # Créer le dossier cache s'il n'existe pas os.makedirs(CACHE_DIR, exist_ok=True) # Charger le cache existant self._load_cache() def _load_cache(self) -> None: """Charge le cache depuis le fichier JSON""" try: if os.path.exists(CACHE_FILE): with open(CACHE_FILE, 'r', encoding='utf-8') as f: self.cache = json.load(f) print(f"[CACHE] OK - Cache charge: {len(self.cache.get('symbols', {}))} cryptos") if os.path.exists(METADATA_FILE): with open(METADATA_FILE, 'r', encoding='utf-8') as f: metadata = json.load(f) self.last_update = datetime.fromisoformat(metadata.get('last_update', '')) print(f"[CACHE] Derniere mise a jour: {self.last_update}") except Exception as e: print(f"[CACHE] ERREUR chargement cache: {e}") self.cache = {} self.last_update = None def _save_cache(self) -> None: """Sauvegarde le cache dans le fichier JSON""" try: with open(CACHE_FILE, 'w', encoding='utf-8') as f: json.dump(self.cache, f, indent=2, ensure_ascii=False) metadata = { 'last_update': self.last_update.isoformat() if self.last_update else None, 'symbols_count': len(self.cache.get('symbols', {})), 'cache_expiry_minutes': CACHE_EXPIRY_MINUTES } with open(METADATA_FILE, 'w', encoding='utf-8') as f: json.dump(metadata, f, indent=2) print(f"[CACHE] SAVED - Cache sauvegarde: {len(self.cache.get('symbols', {}))} cryptos") except Exception as e: print(f"[CACHE] ERREUR sauvegarde cache: {e}") def _get_watchlist(self) -> List[str]: """Récupère la liste COMPLÈTE des cryptos tradables (bot + spy testnet + spy prod)""" symbols_set = set() # 1) Bot watchlist: symbols + auto_added + spy_injected try: if os.path.exists(WATCHLIST_FILE): with open(WATCHLIST_FILE, 'r', encoding='utf-8') as f: data = json.load(f) symbols_set.update(data.get('symbols', [])) symbols_set.update(data.get('auto_added', {}).keys()) symbols_set.update(data.get('spy_injected', {}).keys()) except Exception as e: print(f"[CACHE] WARN - Erreur lecture watchlist: {e}") # 2) spy_coin_scores.json (testnet + prod) — toutes les valeurs tradées par le spy base_dir = os.path.dirname(__file__) spy_scores_paths = [ os.path.join(base_dir, 'spy_coin_scores.json'), os.path.normpath(os.path.join(base_dir, '..', 'crypto_trading_prod', 'data', 'spy_coin_scores.json')), ] for path in spy_scores_paths: try: if os.path.exists(path): with open(path, 'r', encoding='utf-8') as f: symbols_set.update(json.load(f).keys()) except Exception as e: print(f"[CACHE] WARN - Erreur lecture spy scores {path}: {e}") # 3) Prod watchlist symbols (spy prod) prod_wl = os.path.normpath(os.path.join(base_dir, '..', 'crypto_trading_prod', 'watchlist.json')) try: if os.path.exists(prod_wl): with open(prod_wl, 'r', encoding='utf-8') as f: symbols_set.update(json.load(f).get('symbols', [])) except Exception as e: print(f"[CACHE] WARN - Erreur lecture prod watchlist: {e}") # Filtrer: ASCII seulement, USDT/USDC, longueur min 5 valid = [s for s in symbols_set if s.isascii() and s.isalnum() and len(s) >= 5 and (s.endswith('USDT') or s.endswith('USDC'))] valid.sort() if valid: print(f"[CACHE] Watchlist complète: {len(valid)} cryptos (bot + spy)") return valid # Liste par défaut si pas de watchlist return [ "BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT", "XRPUSDT", "ADAUSDT", "DOGEUSDT", "AVAXUSDT", "DOTUSDT", "LINKUSDT" ] def is_cache_valid(self) -> bool: """Vérifie si le cache est encore valide""" if not self.last_update or not self.cache: return False expiry_time = self.last_update + timedelta(minutes=CACHE_EXPIRY_MINUTES) return datetime.now() < expiry_time async def _fetch_ticker_24h(self, session: aiohttp.ClientSession, symbol: str) -> Optional[Dict]: """Récupère les données 24h pour un symbole""" try: url = f"{self.base_url}/ticker/24hr?symbol={symbol}" async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: if response.status == 200: return await response.json() except Exception as e: print(f"[CACHE] WARN - Erreur ticker {symbol}: {e}") return None async def _fetch_klines(self, session: aiohttp.ClientSession, symbol: str, interval: str = "1h", limit: int = 100) -> Optional[List]: """Récupère les données OHLCV (klines) pour un symbole""" try: url = f"{self.base_url}/klines?symbol={symbol}&interval={interval}&limit={limit}" async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: if response.status == 200: return await response.json() except Exception as e: print(f"[CACHE] WARN - Erreur klines {symbol}: {e}") return None def _calculate_rsi(self, closes: List[float], period: int = 14) -> float: """Calcule le RSI""" if len(closes) < period + 1: return 50.0 gains = [] losses = [] for i in range(1, len(closes)): change = closes[i] - closes[i-1] if change > 0: gains.append(change) losses.append(0) else: gains.append(0) losses.append(abs(change)) if len(gains) < period: return 50.0 avg_gain = sum(gains[-period:]) / period avg_loss = sum(losses[-period:]) / period if avg_loss == 0: return 100.0 rs = avg_gain / avg_loss rsi = 100 - (100 / (1 + rs)) return round(rsi, 2) def _calculate_ema(self, values: List[float], period: int) -> float: """Calcule l'EMA""" if len(values) < period: return values[-1] if values else 0 multiplier = 2 / (period + 1) ema = values[0] for value in values[1:]: ema = (value - ema) * multiplier + ema return round(ema, 8) def _calculate_bollinger_bands(self, closes: List[float], period: int = 20, std_dev: float = 2.0) -> Dict: """Calcule les bandes de Bollinger""" if len(closes) < period: return {'upper': 0, 'middle': 0, 'lower': 0, 'bandwidth': 0} recent = closes[-period:] middle = sum(recent) / period variance = sum((x - middle) ** 2 for x in recent) / period std = variance ** 0.5 upper = middle + (std * std_dev) lower = middle - (std * std_dev) bandwidth = ((upper - lower) / middle) * 100 if middle > 0 else 0 return { 'upper': round(upper, 8), 'middle': round(middle, 8), 'lower': round(lower, 8), 'bandwidth': round(bandwidth, 2) } def _calculate_macd(self, closes: List[float], fast: int = 12, slow: int = 26, signal: int = 9) -> Dict: """Calcule le MACD""" if len(closes) < slow: return {'macd': 0, 'signal': 0, 'histogram': 0} ema_fast = self._calculate_ema(closes, fast) ema_slow = self._calculate_ema(closes, slow) macd_line = ema_fast - ema_slow # Pour le signal, on devrait calculer l'EMA du MACD sur plusieurs périodes # Simplification ici signal_line = macd_line * 0.9 # Approximation histogram = macd_line - signal_line return { 'macd': round(macd_line, 8), 'signal': round(signal_line, 8), 'histogram': round(histogram, 8) } async def _process_symbol(self, session: aiohttp.ClientSession, symbol: str) -> Optional[Dict]: """Traite un symbole: récupère les données et calcule les indicateurs""" try: # Récupérer les données en parallèle ticker_task = self._fetch_ticker_24h(session, symbol) klines_task = self._fetch_klines(session, symbol, "1h", 100) ticker, klines = await asyncio.gather(ticker_task, klines_task) if not ticker: return None # Extraire les prix de clôture des klines closes = [] if klines: closes = [float(k[4]) for k in klines] # Index 4 = close price # Calculer les indicateurs current_price = float(ticker.get('lastPrice', 0)) data = { 'symbol': symbol, 'price': current_price, 'priceChange': float(ticker.get('priceChange', 0)), 'priceChangePercent': float(ticker.get('priceChangePercent', 0)), 'volume': float(ticker.get('volume', 0)), 'quoteVolume': float(ticker.get('quoteVolume', 0)), 'high24h': float(ticker.get('highPrice', 0)), 'low24h': float(ticker.get('lowPrice', 0)), 'openPrice': float(ticker.get('openPrice', 0)), 'weightedAvgPrice': float(ticker.get('weightedAvgPrice', 0)), # Indicateurs techniques 'indicators': { 'rsi': self._calculate_rsi(closes, 14) if closes else 50, 'rsi_7': self._calculate_rsi(closes, 7) if closes else 50, 'ema_short': self._calculate_ema(closes, 9) if closes else current_price, 'ema_long': self._calculate_ema(closes, 21) if closes else current_price, 'ema_50': self._calculate_ema(closes, 50) if len(closes) >= 50 else current_price, 'bollinger': self._calculate_bollinger_bands(closes, 20, 2.0) if closes else {}, 'macd': self._calculate_macd(closes) if closes else {} }, # Signaux de trading 'signals': { 'trend': 'bullish' if closes and closes[-1] > self._calculate_ema(closes, 21) else 'bearish', 'rsi_signal': 'oversold' if self._calculate_rsi(closes, 14) < 30 else ('overbought' if self._calculate_rsi(closes, 14) > 70 else 'neutral'), 'volatility': 'high' if self._calculate_bollinger_bands(closes).get('bandwidth', 0) > 5 else 'normal' }, 'updated_at': datetime.now().isoformat() } return data except Exception as e: print(f"[CACHE] ERREUR traitement {symbol}: {e}") return None async def fetch_all_data(self, force: bool = False) -> Dict[str, Any]: """Récupère toutes les données pour tous les symboles de la watchlist""" # Vérifier si le cache est valide if not force and self.is_cache_valid(): print("[CACHE] OK - Cache valide, utilisation des donnees en cache") return self.cache if self.is_updating: print("[CACHE] WAIT - Mise a jour en cours...") return self.cache self.is_updating = True start_time = time.time() try: symbols = self._get_watchlist() print(f"[CACHE] UPDATE - Mise a jour de {len(symbols)} cryptos depuis Binance...") async with aiohttp.ClientSession() as session: # Traiter les symboles par lots de 20 pour performance (rate limit Binance: 1200/min) batch_size = 20 results = {} for i in range(0, len(symbols), batch_size): batch = symbols[i:i+batch_size] tasks = [self._process_symbol(session, symbol) for symbol in batch] batch_results = await asyncio.gather(*tasks) for result in batch_results: if result: results[result['symbol']] = result # Pause entre les lots pour éviter le rate limiting if i + batch_size < len(symbols): await asyncio.sleep(0.5) # Construire le cache final self.cache = { 'symbols': results, 'count': len(results), 'updated_at': datetime.now().isoformat(), 'source': 'binance_testnet' if self.use_testnet else 'binance', 'cache_expiry_minutes': CACHE_EXPIRY_MINUTES } self.last_update = datetime.now() # Sauvegarder le cache self._save_cache() elapsed = time.time() - start_time print(f"[CACHE] OK - Mise a jour terminee en {elapsed:.2f}s - {len(results)} cryptos") return self.cache except Exception as e: print(f"[CACHE] ERREUR mise a jour: {e}") return self.cache finally: self.is_updating = False def get_cached_data(self) -> Dict[str, Any]: """Retourne les données en cache (sans mise à jour)""" return self.cache def get_symbol_data(self, symbol: str) -> Optional[Dict]: """Retourne les données d'un symbole spécifique""" symbols = self.cache.get('symbols', {}) return symbols.get(symbol) def get_summary(self) -> Dict[str, Any]: """Retourne un résumé du cache""" symbols = self.cache.get('symbols', {}) if not symbols: return { 'count': 0, 'valid': False, 'last_update': None } # Calculer les statistiques bullish_count = sum(1 for s in symbols.values() if s.get('signals', {}).get('trend') == 'bullish') bearish_count = len(symbols) - bullish_count oversold = [s['symbol'] for s in symbols.values() if s.get('signals', {}).get('rsi_signal') == 'oversold'] overbought = [s['symbol'] for s in symbols.values() if s.get('signals', {}).get('rsi_signal') == 'overbought'] # Top gainers et losers sorted_by_change = sorted(symbols.values(), key=lambda x: x.get('priceChangePercent', 0), reverse=True) top_gainers = [{'symbol': s['symbol'], 'change': s.get('priceChangePercent', 0)} for s in sorted_by_change[:5]] top_losers = [{'symbol': s['symbol'], 'change': s.get('priceChangePercent', 0)} for s in sorted_by_change[-5:]] return { 'count': len(symbols), 'valid': self.is_cache_valid(), 'last_update': self.cache.get('updated_at'), 'expiry_minutes': CACHE_EXPIRY_MINUTES, 'market_sentiment': { 'bullish': bullish_count, 'bearish': bearish_count, 'ratio': round(bullish_count / len(symbols) * 100, 1) if symbols else 0 }, 'signals': { 'oversold': oversold[:10], 'overbought': overbought[:10] }, 'top_gainers': top_gainers, 'top_losers': top_losers } def start_auto_update(self, interval_seconds: int = UPDATE_INTERVAL_SECONDS) -> None: """Démarre la mise à jour automatique en arrière-plan""" if self._update_thread and self._update_thread.is_alive(): print("[CACHE] WARN - Mise a jour automatique deja active") return self._stop_event.clear() def update_loop(): while not self._stop_event.is_set(): # Attendre l'intervalle EN PREMIER pour ne pas refaire tout de suite le fetch du démarrage self._stop_event.wait(interval_seconds) if self._stop_event.is_set(): break try: # force=True: relit _get_watchlist() pour capter les nouveaux coins ajoutés par le spy asyncio.run(self.fetch_all_data(force=True)) except Exception as e: print(f"[CACHE] ERREUR auto-update: {e}") self._update_thread = threading.Thread(target=update_loop, daemon=True) self._update_thread.start() print(f"[CACHE] AUTO-UPDATE active (intervalle: {interval_seconds}s)") def stop_auto_update(self) -> None: """Arrête la mise à jour automatique""" self._stop_event.set() if self._update_thread: self._update_thread.join(timeout=5) print("[CACHE] AUTO-UPDATE arretee") # Instance globale _fetcher: Optional[CryptoDataFetcher] = None def get_fetcher(use_testnet: bool = True) -> CryptoDataFetcher: """Retourne l'instance globale du fetcher""" global _fetcher if _fetcher is None: _fetcher = CryptoDataFetcher(use_testnet=use_testnet) return _fetcher # ═══════════════════════════════════════════════════════════════════════════════ # SYSTÈME DE SCORING INTELLIGENT # ═══════════════════════════════════════════════════════════════════════════════ class TradingOpportunityScorer: """ Système de scoring pour identifier les meilleures opportunités de trading. Utilise les paramètres de config.py pour être cohérent avec le bot. Critères de scoring (sur 100 points): - RSI Signal (25 pts): Survente = achat, Surachat = vente - Tendance EMA (20 pts): Prix au-dessus/dessous des EMA - Momentum (20 pts): Force du mouvement récent - Volatilité (15 pts): Bandes de Bollinger favorables - Volume (10 pts): Volume supérieur à la moyenne - Stabilité (10 pts): Pas de mouvements extrêmes Seuils utilisés depuis config.py: - MIN_AI_SCORE_FOR_BUY: {min_score} (score minimum pour achat) - RSI_OVERSOLD: {rsi_oversold} (seuil survente) - RSI_OVERBOUGHT: {rsi_overbought} (seuil surachat) """ # Poids des critères (total = 100) WEIGHTS = { 'rsi': 25, 'trend': 20, 'momentum': 20, 'volatility': 15, 'volume': 10, 'stability': 10 } @staticmethod def calculate_score(crypto_data: Dict) -> Dict[str, Any]: """ Calcule le score d'opportunité pour une crypto. Retourne un dict avec le score total et les détails. Utilise les seuils de config.py. """ if not crypto_data: return {'score': 0, 'signal': 'none', 'details': {}} indicators = crypto_data.get('indicators', {}) signals = crypto_data.get('signals', {}) details = {} total_score = 0 # 1. Score RSI (25 pts) - Utilise RSI_OVERSOLD et RSI_OVERBOUGHT de config rsi = indicators.get('rsi', 50) rsi_oversold = RSI_OVERSOLD # Depuis config.py (30) rsi_overbought = RSI_OVERBOUGHT # Depuis config.py (70) if rsi < rsi_oversold - 5: rsi_score = 25 # Très survendu = excellente opportunité elif rsi < rsi_oversold: rsi_score = 22 elif rsi < rsi_oversold + 5: rsi_score = 18 elif rsi < rsi_oversold + 10: rsi_score = 14 elif rsi > rsi_overbought + 5: rsi_score = 5 # Très surachat = risqué pour achat elif rsi > rsi_overbought: rsi_score = 8 elif rsi > rsi_overbought - 5: rsi_score = 10 else: rsi_score = 12 # Zone neutre details['rsi'] = {'value': rsi, 'score': rsi_score, 'max': 25} total_score += rsi_score # 2. Score Tendance EMA (20 pts) price = crypto_data.get('price', 0) ema_short = indicators.get('ema_short', price) ema_long = indicators.get('ema_long', price) ema_9 = indicators.get('ema_9', 0) ema_21 = indicators.get('ema_21', 0) ema_50 = indicators.get('ema_50', 0) # Déterminer l'alignement EMA (nouveau) if ema_9 > 0 and ema_21 > 0 and ema_50 > 0: # Alignement parfait haussier: EMA9 > EMA21 > EMA50 if ema_9 > ema_21 > ema_50: ema_alignment = 'bullish' # Alignement parfait baissier: EMA50 > EMA21 > EMA9 elif ema_50 > ema_21 > ema_9: ema_alignment = 'bearish' else: ema_alignment = 'mixed' elif ema_short > 0 and ema_long > 0: # Fallback sur EMA court/long ema_alignment = 'bullish' if ema_short > ema_long else 'bearish' else: ema_alignment = 'unknown' if price > 0 and ema_short > 0 and ema_long > 0: # Cross haussier: EMA court > EMA long et prix au-dessus if ema_short > ema_long and price > ema_short: trend_score = 20 # Tendance haussière confirmée elif ema_short > ema_long: trend_score = 16 # Début de tendance haussière elif price > ema_long: trend_score = 12 # Possible retournement elif ema_short < ema_long and price < ema_short: trend_score = 5 # Tendance baissière else: trend_score = 10 # Neutre else: trend_score = 10 trend_direction = 'bullish' if ema_short > ema_long else 'bearish' details['trend'] = {'direction': trend_direction, 'score': trend_score, 'max': 20, 'ema_alignment': ema_alignment} total_score += trend_score # 3. Score Momentum (20 pts) - Basé sur le changement 24h change_24h = crypto_data.get('priceChangePercent', 0) # Momentum positif modéré = bon (pas trop extrême) if 2 <= change_24h <= 8: momentum_score = 20 # Hausse saine elif 0 <= change_24h < 2: momentum_score = 15 # Légère hausse elif 8 < change_24h <= 15: momentum_score = 12 # Hausse forte (risque de correction) elif -3 <= change_24h < 0: momentum_score = 14 # Légère baisse (opportunité) elif -8 <= change_24h < -3: momentum_score = 18 # Baisse = opportunité d'achat elif change_24h < -8: momentum_score = 10 # Chute importante (risqué) else: momentum_score = 8 # Hausse extrême (très risqué) details['momentum'] = {'change_24h': change_24h, 'score': momentum_score, 'max': 20} total_score += momentum_score # 4. Score Volatilité (15 pts) - Bandes de Bollinger bb = indicators.get('bollinger', {}) bandwidth = bb.get('bandwidth', 0) bb_lower = bb.get('lower', 0) bb_upper = bb.get('upper', 0) bb_middle = bb.get('middle', 0) # Déterminer la direction BB (nouveau) bb_prev_middle = bb.get('prev_middle', 0) if bb_middle > 0 and bb_prev_middle > 0: bb_slope = ((bb_middle - bb_prev_middle) / bb_prev_middle) * 100 if bb_slope > 0.1: bb_direction = 'up' elif bb_slope < -0.1: bb_direction = 'down' else: bb_direction = 'flat' elif change_24h > 1: bb_direction = 'up' # Estimation basée sur le momentum elif change_24h < -1: bb_direction = 'down' else: bb_direction = 'flat' # Volatilité modérée = meilleur pour trading if 2 <= bandwidth <= 5: vol_score = 15 # Volatilité idéale elif 1 <= bandwidth < 2: vol_score = 12 # Faible volatilité (squeeze possible) elif 5 < bandwidth <= 8: vol_score = 10 # Volatilité élevée elif bandwidth > 8: vol_score = 5 # Trop volatile else: vol_score = 8 # Calculer la position BB pour bonus et détection de creux bb_position = 0.5 # Position par défaut (milieu) if price > 0 and bb_lower > 0 and bb_upper > 0: bb_position = (price - bb_lower) / (bb_upper - bb_lower) if bb_upper > bb_lower else 0.5 if bb_position < 0.2: vol_score = min(15, vol_score + 5) # Près de la bande basse elif bb_position > 0.8: vol_score = max(0, vol_score - 3) # Près de la bande haute details['volatility'] = {'bandwidth': bandwidth, 'score': vol_score, 'max': 15, 'bb_direction': bb_direction, 'bb_position': bb_position} total_score += vol_score # 5. Score Volume (10 pts) volume = crypto_data.get('quoteVolume', 0) # Volume élevé = bon signe de liquidité if volume > 100_000_000: volume_score = 10 # Très liquide elif volume > 50_000_000: volume_score = 8 elif volume > 10_000_000: volume_score = 6 elif volume > 1_000_000: volume_score = 4 else: volume_score = 2 # Faible liquidité details['volume'] = {'quote_volume': volume, 'score': volume_score, 'max': 10} total_score += volume_score # 6. Score Stabilité (10 pts) - Écart high/low raisonnable high = crypto_data.get('high24h', 0) low = crypto_data.get('low24h', 0) if high > 0 and low > 0: daily_range = ((high - low) / low) * 100 if daily_range < 3: stability_score = 10 # Très stable elif daily_range < 5: stability_score = 8 elif daily_range < 10: stability_score = 6 elif daily_range < 15: stability_score = 4 else: stability_score = 2 # Très volatile else: stability_score = 5 details['stability'] = {'daily_range': daily_range if 'daily_range' in dir() else 0, 'score': stability_score, 'max': 10} total_score += stability_score # ═══════════════════════════════════════════════════════════════════ # CRITÈRES STRICTS POUR ACHAT (alignés avec ai_predictor.py) # ═══════════════════════════════════════════════════════════════════ # ════════════════════════════════════════════════════════════════════ # RÈGLE STRICTE ABSOLUE: JAMAIS ACHETER SI EMA9 >= EMA21 (ema_short >= ema_long) # ════════════════════════════════════════════════════════════════════ ema_blocks_buy = False # Utiliser ema_9/ema_21 ou fallback sur ema_short/ema_long ema_9_val = ema_9 if ema_9 > 0 else ema_short ema_21_val = ema_21 if ema_21 > 0 else ema_long if ema_9_val > 0 and ema_21_val > 0: ema_diff_pct = ((ema_9_val - ema_21_val) / ema_21_val) * 100 if ema_diff_pct >= 0: # EMA9 >= EMA21 = tendance haussière/correction à venir ema_blocks_buy = True # Critères secondaires (indicatifs mais ne bloquent pas) is_bearish_secondary = ( ema_alignment == 'bearish' or # EMA50 > EMA21 > EMA9 trend_direction == 'bearish' or # Tendance générale baissière bb_direction == 'down' or # BB en baisse rsi > RSI_OVERBOUGHT or # RSI trop haut change_24h < -1 # Prix en baisse sur 24h ) # is_bearish pour l'affichage is_bearish = ema_blocks_buy or is_bearish_secondary # Vérifier momentum (prix doit monter) - pour info seulement is_falling = change_24h < 0 # Prix en baisse # Utiliser MIN_AI_SCORE_FOR_BUY de config.py pour les seuils min_score_buy = MIN_AI_SCORE_FOR_BUY # 70 par défaut # ═══════════════════════════════════════════════════════════════════ # RÈGLE STRICTE UNIQUE: BLOQUER SEULEMENT SI EMA9 >= EMA21 # Les autres conditions (RSI, BB, etc.) réduisent le score mais ne bloquent pas # ═══════════════════════════════════════════════════════════════════ if ema_blocks_buy: signal = 'blocked' recommendation = '🚫 NO BUY (EMA9 >= EMA21)' elif total_score >= min_score_buy + 5: # 75+ signal = 'strong_buy' recommendation = '🚀 Achat Fort' elif total_score >= min_score_buy: # 70+ signal = 'buy' recommendation = '✅ Achat' elif total_score >= 45: signal = 'neutral' recommendation = '⏸️ HOLD' elif total_score >= 30: signal = 'sell' recommendation = '📉 Vente' else: signal = 'strong_sell' recommendation = '🔻 Vente Forte' return { 'score': total_score, 'signal': signal, 'recommendation': recommendation, 'details': details, 'rsi': rsi, 'trend': trend_direction, 'ema_alignment': ema_alignment, 'bb_direction': bb_direction, 'is_bearish': is_bearish, 'change_24h': change_24h } @staticmethod def get_top_opportunities(cache_data: Dict, limit: int = 10, min_score: int = None) -> List[Dict]: """ Retourne les meilleures opportunités de trading triées par score. EXCLUT les cryptos où EMA9 >= EMA21 (signal='blocked'). Utilise MIN_AI_SCORE_FOR_BUY - 20 comme seuil par défaut (50 si MIN=70). """ # Utiliser le seuil de config si non spécifié if min_score is None: min_score = max(50, MIN_AI_SCORE_FOR_BUY - 20) symbols = cache_data.get('symbols', {}) opportunities = [] for symbol, data in symbols.items(): scoring = TradingOpportunityScorer.calculate_score(data) # FILTRER les signaux bloqués (EMA9 >= EMA21) if scoring.get('signal') == 'blocked': continue if scoring['score'] >= min_score: opportunities.append({ 'symbol': symbol, 'price': data.get('price', 0), 'change_24h': data.get('priceChangePercent', 0), **scoring }) # Trier par score décroissant opportunities.sort(key=lambda x: x['score'], reverse=True) return opportunities[:limit] @staticmethod def get_buy_signals(cache_data: Dict, rsi_threshold: float = None, min_score: int = None) -> List[Dict]: """ Retourne les cryptos avec signal d'achat (RSI bas + bon score). EXCLUT les cryptos où EMA9 >= EMA21 (signal='blocked'). Accepte les creux BB avec un score minimum réduit. """ # Utiliser les seuils de config si non spécifiés if rsi_threshold is None: rsi_threshold = 45 # Plus permissif pour détecter les opportunités if min_score is None: min_score = max(45, MIN_AI_SCORE_FOR_BUY - 25) # 45 si MIN=70 (plus permissif) symbols = cache_data.get('symbols', {}) buy_signals = [] for symbol, data in symbols.items(): indicators = data.get('indicators', {}) rsi = indicators.get('rsi', 50) if rsi < rsi_threshold: scoring = TradingOpportunityScorer.calculate_score(data) # FILTRER les signaux bloqués (EMA9 >= EMA21) if scoring.get('signal') == 'blocked': continue if scoring['score'] >= min_score: buy_signals.append({ 'symbol': symbol, 'price': data.get('price', 0), 'rsi': rsi, 'change_24h': data.get('priceChangePercent', 0), **scoring }) # Trier par RSI croissant (plus survendu = meilleur) buy_signals.sort(key=lambda x: x['rsi']) return buy_signals @staticmethod def get_momentum_plays(cache_data: Dict, min_change: float = 3, max_change: float = 12) -> List[Dict]: """ Retourne les cryptos avec bon momentum (hausse modérée, pas extrême). """ symbols = cache_data.get('symbols', {}) momentum_plays = [] for symbol, data in symbols.items(): change = data.get('priceChangePercent', 0) if min_change <= change <= max_change: scoring = TradingOpportunityScorer.calculate_score(data) momentum_plays.append({ 'symbol': symbol, 'price': data.get('price', 0), 'change_24h': change, **scoring }) # Trier par score décroissant momentum_plays.sort(key=lambda x: x['score'], reverse=True) return momentum_plays[:10] # Ajouter les méthodes au fetcher def get_opportunities(use_testnet: bool = True) -> Dict[str, Any]: """Retourne les opportunités de trading depuis le cache""" fetcher = get_fetcher(use_testnet) cache = fetcher.get_cached_data() if not cache or not cache.get('symbols'): return {'error': 'Cache vide', 'opportunities': []} return { 'top_opportunities': TradingOpportunityScorer.get_top_opportunities(cache, limit=10, min_score=50), 'buy_signals': TradingOpportunityScorer.get_buy_signals(cache, rsi_threshold=35, min_score=50), 'momentum_plays': TradingOpportunityScorer.get_momentum_plays(cache, min_change=2, max_change=10), 'updated_at': cache.get('updated_at'), 'total_cryptos': len(cache.get('symbols', {})) } # CLI pour tests if __name__ == "__main__": import sys print("=" * 60) print("Crypto Data Fetcher - Test") print("=" * 60) fetcher = get_fetcher(use_testnet=True) # Forcer une mise à jour print("\nRecuperation des donnees depuis Binance...") data = asyncio.run(fetcher.fetch_all_data(force=True)) print(f"\nResume du cache:") summary = fetcher.get_summary() print(f" - Cryptos en cache: {summary['count']}") print(f" - Cache valide: {summary['valid']}") print(f" - Derniere MAJ: {summary['last_update']}") if summary.get('market_sentiment'): sentiment = summary['market_sentiment'] print(f"\nSentiment du marche:") print(f" - Bullish: {sentiment['bullish']} ({sentiment['ratio']}%)") print(f" - Bearish: {sentiment['bearish']}") # Test du scoring print("\n" + "=" * 60) print("OPPORTUNITES DE TRADING (Nouveau Systeme de Scoring)") print("=" * 60) opps = get_opportunities(use_testnet=True) if opps.get('top_opportunities'): print(f"\nTop 5 Opportunites (Score > 50):") for i, opp in enumerate(opps['top_opportunities'][:5], 1): print(f" {i}. {opp['symbol'].replace('USDT', '')}: Score {opp['score']}/100 | {opp.get('signal', 'N/A')}") print(f" RSI: {opp['rsi']:.1f} | Trend: {opp['trend']} | 24h: {opp['change_24h']:+.2f}%") if opps.get('buy_signals'): print(f"\nSignaux d'Achat (RSI < 35):") for sig in opps['buy_signals'][:5]: print(f" - {sig['symbol'].replace('USDT', '')}: RSI={sig['rsi']:.1f} | Score={sig['score']}/100") if opps.get('momentum_plays'): print(f"\nMomentum Plays (+2% a +10%):") for mp in opps['momentum_plays'][:5]: print(f" - {mp['symbol'].replace('USDT', '')}: {mp['change_24h']:+.2f}% | Score={mp['score']}/100") print("\nTest termine!")