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

Розроблення скринера нових листингів криптовалют

Скринер нових листингів відслідковує появу нових торговельних пар на крупних біржах та фільтрує їх за заданими критеріями. Це інструмент для пошуку моментум-можливостей: нові листинги часто показують різкий ріст у перші години через ажіотаж та приплив ліквідності. Технічно завдання ділиться на: моніторинг джерел листингів, парсинг даних, 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 за останні 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} **New listing: {listing.symbol}**\n\n"
            f"📊 Exchange: {listing.exchange.upper()}\n"
            f"⏰ Trading starts: {listing.trading_start.strftime('%H:%M UTC')}\n"
            f"💯 Score: {score.score}/30 (Grade {score.grade})\n\n"
            f"**Signals:**\n" + '\n'.join(f"• {s}" for s in score.signals) +
            f"\n\n[Trade]({listing.trading_url})"
        )

Frontend таблиця листингів

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

  return (
    <table className="w-full">
      <thead>
        <tr>
          <th>Symbol</th>
          <th>Exchange</th>
          <th>Trading Starts</th>
          <th>Score</th>
          <th>Signals</th>
          <th>Action</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>
  );
};

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