Разработка модели классификации фейковых новостей в крипто

Разработка модели классификации фейковых новостей в крипто

Крипто-пространство — идеальная среда для дезинформации. Высокая волатильность означает, что один фейковый твит о листинге на Binance или о взломе протокола может двигать цену на десятки процентов. Pump-and-dump схемы начинаются с информационной манипуляции. Scam-проекты живут за счёт сфабрикованных новостей о партнёрствах.

Разработка классификатора фейковых новостей в крипто — это задача NLP с несколькими специфическими сложностями: узкодоменная терминология, скорость распространения информации (новость устаревает за часы), мультиязычность, намеренная обфускация текста авторами фейков.

Постановка задачи и типология фейков

Прежде чем строить модель, нужно понять, что именно классифицируем. «Фейковые новости» — слишком широкое понятие.

Категории дезинформации в крипто:

• Fake listings: ложные заявления о листинге на крупной бирже (CEX или DEX пул с фейковым токеном)
• Fake partnerships: несуществующие партнёрства (X протокол интегрируется с Y)
• Fabricated exploits: ложные сообщения о взломе для паники и продаж
• Shill content: манипулятивное продвижение без раскрытия финансового интереса
• Price manipulation narrative: скоординированные нарративы для pump/dump
• Impersonation: контент от аккаунтов, имитирующих официальные (Vitalik_Buterin_ вместо VitalikButerin)

Каждая категория имеет свои текстовые паттерны, источники и методы верификации. Модель должна классифицировать по категориям, а не просто binary «fake/real».

Сбор и разметка данных

Главная проблема — отсутствие готового датасета. Существующие датасеты фейков (LIAR, FakeNewsNet) не охватывают крипто-специфику.

Источники данных

