Разработка скринера новых листингов

Разработка скринера новых листингов

Скринер новых листингов отслеживает появление новых торговых пар на крупных биржах и фильтрует их по заданным критериям. Это инструмент для поиска моментум-возможностей: новые листинги часто показывают резкий рост в первые часы из-за ажиотажа и притока ликвидности. Технически задача делится на: мониторинг источников листингов, парсинг данных, real-time алерты.

Источники данных о листингах

Биржевые API

Крупные биржи предоставляют REST API для получения списка торговых пар:

import asyncio
import httpx
from datetime import datetime

class ListingMonitor:
    EXCHANGES = {
        'binance': 'https://api.binance.com/api/v3/exchangeInfo',
        'bybit': 'https://api.bybit.com/v5/market/instruments-info?category=spot',
        'okx': 'https://www.okx.com/api/v5/public/instruments?instType=SPOT',
        'kucoin': 'https://api.kucoin.com/api/v1/symbols',
    }

    def __init__(self):
        self.known_pairs: dict[str, set] = {ex: set() for ex in self.EXCHANGES}

    async def check_new_listings(self, exchange: str) -> list[str]:
        url = self.EXCHANGES[exchange]
        async with httpx.AsyncClient() as client:
            response = await client.get(url, timeout=10)
            data = response.json()

        current_pairs = self.extract_pairs(exchange, data)
        new_pairs = current_pairs - self.known_pairs[exchange]

        if self.known_pairs[exchange]:  # не первый запуск
            for pair in new_pairs:
                await self.on_new_listing(exchange, pair)

        self.known_pairs[exchange] = current_pairs
        return list(new_pairs)

    def extract_pairs(self, exchange: str, data: dict) -> set:
        if exchange == 'binance':
            return {s['symbol'] for s in data['symbols'] if s['status'] == 'TRADING'}
        elif exchange == 'bybit':
            return {s['symbol'] for s in data['result']['list'] if s['status'] == 'Trading'}
        elif exchange == 'okx':
            return {s['instId'] for s in data['data'] if s['state'] == 'live'}
        return set()

Мониторинг официальных анонсов

Биржи публикуют анонсы листингов за несколько часов до старта торгов. Мониторинг RSS-фидов и официальных страниц:

class AnnouncementScraper:
    ANNOUNCEMENT_FEEDS = {
        'binance': 'https://www.binance.com/en/support/announcement/new-cryptocurrency-listing',
        'bybit': 'https://announcements.bybit.com/en-US/?category=new_crypto',
    }

    async def fetch_latest_announcements(self, exchange: str) -> list[Announcement]:
        # Используем RSS или scraping с rate limiting
        url = self.ANNOUNCEMENT_FEEDS[exchange]
        html = await self.fetch_page(url)
        announcements = self.parse_listings(html, exchange)

        new_ones = [a for a in announcements if a.id not in self.seen_ids]
        self.seen_ids.update(a.id for a in new_ones)

        return new_ones

Анонсы Binance выходят обычно за 1-4 часа до листинга. Это даёт время для подготовки.

Фильтрация и скоринг листингов

Алгоритм скоринга

class ListingScorer:
    def score(self, listing: NewListing) -> ListingScore:
        score = 0
        signals = []

        # Биржа — чем крупнее, тем выше потенциальный pump
        exchange_weights = {'binance': 10, 'okx': 7, 'bybit': 7, 'kucoin': 5}
        score += exchange_weights.get(listing.exchange, 3)

        # Категория токена
        if listing.category in ['defi', 'ai', 'rwa']:
            score += 5
            signals.append(f"Trending sector: {listing.category}")

        # Наличие листинга на нескольких биржах одновременно
        if listing.simultaneous_exchanges >= 2:
            score += 8
            signals.append(f"Multi-exchange listing: {listing.simultaneous_exchanges} exchanges")

        # Наличие фьючерсных пар (обычно листятся после спота — сигнал институционального интереса)
        if listing.has_futures:
            score += 5
            signals.append("Futures listing included")

        # Социальная активность (Twitter mentions за последние 24ч)
        if listing.twitter_mentions_24h > 10000:
            score += 4
            signals.append(f"High social activity: {listing.twitter_mentions_24h} mentions")

        # Маркет кап (если уже торгуется на других биржах)
        if listing.market_cap_usd:
            if listing.market_cap_usd < 50_000_000:
                score += 3  # малая кап = больше потенциал роста
                signals.append("Small cap token")

        return ListingScore(
            listing=listing,
            score=score,
            signals=signals,
            grade='A' if score >= 20 else 'B' if score >= 12 else 'C'
        )

