Граббінг даних про лістинги/деліcтинги з бірж

Парсинг даних про листинги/делістинги з бірж

Об'явлення про листинг нового токена на Binance історично дає ріст ціни на 20–100% в першові хвилини. Делістинг — симетричне падіння. Швидкість отримання цієї інформації напрямку конвертується в P&L для торговельних систем. Задача парсингу листингів — це задача latency, а не просто сбору даних.

Джерела даних про листинги

Біржи об'являють про листинги через кілька каналів з різною затримкою:

Найшвидший спосіб дізнатися про листинг часто — не офіційне об'ява, а поява нового торгового інструмента в API.

Моніторинг API на появу нових інструментів

Більшість бірж мають endpoint зі списком усіх торгових пар. Періодичне порівняння поточного списку з попереднім дає дельту — нові листинги:

import asyncio
import aiohttp
from datetime import datetime

class ListingMonitor:
    def __init__(self):
        self.known_symbols: dict[str, set] = {}
        self.poll_interval = 30  # секунд
    
    async def get_binance_symbols(self, session: aiohttp.ClientSession) -> set:
        async with session.get(
            "https://api.binance.com/api/v3/exchangeInfo",
            timeout=aiohttp.ClientTimeout(total=5)
        ) as resp:
            data = await resp.json()
            return {
                s['symbol'] 
                for s in data['symbols'] 
                if s['status'] == 'TRADING'
            }
    
    async def check_for_new_listings(self, exchange: str, session):
        current = await self.get_symbols(exchange, session)
        previous = self.known_symbols.get(exchange, set())
        
        new_listings = current - previous
        delistings = previous - current
        
        if new_listings:
            for symbol in new_listings:
                await self.on_new_listing(exchange, symbol)
        
        if delistings:
            for symbol in delistings:
                await self.on_delisting(exchange, symbol)
        
        self.known_symbols[exchange] = current
    
    async def on_new_listing(self, exchange: str, symbol: str):
        event = {
            'type': 'listing',
            'exchange': exchange,
            'symbol': symbol,
            'detected_at': datetime.utcnow().isoformat(),
        }
        await self.notify(event)

Інтервал polling повинен балансувати швидкість виявлення та rate limits: 15–30 секунд — розумний компромісс. Для Binance futures — окремий endpoint /fapi/v1/exchangeInfo.

Парсинг офіційних сторінок об'янь

Біржи публікують текстові об'яви про листинги на своїх сайтах. Парсинг HTML-сторінок — додатковий джерело:

from bs4 import BeautifulSoup
import re

async def scrape_binance_announcements(session: aiohttp.ClientSession) -> list:
    url = "https://www.binance.com/en/support/announcement/new-cryptocurrency-listing"
    
    headers = {
        'User-Agent': 'Mozilla/5.0 (compatible; research bot)',
        'Accept-Language': 'en-US,en;q=0.9',
    }
    
    async with session.get(url, headers=headers) as resp:
        html = await resp.text()
    
    soup = BeautifulSoup(html, 'html.parser')
    announcements = []
    
    for article in soup.select('a[href*="/support/announcement/"]'):
        title = article.get_text(strip=True)
        href = article.get('href')
        
        # Шукаємо упоминание тікерів у заголовку
        tickers = re.findall(r'\(([A-Z]{2,10})\)', title)
        
        if tickers:
            announcements.append({
                'title': title,
                'url': href,
                'tickers': tickers,
                'scraped_at': datetime.utcnow().isoformat(),
            })
    
    return announcements

Проблема: Binance та Bybit активно використовують JavaScript rendering та anti-bot захист (Cloudflare). Playwright або Puppeteer для headless Chrome — стандартне рішення для JS-heavy сторінок:

from playwright.async_api import async_playwright

async def scrape_with_playwright(url: str) -> str:
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        await page.goto(url, wait_until='networkidle')
        content = await page.content()
        
        await browser.close()
        return content

RSS та Telegram

Деякі біржи (Kraken, KuCoin) публікують об'яви через RSS — найпростіший та найнадійніший джерело:

import feedparser

def parse_exchange_rss(feed_url: str) -> list:
    feed = feedparser.parse(feed_url)
    listings = []
    
    for entry in feed.entries:
        if any(word in entry.title.lower() 
               for word in ['listing', 'adds', 'new trading pair']):
            listings.append({
                'title': entry.title,
                'link': entry.link,
                'published': entry.published,
            })
    
    return listings

Telegram-моніторинг офіційних каналів бірж через Telethon:

from telethon import TelegramClient, events

client = TelegramClient('session', api_id, api_hash)

@client.on(events.NewMessage(chats=['@binance', '@kucoincom']))
async def handle_announcement(event):
    text = event.message.text
    if 'listing' in text.lower() or 'will list' in text.lower():
        tickers = re.findall(r'\$([A-Z]{2,10})', text)
        await process_listing_announcement(tickers, text, event.date)

Зберігання та дедупліциія

Листинг може бути виявлений через кілька каналів одночасно — потрібна дедупліциія:

CREATE TABLE listing_events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    exchange TEXT NOT NULL,
    symbol TEXT NOT NULL,
    event_type TEXT NOT NULL, -- 'listing' | 'delisting' | 'suspension'
    detected_at TIMESTAMPTZ NOT NULL,
    source TEXT NOT NULL,    -- 'api_poll' | 'announcement' | 'rss' | 'telegram'
    raw_data JSONB,
    UNIQUE(exchange, symbol, event_type, date_trunc('hour', detected_at))
);

Нотифікації при подій — через webhook (Slack, Discord, Telegram бот, власний endpoint). Для трейдинг-систем — Kafka топік exchange.listings для downstream consumers.

Обмеження та точність

Ні один джерело не дає 100% покриття та нульову затримку. Комбінація API polling + RSS + Telegram дає ~95% подій в перші 5 хвилин для топ-10 бірж. Мілкі біржи — тільки через парсинг сторінок.

Ложні спрацювання: поява символа в API не завжди означає відкриття торгів — біржа може додати пару у "предторговий" режим. Потрібна перевірка статусу (status: TRADING vs PRE_DELIVERING).

Розробка системи моніторингу листингів для 10–15 бірж з multi-channel оповіщенням та зберіганням історії — 2–3 тижні.