Twitter/X API: основная площадка для крипто-новостей и манипуляций. Academic Research API даёт доступ к историческим данным. Ключевые фильтры: аккаунты с > 1,000 follower в крипто-нише, хэштеги (#bitcoin, #defi, #altcoin и др.), ключевые слова протоколов.

Telegram channels: Telethon для парсинга публичных каналов. Важный источник — pump-and-dump каналы (многие публичны).

Reddit: r/CryptoCurrency, r/Bitcoin, r/CryptoMoonShots. Pushshift API или Reddit API для исторических данных.

Crypto news агрегаторы: CoinDesk, Cointelegraph, Decrypt — верифицированные новости (позитивный класс). Contrast со слухами в Twitter.

Стратегия разметки

Автоматическая разметка через кросс-верификацию: если новость появляется в Twitter, но не подтверждается официальными каналами проекта за 24 часа — потенциальный фейк. Если новость противоречит on-chain данным (заявлен листинг, но пула нет) — фейк с высокой вероятностью.

Human labeling через crowdsourcing (Scale AI, Appen) с доменными экспертами для финальной верификации. Каждый пример размечается минимум тремя аннотаторами, используем inter-annotator agreement (Cohen's kappa > 0.7 как порог качества).

from datasets import Dataset
import pandas as pd

# Структура датасета
example_schema = {
    'id': str,                    # уникальный идентификатор
    'text': str,                  # текст новости/поста
    'source': str,                # twitter / telegram / reddit / news_site
    'author': str,                # аккаунт-источник
    'timestamp': str,             # время публикации
    'label': int,                 # 0=real, 1=fake
    'category': str,              # fake_listing / fake_partnership / etc.
    'confidence': float,          # уверенность разметчиков (0-1)
    'verification_sources': list, # ссылки на верификацию
    'mentioned_tokens': list,     # упомянутые токены/проекты
    'mentioned_exchanges': list,  # упомянутые биржи
}

# Балансировка датасета важна: реальных новостей обычно больше
def balance_dataset(df: pd.DataFrame, target_ratio: float = 0.4) -> pd.DataFrame:
    """target_ratio: доля fake в финальном датасете"""
    fake = df[df['label'] == 1]
    real = df[df['label'] == 0]
    
    n_fake = len(fake)
    n_real_target = int(n_fake / target_ratio * (1 - target_ratio))
    
    real_sampled = real.sample(n=min(n_real_target, len(real)), random_state=42)
    return pd.concat([fake, real_sampled]).sample(frac=1, random_state=42)

Архитектура модели

Feature Engineering: что важно для крипто-фейков

Прежде чем выбрать архитектуру, определяем признаки, специфичные для домена.

Текстовые сигналы фейков:

• Чрезмерный hype без конкретики («100x guaranteed», «next bitcoin»)
• Срочность («buy NOW», «last chance»)
• Имена известных проектов/личностей без контекста
• Грамматические ошибки (импersonation аккаунты часто небрежны)
• Несоответствие заявленного источника реальному

Metadata признаки:

• Возраст аккаунта и история публикаций
• Follower/following ratio (0.01 = подозрительно)
• Скорость распространения (viral за первый час = подозрительно)
• Временной паттерн (публикация в 3:00 UTC)
• Количество похожих постов с одинаковым нарративом

Модельный стек

Baseline: TF-IDF + Logistic Regression / XGBoost. Быстро обучается, хорошо интерпретируется, устанавливает baseline F1.

Transformer-based: FinBERT (финансово-ориентированный BERT) или криптo-fine-tuned BERT. Понимает контекст и нюансы языка.

Ensemble: комбинация transformer features + metadata features через gradient boosting. Практически всегда превосходит одиночную модель.

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier

class CryptoFakeNewsClassifier:
    def __init__(self, model_name: str = 'ProsusAI/finbert'):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.text_model = AutoModelForSequenceClassification.from_pretrained(
            model_name, 
            num_labels=6  # real + 5 категорий фейков
        )
        self.meta_classifier = GradientBoostingClassifier(
            n_estimators=300,
            max_depth=6,
            learning_rate=0.05
        )
        
    def extract_text_features(self, texts: list[str]) -> np.ndarray:
        """Извлекает CLS embeddings из transformer модели"""
        self.text_model.eval()
        all_embeddings = []
        
        batch_size = 32
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i+batch_size]
            inputs = self.tokenizer(
                batch,
                max_length=512,
                truncation=True,
                padding=True,
                return_tensors='pt'
            )
            
            with torch.no_grad():
                outputs = self.text_model(**inputs, output_hidden_states=True)
                # CLS token из последнего слоя
                cls_embeddings = outputs.hidden_states[-1][:, 0, :]
                all_embeddings.append(cls_embeddings.numpy())
        
        return np.vstack(all_embeddings)
    
    def extract_meta_features(self, posts: list[dict]) -> np.ndarray:
        """Извлекает metadata признаки из поста"""
        features = []
        for post in posts:
            account_age_days = (
                pd.Timestamp.now() - pd.Timestamp(post['account_created'])
            ).days
            
            feature_vector = [
                account_age_days,
                post.get('followers_count', 0),
                post.get('following_count', 1),
                post.get('followers_count', 0) / max(post.get('following_count', 1), 1),
                post.get('tweet_count', 0),
                post.get('retweet_count', 0),
                post.get('like_count', 0),
                int(post.get('verified', False)),
                len(post.get('text', '')),
                post.get('text', '').count('!'),
                post.get('text', '').count('$'),
                # Час публикации (цикличность через sin/cos)
                np.sin(2 * np.pi * pd.Timestamp(post['created_at']).hour / 24),
                np.cos(2 * np.pi * pd.Timestamp(post['created_at']).hour / 24),
                # Наличие URL
                int('http' in post.get('text', '')),
                # Число упоминаний известных проектов
                sum(1 for token in KNOWN_TOKENS if token.lower() in post.get('text', '').lower()),
            ]
            features.append(feature_vector)
        
        return np.array(features)
    
    def predict(self, posts: list[dict]) -> dict:
        texts = [p['text'] for p in posts]
        
        text_features = self.extract_text_features(texts)
        meta_features = self.extract_meta_features(posts)
        
        # Конкатенация features
        combined = np.hstack([text_features, meta_features])
        
        # Финальная классификация
        probabilities = self.meta_classifier.predict_proba(combined)
        predictions = self.meta_classifier.predict(combined)
        
        return {
            'predictions': predictions,
            'probabilities': probabilities,
            'labels': ['real', 'fake_listing', 'fake_partnership', 
                      'fake_exploit', 'shill', 'impersonation']
        }

Fine-tuning на крипто-домен

FinBERT обучался на финансовых новостях, но не специализирован для крипто. Fine-tuning на крипто-корпусе значительно улучшает качество.

from transformers import Trainer, TrainingArguments

training_args = TrainingArguments(
    output_dir='./crypto-fake-news-model',
    num_train_epochs=5,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=32,
    warmup_steps=500,
    weight_decay=0.01,
    logging_dir='./logs',
    evaluation_strategy='epoch',
    save_strategy='epoch',
    load_best_model_at_end=True,
    metric_for_best_model='f1_macro',  # важнее accuracy на несбалансированных данных
    
    # Class weights для несбалансированных классов
    # fake примеры в minority — нужно больший штраф за FN
)

# Кастомные weights: fake классы получают больший вес
class_weights = compute_class_weight(
    class_weight='balanced',
    classes=np.unique(train_labels),
    y=train_labels
)

Верификация через on-chain данные

Уникальное преимущество крипто-домена: многие заявления верифицируемы on-chain.

Заявлен листинг на Uniswap V3: проверяем через Uniswap Subgraph — существует ли пул. Заявлен exploit: проверяем изменение TVL в DeFiLlama API за указанный период. Заявлено партнёрство с протоколом: ищем on-chain взаимодействие между контрактами.

import aiohttp

async def verify_listing_claim(token_address: str, dex: str = 'uniswap_v3') -> dict:
    """Верификация заявления о листинге через on-chain данные"""
    
    if dex == 'uniswap_v3':
        query = """
        query PoolsForToken($token: String!) {
            pools(where: { 
                or: [
                    { token0: $token },
                    { token1: $token }
                ]
            }, first: 5) {
                id
                token0 { symbol }
                token1 { symbol }
                liquidity
                totalValueLockedUSD
                createdAtTimestamp
            }
        }
        """
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                'https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3',
                json={'query': query, 'variables': {'token': token_address.lower()}}
            ) as response:
                data = await response.json()
                pools = data.get('data', {}).get('pools', [])
                
                return {
                    'listing_exists': len(pools) > 0,
                    'pools': pools,
                    'total_tvl': sum(float(p['totalValueLockedUSD']) for p in pools)
                }

