Граббинг данных о листингах/делистингах с бирж

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

Объявление о листинге нового токена на 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 недели.