Whale transaction alerts system development

Development of Whale Transaction Alerts System

Whale transactions — large movements of cryptocurrency capable of influencing market dynamics. Transfer of thousands of BTC to an exchange may signal upcoming sales; transfer from exchange to cold storage — long-term accumulation. Monitoring such transactions gives traders an information advantage.

Data Sources

On-chain monitoring — connection to blockchain nodes and real-time tracking of mempool and confirmed transactions.

Whale Alert API — ready-made service with webhooks and REST API. Covers BTC, ETH, major tokens and stablecoins. The fastest way to get started.

Blockchain-specific APIs — Etherscan API (Ethereum), Blockcypher (BTC/ETH), QuickNode (multiple networks).

Whale Alert Integration

import httpx
from typing import Optional

class WhaleAlertClient:
    BASE_URL = "https://api.whale-alert.io/v1"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = httpx.AsyncClient()

    async def get_recent_transactions(
        self,
        min_value: int = 500_000,  # USD
        blockchain: Optional[str] = None,
        limit: int = 100,
    ) -> list[dict]:
        params = {
            "api_key": self.api_key,
            "min_value": min_value,
            "limit": limit,
        }
        if blockchain:
            params["blockchain"] = blockchain

        resp = await self.session.get(
            f"{self.BASE_URL}/transactions",
            params=params
        )
        resp.raise_for_status()
        return resp.json().get("transactions", [])

    async def get_transaction(self, tx_hash: str, blockchain: str) -> dict:
        resp = await self.session.get(
            f"{self.BASE_URL}/transaction/{blockchain}/{tx_hash}",
            params={"api_key": self.api_key}
        )
        return resp.json()

On-chain Bitcoin Monitoring

For direct monitoring without third-party APIs — connection to Bitcoin node via ZeroMQ:

import zmq.asyncio
import asyncio
import struct

class BitcoinTxMonitor:
    """Connection to Bitcoin node via ZMQ for transaction monitoring"""

    def __init__(self, zmq_address: str = "tcp://127.0.0.1:28332"):
        self.address = zmq_address

    async def monitor(self, min_btc: float = 100.0):
        context = zmq.asyncio.Context()
        socket = context.socket(zmq.SUB)
        socket.connect(self.address)
        socket.setsockopt(zmq.SUBSCRIBE, b"rawtx")

        while True:
            topic, data, seq = await socket.recv_multipart()
            tx = self.parse_transaction(data)

            total_output = sum(out.value for out in tx.outputs) / 1e8
            if total_output >= min_btc:
                await self.on_large_transaction(tx, total_output)

    async def on_large_transaction(self, tx, btc_amount: float):
        # Classify transaction
        classification = await self.classify_transaction(tx)

        whale_tx = WhaleTransaction(
            tx_hash=tx.hash,
            blockchain="bitcoin",
            amount_btc=btc_amount,
            amount_usd=btc_amount * await self.get_btc_price(),
            from_type=classification.from_type,  # exchange, whale_wallet, unknown
            to_type=classification.to_type,
            from_address=tx.inputs[0].address,
            to_address=tx.outputs[0].address,
            timestamp=datetime.utcnow(),
        )

        await self.alert_engine.process_whale_transaction(whale_tx)

Address Classification

The value of whale alerts is in context: "10,000 ETH transferred" — interesting; "10,000 ETH transferred from Binance cold wallet to FTX hot wallet" — very interesting.

class AddressClassifier:
    # Databases of known addresses
    EXCHANGE_ADDRESSES = {
        "0x3f5ce5fbfe3e9af3971dd833d26ba9b5c936f0be": ("binance", "hot_wallet"),
        "0xd551234ae421e3bcba99a0da6d736074f22192ff": ("binance", "cold_wallet"),
        # ... thousands of addresses of major platforms
    }

    def __init__(self, chainalysis_client=None):
        self.chainalysis = chainalysis_client

    async def classify(self, address: str, blockchain: str) -> AddressInfo:
        # 1. Check local database
        if address.lower() in self.EXCHANGE_ADDRESSES:
            name, wallet_type = self.EXCHANGE_ADDRESSES[address.lower()]
            return AddressInfo(address=address, entity=name, type=wallet_type, confidence=1.0)

        # 2. Chainalysis API for unknown addresses
        if self.chainalysis:
            result = await self.chainalysis.identify_address(address, blockchain)
            if result.entity:
                return AddressInfo(
                    address=address,
                    entity=result.entity,
                    type=result.category,
                    confidence=result.confidence
                )

        return AddressInfo(address=address, entity="unknown", type="unknown", confidence=0.0)

Alert Formatting

def format_whale_alert(tx: WhaleTransaction) -> str:
    amount_str = f"${tx.amount_usd:,.0f} ({tx.amount_crypto:.0f} {tx.blockchain.upper()})"

    from_label = tx.from_entity or f"{tx.from_address[:6]}...{tx.from_address[-4:]}"
    to_label = tx.to_entity or f"{tx.to_address[:6]}...{tx.to_address[-4:]}"

    # Signal interpretation
    signal = ""
    if tx.from_type == "exchange" and tx.to_type in ("unknown", "whale_wallet"):
        signal = "🐂 Withdrawal from exchange (bullish signal)"
    elif tx.from_type in ("unknown", "whale_wallet") and tx.to_type == "exchange":
        signal = "🐻 Deposit to exchange (potential selling pressure)"
    elif tx.from_type == "exchange" and tx.to_type == "exchange":
        signal = "🔄 Exchange-to-exchange transfer"

    return f"""🐋 Whale Alert

{amount_str}
{from_label} → {to_label}

{signal}
🔗 {tx.explorer_url}"""

Filters to Reduce Noise

Without filtering, whale alerts turn into spam. Settings:

• Minimum USD threshold: $500K, $1M, $5M, $10M to choose
• Specific assets: only BTC and ETH, or all top-20
• Direction: only exchange deposits, only withdrawals, or all
• Exclude known transfers: internal transfers between one exchange's wallets
• Cooldown: no more than one alert per symbol per N minutes