Build a Whale Alerts Telegram Bot with thirdweb Insight

In blockchain, the ability to track transactions in real-time can give you a significant advantage. Whether you're monitoring whale activity or keeping tabs on specific wallets, having instant notifications can help you make better decisions.

Let's build a Telegram bot that does exactly that using thirdweb's Insight service.

In this tutorial, you'll learn how to:

• Use thirdweb Insight to fetch blockchain data
• Create a Telegram bot from scratch
• Combine them to track wallet transactions in real-time

What You'll Need

• A third web account
• Python installed on your machine
• Telegram account
• Basic understanding of Python programming

Check out the video version of this tutorial here:

Getting Started with thirdweb Insight

thirdweb Insight provides an easy way to access blockchain data through simple API requests. Here's how to use it:

1. Log into your thirdweb dashboard
2. Select any project and navigate to the Insight tab
3. In the playground, select "Get All Transactions"
4. Choose your blockchain network (e.g., Ethereum mainnet, Arbitrum Sepolia)
5. Set parameters like limit (e.g., 5 transactions)
6. Click "Run" to see results

The important part is the URL that Insight generates. You can use this URL in your applications to fetch blockchain data programmatically.

Fetching Blockchain Data with Python

Let's create a Python script to fetch transaction data:

import requests
import json

# Set parameters
chain_id = "421614"  # Arbitrum Sepolia
client_id = "YOUR_CLIENT_ID"  # Get from thirdweb dashboard
wallet_address = "YOUR_WALLET_ADDRESS"
limit = 1
sort_by = "block"
sort_order = "desc"

# Create URL
transactions_url = f"https://{chain_id}.insight.thirdweb.com/v1/transactions?limit={limit}&client_id={client_id}&sort_by={sort_by}&sort_order={sort_order}&from_address={wallet_address}"

def fetch_data(url):
    response = requests.get(url)
    return response.json()

# Fetch and display transaction
transactions = fetch_data(transactions_url)
print(json.dumps(transactions, indent=2))

This script will fetch the latest transaction from a specific wallet address on the Arbitrum Sepolia network.

Creating a Basic Telegram Bot

Before we combine everything, let's create a simple Telegram bot:

1. Open Telegram and search for "BotFather"
2. Start a chat and send /newbot
3. Follow the prompts to create your bot (name and username)
4. BotFather will give you a token - save this securely!

Now, let's create a basic bot using Python:

from telegram import Update
from telegram.ext import (
    Application,
    CommandHandler,
    MessageHandler,
    filters,
    ContextTypes
)

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("Hello! Welcome to the thirdweb Transaction Alerts bot!")

async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    text = update.message.text
    await update.message.reply_text(f"You said: {text}")

def main():
    bot_token = "YOUR_BOT_TOKEN"  # From BotFather
    
    # Create application
    application = Application.builder().token(bot_token).build()
    
    # Add handlers
    application.add_handler(CommandHandler("start", start))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
    
    # Run the bot
    application.run_polling()

if __name__ == "__main__":
    main()

This creates a basic bot that responds to the /start command and echoes back any messages you send.

Building the Complete Transaction Tracker Bot

Now, let's combine both functionalities to create our transaction tracker. The complete bot will:

1. Allow users to set up tracking preferences:
   • Choose a blockchain network
   • Enter a wallet address to track
   • Set an update interval
2. Track wallet activity and send notifications based on the interval

Here's the structure for our complete bot:

import requests
import asyncio
from telegram import Update, ReplyKeyboardMarkup, ReplyKeyboardRemove
from telegram.ext import (
    Application,
    CommandHandler,
    MessageHandler,
    filters,
    ContextTypes,
    ConversationHandler
)

# Define conversation states
CHAIN, WALLET, INTERVAL = range(3)

# Store user data
user_data = {}

# Function to fetch transactions
def fetch_transactions(chain_id, client_id, wallet_address, limit=1, sort_by="block", sort_order="desc"):
    try:
        transactions_url = f"https://{chain_id}.insight.thirdweb.com/v1/transactions"
        params = {
            "limit": limit,
            "client_id": client_id,
            "sort_by": sort_by,
            "sort_order": sort_order,
            "from_address": wallet_address
        }
        response = requests.get(transactions_url, params=params)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        return {"error": str(e)}

# Bot command handlers
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    welcome_text = (
        "Welcome to thirdweb Tracker Bot! 👋\n\n"
        "With this bot, you can track transactions for any wallet on various blockchains.\n\n"
        "Use /setup to configure your tracking preferences."
    )
    await update.message.reply_text(welcome_text)

