#!/usr/bin/env python3 """ Système dynamique de classification des marchés Hyperliquid Détecte automatiquement la qualité des tokens selon des critères objectifs """ import json import logging import time from datetime import datetime, timedelta from typing import Dict, List, Set, Tuple from hyperliquid_client import HyperliquidClient logger = logging.getLogger(__name__) class DynamicMarketClassifier: """ Classifie automatiquement les tokens selon leur qualité et métriques """ def __init__(self, cache_duration_hours: int = 6): """ Args: cache_duration_hours: Durée de cache pour éviter les re-calculs """ self.cache_duration = cache_duration_hours * 3600 self.cache_file = "market_classification_cache.json" self.client = None # Critères de classification self.quality_thresholds = { 'high': { 'min_volume_24h_usd': 10_000_000, # 10M USD volume minimum 'min_market_cap_rank': 200, # Top 200 cryptos (approximatif) 'min_age_days': 365, # Au moins 1 an d'âge 'min_liquidity_score': 0.8, # Score de liquidité élevé 'known_quality_tokens': True # Dans notre liste connue }, 'medium': { 'min_volume_24h_usd': 1_000_000, # 1M USD volume minimum 'min_market_cap_rank': 1000, # Top 1000 cryptos 'min_age_days': 90, # Au moins 3 mois 'min_liquidity_score': 0.5, # Score de liquidité moyen } # Tout le reste = shitcoin } # Tokens connus de haute qualité (base statique) self.known_high_quality = { 'BTC', 'ETH', 'BNB', 'XRP', 'ADA', 'SOL', 'DOGE', 'DOT', 'MATIC', 'AVAX', 'UNI', 'AAVE', 'COMP', 'LINK', 'SUSHI', 'CRV', 'YFI', 'MKR', 'SNX', 'ATOM', 'FTM', 'ALGO', 'NEAR', 'FLOW', 'ICP', 'EOS', 'TRX', 'XTZ', 'RENDER', 'FIL', 'GRT', 'ENS', 'LDO', 'RPL', 'OP', 'ARB' } # Tokens connus de qualité moyenne (base statique) self.known_medium_quality = { 'PENDLE', 'JUP', 'WLD', 'STX', 'SEI', 'SUI', 'APT', 'TIA', 'INJ', 'TAO', 'THETA', 'VET', 'HBAR', 'LTC', 'BCH', 'ETC' } # Memecoins/Shitcoins forcés (base statique) self.known_shitcoins = { 'PEPE', 'SHIB', 'WIF', 'BONK', 'FLOKI', 'BRETT', 'POPCAT', 'MEME', 'DOGE', 'BOME', 'MEW', 'TURBO', 'PNUT', 'GOAT', 'MOODENG', 'CHILLGUY', 'FARTCOIN', 'kPEPE', 'kSHIB', 'kBONK', 'kFLOKI' } def initialize_client(self): """Initialise le client Hyperliquid si nécessaire""" if not self.client: try: self.client = HyperliquidClient() logger.info("Client Hyperliquid initialisé pour classification") except Exception as e: logger.error(f"Erreur init client: {e}") def load_cache(self) -> Dict: """Charge le cache de classification""" try: with open(self.cache_file, 'r') as f: cache = json.load(f) # Vérifier si le cache est encore valide cache_time = datetime.fromisoformat(cache.get('timestamp', '2000-01-01')) if datetime.now() - cache_time < timedelta(seconds=self.cache_duration): logger.info(f"Cache de classification valide (age: {datetime.now() - cache_time})") return cache except (FileNotFoundError, json.JSONDecodeError, KeyError): pass logger.info("Cache de classification expiré ou invalide") return {} def save_cache(self, classification: Dict): """Sauvegarde le cache de classification""" try: cache_data = { 'timestamp': datetime.now().isoformat(), 'classification': classification, 'update_count': len(classification) } with open(self.cache_file, 'w') as f: json.dump(cache_data, f, indent=2) logger.info(f"Cache sauvegardé: {len(classification)} tokens classifiés") except Exception as e: logger.error(f"Erreur sauvegarde cache: {e}") def get_market_metrics(self, symbol: str) -> Dict: """ Récupère les métriques d'un marché depuis Hyperliquid Returns: Dict avec volume_24h, prix, liquidité, etc. """ if not self.client: self.initialize_client() try: # Récupérer les stats 24h stats = self.client.info.all_mids() # Prix actuels # Volume et liquidité depuis meta meta = self.client.info.meta() universe = meta.get('universe', []) for token_info in universe: if token_info.get('name') == symbol: return { 'symbol': symbol, 'volume_24h_usd': float(token_info.get('prevDayPx', {}).get('volume', 0)), 'price_usd': float(stats.get(symbol, 0)) if symbol in stats else 0, 'max_leverage': int(token_info.get('maxLeverage', 1)), 'only_isolated': token_info.get('onlyIsolated', False), 'listing_time': token_info.get('genesis', 0), } except Exception as e: logger.error(f"Erreur métriques {symbol}: {e}") return {'symbol': symbol, 'volume_24h_usd': 0, 'price_usd': 0, 'max_leverage': 1} def calculate_liquidity_score(self, metrics: Dict) -> float: """ Calcule un score de liquidité basé sur volume, spread, depth Returns: Score entre 0 et 1 """ volume_24h = metrics.get('volume_24h_usd', 0) max_leverage = metrics.get('max_leverage', 1) # Score basé sur le volume (log scale) if volume_24h > 0: volume_score = min(1.0, (volume_24h / 50_000_000) ** 0.3) # Normalisé sur 50M else: volume_score = 0 # Score basé sur le leverage max (indicateur de confiance Hyperliquid) leverage_score = min(1.0, max_leverage / 50) # Normalisé sur 50x # Score composite liquidity_score = (volume_score * 0.7 + leverage_score * 0.3) return liquidity_score def classify_token(self, symbol: str, metrics: Dict = None) -> str: """ Classifie un token selon sa qualité Returns: 'high', 'medium', ou 'shitcoin' """ clean_symbol = symbol.replace('USDT', '').replace('-PERP', '').upper() # 1. Vérifier les listes connues d'abord (priorité aux shitcoins) if clean_symbol in self.known_shitcoins: return 'shitcoin' if clean_symbol in self.known_high_quality: return 'high' if clean_symbol in self.known_medium_quality: return 'medium' # 2. Analyse dynamique si métriques disponibles if not metrics: metrics = self.get_market_metrics(symbol) volume_24h = metrics.get('volume_24h_usd', 0) liquidity_score = self.calculate_liquidity_score(metrics) max_leverage = metrics.get('max_leverage', 1) # 3. Classification selon critères high_criteria = self.quality_thresholds['high'] medium_criteria = self.quality_thresholds['medium'] # Haute qualité ULTRA-STRICTE if (volume_24h >= high_criteria['min_volume_24h_usd'] and liquidity_score >= high_criteria['min_liquidity_score'] and max_leverage >= 15): # AUGMENTÉ: Minimum 15x leverage pour haute qualité return 'high' # Qualité moyenne STRICTE if (volume_24h >= medium_criteria['min_volume_24h_usd'] and liquidity_score >= medium_criteria['min_liquidity_score'] and max_leverage >= 10): # AUGMENTÉ: Minimum 10x leverage pour qualité moyenne return 'medium' # Shitcoin par défaut return 'shitcoin' def scan_all_markets(self) -> Dict[str, str]: """ Scanne tous les marchés Hyperliquid et les classifie Returns: Dict {symbol: quality} """ logger.info("Début du scan dynamique des marchés Hyperliquid...") if not self.client: self.initialize_client() classification = {} try: # Récupérer tous les marchés disponibles meta = self.client.info.meta() universe = meta.get('universe', []) logger.info(f"Analyse de {len(universe)} marchés...") # Analyser chaque marché for i, token_info in enumerate(universe): symbol = token_info.get('name', '') if not symbol: continue # Délai pour éviter rate limiting if i > 0 and i % 10 == 0: time.sleep(1) logger.info(f"Progression: {i}/{len(universe)} marchés analysés") # Récupérer métriques et classifier metrics = { 'symbol': symbol, 'volume_24h_usd': float(token_info.get('prevDayPx', {}).get('volume', 0)), 'max_leverage': int(token_info.get('maxLeverage', 1)), 'only_isolated': token_info.get('onlyIsolated', False), } quality = self.classify_token(symbol, metrics) classification[symbol] = quality logger.debug(f"{symbol}: {quality} (vol: ${metrics['volume_24h_usd']:,.0f}, lev: {metrics['max_leverage']}x)") except Exception as e: logger.error(f"Erreur scan marchés: {e}") # Résumé summary = {} for quality in ['high', 'medium', 'shitcoin']: count = sum(1 for q in classification.values() if q == quality) summary[quality] = count logger.info(f"Classification terminée: {summary}") return classification def get_updated_classification(self) -> Tuple[Set[str], Set[str]]: """ Récupère une classification mise à jour des marchés Returns: Tuple (high_quality_tokens, medium_quality_tokens) """ # Vérifier le cache cache = self.load_cache() if cache and 'classification' in cache: classification = cache['classification'] else: # Scanner tous les marchés classification = self.scan_all_markets() self.save_cache(classification) # Extraire les sets par qualité high_quality = set() medium_quality = set() for symbol, quality in classification.items(): clean_symbol = symbol.replace('USDT', '').replace('-PERP', '').upper() if quality == 'high': high_quality.add(clean_symbol) elif quality == 'medium': medium_quality.add(clean_symbol) logger.info(f"Classification: {len(high_quality)} high, {len(medium_quality)} medium") return high_quality, medium_quality def update_trading_config_with_dynamic_markets(): """Met à jour trading_config.py avec les marchés détectés dynamiquement""" classifier = DynamicMarketClassifier() high_quality, medium_quality = classifier.get_updated_classification() # Créer les nouvelles listes high_list = sorted(list(high_quality)) medium_list = sorted(list(medium_quality)) print(f"📊 MARCHÉS DÉTECTÉS DYNAMIQUEMENT:") print(f"✅ Haute qualité: {len(high_list)} tokens") print(f"⚡ Qualité moyenne: {len(medium_list)} tokens") print(f"\nHaute qualité: {high_list[:10]}..." if len(high_list) > 10 else f"\nHaute qualité: {high_list}") print(f"Qualité moyenne: {medium_list[:10]}..." if len(medium_list) > 10 else f"\nQualité moyenne: {medium_list}") return high_quality, medium_quality if __name__ == "__main__": logging.basicConfig(level=logging.INFO) update_trading_config_with_dynamic_markets()