import yaml import os import requests import time from datetime import datetime import mysql.connector from fastapi import FastAPI, HTTPException import uvicorn import telegram import threading import asyncio import logging import queue # Configuration du logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # Charger le fichier config.yaml CONFIG_PATH = os.path.join(os.path.dirname(__file__), "config.yaml") def load_config(): with open(CONFIG_PATH, "r") as file: return yaml.safe_load(file) config = load_config() # Récupérer les variables depuis config.yaml MYSQL_HOST = config["mysql"]["host"] MYSQL_USER = config["mysql"]["user"] MYSQL_PASSWORD = config["mysql"]["password"] MYSQL_DATABASE = config["mysql"]["database"] TELEGRAM_BOT_TOKEN = config["telegram"]["bot_token"] TELEGRAM_CHAT_ID = config["telegram"]["chat_id"] # Récupérer les filtres (liquidité et volume) FILTER_MIN_LIQUIDITY = config.get("filters", {}).get("min_liquidity", 0) FILTER_MIN_VOLUME = config.get("filters", {}).get("min_volume", 0) logging.info("✅ Configuration chargée avec succès:") logging.info(f"MYSQL_HOST: {MYSQL_HOST}") logging.info(f"MYSQL_USER: {MYSQL_USER}") logging.info(f"MYSQL_DATABASE: {MYSQL_DATABASE}") logging.info(f"TELEGRAM_BOT_TOKEN: {TELEGRAM_BOT_TOKEN}") logging.info(f"TELEGRAM_CHAT_ID: {TELEGRAM_CHAT_ID}") logging.info(f"Filtres : Liquidité min = {FILTER_MIN_LIQUIDITY}, Volume min = {FILTER_MIN_VOLUME}") if not all([MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE, TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID]): raise ValueError("❌ Erreur: Une ou plusieurs variables de configuration sont manquantes.") # --- Vérification via RugCheck --- # URL d'appel pour RugCheck (à adapter si nécessaire) RUG_CHECK_URL = config.get("pocket_universe", {}).get("api_url", "https://api.rugcheck.xyz/v1/tokens/{}/report") # Critères de vérification min_lp_locked_amount = 0 min_lp_locked_pct = 5 max_holder_pct = 90 def check_top_holders(holders): """Vérifie que pour chaque détenteur dans 'topHolders', le pourcentage détenu ne dépasse pas max_holder_pct.""" if not holders: return False for holder in holders: if holder.get('pct', 0) > max_holder_pct: logging.warning(f"Holder {holder.get('address')} possède {holder.get('pct', 0)}% (> {max_holder_pct}%)") return False return True def check_lp_burned(markets): """ Vérifie si la LP est considérée comme verrouillée. Pour un marché de type "raydium", on vérifie : - lpLocked > 0, - lpLockedUSD > min_lp_locked_amount, - lpLockedPct > min_lp_locked_pct. """ if not markets: return False raydium_market = next((market for market in markets if market.get("marketType") == "raydium"), None) if not raydium_market: return False lp = raydium_market.get('lp', {}) is_lp_locked = lp.get('lpLocked', 0) lp_amount = lp.get('lpLockedUSD', 0) lp_locked_pct = lp.get('lpLockedPct', 0) if is_lp_locked > 0 and lp_amount > min_lp_locked_amount and lp_locked_pct > min_lp_locked_pct: return True return False def rugcheck_rate_limit(func): def wrapper(*args, **kwargs): result = func(*args, **kwargs) time.sleep(2) # Délai de 2 secondes entre deux appels return result return wrapper @rugcheck_rate_limit def verify_with_rugcheck(token_address): """ Vérifie le token via l'API RugCheck. Le token est accepté si les listes "topHolders" et "markets" sont non vides et que check_top_holders() et check_lp_burned() renvoient True. """ try: url = RUG_CHECK_URL.format(token_address) response = requests.get(url, timeout=10) response.raise_for_status() data = response.json() # Utiliser "topHolders" pour les détenteurs holders = data.get("topHolders", []) markets = data.get("markets", []) if not holders or not markets: return False if not check_top_holders(holders): return False if not check_lp_burned(markets): return False return True except requests.exceptions.RequestException as e: logging.error(f"Erreur lors de la vérification RugCheck pour {token_address}: {e}") return False # --- Récupération des tokens depuis différents endpoints --- def fetch_tokens_from_dexscreener(): url = "https://api.dexscreener.com/token-profiles/latest/v1?limit=1000" try: response = requests.get(url, timeout=10) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logging.error(f"Erreur dans fetch_tokens_from_dexscreener: {e}") return None def fetch_tokens_boosts_latest(): url = "https://api.dexscreener.com/token-boosts/latest/v1?limit=1000" try: response = requests.get(url, timeout=10) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logging.error(f"Erreur dans fetch_tokens_boosts_latest: {e}") return None def fetch_tokens_boosts_top(): url = "https://api.dexscreener.com/token-boosts/top/v1?limit=1000" try: response = requests.get(url, timeout=10) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logging.error(f"Erreur dans fetch_tokens_boosts_top: {e}") return None def combine_tokens(): tokens = [] profiles = fetch_tokens_from_dexscreener() boosts_latest = fetch_tokens_boosts_latest() boosts_top = fetch_tokens_boosts_top() if isinstance(profiles, dict): tokens.extend(profiles.get("data", [])) elif isinstance(profiles, list): tokens.extend(profiles) if isinstance(boosts_latest, dict): tokens.extend(boosts_latest.get("data", [])) elif isinstance(boosts_latest, list): tokens.extend(boosts_latest) if isinstance(boosts_top, dict): tokens.extend(boosts_top.get("data", [])) elif isinstance(boosts_top, list): tokens.extend(boosts_top) # Éliminer les doublons par adresse unique = {} for token in tokens: addr = token.get("tokenAddress") or token.get("baseToken", {}).get("address") if addr: unique[addr] = token return list(unique.values()) # --- Récupération des détails d'un token --- def rate_limit_detail(func): def wrapper(*args, **kwargs): result = func(*args, **kwargs) time.sleep(0.2) # Limite de 300 appels par minute return result return wrapper @rate_limit_detail def fetch_dexscreener_detail(token_address): url = f"https://api.dexscreener.com/tokens/v1/solana/{token_address}" try: response = requests.get(url, timeout=10) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logging.error(f"Erreur Dexscreener (détail) pour {token_address}: {e}") return None # --- Insertion en bulk en MySQL --- def bulk_save_token_data(tokens_list): if not tokens_list: return try: mydb = mysql.connector.connect( host=MYSQL_HOST, user=MYSQL_USER, password=MYSQL_PASSWORD, database=MYSQL_DATABASE ) mycursor = mydb.cursor() sql = ("INSERT INTO token_data (pair_address, nom, liquidity, volume, price, priceChange, timestamp) " "VALUES (%s, %s, %s, %s, %s, %s, NOW())") mycursor.executemany(sql, tokens_list) mydb.commit() logging.info(f"{mycursor.rowcount} records insérées en bulk.") except mysql.connector.Error as err: logging.error(f"Erreur MySQL lors de l'insertion en bulk: {err}") finally: if mydb.is_connected(): mycursor.close() mydb.close() def detect_pump(data): if not data: return if isinstance(data, dict): pairs = data.get('pairs', []) elif isinstance(data, list): pairs = data else: pairs = [] prices = [] volumes = [] for pair in pairs: try: price = pair.get('priceUsd', 0) volume = pair.get('volume', {}).get('h24', 0) price = float(price) if price is not None else 0 volume = float(volume) if volume is not None else 0 if price > 0 and volume > 0: prices.append(price) volumes.append(volume) except (ValueError, TypeError): continue if len(prices) < 2 or len(volumes) < 2: return avg_price = sum(prices) / len(prices) avg_volume = sum(volumes) / len(volumes) price_jumps = [abs(prices[i] - prices[i-1]) / avg_price for i in range(1, len(prices))] volume_jumps = [abs(volumes[i] - volumes[i-1]) / avg_volume for i in range(1, len(volumes))] threshold = 0.3 pump_events = [] for i in range(1, len(prices)): if price_jumps[i-1] > threshold and volume_jumps[i-1] > threshold: pump_events.append({"price": prices[i], "volume": volumes[i]}) return pump_events def send_telegram_alert(message): try: bot = telegram.Bot(token=TELEGRAM_BOT_TOKEN) asyncio.run(bot.send_message(chat_id=TELEGRAM_CHAT_ID, text=message)) logging.info("📢 Alerte envoyée sur Telegram !") except Exception as e: logging.error(f"❌ Erreur lors de l'envoi de l'alerte Telegram : {e}") app = FastAPI() @app.get("/analyze/{pair_address}") async def analyze_pair(pair_address: str): detail = fetch_dexscreener_detail(pair_address) if detail is None: raise HTTPException(status_code=500, detail="Failed to fetch details from Dexscreener") pump_events = detect_pump(detail) if pump_events: message = f"🚀 Pump détecté pour {pair_address} !\n\n" for event in pump_events: message += f"💰 Prix: {event['price']}\n📈 Volume: {event['volume']}\n" send_telegram_alert(message) for event in pump_events: bulk_save_token_data([(pair_address, detail.get("baseToken", {}).get("name") or detail.get("name", "Inconnu"), detail.get("liquidity", {}).get("usd", 0), detail.get("volume", {}).get("h24", 0), float(detail.get("priceUsd", 0)), float(detail.get("priceChange", {}).get("h1", 0)))]) return {"status": "Alerte envoyée sur Telegram et données enregistrées"} else: return {"status": "Aucun pump détecté"} # --- Gestion par file d'attente pour limiter les appels à RugCheck --- token_queue = queue.Queue() def worker(): while True: try: token_data = token_queue.get(timeout=5) except queue.Empty: break # Fin du traitement si la file est vide token_address, token_name, liquidity_detail, volume_detail, detail_obj = token_data if verify_with_rugcheck(token_address): valid_tokens.append((token_address, token_name, liquidity_detail, volume_detail, float(detail_obj.get("priceUsd", 0)), float(detail_obj.get("priceChange", {}).get("h1", 0)))) else: non_validated_count[0] += 1 logging.info(f"Token non validé par RugCheck : {non_validated_count[0]}") token_queue.task_done() def continuous_token_search(interval=10): global valid_tokens, non_validated_count while True: logging.info("Recherche des tokens en cours...") tokens = combine_tokens() logging.info(f"Nombre de tokens combinés: {len(tokens)}") tokens = [token for token in tokens if token.get("url", "").startswith("https://dexscreener.com/solana/")] total_tokens_analyzed = len(tokens) tokens_for_rugcheck = [] # Tokens ayant passé les filtres internes valid_tokens = [] # Tokens validés par RugCheck non_validated_count = [0] # Utilisation d'une liste pour pouvoir modifier la valeur dans les threads # Première passe : appliquer les filtres internes for i, token in enumerate(tokens): token_address = token.get("tokenAddress") or token.get("baseToken", {}).get("address") if token_address: token["tokenAddress"] = token_address else: logging.warning(f"Adresse non trouvée pour le token index {i}") continue # Extraction du nom depuis le profil token_name = token.get("nom") if not token_name and token.get("baseToken"): token_name = token["baseToken"].get("name") or token["baseToken"].get("symbol", "Inconnu") # Récupérer les détails pour affiner l'extraction detail = fetch_dexscreener_detail(token_address) if not detail: logging.warning(f"Aucune donnée DexScreener pour {token_address}") continue if isinstance(detail, dict): pairs = detail.get("pairs", []) detail_obj = detail elif isinstance(detail, list): pairs = detail detail_obj = detail[0] if len(detail) > 0 else {} else: pairs = [] detail_obj = {} if not pairs: logging.warning(f"La liste 'pairs' est vide pour {token_address}") continue try: pair_info = pairs[0] liquidity_str = pair_info.get("liquidity", {}).get("usd") volume_str = pair_info.get("volume", {}).get("h24") liquidity_detail = float(liquidity_str) if liquidity_str is not None else 0 volume_detail = float(volume_str) if volume_str is not None else 0 except Exception as e: logging.error(f"Erreur de conversion des infos DexScreener pour {token_address}: {e}") continue if liquidity_detail < FILTER_MIN_LIQUIDITY or volume_detail < FILTER_MIN_VOLUME: continue # Si le nom est toujours vide, tenter de le récupérer depuis detail_obj if not token_name and isinstance(detail_obj, dict): token_name = detail_obj.get("baseToken", {}).get("name") or detail_obj.get("baseToken", {}).get("symbol", "Inconnu") token["nom"] = token_name tokens_for_rugcheck.append((token_address, token_name, liquidity_detail, volume_detail, detail_obj)) logging.info(f"Tokens à traiter par RugCheck : {len(tokens_for_rugcheck)}") # Placer les tokens dans la file d'attente for token_data in tokens_for_rugcheck: token_queue.put(token_data) # Lancer quelques workers pour traiter la file (ici 4 threads) threads = [] for _ in range(4): t = threading.Thread(target=worker) t.start() threads.append(t) token_queue.join() for t in threads: t.join() logging.info(f"Nombre de tokens analysés: {total_tokens_analyzed}. Nombre de tokens validés: {len(valid_tokens)}. Tokens non validés par RugCheck : {non_validated_count[0]}") bulk_save_token_data(valid_tokens) # Compte à rebours pour la fin de la boucle for sec in range(interval, 0, -1): logging.info(f"Fin de la boucle dans {sec} s") time.sleep(1) @app.on_event("startup") def start_background_tasks(): thread = threading.Thread(target=continuous_token_search, args=(10,), daemon=True) thread.start() if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)