Розробка системи алертів за 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-алерти перетворюються в потік спаму. Налаштування:

• Мінімальний поріг у доларах США: $500K, $1M, $5M, $10M на вибір
• Конкретні активи: лише BTC та ETH, або всі top-20
• Напрямок: лише депозити на біржі, лише виводи, або все
• Виключити відомі переводи: внутрішні переводи між гаманцями однієї біржі
• Cooldown: не частіше одного алерту за символ за N хвилин