pumpbot.py

#!/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()