On-Chain Whale Tracker Bot: Build Your Own with Python and Etherscan

Why Whale Tracking Works

When a wallet holding 10,000+ BTC starts moving funds to an exchange, it often precedes a sell-off. When a known DeFi whale moves $50M into Aave, it signals confidence in the market. On-chain data is the most transparent market data in existence — every trade is public, permanent, and machine-readable.

Institutional traders have teams monitoring this. Individual traders now have Python.

What to Track

Tier 1: Exchange Flow (Most Actionable)

• Large deposits to Binance/Coinbase/Kraken → potential sell pressure
• Large withdrawals from exchanges → potential accumulation
• Exchange reserve levels declining over weeks → bullish signal

Tier 2: Whale Wallet Movements

• Known early Bitcoin wallets moving for the first time in years
• Satoshi-era wallets (technically impossible to track reliably, but monitored obsessively)
• Publicized fund wallets (a16z, Pantera, Jump Trading addresses)

Tier 3: DeFi Activity

• Large Uniswap swaps (>$500K) often move price significantly
• Aave large borrows can signal leveraged positions
• Curve pool composition shifts signal stablecoin flows

Setting Up: Etherscan + Alchemy

import requests
import asyncio
from web3 import Web3
from typing import Optional

ETHERSCAN_API_KEY = 'YOUR_KEY'
ALCHEMY_URL = f"https://eth-mainnet.g.alchemy.com/v2/{ALCHEMY_KEY}"

w3 = Web3(Web3.HTTPProvider(ALCHEMY_URL))

# Known exchange deposit addresses (public info)
EXCHANGE_ADDRESSES = {
    '0x28c6c06298d514db089934071355e5743bf21d60': 'Binance Hot Wallet',
    '0x21a31ee1afc51d94c2efccaa2092ad1028285549': 'Binance Cold Wallet',
    '0x8315177ab297ba92a06054ce80a67ed4dbd7ed3a': 'Coinbase Hot Wallet',
    '0x71660c4005ba85c37ccec55d0c4493e66fe775d3': 'Coinbase 2',
}

def check_exchange_address(address: str) -> Optional[str]:
    """Check if address belongs to a known exchange"""
    return EXCHANGE_ADDRESSES.get(address.lower())

Building the Whale Alert Bot

from decimal import Decimal
import json
from datetime import datetime

MIN_ETH_ALERT = 500    # Alert on 500+ ETH transfers
MIN_USD_ALERT = 1_000_000  # Or $1M+

class WhaleTracker:
    def __init__(self, w3: Web3, alert_callback=None):
        self.w3 = w3
        self.alert_callback = alert_callback or self.default_alert
        self.seen_txs = set()
    
    async def process_block(self, block_number: int):
        """Process all transactions in a block for whale activity"""
        block = self.w3.eth.get_block(block_number, full_transactions=True)
        
        for tx in block.transactions:
            if tx.hash.hex() in self.seen_txs:
                continue
            self.seen_txs.add(tx.hash.hex())
            
            # Check ETH transfer size
            value_eth = Decimal(str(self.w3.from_wei(tx.value, 'ether')))
            
            if value_eth >= MIN_ETH_ALERT:
                await self.analyze_large_transfer(tx, value_eth)
    
    async def analyze_large_transfer(self, tx, value_eth: Decimal):
        """Deep analysis of a large transfer"""
        from_label = check_exchange_address(tx['from']) or await self.label_address(tx['from'])
        to_label = check_exchange_address(tx['to']) or await self.label_address(tx['to'])
        
        # Determine significance
        significance = self.assess_significance(from_label, to_label)
        
        if significance:
            await self.alert_callback({
                'type': 'ETH_TRANSFER',
                'amount_eth': float(value_eth),
                'from': tx['from'],
                'from_label': from_label,
                'to': tx['to'],
                'to_label': to_label,
                'tx_hash': tx.hash.hex(),
                'significance': significance,
                'timestamp': datetime.now().isoformat()
            })
    
    def assess_significance(self, from_label: str, to_label: str) -> Optional[str]:
        """Determine if this transfer is worth alerting on"""
        
        if to_label and 'Exchange' in to_label:
            return f"⚠️ INFLOW to {to_label} — potential sell pressure"
        
        if from_label and 'Exchange' in from_label:
            return f"📈 OUTFLOW from {from_label} — potential accumulation"
        
        if not from_label and not to_label:
            return "🐋 Unknown wallet → Unknown wallet (large whale move)"
        
        return None
    
    async def label_address(self, address: str) -> str:
        """Try to label an address using Etherscan"""
        url = f"https://api.etherscan.io/api?module=account&action=txlist&address={address}&limit=1&apikey={ETHERSCAN_API_KEY}"
        
        try:
            response = requests.get(url, timeout=5).json()
            # Check if it's a contract
            code = self.w3.eth.get_code(self.w3.to_checksum_address(address))
            if len(code) > 2:
                return "Smart Contract"
        except:
            pass
        
        return ""
    
    async def default_alert(self, event: dict):
        print(f"\n🐋 WHALE ALERT [{event['type']}]")
        print(f"   Amount: {event['amount_eth']:,.1f} ETH")
        print(f"   From: {event.get('from_label') or event['from'][:12]}...")
        print(f"   To: {event.get('to_label') or event['to'][:12]}...")
        print(f"   {event['significance']}")
        print(f"   Tx: https://etherscan.io/tx/{event['tx_hash']}")

