Парсинг Twitter API и сентимент-анализ на Python NLTK для предсказания крипто-пампов

import tweepy
import pandas as pd
import numpy as np
import nltk
from nltk.sentiment.vader import SentimentIntensityAnalyzer

# Инициализация лексикона VADER
nltk.download('vader_lexicon', quiet=True)

class CryptoPumpPredictor:
    def __init__(self, bearer_token: str):
        """
        Инициализация клиента Twitter API v2 и анализатора настроений.
        """
        self.client = tweepy.Client(bearer_token=bearer_token)
        self.sia = SentimentIntensityAnalyzer()

    def fetch_tweets(self, query: str, max_results: int = 100):
        """
        Получение недавних твитов по поисковому запросу.
        """
        try:
            tweets = self.client.search_recent_tweets(
                query=query,
                max_results=max_results,
                tweet_fields=['created_at', 'public_metrics', 'text'],
                expansions=['author_id'],
                user_fields=['public_metrics']
            )
            return tweets
        except Exception as e:
            print(f"Ошибка при получении твитов: {e}")
            return None

    def analyze_sentiment(self, text: str) -> float:
        """
        Анализ тональности текста с помощью VADER.
        Возвращает compound score от -1.0 (негативный) до 1.0 (позитивный).
        """
        scores = self.sia.polarity_scores(text)
        return scores['compound']

    def calculate_pump_score(self, tweets_data) -> tuple:
        """
        Расчет вероятности пампа на основе объема, тональности и веса авторов.
        """
        if not tweets_data or not tweets_data.data:
            return 0.0, pd.DataFrame()

        # Создаем словарь пользователей для быстрого доступа к их метрикам
        users_dict = {u.id: u for u in tweets_data.includes['users']}
        processed_tweets = []

        for tweet in tweets_data.data:
            author = users_dict.get(tweet.author_id)
            followers = author.public_metrics['followers_count'] if author else 0
            
            sentiment = self.analyze_sentiment(tweet.text)
            
            # Логарифмическое взвешивание подписчиков для сглаживания влияния ботов-миллионников
            weight = np.log1p(followers) + 1.0
            weighted_sentiment = sentiment * weight
            
            processed_tweets.append({
                'text': tweet.text,
                'sentiment': sentiment,
                'followers': followers,
                'weighted_sentiment': weighted_sentiment
            })

        df = pd.DataFrame(processed_tweets)
        
        # Расчет агрегированных метрик
        avg_sentiment = df['sentiment'].mean()
        total_weighted_sentiment = df['weighted_sentiment'].sum()
        tweet_volume = len(df)
        
        # Формула Pump Score: учитывает средний сентимент и взвешенный объем
        pump_score = (avg_sentiment * 0.3) + (total_weighted_sentiment / (tweet_volume + 1) * 0.7)
        
        return float(pump_score), df

# Пример использования
if __name__ == '__main__':
    BEARER_TOKEN = "YOUR_TWITTER_BEARER_TOKEN"
    QUERY = "$DOGE OR #Dogecoin -is:retweet"
    
    predictor = CryptoPumpPredictor(bearer_token=BEARER_TOKEN)
    print(f"Сбор твитов по запросу: {QUERY}...")
    
    raw_tweets = predictor.fetch_tweets(query=QUERY, max_results=50)
    score, details_df = predictor.calculate_pump_score(raw_tweets)
    
    print(f"\n--- Результаты анализа ---")
    print(f"Итоговый Pump Score: {score:.4f}")
    if not details_df.empty:
        print(f"Средний сентимент: {details_df['sentiment'].mean():.4f}")
        print(f"Всего проанализировано твитов: {len(details_df)}")
        print("\nТоп-3 наиболее влиятельных позитивных твита:")
        top_tweets = details_df.sort_values(by='weighted_sentiment', ascending=False).head(3)
        for idx, row in top_tweets.iterrows():
            print(f"- [{row['followers']} фолловеров, Sentiment: {row['sentiment']:.2f}]: {row['text'][:100]}...")

pip install tweepy pandas numpy nltk