async def verify_exploit_claim(protocol: str, claimed_amount_usd: float, 
                                claim_timestamp: int) -> dict:
    """Верификация заявления о взломе через TVL данные"""
    
    # DeFiLlama API для исторического TVL
    async with aiohttp.ClientSession() as session:
        async with session.get(
            f'https://api.llama.fi/protocol/{protocol}'
        ) as response:
            data = await response.json()
            tvl_history = data.get('tvl', [])
    
    # Найти изменение TVL вокруг timestamp
    before_tvl = get_tvl_at_timestamp(tvl_history, claim_timestamp - 3600)
    after_tvl = get_tvl_at_timestamp(tvl_history, claim_timestamp + 3600)
    
    tvl_drop = before_tvl - after_tvl if before_tvl > after_tvl else 0
    
    return {
        'tvl_drop_detected': tvl_drop > 0,
        'detected_amount': tvl_drop,
        'claimed_amount': claimed_amount_usd,
        'plausible': abs(tvl_drop - claimed_amount_usd) / claimed_amount_usd < 0.3
    }

On-chain верификация превращает часть задачи из NLP в детерминированную проверку. Это значительно повышает точность для категорий, имеющих on-chain след.

Оценка качества модели

Для задачи детекции фейков accuracy — неправильная метрика. Если 90% примеров реальные новости, модель, всегда предсказывающая «real», получит 90% accuracy.

Precision, Recall, F1 по классам — основные метрики. Особое внимание к recall для fake классов: FN (пропущенный фейк) хуже FP (ложная тревога).

Area Under ROC Curve (AUC-ROC): порогово-независимая оценка.

Temporal stability: модель должна сохранять качество на новых данных. Крипто-нарративы меняются быстро — необходимо регулярное переобучение.

from sklearn.metrics import classification_report, roc_auc_score
import pandas as pd

def evaluate_model(y_true, y_pred, y_proba, class_names):
    # Детальный отчёт по классам
    report = classification_report(
        y_true, y_pred, 
        target_names=class_names,
        output_dict=True
    )
    
    df_report = pd.DataFrame(report).T
    
    # Особый акцент на fake классы
    fake_classes = [c for c in class_names if c != 'real']
    fake_f1_avg = df_report.loc[fake_classes, 'f1-score'].mean()
    
    print(f"Fake detection F1 (macro avg): {fake_f1_avg:.3f}")
    print(f"Real precision: {df_report.loc['real', 'precision']:.3f}")
    
    # AUC для бинарной задачи (fake vs real)
    binary_labels = (y_true > 0).astype(int)  # все fake категории = 1
    binary_proba = 1 - y_proba[:, 0]           # вероятность "not real"
    auc = roc_auc_score(binary_labels, binary_proba)
    print(f"AUC-ROC (fake vs real): {auc:.3f}")
    
    return df_report

Целевые показатели для production системы: Fake detection recall > 0.85 (пропускаем не более 15% фейков), Real precision > 0.90 (не более 10% ложных тревог), F1 macro > 0.82.

Deployment и реальное использование

Модель в production должна обрабатывать потоковые данные в реальном времени. Архитектура: Kafka для ingestion Twitter/Telegram stream → inference service (FastAPI + GPU или CPU для небольшой нагрузки) → PostgreSQL + Elasticsearch для хранения и поиска → alerting при обнаружении фейков.

Concept drift: крипто-нарративы меняются быстро. Раз в месяц переобучение на новых данных. Мониторинг distribution shift в inference (если распределение входных данных значительно меняется — сигнал для переобучения).

Человеческая верификация: high-confidence фейки (>0.95 вероятность) публикуются автоматически, borderline (0.6–0.95) идут на human review. Это снижает ущерб от ошибок модели.

Правовые и этические аспекты

Автоматическая классификация контента как «фейкового» несёт правовые риски. Рекомендации: система должна работать как «risk score» для дополнительной проверки, а не как финальный вердикт. Публичная система должна иметь механизм апелляции. Использование только публично доступных данных.

Стек и сроки разработки

Инфраструктура данных: Python (pandas, datasets), Twitter/Telegram API, PostgreSQL.

ML стек: PyTorch + HuggingFace Transformers, scikit-learn, Weights & Biases для экспериментов.

Deployment: FastAPI, Docker, Redis для кэширования, опционально Triton Inference Server для GPU inference.

Сроки: сбор и разметка 50,000 примеров — 6–8 недель, обучение baseline + transformer моделей + evaluation — 4–6 недель, production deployment + мониторинг — 3–4 недели. Итого: 3–4 месяца до production-ready системы.