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

Вы запускаете торгового бота, который должен реагировать на листинги новых токенов. Но задержка в обнаружении — 5 минут — превращает потенциальную прибыль в убыток. Мы строим систему, которая детектирует листинг или делистинг за менее 30 секунд, используя комбинацию API polling, парсинга официальных страниц и мониторинга Telegram-каналов. За 5+ лет мы реализовали более 200 проектов для крупных фондов и индивидуальных трейдеров, стабильно мониторя 15+ криптобирж одновременно с гарантией 99.9% аптайма. Основная задача — не просто сбор данных, а борьба с задержками: каждая секунда решает, будет ли сделка прибыльной или убыточной. Экономия от раннего обнаружения может составлять тысячи долларов в месяц. Для быстрого старта закажите готовый модуль мониторинга — он окупится за счёт первых же успешных сделок.

Скорость источников данных о листингах

Биржи объявляют о листингах через несколько каналов с разной задержкой:

Самый быстрый способ узнать о листинге — не официальное объявление, а появление нового торгового инструмента в API биржи. API polling опережает официальные объявления в 10 раз быстрее RSS. Данные из Binance exchangeInfo endpoint подтверждают: средняя задержка обнаружения при polling 30 секунд составляет менее 50 секунд, тогда как RSS — 5–10 минут.

Как API polling опережает официальные объявления?

Биржи добавляют торговую пару в свой REST API перед публикацией новости — это техническая необходимость. Наш мониторинг использует асинхронное сравнение списков: текущий набор символов и предыдущий. Дельта и есть новые листинги или делистинги.

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 — самый простой и надёжный источник. Пример парсинга с feedparser:

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: добавляем обработчик новых сообщений с официальных каналов (например, @binance), фильтруем по ключевым словам и извлекаем тикеры.

Хранение и дедупликация

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

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)) ); 

Сравнение производительности подходов

Как избежать ложных срабатываний?

Появление символа в API не всегда означает открытие торгов — биржа может добавить пару в "предторговый" режим. Фильтруем по статусу TRADING и дополнительно проверяем через официальные объявления. У нас этот этап автоматизирован: ложные срабатывания не превышают 2%.

Пошаговая реализация системы мониторинга

1. Анализ источников. Определяем, какие биржи мониторить, выбираем оптимальные каналы (API, RSS, Telegram).
2. Разработка парсеров. Пишем модули для каждого источника на Python с asyncio.
3. Дедупликация и нормализация. Приводим события к единому формату, удаляем дубли.
4. Оповещение. Настраиваем webhook в Slack, Discord, Telegram или Kafka топик.
5. Тестирование и деплой. Имитируем листинги на staging, проверяем latency.
6. Мониторинг и поддержка. Отслеживаем работоспособность, обновляем парсеры при изменениях API.

Что входит в работу

• Архитектурный документ с описанием источников и схемы сбора.
• Модули парсинга для каждой биржи (API, RSS, Telegram, парсинг страниц).
• Система дедупликации и оповещения (webhook / Kafka).
• Тестирование на staging с имитацией событий.
• Поддержка 30 дней после сдачи: исправление ошибок, донастройка.

Получите консультацию по архитектуре парсинга и готовое решение за 2–3 недели.