Фильтры пользователя

interface ListingFilters {
  exchanges: string[];           // ['binance', 'bybit', 'okx']
  minScore: number;              // минимальный score для алерта
  categories: string[];          // ['defi', 'ai', 'gaming']
  minMarketCapUsd?: number;
  maxMarketCapUsd?: number;
  requireFutures: boolean;       // только если есть futures листинг
  minSocialMentions?: number;
}

Real-time алерты

class ListingAlertSystem:
    async def dispatch_alert(self, listing: NewListing, score: ListingScore, users: list):
        for user in users:
            if not self.matches_filters(listing, score, user.filters):
                continue

            # Telegram
            if user.telegram_enabled:
                message = self.format_telegram_message(listing, score)
                await self.telegram.send(user.telegram_chat_id, message)

            # Push уведомление
            if user.push_enabled:
                await self.push.send(user.push_token, {
                    'title': f"New listing: {listing.symbol}",
                    'body': f"{listing.exchange.upper()} • Score: {score.score} • {', '.join(score.signals[:2])}"
                })

    def format_telegram_message(self, listing: NewListing, score: ListingScore) -> str:
        grade_emoji = {'A': '🔥', 'B': '⚡', 'C': 'ℹ️'}.get(score.grade, '')
        return (
            f"{grade_emoji} **Новый листинг: {listing.symbol}**\n\n"
            f"📊 Биржа: {listing.exchange.upper()}\n"
            f"⏰ Старт торгов: {listing.trading_start.strftime('%H:%M UTC')}\n"
            f"💯 Score: {score.score}/30 (Grade {score.grade})\n\n"
            f"**Сигналы:**\n" + '\n'.join(f"• {s}" for s in score.signals) +
            f"\n\n[Торговать]({listing.trading_url})"
        )

Frontend таблица листингов

const ListingsTable: React.FC = () => {
  const [listings, setListings] = useState<Listing[]>([]);

  return (
    <table className="w-full">
      <thead>
        <tr>
          <th>Символ</th>
          <th>Биржа</th>
          <th>Старт торгов</th>
          <th>Score</th>
          <th>Сигналы</th>
          <th>Действие</th>
        </tr>
      </thead>
      <tbody>
        {listings.map(l => (
          <tr key={`${l.exchange}-${l.symbol}`}>
            <td className="font-bold">{l.symbol}</td>
            <td>{l.exchange.toUpperCase()}</td>
            <td>{formatDistanceToNow(l.tradingStart, {addSuffix: true})}</td>
            <td>
              <span className={`badge ${l.grade === 'A' ? 'bg-green-100 text-green-800' : 'bg-gray-100'}`}>
                {l.score}
              </span>
            </td>
            <td className="text-sm text-gray-600">{l.signals.slice(0,2).join(', ')}</td>
            <td>
              <a href={l.url} target="_blank" className="btn btn-sm">Открыть</a>
            </td>
          </tr>
        ))}
      </tbody>
    </table>
  );
};

Скринер листингов — это конкурентное преимущество для активных трейдеров. Важно мониторить не только спот, но и futures листинги, а также листинги на DEX (через Uniswap/PancakeSwap event мониторинг) — там возможности появляются ещё раньше, до попадания токена на централизованные биржи.