How to Build a Crypto Whale Tracker with On-Chain and Off-Chain Data

Whale watching in crypto is a legitimate trading edge. When wallets holding millions in tokens start moving, the market often follows. By combining on-chain transaction data with off-chain signals like exchange announcements and social media, you can build an early warning system.

Here's how to build a complete whale tracker in Python.

What Makes a Whale?

In crypto, a "whale" is typically:

• Top 100 holders of any given token
• Wallets with >$1M in a single asset
• Wallets associated with known funds, exchanges, or early investors

Setting Up

pip install web3 requests pandas websockets

On-Chain: Monitoring Large Transactions

from web3 import Web3
from datetime import datetime
import time

class WhaleMonitor:
    def __init__(self, rpc_url: str, min_value_eth: float = 100.0):
        self.w3 = Web3(Web3.HTTPProvider(rpc_url))
        self.min_value_wei = Web3.to_wei(min_value_eth, "ether")
        self.known_wallets = {
            "0x28C6c06298d514Db089934071355E5743bf21d60": "Binance Hot Wallet",
            "0x21a31Ee1afC51d94C2eFcCAa2092aD1028285549": "Binance Cold",
            "0x1B3cB81E51011b549d78bf720b0d924ac763A7C2": "Coinbase Commerce",
            "0xDFd5293D8e347dFe59E90eFd55b2956a1343963d": "Ethereum Foundation",
        }

    def scan_recent_blocks(self, num_blocks: int = 10) -> list[dict]:
        latest = self.w3.eth.block_number
        whale_txs = []

        for block_num in range(latest - num_blocks, latest + 1):
            block = self.w3.eth.get_block(block_num, full_transactions=True)

            for tx in block.transactions:
                if tx.value >= self.min_value_wei:
                    whale_txs.append({
                        "hash": tx.hash.hex(),
                        "from": tx["from"],
                        "to": tx["to"],
                        "value_eth": float(Web3.from_wei(tx.value, "ether")),
                        "block": block_num,
                        "timestamp": datetime.fromtimestamp(block.timestamp).isoformat(),
                        "from_label": self.known_wallets.get(tx["from"], "Unknown"),
                        "to_label": self.known_wallets.get(tx["to"], "Unknown"),
                    })

        return whale_txs

ERC-20 Token Transfer Monitoring

ETH transfers are just the start. Most whale activity involves tokens:

    TRANSFER_TOPIC = Web3.keccak(text="Transfer(address,address,uint256)").hex()

    def monitor_token_transfers(self, token_address: str, 
                                 min_value: float,
                                 decimals: int = 18,
                                 blocks: int = 100) -> list[dict]:
        latest = self.w3.eth.block_number

        logs = self.w3.eth.get_logs({
            "address": Web3.to_checksum_address(token_address),
            "topics": [self.TRANSFER_TOPIC],
            "fromBlock": latest - blocks,
            "toBlock": latest
        })

        large_transfers = []
        min_raw = int(min_value * (10 ** decimals))

        for log in logs:
            value = int(log["data"].hex(), 16)

            if value >= min_raw:
                sender = "0x" + log["topics"][1].hex()[-40:]
                receiver = "0x" + log["topics"][2].hex()[-40:]

                large_transfers.append({
                    "token": token_address,
                    "from": sender,
                    "to": receiver,
                    "value": value / (10 ** decimals),
                    "tx_hash": log["transactionHash"].hex(),
                    "block": log["blockNumber"],
                })

        return large_transfers

Off-Chain: Exchange and Social Signals

import requests

class OffChainTracker:
    def __init__(self, scraper_api_key: str):
        self.api_key = scraper_api_key

    def get_exchange_volumes(self, token: str) -> dict:
        url = f"https://api.coingecko.com/api/v3/coins/{token}/tickers"
        resp = requests.get(url, timeout=30)
        data = resp.json()

        volumes = {}
        for ticker in data.get("tickers", []):
            exchange = ticker["market"]["name"]
            vol = ticker.get("converted_volume", {}).get("usd", 0)
            volumes[exchange] = volumes.get(exchange, 0) + vol

        return dict(sorted(volumes.items(), key=lambda x: x[1], reverse=True)[:10])

    def scrape_whale_alerts(self) -> list[dict]:
        from bs4 import BeautifulSoup
        params = {"api_key": self.api_key, "url": "https://whale-alert.io/recent-transactions", "render": "true"}
        resp = requests.get("https://api.scraperapi.com", params=params, timeout=60)
        soup = BeautifulSoup(resp.text, "html.parser")

        return [
            {"amount": r.select_one(".amount").get_text(strip=True),
             "asset": r.select_one(".asset").get_text(strip=True)}
            for r in soup.select(".transaction-row") if r.select_one(".amount")
        ]

Alert System

def run_whale_tracker(rpc_url: str, api_key: str):
    onchain = WhaleMonitor(rpc_url, min_value_eth=50)
    offchain = OffChainTracker(api_key)

    print("Starting whale tracker...")

    while True:
        whale_txs = onchain.scan_recent_blocks(num_blocks=5)

        for tx in whale_txs:
            direction = "Unknown"
            if "Binance" in tx["to_label"] or "Coinbase" in tx["to_label"]:
                direction = "EXCHANGE DEPOSIT (potential sell)"
            elif "Binance" in tx["from_label"] or "Coinbase" in tx["from_label"]:
                direction = "EXCHANGE WITHDRAWAL (accumulation)"

            print(f"WHALE ALERT: {tx['value_eth']:.2f} ETH")
            print(f"  From: {tx['from_label']} -> To: {tx['to_label']}")
            print(f"  Signal: {direction}")
            print(f"  TX: {tx['hash']}")

        time.sleep(15)

run_whale_tracker("https://eth.llamarpc.com", "YOUR_SCRAPERAPI_KEY")

Scaling with Proxy Infrastructure

For scraping multiple whale alert services and exchange data simultaneously, ScraperAPI handles the proxy rotation. Use ThorData for residential proxies when accessing geo-restricted exchange APIs. Monitor your data pipeline with ScrapeOps.

What's Next

• Add Telegram/Discord bot integration for real-time alerts
• Build historical whale behavior profiles
• Correlate whale movements with price action for backtesting
• Track NFT whale wallets for collection launch signals

Whale tracking is one of the few crypto tools with demonstrable alpha. Build it, use it, profit from it.