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

Ви запускаєте торгового бота, який має реагувати на лістинги нових токенів. Але затримка у виявленні — 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 тижні.