Build a Solana Wallet Tracker Bot with Telegram Alerts

Why Track Solana Wallets?

Wallet tracking is intelligence gathering. The most profitable traders in Solana consistently:

• Copy trades from verified on-chain "smart money" wallets
• Front-run obvious accumulation patterns
• Detect insider activity before public announcements
• Monitor their own portfolio in real-time

The Solana on-chain environment is particularly rich for this — high speed, low fees mean wallets trade frequently, providing dense signal.

Setting Up Helius for Real-Time Notifications

Helius is the best Solana RPC for wallet tracking — they offer webhooks that fire on any wallet transaction.

import requests
import json

HELIUS_API_KEY = 'your-helius-api-key'
WEBHOOK_URL = 'https://your-webhook-server.com/webhook'  # Your server URL

def create_wallet_webhook(wallet_addresses: list[str]) -> dict:
    """Set up Helius webhook to monitor multiple wallets"""
    
    webhook_data = {
        "webhookURL": WEBHOOK_URL,
        "transactionTypes": ["SWAP", "TRANSFER", "NFT_SALE"],  # What to track
        "accountAddresses": wallet_addresses,
        "webhookType": "enhanced",  # Rich transaction data
        "txnStatus": "all",
    }
    
    response = requests.post(
        f'https://api.helius.xyz/v0/webhooks?api-key={HELIUS_API_KEY}',
        json=webhook_data
    )
    
    webhook = response.json()
    print(f"✅ Webhook created: {webhook['webhookID']}")
    return webhook

# Track these wallets
wallets_to_track = [
    '9WzDXwBbmkg8ZTbNMqUxvQRAyrZzDsGYdLVL9zYtAWWM',  # Known Solana trader
    '7YttLkHDoNj9wyDur5pM1ejNaAvT9X4eqaYcHQqtj2G5',  # Another example
]

webhook = create_wallet_webhook(wallets_to_track)

Building the Webhook Server

from flask import Flask, request, jsonify
import asyncio
import telegram

app = Flask(__name__)

# Telegram bot setup
TELEGRAM_BOT_TOKEN = 'your-telegram-bot-token'
TELEGRAM_CHAT_ID = 'your-chat-id'

bot = telegram.Bot(token=TELEGRAM_BOT_TOKEN)

def format_transaction_alert(tx_data: dict) -> str:
    """Format a transaction into a readable Telegram message"""
    
    tx_type = tx_data.get('type', 'UNKNOWN')
    signature = tx_data.get('signature', '')[:20] + '...'
    fee = tx_data.get('fee', 0) / 1e9  # lamports → SOL
    
    msg = f"🔔 **New Transaction Detected**\n"
    msg += f"**Type:** {tx_type}\n"
    msg += f"**Sig:** `{signature}`\n"
    msg += f"**Fee:** {fee:.4f} SOL\n"
    
    # Handle swap events (most interesting for traders)
    if tx_type == 'SWAP':
        events = tx_data.get('events', {}).get('swap', {})
        if events:
            token_in = events.get('tokenInputs', [{}])[0]
            token_out = events.get('tokenOutputs', [{}])[0]
            
            amount_in = token_in.get('rawTokenAmount', {}).get('tokenAmount', 0)
            amount_out = token_out.get('rawTokenAmount', {}).get('tokenAmount', 0)
            
            mint_in = token_in.get('mint', 'unknown')
            mint_out = token_out.get('mint', 'unknown')
            
            # Resolve token names (simplified)
            known_tokens = {
                'So11111111111111111111111111111111111111112': 'SOL',
                'EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v': 'USDC',
                'Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB': 'USDT',
            }
            
            name_in = known_tokens.get(mint_in, mint_in[:8] + '...')
            name_out = known_tokens.get(mint_out, mint_out[:8] + '...')
            
            msg += f"\n💱 **SWAP DETECTED**\n"
            msg += f"{amount_in:,.2f} {name_in} → {amount_out:,.2f} {name_out}\n"
            
            # Check if this is a large swap (potentially interesting)
            if mint_in == 'EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v':  # USDC in
                usd_value = float(amount_in)
                if usd_value > 10000:
                    msg += f"💰 **BIG TRADE: ${usd_value:,.0f} USDC**\n"
    
    # Link to explorer
    msg += f"\n🔗 [View on Solscan](https://solscan.io/tx/{tx_data.get('signature', '')})"
    
    return msg

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    """Receive Helius webhook and send Telegram alert"""
    
    transactions = request.json
    
    if not transactions:
        return jsonify({'status': 'no data'}), 200
    
    for tx in transactions:
        try:
            message = format_transaction_alert(tx)
            
            # Send to Telegram
            asyncio.run(
                bot.send_message(
                    chat_id=TELEGRAM_CHAT_ID,
                    text=message,
                    parse_mode='Markdown'
                )
            )
            
        except Exception as e:
            print(f"Error processing transaction: {e}")
    
    return jsonify({'status': 'ok'}), 200

