Граббінг даних із соціальних мереж (Twitter/X, Telegram, Discord)

Парсинг даних із соціальних мереж (Twitter/X, Telegram, Discord)

Крипто-спільнота існує в Twitter/X, Telegram та Discord. Там раніше за все з'являються сигнали: слив інсайду за кілька годин до оголошення, нарастання паніки при депегінгу, координація pump-групи, ранні обговорення уязвимостей. Для trading signals, sentiment analysis та мониторингу безпеки потрібен надійний pipeline сбору цих даних. У кожної платформи своя специфіка доступу.

Twitter/X: API та обходні шляхи

Офіційний API

Twitter API v2 — єдиний легальний шлях. Після реструктуризації X Corp тарифи стали агресивними:

• Free tier — тільки запис, читання обмежене. Для парсингу бесполезен.
• Basic ($100/місяць) — 10,000 постів/місяць. Едва хватає на мониторинг одного аккаунту.
• Pro ($5000/місяць) — 1M твітів/місяць, доступ до Filtered Stream. Реально для серйозної аналітики.
• Enterprise — Full Archive Search, Firehose. Ціна по запиту, десятки тисяч на місяць.

Для crypto sentiment на Pro тарифі:

import tweepy

client = tweepy.Client(bearer_token=BEARER_TOKEN)

# Filtered Stream для real-time мониторингу
class CryptoStreamListener(tweepy.StreamingClient):
    def on_tweet(self, tweet):
        if tweet.data:
            asyncio.create_task(self.process_tweet(tweet))

    async def process_tweet(self, tweet):
        await self.queue.put({
            "id": tweet.data.id,
            "text": tweet.data.text,
            "author_id": tweet.data.author_id,
            "created_at": tweet.data.created_at,
            "source": "twitter",
        })

stream = CryptoStreamListener(bearer_token=BEARER_TOKEN, queue=event_queue)
# Правила фільтрації (оператори AND/OR/NOT)
stream.add_rules(tweepy.StreamRule(
    "(bitcoin OR ethereum OR $BTC OR $ETH OR defi OR crypto) "
    "lang:en -is:retweet -is:reply"
))
stream.filter(tweet_fields=["created_at", "author_id", "public_metrics"])

Recent Search для історичних даних (до 7 днів на Pro):

# Пагінація через next_token
tweets = []
paginator = tweepy.Paginator(
    client.search_recent_tweets,
    query="$BTC OR bitcoin lang:en -is:retweet",
    tweet_fields=["created_at", "public_metrics", "author_id"],
    max_results=100,
    limit=10,  # 10 сторінок = 1000 твітів
)
async for tweet in paginator:
    tweets.append(tweet)

Альтернативи та обмеження

При обмеженому бюджеті — сторонні провайдери даних Twitter: Brandwatch, Sprinklr, Tweetbinder. Продають доступ до історичних даних та стримів по більш доступних цінах.

Scraping через unofficial API (без ключів) — порушення ToS, юридично рискованно для комерційних проектів. Технічно можливо через сессіонні cookies та reverse-engineered endpoints, але X Corp активно блокує.

Telegram: MTProto API

Telegram — основна платформа крипто-анонсів. Більшість проектів ведуть офіційні канали в Telegram.

Telethon: User account API

Telegram надає два типи API: Bot API (обмежений) та MTProto API (повний доступ через user account). Для парсингу каналів потрібен MTProto через бібліотеку Telethon:

from telethon import TelegramClient, events
from telethon.tl.types import Channel

API_ID = int(os.getenv("TELEGRAM_API_ID"))
API_HASH = os.getenv("TELEGRAM_API_HASH")

async def monitor_channels(channel_usernames: list[str]):
    async with TelegramClient("session", API_ID, API_HASH) as client:
        # Підписка на нові повідомлення
        @client.on(events.NewMessage(chats=channel_usernames))
        async def handler(event):
            msg = event.message
            await process_message({
                "channel": event.chat.username,
                "message_id": msg.id,
                "text": msg.text or "",
                "date": msg.date,
                "views": msg.views,
                "forwards": msg.forwards,
                "has_media": bool(msg.media),
            })

        # Отримання історії канала
        async def fetch_history(channel: str, limit: int = 1000):
            messages = []
            async for msg in client.iter_messages(channel, limit=limit):
                messages.append({
                    "id": msg.id,
                    "text": msg.text or "",
                    "date": msg.date,
                    "views": msg.views,
                })
            return messages

        await client.run_until_disconnected()

Важливо: Telethon використовує аккаунт реального користувача. Telegram блокує аккаунти при підозрілій активності (забагато запитів, парсинг забагато каналів). Використовуйте виділений аккаунт, дотримуйтесь rate limits, не парсьте закриті групи без дозволу.

