Граббинг данных из социальных сетей (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 и bot добавленный на сервер:

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 недели.