if __name__ == '__main__':
    app.run(port=5000)

WebSocket Alternative (No Server Required)

If you don't want to run a server, use WebSocket subscriptions directly:

import websockets
import asyncio
import json
import telegram

HELIUS_WS = f"wss://atlas-mainnet.helius-rpc.com?api-key={HELIUS_API_KEY}"

WALLETS_TO_TRACK = [
    '9WzDXwBbmkg8ZTbNMqUxvQRAyrZzDsGYdLVL9zYtAWWM',
]

async def track_wallets_ws():
    """Track wallets using WebSocket — no server needed"""
    
    bot = telegram.Bot(token=TELEGRAM_BOT_TOKEN)
    
    async with websockets.connect(HELIUS_WS) as ws:
        print("Connected to Helius WebSocket")
        
        # Subscribe to transaction logs for each wallet
        for wallet in WALLETS_TO_TRACK:
            subscribe_msg = {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "logsSubscribe",
                "params": [
                    {"mentions": [wallet]},
                    {"commitment": "confirmed"}
                ]
            }
            await ws.send(json.dumps(subscribe_msg))
            print(f"👀 Watching: {wallet[:8]}...")
        
        while True:
            msg = await ws.recv()
            data = json.loads(msg)
            
            if 'params' not in data:
                continue
            
            result = data['params']['result']
            
            # Check for swap-related logs
            logs = result['value']['logs']
            signature = result['value']['signature']
            
            is_swap = any('Instruction: Swap' in log or 'Program log: Ray' in log for log in logs)
            
            if is_swap:
                alert = f"🔄 Swap detected from tracked wallet!\n"
                alert += f"TX: https://solscan.io/tx/{signature}"
                
                await bot.send_message(
                    chat_id=TELEGRAM_CHAT_ID,
                    text=alert
                )
                print(f"Alert sent: {signature[:20]}")

# Run the tracker
asyncio.run(track_wallets_ws())

Finding Wallets Worth Tracking

def analyze_wallet_performance(wallet_address: str) -> dict:
    """Analyze a wallet's trading performance using Helius"""
    
    response = requests.get(
        f'https://api.helius.xyz/v0/addresses/{wallet_address}/transactions',
        params={
            'api-key': HELIUS_API_KEY,
            'limit': 100,
            'type': 'SWAP',
        }
    )
    
    transactions = response.json()
    
    wins = 0
    total_pnl_usd = 0
    trade_count = len(transactions)
    
    for tx in transactions:
        # Simplified PnL calculation
        # A real implementation would track entry/exit pairs
        pass
    
    return {
        'wallet': wallet_address,
        'total_trades': trade_count,
        'wins': wins,
        'estimated_pnl': total_pnl_usd,
    }

# Start with known alpha wallets from social media
# and backtest their performance before tracking live
alpha_wallets = [
    # Add addresses of traders you want to monitor
]

Setting Up Telegram Bot

# 1. Message @BotFather on Telegram
# 2. Create new bot: /newbot
# 3. Save the API token
# 4. Get your chat ID by messaging @userinfobot

# Install dependencies
pip install python-telegram-bot helius-sdk flask

Wallet tracking bots are among the most immediately useful tools for active DeFi traders. Pair this with automated copy trading (execute the same swap when you detect it) for a fully automated strategy — though always backtest first and manage size carefully.