Monitoring DEX Whale Swaps

Large Uniswap swaps often move prices. Track them:

from web3 import Web3

UNISWAP_V3_SWAP_TOPIC = w3.keccak(text='Swap(address,address,int256,int256,uint160,uint128,int24)').hex()

def monitor_large_swaps(min_usd: float = 500_000):
    """Monitor Uniswap V3 for large swaps via event logs"""
    
    # Get recent blocks
    latest = w3.eth.block_number
    logs = w3.eth.get_logs({
        'fromBlock': latest - 10,
        'toBlock': latest,
        'topics': [UNISWAP_V3_SWAP_TOPIC],
    })
    
    for log in logs:
        # Parse the swap event
        swap_data = decode_uniswap_swap(log)
        if swap_data and swap_data['amount_usd'] >= min_usd:
            print(f"🔄 Large Swap: ${swap_data['amount_usd']:,.0f}")
            print(f"   {swap_data['token_in']} → {swap_data['token_out']}")
            print(f"   Pool: {log['address'][:12]}...")

Connecting to Telegram for Real-Time Alerts

import telegram
import asyncio

bot = telegram.Bot(token=TELEGRAM_BOT_TOKEN)

async def send_whale_alert_telegram(event: dict):
    """Send formatted whale alert to Telegram"""
    
    emoji_map = {
        'INFLOW': '🔴',
        'OUTFLOW': '🟢',
        'UNKNOWN': '🟡'
    }
    
    message = f"""
{emoji_map.get(event.get('flow', 'UNKNOWN'))} **Whale Alert**

💰 Amount: `{event['amount_eth']:,.0f} ETH`
📤 From: `{event.get('from_label') or event['from'][:12]}...`
📥 To: `{event.get('to_label') or event['to'][:12]}...`

{event['significance']}

🔗 [View on Etherscan](https://etherscan.io/tx/{event['tx_hash']})
"""
    
    await bot.send_message(
        chat_id=TELEGRAM_CHAT_ID,
        text=message,
        parse_mode='Markdown'
    )

Combining On-Chain Signals with Price Action

Whale data is most useful when combined with price context:

def generate_composite_signal(whale_events: list, price_data: dict) -> str:
    """Combine whale activity with technical signals"""
    
    # Count inflows vs outflows in last hour
    inflows = sum(1 for e in whale_events if 'INFLOW' in e.get('significance', ''))
    outflows = sum(1 for e in whale_events if 'OUTFLOW' in e.get('significance', ''))
    
    # Get technical trend
    rsi = calculate_rsi(price_data['close'])
    above_200ma = price_data['close'][-1] > calculate_ma(price_data['close'], 200)
    
    # Generate signal
    if outflows > inflows * 2 and rsi > 70:
        return "STRONG SELL: Exchange inflows + overbought RSI"
    elif inflows > outflows * 2 and rsi < 40:
        return "STRONG BUY: Exchange outflows + oversold RSI"
    else:
        return f"NEUTRAL: {inflows} inflows, {outflows} outflows, RSI {rsi:.0f}"

Whale tracking is one of the most powerful edges available to retail traders. The on-chain data is free, the tools are open source, and most traders haven't automated it yet.