Разработка системы алертов по whale-транзакциям

Разработка системы алертов по whale-транзакциям

Whale-транзакции — крупные перемещения криптовалюты, способные влиять на рыночную динамику. Перевод тысяч BTC на биржу может сигнализировать о готовящихся продажах; перевод с биржи на холодное хранение — о долгосрочном накоплении. Система мониторинга таких транзакций даёт трейдерам информационное преимущество.

Источники данных

On-chain мониторинг — подключение к blockchain nodes и отслеживание мемпула и подтверждённых транзакций в реальном времени.

Whale Alert API — готовый сервис с webhooks и REST API. Покрывает BTC, ETH, крупные токены и стейблкоины. Самый быстрый способ начать.

Blockchain-specific APIs — Etherscan API (Ethereum), Blockcypher (BTC/ETH), QuickNode (несколько сетей).

Whale Alert интеграция

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

Для прямого мониторинга без сторонних API — подключение к Bitcoin node через ZeroMQ:

import zmq.asyncio
import asyncio
import struct

class BitcoinTxMonitor:
    """Подключение к Bitcoin node через ZMQ для мониторинга транзакций"""

    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):
        # Классифицируем транзакцию
        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)

Классификация адресов

Ценность whale-алертов в контексте: "10,000 ETH transferred" — интересно; "10,000 ETH transferred from Binance cold wallet to FTX hot wallet" — очень интересно.

class AddressClassifier:
    # Базы данных известных адресов
    EXCHANGE_ADDRESSES = {
        "0x3f5ce5fbfe3e9af3971dd833d26ba9b5c936f0be": ("binance", "hot_wallet"),
        "0xd551234ae421e3bcba99a0da6d736074f22192ff": ("binance", "cold_wallet"),
        # ... тысячи адресов крупных платформ
    }

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

    async def classify(self, address: str, blockchain: str) -> AddressInfo:
        # 1. Проверяем локальную базу
        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 для неизвестных адресов
        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)

Форматирование алертов

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 = ""
    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}"""

Фильтры для снижения шума

Без фильтрации whale-алерты превращаются в поток спама. Настройки:

• Минимальный порог в USD: $500K, $1M, $5M, $10M на выбор
• Конкретные активы: только BTC и ETH, или все top-20
• Направление: только депозиты на биржи, только выводы, или всё
• Исключить известные переводы: внутренние переводы между кошельками одной биржи
• Cooldown: не чаще одного алерта по символу за N минут