async def setup(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(
        "Which blockchain network do you want to track? (For example: Ethereum, Sepolia, Arbitrum Sepolia)",
        reply_markup=ReplyKeyboardMarkup([
            ["Ethereum Sepolia", "Arbitrum Sepolia"]
        ], one_time_keyboard=True)
    )
    return CHAIN

async def set_chain(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chain = update.message.text
    user_id = update.effective_user.id
    user_data[user_id] = {"chain": chain}
    
    await update.message.reply_text(
        "Please enter the wallet address you want to track:",
        reply_markup=ReplyKeyboardRemove()
    )
    return WALLET

async def set_wallet(update: Update, context: ContextTypes.DEFAULT_TYPE):
    wallet = update.message.text
    user_id = update.effective_user.id
    user_data[user_id]["wallet"] = wallet
    
    await update.message.reply_text(
        "How often do you want to receive updates?",
        reply_markup=ReplyKeyboardMarkup([
            ["1 minute", "1 hour"],
            ["4 hours", "12 hours"]
        ], one_time_keyboard=True)
    )
    return INTERVAL

async def set_interval(update: Update, context: ContextTypes.DEFAULT_TYPE):
    interval_text = update.message.text
    user_id = update.effective_user.id
    
    # Convert text intervals to seconds
    interval_mapping = {
        "1 minute": 60,
        "1 hour": 3600,
        "4 hours": 14400,
        "12 hours": 43200
    }
    
    interval = interval_mapping.get(interval_text, 60)  # Default to 1 minute if invalid
    user_data[user_id]["interval"] = interval
    
    await update.message.reply_text(
        f"Tracking setup complete!\n\n"
        f"Chain: {user_data[user_id]['chain']}\n"
        f"Wallet: {user_data[user_id]['wallet']}\n"
        f"Update interval: {interval_text}"
    )
    
    # Start tracking for this user
    asyncio.create_task(track_wallet_activity(context.bot, user_id))
    return ConversationHandler.END

async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("Setup cancelled.", reply_markup=ReplyKeyboardRemove())
    return ConversationHandler.END

# Transaction tracking function
async def track_wallet_activity(bot, user_id):
    # Chain ID mapping
    chain_mapping = {
        "Ethereum Sepolia": "11155111",
        "Arbitrum Sepolia": "421614"
    }
    
    # Block explorer mapping
    block_explorers = {
        "Ethereum Sepolia": "https://sepolia.etherscan.io/tx/",
        "Arbitrum Sepolia": "https://sepolia-explorer.arbitrum.io/tx/"
    }
    
    client_id = "YOUR_CLIENT_ID"  # Replace with your client ID
    
    while user_id in user_data:
        user_chain = user_data[user_id]["chain"]
        wallet = user_data[user_id]["wallet"]
        
        chain_id = chain_mapping.get(user_chain, "1")
        block_explorer = block_explorers.get(user_chain, "https://etherscan.io/tx/")
        
        transactions = fetch_transactions(chain_id, client_id, wallet)
        
        if "error" in transactions:
            await bot.send_message(user_id, f"Error fetching transactions: {transactions['error']}")
        elif "data" in transactions:
            messages = []
            
            for tx in transactions["data"]:
                if tx["value"] == "0":
                    continue  # Skip zero-value transactions
                
                # Convert from wei to ETH
                eth_value = float(tx["value"]) / 1000000000000000000
                
                to_address = tx["to_address"]
                from_address = tx["from_address"]
                
                # Determine if this is incoming or outgoing
                if from_address.lower() == wallet.lower():
                    direction = "sent"
                    counterparty = to_address
                else:
                    direction = "received"
                    counterparty = from_address
                
                # Format message
                message = (
                    f"🔔 *Transaction Alert*\n\n"
                    f"Wallet {direction} *{eth_value:.4f} ETH*\n"
                    f"{'To' if direction == 'sent' else 'From'}: `{counterparty[:6]}...{counterparty[-4:]}`\n\n"
                    f"[View on Block Explorer]({block_explorer}{tx['transaction_hash']})"
                )
                messages.append(message)
            
            # Send messages
            for message in messages:
                await bot.send_message(user_id, message, parse_mode="Markdown")
        
        # Wait for the specified interval before checking again
        await asyncio.sleep(user_data[user_id]["interval"])

def main():
    bot_token = "YOUR_BOT_TOKEN"  # Replace with your bot token
    
    # Create application
    application = Application.builder().token(bot_token).build()
    print("Bot started successfully!")
    
    # Define conversation handler
    setup_handler = ConversationHandler(
        entry_points=[CommandHandler("setup", setup)],
        states={
            CHAIN: [MessageHandler(filters.TEXT & ~filters.COMMAND, set_chain)],
            WALLET: [MessageHandler(filters.TEXT & ~filters.COMMAND, set_wallet)],
            INTERVAL: [MessageHandler(filters.TEXT & ~filters.COMMAND, set_interval)]
        },
        fallbacks=[CommandHandler("cancel", cancel)]
    )
    
    # Add handlers
    application.add_handler(CommandHandler("start", start))
    application.add_handler(setup_handler)
    
    # Run the bot
    application.run_polling()

if __name__ == "__main__":
    main()

How to Use the Bot

1. Start a chat with your bot
2. Send /start to get the welcome message
3. Send /setup to begin configuration
4. Select your blockchain network
5. Enter the wallet address you want to track
6. Choose how often you want to receive updates

The bot will then start sending you notifications about transactions involving the wallet you're tracking. Each notification includes:

• Transaction direction (sent/received)
• Amount in ETH
• Counterparty address (abbreviated)
• Link to view the transaction on a block explorer

Customization Options

Here are some ways you can extend the bot:

• Add more blockchain networks
• Track multiple wallets simultaneously
• Filter by transaction type or value
• Add notifications for specific contract interactions
• Create alerts for certain value thresholds

Conclusion

Congratulations! You've built a custom Telegram bot that leverages thirdweb Insight to track blockchain transactions in real-time. This is just the beginning of what you can do with blockchain data. You could extend this to track specific NFT collections, monitor DeFi positions, or create custom alerts for market movements.

The possibilities are endless, and thirdweb Insight makes it easy to access the data you need. Try customizing this bot to fit your specific use case and share your creations with the community!