Навчання NLP-моделі для аналізу Telegram-каналів

Навчання NLP-моделі для аналізу Telegram-каналів

Telegram — центральне середовище крипто-комунікацій. Великі інфлюенсери ведуть канали зі сотнями тисяч підписників. Анонімні аналітики публікують торгові ідеї. Команди проектів анонсують оновлення. Моніторинг цих каналів дає ранній доступ до інформації.

Збір даних через 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 аналітики): глибокий аналіз ринку. Якісний сигнал.

Project official channels (Ethereum Foundation, Binance, Uniswap): офіційні оголошення. Дуже високий impact на несподіваних новинах.

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

Community chats: великі групи, багато шуму, мало сигналу.

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» видобуємо структуровані торгові параметри:

import re

def extract_trade_signal(text):
    """Видобування структурованих торгових сигналів з Telegram повідомлень"""
    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

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

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

Tech стек

Python (Telethon для Telegram API), PostgreSQL для зберігання повідомлень, Redis для дедуплікації, FastAPI для serving NLP предсказань, React дашборд з історією повідомлень каналів та timeline настроїв.

Розроблення системи моніторингу Telegram-каналів з realtime збором, мультимовним NLP, видобуванням торгових сигналів, scoring репутації каналів та P&D виявленням.