Building a Powerful EVM Token Analysis Telegram Bot

def _initialize_web3(self, network: str) -> Web3:
    """Initialize Web3 connection for the specified network."""
    provider_url = NETWORKS[network]["provider_url"]
    logger.info(f"Initializing Web3 for {network} with provider URL: {provider_url}")

    web3 = Web3(Web3.HTTPProvider(provider_url))

    # Add PoA middleware for networks like BSC
    web3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0)

    # For BNB Chain (BSC), we need additional configuration
    if network == "BNB":
        # BSC uses a different gas price strategy
        from web3.gas_strategies.rpc import rpc_gas_price_strategy
        web3.eth.set_gas_price_strategy(rpc_gas_price_strategy)

    if not web3.is_connected():
        logger.error(f"Failed to connect to {network} network")
        raise ConnectionError(f"Cannot connect to {network} network")

    logger.info(f"Connected to {network} network: Chain ID {web3.eth.chain_id}")
    return web3

class TokenData:
    """Model for cached token data"""
    def __init__(
        self,
        address: str,
        name: Optional[str] = None,
        symbol: Optional[str] = None,
        deployer: Optional[str] = None,
        deployment_date: Optional[datetime] = None,
        current_price: Optional[float] = None,
        current_market_cap: Optional[float] = None,
        ath_market_cap: Optional[float] = None,
        ath_date: Optional[datetime] = None,
        last_updated: Optional[datetime] = None
    ):
        self.address = address
        self.name = name
        self.symbol = symbol
        self.deployer = deployer
        self.deployment_date = deployment_date
        self.current_price = current_price
        self.current_market_cap = current_market_cap
        self.ath_market_cap = ath_market_cap
        self.ath_date = ath_date
        self.last_updated = last_updated or datetime.now()

def init_database() -> bool:
    """Initialize the database connection and set up indexes"""
    global _db

    try:
        # Connect to MongoDB
        client = MongoClient(MONGODB_URI) 
        _db = client[DB_NAME]

        # Set up indexes for collections
        # Users collection
        _db.users.create_index([("user_id", ASCENDING)], unique=True)

        # User scans collection
        _db.user_scans.create_index([
            ("user_id", ASCENDING),
            ("scan_type", ASCENDING),
            ("date", ASCENDING)
        ], unique=True)

        # Token data collection
        _db.token_data.create_index([("address", ASCENDING)], unique=True)
        _db.token_data.create_index([("deployer", ASCENDING)])

        # Additional indexes for watchlists
        # ...

        return True
    except Exception as e:
        logging.error(f"Failed to initialize database: {e}")
        return False

def check_subscription_payment(self, user_id: int) -> Dict[str, Any]:
    if user_id not in self.subscriptions:
        logger.warning(f"No subscription found for user {user_id}")
        return {
            "success": False,
            "error": {
                "code": "SUBSCRIPTION_NOT_FOUND",
                "message": f"No subscription found for user {user_id}"
            }
        }

    subscription = self.subscriptions[user_id]
    network = subscription["network"]
    address = subscription["wallet_address"]
    required_amount = subscription["amount_required"]

    balance_result = self.check_wallet_balance(address, network)
    if not balance_result["success"]:
        return balance_result

    current_balance = balance_result["data"]["native_balance"]
    formatted_balance = balance_result["data"]["formatted_native_balance"]

    # Process payment logic
    # ...

async def handle_start_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle the main menu display with only token analysis functionality"""

    if "default_network" not in context.user_data:
        context.user_data["default_network"] = "eth"

    selected_network = context.user_data.get("default_network")
    network_display = {
        "eth": "🌐 Ethereum",
        "base": "🛡️ Base",
        "bsc": "🔶 BSC"
    }

    welcome_message = (
        f"🆘 Welcome to EVM Token Analysis Bot, {update.effective_user.first_name}! 🎉\n\n"
        f"📊 Token Analysis Features:\n"
        # Feature list...
    )

    token_analysis_keyboard = [
        [InlineKeyboardButton("🛒 First Buyers & Profits of a token", callback_data="token_first_buyers")],
        [InlineKeyboardButton("💰 Most Profitable Wallets of a token", callback_data="token_most_profitable_wallets")],
        # Additional buttons...
    ]

async def handle_telegram_error(update: Optional[Update], context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle generic TelegramError errors."""
    if update and update.effective_message:
        error_message = "An error occurred while processing your request. Please try again later."

        try:
            await update.effective_message.reply_text(error_message)
        except Exception as e:
            logger.error(f"Failed to send error message: {e}")

