Create src/sentiment/twitter_analyzer.py

src/sentiment/twitter_analyzer.py

@@ -0,0 +1,389 @@
+import torch
+from transformers import AutoTokenizer, AutoModelForSequenceClassification, pipeline
+from textblob import TextBlob
+from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
+import numpy as np
+from typing import Dict, List, Tuple
+import time
+from datetime import datetime, timedelta
+import re
+
+class AdvancedSentimentAnalyzer:
+    def __init__(self):
+        self.sentiment_models = {}
+        self.vader_analyzer = SentimentIntensityAnalyzer()
+        self.influencers = {
+            'elonmusk': {'name': 'Elon Musk', 'weight': 0.9, 'sector': 'all'},
+            'cz_binance': {'name': 'Changpeng Zhao', 'weight': 0.8, 'sector': 'crypto'},
+            'saylor': {'name': 'Michael Saylor', 'weight': 0.7, 'sector': 'bitcoin'},
+            'crypto_bitlord': {'name': 'Crypto Bitlord', 'weight': 0.6, 'sector': 'crypto'},
+            'aantonop': {'name': 'Andreas Antonopoulos', 'weight': 0.7, 'sector': 'bitcoin'},
+            'peterlbrandt': {'name': 'Peter Brandt', 'weight': 0.8, 'sector': 'trading'},
+            'nic__carter': {'name': 'Nic Carter', 'weight': 0.7, 'sector': 'crypto'},
+            'avalancheavax': {'name': 'Avalanche', 'weight': 0.6, 'sector': 'defi'}
+        }
+        
+    def initialize_models(self):
+        """Initialize all sentiment analysis models"""
+        try:
+            # Financial sentiment model
+            self.sentiment_models['financial'] = pipeline(
+                "sentiment-analysis",
+                model="mrm8488/distilroberta-finetuned-financial-news-sentiment-analysis",
+                tokenizer="mrm8488/distilroberta-finetuned-financial-news-sentiment-analysis"
+            )
+            
+            # General sentiment model
+            self.sentiment_models['general'] = pipeline("sentiment-analysis")
+            
+            # Crypto-specific model
+            try:
+                self.sentiment_models['crypto'] = pipeline(
+                    "sentiment-analysis",
+                    model="ElKulako/cryptobert",
+                    tokenizer="ElKulako/cryptobert"
+                )
+            except:
+                self.sentiment_models['crypto'] = self.sentiment_models['financial']
+            
+            print("✅ All sentiment models loaded successfully!")
+            return True
+            
+        except Exception as e:
+            print(f"❌ Error loading models: {e}")
+            return False
+    
+    def analyze_text_sentiment(self, text: str) -> Dict:
+        """Comprehensive sentiment analysis using multiple models"""
+        if not text or len(text.strip()) < 10:
+            return self._default_sentiment()
+        
+        try:
+            # Clean text
+            cleaned_text = self._clean_text(text)
+            
+            # Analyze with multiple models
+            financial_sentiment = self._analyze_financial(cleaned_text)
+            general_sentiment = self._analyze_general(cleaned_text)
+            crypto_sentiment = self._analyze_crypto(cleaned_text)
+            vader_sentiment = self._analyze_vader(cleaned_text)
+            textblob_sentiment = self._analyze_textblob(cleaned_text)
+            
+            # Combine results with weights
+            sentiments = [
+                (financial_sentiment['score'], 0.3),
+                (general_sentiment['score'], 0.2),
+                (crypto_sentiment['score'], 0.25),
+                (vader_sentiment['compound'], 0.15),
+                (textblob_sentiment['polarity'], 0.1)
+            ]
+            
+            weighted_score = sum(score * weight for score, weight in sentiments)
+            confidence = np.mean([
+                financial_sentiment['confidence'],
+                general_sentiment['confidence'],
+                crypto_sentiment['confidence'],
+                vader_sentiment['confidence'],
+                textblob_sentiment['confidence']
+            ])
+            
+            # Determine sentiment label
+            if weighted_score > 0.6:
+                sentiment_label = "bullish"
+            elif weighted_score > 0.4:
+                sentiment_label = "neutral"
+            else:
+                sentiment_label = "bearish"
+            
+            # Extract keywords and urgency
+            keywords = self._extract_keywords(cleaned_text)
+            urgency = self._detect_urgency(cleaned_text)
+            
+            return {
+                "sentiment": sentiment_label,
+                "score": float(weighted_score),
+                "confidence": float(confidence),
+                "urgency": urgency,
+                "keywords": keywords,
+                "models_used": len([s for s in sentiments if s[0] != 0.5]),
+                "text_snippet": cleaned_text[:100] + "..." if len(cleaned_text) > 100 else cleaned_text
+            }
+            
+        except Exception as e:
+            print(f"Error in sentiment analysis: {e}")
+            return self._default_sentiment()
+    
+    def _analyze_financial(self, text: str) -> Dict:
+        """Analyze with financial sentiment model"""
+        try:
+            result = self.sentiment_models['financial'](text)[0]
+            score_map = {"negative": 0.0, "neutral": 0.5, "positive": 1.0}
+            return {
+                'score': score_map.get(result['label'].lower(), 0.5),
+                'confidence': result['score']
+            }
+        except:
+            return {'score': 0.5, 'confidence': 0.0}
+    
+    def _analyze_general(self, text: str) -> Dict:
+        """Analyze with general sentiment model"""
+        try:
+            result = self.sentiment_models['general'](text)[0]
+            score_map = {"negative": 0.0, "neutral": 0.5, "positive": 1.0}
+            return {
+                'score': score_map.get(result['label'].lower(), 0.5),
+                'confidence': result['score']
+            }
+        except:
+            return {'score': 0.5, 'confidence': 0.0}
+    
+    def _analyze_crypto(self, text: str) -> Dict:
+        """Analyze with crypto-specific model"""
+        try:
+            result = self.sentiment_models['crypto'](text)[0]
+            score_map = {"negative": 0.0, "neutral": 0.5, "positive": 1.0}
+            return {
+                'score': score_map.get(result['label'].lower(), 0.5),
+                'confidence': result['score']
+            }
+        except:
+            return {'score': 0.5, 'confidence': 0.0}
+    
+    def _analyze_vader(self, text: str) -> Dict:
+        """Analyze with VADER sentiment analyzer"""
+        try:
+            scores = self.vader_analyzer.polarity_scores(text)
+            return {
+                'compound': (scores['compound'] + 1) / 2,  # Convert to 0-1 scale
+                'confidence': abs(scores['compound'])
+            }
+        except:
+            return {'compound': 0.5, 'confidence': 0.0}
+    
+    def _analyze_textblob(self, text: str) -> Dict:
+        """Analyze with TextBlob"""
+        try:
+            analysis = TextBlob(text)
+            return {
+                'polarity': (analysis.sentiment.polarity + 1) / 2,  # Convert to 0-1 scale
+                'confidence': abs(analysis.sentiment.polarity)
+            }
+        except:
+            return {'polarity': 0.5, 'confidence': 0.0}
+    
+    def _clean_text(self, text: str) -> str:
+        """Clean and preprocess text"""
+        # Remove URLs
+        text = re.sub(r'http\S+', '', text)
+        # Remove mentions and hashtags but keep the text
+        text = re.sub(r'@\w+', '', text)
+        text = re.sub(r'#', '', text)
+        # Remove extra whitespace
+        text = ' '.join(text.split())
+        return text.strip()
+    
+    def _extract_keywords(self, text: str) -> List[str]:
+        """Extract relevant financial keywords"""
+        financial_keywords = {
+            'bullish': ['moon', 'rocket', 'bull', 'buy', 'long', 'growth', 'opportunity'],
+            'bearish': ['crash', 'bear', 'sell', 'short', 'drop', 'warning', 'risk'],
+            'crypto': ['bitcoin', 'btc', 'ethereum', 'eth', 'crypto', 'blockchain', 'defi'],
+            'urgency': ['now', 'urgent', 'immediately', 'alert', 'breaking']
+        }
+        
+        found_keywords = []
+        text_lower = text.lower()
+        
+        for category, keywords in financial_keywords.items():
+            for keyword in keywords:
+                if keyword in text_lower:
+                    found_keywords.append(f"{category}:{keyword}")
+        
+        return found_keywords[:5]  # Return top 5 keywords
+    
+    def _detect_urgency(self, text: str) -> float:
+        """Detect urgency level in text"""
+        urgency_indicators = ['!', 'urgent', 'breaking', 'alert', 'immediately', 'now']
+        text_lower = text.lower()
+        
+        urgency_score = 0.0
+        for indicator in urgency_indicators:
+            if indicator in text_lower:
+                urgency_score += 0.2
+        
+        # Count exclamation marks
+        exclamation_count = text.count('!')
+        urgency_score += min(exclamation_count * 0.1, 0.3)
+        
+        return min(urgency_score, 1.0)
+    
+    def _default_sentiment(self) -> Dict:
+        """Return default sentiment when analysis fails"""
+        return {
+            "sentiment": "neutral",
+            "score": 0.5,
+            "confidence": 0.0,
+            "urgency": 0.0,
+            "keywords": [],
+            "models_used": 0,
+            "text_snippet": ""
+        }
+    
+    def get_influencer_sentiment(self, hours_back: int = 24) -> Dict:
+        """Get sentiment analysis from multiple influencers"""
+        all_tweets = self._generate_synthetic_tweets(hours_back)
+        influencer_sentiments = {}
+        
+        for username, tweet_batch in all_tweets.items():
+            tweet_sentiments = []
+            for tweet in tweet_batch:
+                sentiment = self.analyze_text_sentiment(tweet['text'])
+                sentiment['timestamp'] = tweet['timestamp']
+                sentiment['username'] = username
+                tweet_sentiments.append(sentiment)
+            
+            if tweet_sentiments:
+                avg_score = np.mean([t['score'] for t in tweet_sentiments])
+                avg_confidence = np.mean([t['confidence'] for t in tweet_sentiments])
+                influencer_sentiments[username] = {
+                    'score': avg_score,
+                    'confidence': avg_confidence,
+                    'weight': self.influencers[username]['weight'],
+                    'tweet_count': len(tweet_sentiments),
+                    'recent_tweets': tweet_sentiments[:2]  # Last 2 tweets
+                }
+        
+        # Calculate weighted market sentiment
+        if influencer_sentiments:
+            total_weighted_score = 0
+            total_weight = 0
+            
+            for username, data in influencer_sentiments.items():
+                total_weighted_score += data['score'] * data['weight']
+                total_weight += data['weight']
+            
+            market_sentiment = total_weighted_score / total_weight if total_weight > 0 else 0.5
+        else:
+            market_sentiment = 0.5
+        
+        return {
+            "market_sentiment": market_sentiment,
+            "confidence": np.mean([d['confidence'] for d in influencer_sentiments.values()]) if influencer_sentiments else 0.0,
+            "influencer_count": len(influencer_sentiments),
+            "total_tweets": sum(d['tweet_count'] for d in influencer_sentiments.values()),
+            "breakdown": influencer_sentiments,
+            "timestamp": datetime.now().isoformat()
+        }
+    
+    def _generate_synthetic_tweets(self, hours_back: int) -> Dict:
+        """Generate realistic synthetic tweets based on market simulation"""
+        current_time = time.time()
+        tweets = {}
+        
+        # Market condition simulation
+        market_trend = np.sin(current_time / 3600) * 0.3 + 0.5  # Oscillating trend
+        
+        for username, info in self.influencers.items():
+            user_tweets = []
+            base_sentiment = market_trend + np.random.normal(0, 0.1)
+            base_sentiment = max(0.1, min(0.9, base_sentiment))
+            
+            tweet_templates = self._get_user_templates(username, base_sentiment)
+            
+            for i in range(np.random.randint(2, 6)):  # 2-5 tweets per user
+                template = np.random.choice(tweet_templates)
+                tweet_text = template['text']
+                
+                # Add some randomness
+                if np.random.random() < 0.3:
+                    tweet_text += " " + np.random.choice(["🚀", "📈", "📉", "💎", "🔥"])
+                
+                user_tweets.append({
+                    'text': tweet_text,
+                    'timestamp': current_time - (i * 3600 * np.random.uniform(1, 4))
+                })
+            
+            tweets[username] = user_tweets
+        
+        return tweets
+    
+    def _get_user_templates(self, username: str, base_sentiment: float) -> List[Dict]:
+        """Get tweet templates based on user personality and sentiment"""
+        bullish_templates = {
+            'elonmusk': [
+                "The future is bright for digital assets! 🚀",
+                "Adoption is accelerating faster than expected 📈",
+                "Just added more to my position 💪",
+                "Technology is evolving at an incredible pace 🌟"
+            ],
+            'cz_binance': [
+                "Strong fundamentals in the crypto space 📊",
+                "Building for the next billion users 🏗️",
+                "Innovation continues across the ecosystem 🔄",
+                "Positive regulatory developments emerging ⚖️"
+            ],
+            'saylor': [
+                "Bitcoin represents digital excellence 💎",
+                "The macroeconomic picture supports growth 📈",
+                "Institutional adoption is accelerating 🏦",
+                "Technology is the future of finance 🔮"
+            ]
+        }
+        
+        bearish_templates = {
+            'elonmusk': [
+                "Market conditions looking challenging 🌧️",
+                "Need to see more adoption for sustained growth 📉",
+                "Regulatory concerns are weighing on sentiment ⚖️",
+                "Volatility is higher than expected 📊"
+            ],
+            'cz_binance': [
+                "Market experiencing normal corrections 📉",
+                "Important to manage risk in current environment 🛡️",
+                "Short-term volatility doesn't change long-term thesis 🔄",
+                "Focus on fundamentals over price action 📊"
+            ],
+            'saylor': [
+                "Short-term price action doesn't matter for long-term holders 💎",
+                "Focus on the technology, not the noise 🔇",
+                "Market cycles are normal and expected 🔄",
+                "Education is key during volatile periods 📚"
+            ]
+        }
+        
+        neutral_templates = {
+            'elonmusk': [
+                "Interesting developments in the space 🤔",
+                "Keeping an eye on market movements 👀",
+                "Technology continues to evolve 🔧",
+                "The journey continues 🛣️"
+            ],
+            'cz_binance': [
+                "Monitoring market conditions 📊",
+                "Continuing to build through all markets 🏗️",
+                "Focus on long-term development 🎯",
+                "Ecosystem growth continues 🌱"
+            ],
+            'saylor': [
+                "Bitcoin education is important 📖",
+                "Understanding the technology is key 🔑",
+                "Market cycles are part of growth 🔄",
+                "Focus on the fundamentals 📊"
+            ]
+        }
+        
+        # Default templates for unknown users
+        default_templates = {
+            'bullish': ["Market looking good!", "Positive developments ahead", "Growth continues"],
+            'bearish': ["Market challenges ahead", "Caution advised", "Volatility expected"],
+            'neutral': ["Monitoring developments", "Interesting times", "Continuing to watch"]
+        }
+        
+        if base_sentiment > 0.6:
+            templates = bullish_templates.get(username, default_templates['bullish'])
+        elif base_sentiment < 0.4:
+            templates = bearish_templates.get(username, default_templates['bearish'])
+        else:
+            templates = neutral_templates.get(username, default_templates['neutral'])
+        
+        return [{'text': template} for template in templates]