Bot API підходить тільки якщо бот додан у групу/канал. Публічні канали недоступні боту без вступу.

Обнаружение аномалій у Telegram

Різкий ріст активності у каналі — сигнал:

async def detect_activity_spike(channel: str, window_minutes: int = 60):
    # Рахуємо повідомлення за останню годину vs попередню годину
    now = datetime.utcnow()
    hour_ago = now - timedelta(hours=1)
    two_hours_ago = now - timedelta(hours=2)

    recent_count = await db.count_messages(channel, hour_ago, now)
    prev_count = await db.count_messages(channel, two_hours_ago, hour_ago)

    if prev_count > 0:
        spike_ratio = recent_count / prev_count
        if spike_ratio > 3:  # У 3 рази більше обичного
            await alert(f"Activity spike in {channel}: {spike_ratio:.1f}x")

Discord: Bot API

Більшість DeFi-проектів використовують Discord для community. Там відбуваються технічні обговорення, early announcements, іноді — координація атак.

Discord Bot

Потрібен bot token із Discord Developer Portal та бот доданий на сервер:

import discord
from discord.ext import commands

intents = discord.Intents.default()
intents.message_content = True  # Privileged intent — потребує одобрення Discord
bot = commands.Bot(command_prefix="!", intents=intents)

TARGET_SERVERS = {
    "1234567890": ["general", "announcements", "alpha-calls"],
}

@bot.event
async def on_message(message: discord.Message):
    if message.author.bot:
        return

    guild_id = str(message.guild.id) if message.guild else None
    if guild_id not in TARGET_SERVERS:
        return

    channel_name = message.channel.name
    if channel_name not in TARGET_SERVERS[guild_id]:
        return

    await process_message({
        "platform": "discord",
        "server": message.guild.name,
        "channel": channel_name,
        "author": str(message.author),
        "content": message.content,
        "timestamp": message.created_at,
        "attachments": [a.url for a in message.attachments],
    })

Обмеження: message_content — привілейований intent. Discord потребує верифікації бота (100+ серверів) для його використання. На маленьких серверах працює без верифікації, на крупних — потрібно одобрення.

Історія повідомлень доступна через channel.history(), але тільки для серверів де бот уже присутній. Не можна отримати історію retroactively.

Зберігання та обробка

Єдина схема для повідомлень із всіх платформ:

CREATE TABLE social_messages (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    platform    TEXT NOT NULL,          -- 'twitter', 'telegram', 'discord'
    source_id   TEXT NOT NULL,          -- оригінальний ID повідомлення
    channel     TEXT,                   -- @username, channel_name, server/channel
    author      TEXT,
    content     TEXT NOT NULL,
    metadata    JSONB,                  -- platform-specific: views, likes, reactions
    captured_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at TIMESTAMPTZ,
    UNIQUE (platform, source_id)
);

CREATE INDEX idx_social_platform_channel ON social_messages (platform, channel, published_at DESC);
CREATE INDEX idx_social_content_fts ON social_messages USING gin(to_tsvector('english', content));

GIN індекс для full-text search — потрібен для пошуку упоминання токенів, ключових слів, адрес контрактів.

Sentiment аналіз

Для crypto-специфічного sentiment сирий текст потрібно обробити:

Keyword extraction: упоминання тикерів ($BTC, $ETH, $PEPE), адрес контрактів (0x...), протоколів.

Тональність: спеціалізовані моделі краще за загальні. FinBERT та CryptoBERT — fine-tuned BERT для фінансового/крипто контенту. Через HuggingFace:

from transformers import pipeline

sentiment = pipeline(
    "sentiment-analysis",
    model="ElKulako/cryptobert",
    device=0,  # GPU
)

def analyze_sentiment(text: str) -> dict:
    result = sentiment(text[:512])[0]  # BERT обмежений 512 токенами
    return {
        "label": result["label"],   # Bullish/Bearish/Neutral
        "score": result["score"],
    }

Volume-weighted sentiment — взвешиваємо sentiment по охопленню: твіт з 100k impressions важить більше ніж з 100. Для Telegram — по views повідомлення.

Операційні обмеження

Мониторинг соціальних мереж — юридично чутлива область. ToS більшості платформ забороняють комерційний scraping без офіційного API. Практичні обмеження:

• Twitter: офіційний API обов'язковий для будь-якого комерційного використання
• Telegram: парсинг публічних каналів через MTProto — сіра зона, прямо не запрещено
• Discord: тільки через офіційний Bot API, без scraping через web interface

Розробка pipeline для двох платформ (Twitter + Telegram) з sentiment analysis та зберіганням — 2-3 тижні. Додавання Discord та кастомних ML-моделей — ще 1-2 тижні.