async def handle_period_selection(
    update: Update, 
    context: ContextTypes.DEFAULT_TYPE,
    feature_info:str, 
    scan_type: str,
    callback_prefix: str
) -> None:
    """Generic handler for period selection"""
    query = update.callback_query
    user = await check_callback_user(update)

    # Check if user has reached daily limit
    has_reached_limit, current_count = await check_rate_limit_service(
        user.user_id, scan_type, FREE_WALLET_SCANS_DAILY
    )

    if has_reached_limit and not user.is_premium:
        keyboard = [
            [InlineKeyboardButton("💎 Upgrade to Premium", callback_data="premium_info")],
            [InlineKeyboardButton("🔙 Back", callback_data="wallet_analysis")]
        ]
        reply_markup = InlineKeyboardMarkup(keyboard)

        await query.message.reply_text(
            f"⚠️ Daily Limit Reached\n\n"
            # Limit message...
            f"To unlock unlimited access to powerful wallet analysis features, upgrade to Premium and explore the full potential of on-chain intelligence. 💎\n",
            reply_markup=reply_markup,
            parse_mode=ParseMode.HTML
        )
        return

async def handle_high_net_worth_holders(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle high net worth token holders button callback"""
    query = update.callback_query
    user = await check_callback_user(update)

    # Check if user is premium
    if not user.is_premium:
        keyboard = [
            [InlineKeyboardButton("💎 Upgrade to Premium", callback_data="premium_info")],
            [InlineKeyboardButton("🔙 Back", callback_data="main_menu")]
        ]
        reply_markup = InlineKeyboardMarkup(keyboard)

        await query.message.reply_text(
            "⭐ Premium Feature\n\n"
            "💎 High Net Worth Holders is an exclusive premium feature that identifies:\n"
            # Feature description...
            "💎 Upgrade to Premium for unlimited access to this and other advanced features.",
            reply_markup=reply_markup,
            parse_mode=ParseMode.HTML
        )
        return

    # If premium, proceed with feature
    await handle_token_analysis_token_input(update, context, "high_net_worth_holders")

async def handle_admin_add_premium(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Admin command to manually add premium to a user"""
    # Check if user is admin
    user_id = update.effective_user.id
    user = get_user(user_id)

    if not user or not hasattr(user, 'is_admin') or not user.is_admin:
        await update.message.reply_text("⛔ This command is only available to administrators.")
        return

    # Command logic for adding premium access
    # ...

git clone https://github.com/stevendev0822/EVM-Token-Analysis-Telegram-Bot.git

cd EVM-Token-Analysis-Telegram-Bot

pip install -r requirements.txt

# DB Configuration
MONGODB_URI="mongodb://localhost:27017/"
DB_NAME="evm_token_analysis"

# API Keys
DUNE_API_KEY=your_dune_api_key_here
ZENROWS_API_KEY=your_zenrows_api_key_here
MORALIS_API_KEY=your_moralis_api_key_here

# ===== BLOCKCHAIN CONFIGURATION =====
# Ethereum Mainnet
ETH_PROVIDER_URL=https://mainnet.infura.io/v3/your_infura_key_here
ETH_CHAIN_ID=1
ETH_ADMIN_WALLET=your_admin_wallet_address

# Binance Smart Chain Mainnet
BNB_PROVIDER_URL=https://bsc-dataseed.binance.org/
BNB_CHAIN_ID=56
BNB_ADMIN_WALLET=your_admin_wallet_address

# Subscription Check Interval (in seconds)
CHECK_INTERVAL=60

# TELEGRAM BOT SETTING
TELEGRAM_TOKEN=your_telegram_bot_token_here

# Free tier limits
FREE_TOKEN_SCANS_DAILY=3
FREE_WALLET_SCANS_DAILY=3
FREE_PROFITABLE_WALLETS_LIMIT=5

# Premium tier limits
PREMIUM_TOKEN_SCANS_DAILY=100
PREMIUM_WALLET_SCANS_DAILY=100
PREMIUM_PROFITABLE_WALLETS_LIMIT=50

# Additional API keys
WEB3_PROVIDER_URI=https://mainnet.infura.io/v3/your_infura_key_here
ETHERSCAN_API_KEY=your_etherscan_api_key_here

python src/main.py