Обучение NLP-модели для анализа Telegram-каналов

Обучение NLP-модели для анализа Telegram-каналов

Telegram — центральная среда крипто-коммуникаций. Крупные инфлюенсеры ведут каналы с сотнями тысяч подписчиков. Анонимные аналитики публикуют торговые идеи. Project teams анонсируют обновления. Мониторинг этих каналов даёт ранний доступ к информации.

Сбор данных через Telethon

from telethon import TelegramClient, events
from telethon.tl.functions.channels import GetFullChannelRequest
import asyncio

class TelegramCryptoMonitor:
    def __init__(self, api_id, api_hash, session_name='crypto_monitor'):
        self.client = TelegramClient(session_name, api_id, api_hash)
        self.channels_to_monitor = []
    
    async def add_channel(self, channel_username):
        """Подписываемся на канал для мониторинга"""
        channel = await self.client.get_entity(channel_username)
        self.channels_to_monitor.append(channel)
        return channel
    
    async def fetch_history(self, channel, limit=1000):
        """Загружаем историю сообщений"""
        messages = []
        async for message in self.client.iter_messages(channel, limit=limit):
            if message.text:
                messages.append({
                    'id': message.id,
                    'text': message.text,
                    'date': message.date,
                    'views': message.views,
                    'forwards': message.forwards,
                    'channel': channel.username
                })
        return messages
    
    async def monitor_realtime(self, callback):
        """Realtime мониторинг новых сообщений"""
        @self.client.on(events.NewMessage(chats=self.channels_to_monitor))
        async def handler(event):
            if event.message.text:
                await callback({
                    'text': event.message.text,
                    'channel': event.chat.username,
                    'date': event.message.date,
                    'views': 0  # views обновляются позже
                })
        
        await self.client.run_until_disconnected()

Категории Telegram-каналов

Trading signals (например, Crypto Signals, Whale Alert): конкретные торговые рекомендации с entry/exit/stop. Высокая ценность, но много pump-and-dump.

Analysis channels (Crypto Fear and Greed, on-chain аналитики): глубокий анализ рынка. Качественный signal.

Project official channels (Ethereum Foundation, Binance, Uniswap): официальные анонсы. Extremely high impact при неожиданных новостях.

News aggregators: перепечатки новостей. Средняя ценность.

Community chats: большие группы, много шума, мало signal.

NLP модель для Telegram

Telegram сообщения длиннее твитов, содержат технический анализ, часто на нескольких языках. Особенности:

from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline
from langdetect import detect

class TelegramMessageAnalyzer:
    def __init__(self):
        self.lang_detector = detect
        
        # Мультиязычная модель для Telegram (много русского, английского, китайского)
        self.multilingual_model = pipeline(
            'text-classification',
            model='cardiffnlp/twitter-xlm-roberta-base-sentiment'
        )
        
        # Английская специализированная модель
        self.en_model = pipeline(
            'text-classification',
            model='./crypto_finbert_finetuned'
        )
    
    def analyze(self, text):
        if len(text) < 10:
            return None
        
        # Определяем язык
        try:
            lang = self.lang_detector(text)
        except:
            lang = 'unknown'
        
        # Выбираем модель
        if lang == 'en':
            result = self.en_model(text[:512])[0]
        else:
            result = self.multilingual_model(text[:512])[0]
        
        return {
            'lang': lang,
            'label': result['label'],
            'score': result['score'],
            'text_length': len(text)
        }

Извлечение торговых сигналов

Из сообщений вида «BTC entry: 44500, target: 48000, SL: 43000» извлекаем structured торговые параметры:

import re

def extract_trade_signal(text):
    """Extracting structured trading signals from Telegram messages"""
    patterns = {
        'symbol': r'\b([A-Z]{2,10}(?:USDT|BTC|ETH|USD)?)\b',
        'entry': r'(?:entry|buy|long)\s*[@:=\s]\s*\$?([0-9,\.]+)',
        'target': r'(?:target|tp|take.?profit)\s*[@:=\s]\s*\$?([0-9,\.]+)',
        'stop_loss': r'(?:sl|stop.?loss|stoploss)\s*[@:=\s]\s*\$?([0-9,\.]+)',
        'direction': r'\b(long|short|buy|sell)\b'
    }
    
    results = {}
    for field, pattern in patterns.items():
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            results[field] = match.group(1)
    
    # Валидность сигнала
    is_valid = 'symbol' in results and 'direction' in results
    return results if is_valid else None

Channel reputation scoring

Не все каналы одинаково надёжны. Оцениваем историческую точность:

def calculate_channel_accuracy(historical_signals, price_data):
    """
    Для каждого сигнала из канала проверяем:
    достиг ли target раньше stop loss?
    """
    wins, losses = 0, 0
    for signal in historical_signals:
        if 'entry' not in signal or 'target' not in signal:
            continue
        
        entry = float(signal['entry'])
        target = float(signal.get('target', 0))
        stop = float(signal.get('stop_loss', entry * 0.95))
        
        # Смотрим что произошло в следующие 7 дней
        future_prices = get_future_prices(
            price_data, signal['timestamp'], days=7
        )
        
        for price in future_prices:
            if price >= target:
                wins += 1
                break
            elif price <= stop:
                losses += 1
                break
    
    accuracy = wins / (wins + losses) if (wins + losses) > 0 else 0
    return {'wins': wins, 'losses': losses, 'accuracy': accuracy}

Pump-and-dump detection

Telegram активно используется для организации P&D схем:

def detect_pump_signal(message, channel_history):
    """Признаки P&D сигнала"""
    indicators = []
    text_lower = message['text'].lower()
    
    # 1. Urgency language
    urgency_words = ['hurry', 'now', 'quickly', '🚀🚀🚀', 'last chance', 'don\'t miss']
    if any(w in text_lower for w in urgency_words):
        indicators.append('urgency')
    
    # 2. Low-cap obscure token
    if 'symbol' in message and is_low_cap_token(message['symbol']):
        indicators.append('low_cap')
    
    # 3. Channel post frequency spike
    recent_posts = [m for m in channel_history[-24h] if m['channel'] == message['channel']]
    if len(recent_posts) > 10:  # > 10 постов за 24h подозрительно
        indicators.append('frequency_spike')
    
    return len(indicators) >= 2, indicators

Технический стек

Python (Telethon для Telegram API), PostgreSQL для хранения сообщений, Redis для дедупликации, FastAPI для serving NLP предсказаний, React dashboard с историей сообщений по каналам и sentiment timeline.

Разрабатываем систему мониторинга Telegram-каналов с realtime сбором, мультиязычным NLP, извлечением торговых сигналов, reputation scoring каналов и P&D detection.