#!/usr/bin/env python3 """ PumpFun Bot - All-In-One -------------------------------- Monitors recently migrated coins from PumpFun API, performs security checks, handles basic sentiment analysis, and integrates a Telegram bot for alerts and (optional) trading commands. Requirements: pip install requests web3 python-telegram-bot==20.* pandas sqlalchemy textblob scikit-learn hdbscan Usage: 1. Create a config.ini (see example below). 2. python pumpfun_bot.py """ import os import time import json import asyncio import configparser from datetime import datetime, timedelta import requests import numpy as np import pandas as pd from sqlalchemy import create_engine from textblob import TextBlob # Blockchain from web3 import Web3 # Telegram Bot (async version) from telegram import Bot, Update from telegram.ext import ( ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters ) # Machine Learning from sklearn.cluster import DBSCAN from sklearn.ensemble import IsolationForest ###################################################################### # 1. CONFIGURATION ###################################################################### CONFIG_FILE = "config.ini" EXAMPLE_CONFIG = """ [API] PUMPFUN_KEY = your_pumpfun_api_key_here INFURA_KEY = your_infura_key_here ETHERSCAN_KEY = your_etherscan_key_here POLL_INTERVAL = 60 [FILTERS] MIN_LIQUIDITY = 5.0 MAX_CREATOR_FEE = 10.0 MIN_HOLDERS = 25 BLOCK_NEW_COINS_MINUTES = 10 MAX_COINS_PER_CREATOR = 3 [BLACKLISTS] COIN_ADDRESSES = 0x0000000000000000000000000000000000000000 DEV_ADDRESSES = 0x0000000000000000000000000000000000000000 COIN_BLACKLIST_URL = https://your.service/coin_blacklist DEV_BLACKLIST_URL = https://your.service/dev_blacklist [TWITTER] # If you have a custom Twitter API or a service like TweetScout API_KEY = your_twitter_api_key RATE_LIMIT = 30 [TELEGRAM] BOT_TOKEN = your_telegram_bot_token CHANNEL_ID = your_telegram_channel_id [TRADING] DEFAULT_AMOUNT = 0.1 SLIPPAGE = 1.5 MAX_POSITION = 5.0 STOP_LOSS = -0.15 TAKE_PROFIT = 0.3 [SECURITY] RUGCHECK_API = your_rugcheck_api_key BUNDLED_THRESHOLD = 0.65 """ class PumpFunBot: def __init__(self, config_path: str = CONFIG_FILE): """Initialize everything: config, DB, Web3, etc.""" self.config = self.load_config(config_path) # Database self.db_path = self.config.get("DATABASE", "DB_PATH", fallback="pumpfun.db") self.db_engine = create_engine(f"sqlite:///{self.db_path}") self.create_tables() # Web3 infura_key = self.config["API"].get("INFURA_KEY", "") self.w3 = Web3( Web3.HTTPProvider(f"https://mainnet.infura.io/v3/{infura_key}") ) # PumpFun API settings self.api_base = "https://api.pump.fun" # Example endpoint self.pumpfun_key = self.config["API"].get("PUMPFUN_KEY", "") self.headers = { 'Authorization': f'Bearer {self.pumpfun_key}', 'Content-Type': 'application/json' } # Filters self.filters = { 'min_liquidity': self.config["FILTERS"].getfloat("MIN_LIQUIDITY", 5.0), 'max_creator_fee': self.config["FILTERS"].getfloat("MAX_CREATOR_FEE", 10.0), 'min_holders': self.config["FILTERS"].getint("MIN_HOLDERS", 25), 'block_new_coins_minutes': self.config["FILTERS"].getint("BLOCK_NEW_COINS_MINUTES", 10), 'max_coins_per_creator': self.config["FILTERS"].getint("MAX_COINS_PER_CREATOR", 3), } # Blacklist sets self.blacklisted_coins = self.load_blacklist("COIN_ADDRESSES") self.blacklisted_devs = self.load_blacklist("DEV_ADDRESSES") # Telegram self.telegram_token = self.config["TELEGRAM"].get("BOT_TOKEN", "") self.telegram_channel_id = self.config["TELEGRAM"].get("CHANNEL_ID", "") self.application = None # Will initialize if Telegram token is present # For demonstration, we keep track of the "currently analyzed" contract self.currently_analyzed_contract = None ###################################################################### # 2. CONFIG & DATABASE ###################################################################### @staticmethod def load_config(path: str) -> configparser.ConfigParser: """Load config.ini or create an example if missing.""" if not os.path.exists(path): with open(path, "w") as f: f.write(EXAMPLE_CONFIG) raise FileNotFoundError( f"No config.ini found. An example config has been created at {path}." ) config = configparser.ConfigParser() config.read(path) return config def load_blacklist(self, key: str): """Load blacklisted addresses from config.""" addresses_str = self.config["BLACKLISTS"].get(key, "") addresses = [addr.strip() for addr in addresses_str.split(",") if addr.strip()] # Convert to checksummed addresses if valid checksummed = set() for addr in addresses: try: checksummed.add(Web3.to_checksum_address(addr)) except ValueError: # If it doesn't parse as an address, skip pass return checksummed def create_tables(self): """Initialize database schema if not exists.""" with self.db_engine.connect() as conn: # Coins table conn.execute(""" CREATE TABLE IF NOT EXISTS coins ( id INTEGER PRIMARY KEY AUTOINCREMENT, contract_address TEXT UNIQUE, name TEXT, symbol TEXT, creator_wallet TEXT, migration_time DATETIME, initial_liquidity REAL, creator_fee FLOAT, holders INTEGER, social_score FLOAT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # Transactions table conn.execute(""" CREATE TABLE IF NOT EXISTS transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, contract_address TEXT, tx_hash TEXT UNIQUE, direction TEXT, amount_eth REAL, gas_price REAL, block_number INTEGER, timestamp DATETIME ) """) # Twitter metrics table conn.execute(""" CREATE TABLE IF NOT EXISTS twitter_metrics ( coin_address TEXT PRIMARY KEY, twitter_handle TEXT, follower_count INTEGER, following_count INTEGER, sentiment_score REAL, post_frequency REAL, verified BOOLEAN, account_age_days INTEGER ) """) # Twitter posts table conn.execute(""" CREATE TABLE IF NOT EXISTS twitter_posts ( post_id TEXT PRIMARY KEY, coin_address TEXT, content TEXT, likes INTEGER, retweets INTEGER, timestamp DATETIME, sentiment REAL, hashtags TEXT, links TEXT ) """) # Security checks table conn.execute(""" CREATE TABLE IF NOT EXISTS security_checks ( contract_address TEXT PRIMARY KEY, rugcheck_score REAL, rugcheck_verdict TEXT, top_holder_percent REAL, is_bundled BOOLEAN, check_time DATETIME ) """) # Trades table (if using trading features) conn.execute(""" CREATE TABLE IF NOT EXISTS trades ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, direction TEXT, contract_address TEXT, amount REAL, tx_hash TEXT, profit REAL ) """) ###################################################################### # 3. DATA FETCHING & PARSING ###################################################################### def fetch_migrated_coins(self, limit=100): """Retrieve recently migrated coins from PumpFun API.""" try: params = { 'limit': limit, 'sort': 'desc' } url = f"{self.api_base}/migrations" resp = requests.get(url, headers=self.headers, params=params, timeout=10) resp.raise_for_status() data = resp.json() return data.get("data", []) except Exception as e: print(f"fetch_migrated_coins error: {e}") return [] def parse_coin_data(self, raw_data): """Convert raw PumpFun API data into a consistent dictionary.""" try: # Some data might not exist in raw_data; handle gracefully contract_address = raw_data.get("contractAddress", "") # Convert to checksum contract_address = Web3.to_checksum_address(contract_address) name = raw_data["token"].get("name", "Unknown") symbol = raw_data["token"].get("symbol", "UNK") creator = raw_data.get("creator", "0x0000000000000000000000000000000000000000") creator_wallet = Web3.to_checksum_address(creator) migration_time_str = raw_data.get("migrationTime", datetime.now().isoformat()) migration_time = datetime.fromisoformat(migration_time_str) initial_liquidity = float(raw_data.get("initialLiquidity", 0)) creator_fee = float(raw_data.get("feePercentage", 0)) holders = int(raw_data.get("holderCount", 0)) parsed = { "contract_address": contract_address, "name": name, "symbol": symbol, "creator_wallet": creator_wallet, "migration_time": migration_time, "initial_liquidity": initial_liquidity, "creator_fee": creator_fee, "holders": holders, } return parsed except Exception as e: print(f"parse_coin_data error: {e}") return {} def enhanced_parse_coin_data(self, raw_data): """ Extended parse to add more fields (like contract verification). Adjust or remove if you do not have Etherscan or advanced checks. """ parsed = self.parse_coin_data(raw_data) parsed["is_verified_contract"] = self.check_contract_verification( parsed["contract_address"] ) return parsed def check_contract_verification(self, address: str) -> bool: """Check if contract is verified on Etherscan (optional).""" etherscan_key = self.config["API"].get("ETHERSCAN_KEY", "") if not etherscan_key or not address: return False try: url = ( f"https://api.etherscan.io/api?" f"module=contract&action=getabi&address={address}&apikey={etherscan_key}" ) resp = requests.get(url, timeout=5) data = resp.json() # If 'result' is a string error, contract might not be verified return bool(data.get("status") == "1") except Exception: return False ###################################################################### # 4. SECURITY & FILTERS ###################################################################### def is_blacklisted(self, coin_data) -> bool: """Check if coin or creator wallet is in our blacklists.""" contract_address = coin_data.get("contract_address", "") creator_wallet = coin_data.get("creator_wallet", "") if not contract_address or not creator_wallet: return True if contract_address in self.blacklisted_coins: print(f"[SECURITY] Coin {contract_address} is blacklisted.") return True if creator_wallet in self.blacklisted_devs: print(f"[SECURITY] Dev {creator_wallet} is blacklisted.") return True # Could also implement "suspicious creator" checks: if self.is_suspicious_creator(creator_wallet): print(f"[SECURITY] Creator {creator_wallet} made too many coins.") return True return False def is_suspicious_creator(self, creator_wallet: str) -> bool: """Check how many coins a single creator has made.""" query = f""" SELECT COUNT(*) as created_coins FROM coins WHERE creator_wallet = '{creator_wallet}' """ df = pd.read_sql(query, self.db_engine) return df["created_coins"].iloc[0] > self.filters["max_coins_per_creator"] def apply_filters(self, coin_data): """Apply standard filters to exclude suspicious/low-quality coins.""" if not coin_data: return False # Check liquidity, fee, holders, and age if coin_data["initial_liquidity"] < self.filters["min_liquidity"]: return False if coin_data["creator_fee"] > self.filters["max_creator_fee"]: return False if coin_data["holders"] < self.filters["min_holders"]: return False age_threshold = datetime.now() - timedelta( minutes=self.filters["block_new_coins_minutes"] ) if coin_data["migration_time"] > age_threshold: # It's too new return False return True def perform_security_checks(self, coin_data: dict): """ Placeholder for external security checks (e.g. RugCheck). If you don't have an external service, omit or stub this. """ # Example: Mark 'top_holder_percent' or 'is_bundled' as False # because we don't have a standard way to get top holders security_data = { "contract_address": coin_data["contract_address"], "rugcheck_score": 100.0, # Pretend it's safe "rugcheck_verdict": "Good", "top_holder_percent": 0.0, "is_bundled": False, "check_time": datetime.now() } # In a real scenario, you'd call the external API and parse: # rug_data = self.check_rugcheck_verdict(coin_data["contract_address"]) # is_bundled = self.analyze_token_distribution(coin_data["contract_address"]) # Then store in DB df = pd.DataFrame([security_data]) df.to_sql("security_checks", self.db_engine, if_exists="replace", index=False) return security_data ###################################################################### # 5. STORING & ANALYZING DATA ###################################################################### def save_coins(self, parsed_coin: dict): """Insert or update coin data in the 'coins' table.""" if not parsed_coin: return df = pd.DataFrame([parsed_coin]) df.to_sql("coins", self.db_engine, if_exists="append", index=False) def analyze_transaction_patterns(self): """Example of identifying suspicious transactions using DBSCAN.""" query = """ SELECT contract_address, COUNT(*) as tx_count, SUM(amount_eth) as total_volume, AVG(gas_price) as avg_gas FROM transactions GROUP BY contract_address """ df = pd.read_sql(query, self.db_engine) if df.empty: return pd.DataFrame() X = df[["tx_count", "total_volume", "avg_gas"]].fillna(0) clustering = DBSCAN(eps=0.5, min_samples=3).fit(X) df["cluster"] = clustering.labels_ # Outliers have label == -1 return df[df["cluster"] == -1] def sentiment_analysis_example(self, text: str) -> float: """Basic sentiment polarity with TextBlob.""" if not text: return 0.0 analysis = TextBlob(text) return analysis.sentiment.polarity def analyze_coin(self, coin_data: dict): """ Placeholder for additional analysis steps on a new coin: e.g. analyzing sentiment, transaction patterns, liquidity, etc. """ print(f"[ANALYSIS] Analyzing {coin_data['symbol']} ({coin_data['contract_address']})...") # ... outliers = self.analyze_transaction_patterns() # If outliers exist, you might do some alerting or extra checks if not outliers.empty: print(f"[WARNING] Transaction outliers detected: {outliers.to_dict('records')}") ###################################################################### # 6. TELEGRAM BOT (Optional) ###################################################################### def setup_telegram_bot(self): """Initialize Telegram application if a bot token is configured.""" if not self.telegram_token: print("[TELEGRAM] No bot token provided. Telegram bot is disabled.") return self.application = ApplicationBuilder().token(self.telegram_token).build() self.application.add_handler(CommandHandler("start", self.cmd_start)) self.application.add_handler(CommandHandler("buy", self.cmd_buy)) self.application.add_handler(CommandHandler("sell", self.cmd_sell)) self.application.add_handler(MessageHandler(filters.TEXT, self.cmd_handle_text)) async def cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /start command.""" await update.message.reply_text( "Welcome to PumpFun Bot!\n" "Commands:\n" "/buy [amount] - Execute a mock buy.\n" "/sell [amount] - Execute a mock sell.\n" ) async def cmd_buy(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Mock buy command.""" try: args = context.args amount = float(args[0]) if args else 0.1 message = f"Buying {amount} of {self.currently_analyzed_contract} (mock)..." await update.message.reply_text(message) # Here you'd call your actual trading code except Exception as e: await update.message.reply_text(f"Error: {str(e)}") async def cmd_sell(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Mock sell command.""" try: args = context.args amount = float(args[0]) if args else 0.1 message = f"Selling {amount} of {self.currently_analyzed_contract} (mock)..." await update.message.reply_text(message) # Here you'd call your actual trading code except Exception as e: await update.message.reply_text(f"Error: {str(e)}") async def cmd_handle_text(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle any text message not matching a command.""" await update.message.reply_text("Use /buy or /sell commands, or /start for help.") async def send_telegram_alert(self, message: str): """Send a message to the configured Telegram channel.""" if not self.telegram_channel_id: print("[TELEGRAM] No CHANNEL_ID configured.") return if not self.application: print("[TELEGRAM] Telegram bot is not initialized.") return bot: Bot = self.application.bot await bot.send_message(chat_id=self.telegram_channel_id, text=message) ###################################################################### # 7. MAIN LOOP ###################################################################### async def monitor_coins_loop(self): """Asynchronous loop that fetches, filters, and analyzes new coins.""" poll_interval = self.config["API"].getint("POLL_INTERVAL", 60) while True: try: raw_coins = self.fetch_migrated_coins(limit=10) for raw_coin in raw_coins: parsed = self.enhanced_parse_coin_data(raw_coin) # Security checks if self.is_blacklisted(parsed): continue self.perform_security_checks(parsed) # Filter checks if not self.apply_filters(parsed): continue # Save to DB self.save_coins(parsed) # Perform further analysis self.analyze_coin(parsed) # Example: send a Telegram alert if desired if self.application: msg = (f"New coin found:\n" f"Symbol: {parsed['symbol']}\n" f"Contract: {parsed['contract_address']}\n" f"Liquidity: {parsed['initial_liquidity']}\n") await self.send_telegram_alert(msg) # Keep track of the "current" coin for /buy /sell self.currently_analyzed_contract = parsed["contract_address"] await asyncio.sleep(poll_interval) except Exception as e: print(f"[ERROR] {e}") await asyncio.sleep(60) def run(self): """Entry point to run the bot. If Telegram is configured, run async.""" self.setup_telegram_bot() if self.application: # If Telegram is available, run the async loop loop = asyncio.get_event_loop() loop.create_task(self.monitor_coins_loop()) loop.create_task(self.application.initialize()) loop.create_task(self.application.start_polling()) try: loop.run_forever() except KeyboardInterrupt: print("[SHUTDOWN] Stopping Telegram bot...") loop.run_until_complete(self.application.shutdown()) loop.run_until_complete(self.application.stop()) loop.close() else: # Run a simple synchronous loop if Telegram not configured print("[INFO] Telegram not configured; running simple loop.") while True: try: raw_coins = self.fetch_migrated_coins(limit=10) for raw_coin in raw_coins: parsed = self.enhanced_parse_coin_data(raw_coin) if self.is_blacklisted(parsed): continue self.perform_security_checks(parsed) if not self.apply_filters(parsed): continue self.save_coins(parsed) self.analyze_coin(parsed) time.sleep(self.config["API"].getint("POLL_INTERVAL", 60)) except Exception as e: print(f"[ERROR] {e}") time.sleep(60) ###################################################################### # 8. MAIN ENTRY POINT ###################################################################### if __name__ == "__main__": bot = PumpFunBot() bot.run()