AI Sentiment Analysis: Social Media DeFi Signal Processing That Actually Works

import tweepy
import praw
import asyncio
from datetime import datetime
import pandas as pd

class SocialMediaCollector:
    def __init__(self, twitter_keys, reddit_keys):
        # Twitter API setup
        self.twitter_auth = tweepy.OAuth1UserHandler(
            twitter_keys['consumer_key'],
            twitter_keys['consumer_secret'],
            twitter_keys['access_token'], 
            twitter_keys['access_token_secret']
        )
        self.twitter_api = tweepy.API(self.twitter_auth)
        
        # Reddit API setup
        self.reddit = praw.Reddit(
            client_id=reddit_keys['client_id'],
            client_secret=reddit_keys['client_secret'],
            user_agent=reddit_keys['user_agent']
        )
    
    def collect_twitter_sentiment(self, keywords, count=100):
        """Collect tweets for sentiment analysis"""
        tweets = []
        
        for keyword in keywords:
            # Search recent tweets containing DeFi keywords
            search_results = tweepy.Cursor(
                self.twitter_api.search_tweets,
                q=f"{keyword} -RT",  # Exclude retweets
                lang="en",
                result_type="recent",
                tweet_mode="extended"
            ).items(count)
            
            for tweet in search_results:
                tweets.append({
                    'id': tweet.id,
                    'text': tweet.full_text,
                    'created_at': tweet.created_at,
                    'user': tweet.user.screen_name,
                    'followers': tweet.user.followers_count,
                    'keyword': keyword,
                    'platform': 'twitter'
                })
        
        return pd.DataFrame(tweets)
    
    def collect_reddit_sentiment(self, subreddits, limit=50):
        """Collect Reddit posts for sentiment analysis"""
        posts = []
        
        for subreddit_name in subreddits:
            subreddit = self.reddit.subreddit(subreddit_name)
            
            # Get hot posts from DeFi subreddits
            for post in subreddit.hot(limit=limit):
                posts.append({
                    'id': post.id,
                    'title': post.title,
                    'text': post.selftext,
                    'score': post.score,
                    'comments': post.num_comments,
                    'created_at': datetime.fromtimestamp(post.created_utc),
                    'subreddit': subreddit_name,
                    'platform': 'reddit'
                })
        
        return pd.DataFrame(posts)

# Usage example
collector = SocialMediaCollector(twitter_keys, reddit_keys)

# Collect DeFi-related social media data
defi_keywords = ['#AAVE', '#Uniswap', '#Compound', '#MakerDAO']
subreddits = ['defi', 'aave_official', 'UniSwap']

twitter_data = collector.collect_twitter_sentiment(defi_keywords)
reddit_data = collector.collect_reddit_sentiment(subreddits)

import re
import nltk
from nltk.sentiment import SentimentIntensityAnalyzer
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch

class DeFiSentimentProcessor:
    def __init__(self):
        # Download required NLTK data
        nltk.download('vader_lexicon', quiet=True)
        self.sia = SentimentIntensityAnalyzer()
        
        # Load crypto-specific BERT model
        model_name = "ElKulako/cryptobert"
        self.crypto_tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.crypto_model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.crypto_classifier = pipeline(
            "sentiment-analysis",
            model=self.crypto_model,
            tokenizer=self.crypto_tokenizer,
            return_all_scores=True
        )
    
    def preprocess_text(self, text):
        """Clean and normalize social media text"""
        # Remove URLs, mentions, hashtags for cleaner analysis
        text = re.sub(r'http\S+|www\S+|https\S+', '', text, flags=re.MULTILINE)
        text = re.sub(r'@\w+|#\w+', '', text)
        
        # Remove excessive whitespace and special characters
        text = re.sub(r'\s+', ' ', text).strip()
        text = re.sub(r'[^\w\s]', ' ', text)
        
        # Convert to lowercase for consistency
        return text.lower()
    
    def extract_crypto_entities(self, text):
        """Identify cryptocurrency mentions in text"""
        # Common DeFi token symbols and names
        crypto_patterns = {
            'AAVE': r'\b(aave|$aave)\b',
            'UNI': r'\b(uni|uniswap|$uni)\b', 
            'COMP': r'\b(comp|compound|$comp)\b',
            'MKR': r'\b(mkr|maker|makerdao|$mkr)\b',
            'CRV': r'\b(crv|curve|$crv)\b'
        }
        
        found_tokens = []
        for token, pattern in crypto_patterns.items():
            if re.search(pattern, text, re.IGNORECASE):
                found_tokens.append(token)
        
        return found_tokens
    
    def analyze_sentiment(self, text, method='crypto_bert'):
        """Analyze sentiment using multiple approaches"""
        preprocessed_text = self.preprocess_text(text)
        
        if method == 'vader':
            # VADER sentiment (good baseline)
            scores = self.sia.polarity_scores(preprocessed_text)
            return {
                'compound': scores['compound'],
                'positive': scores['pos'],
                'negative': scores['neg'],
                'neutral': scores['neu'],
                'method': 'vader'
            }
        
        elif method == 'crypto_bert':
            # Crypto-specific BERT model (most accurate)
            try:
                result = self.crypto_classifier(preprocessed_text[:512])  # BERT token limit
                
                # Extract sentiment probabilities
                sentiment_scores = {item['label'].lower(): item['score'] for item in result[0]}
                
                # Calculate compound score similar to VADER
                if 'positive' in sentiment_scores and 'negative' in sentiment_scores:
                    compound = sentiment_scores['positive'] - sentiment_scores['negative']
                else:
                    compound = sentiment_scores.get('label_1', 0) - sentiment_scores.get('label_0', 0)
                
                return {
                    'compound': compound,
                    'scores': sentiment_scores,
                    'method': 'crypto_bert'
                }
            except Exception as e:
                # Fallback to VADER if BERT fails
                return self.analyze_sentiment(text, method='vader')

# Process collected data
processor = DeFiSentimentProcessor()

def process_social_data(df):
    """Add sentiment analysis to social media DataFrame"""
    processed_data = []
    
    for _, row in df.iterrows():
        # Combine title and text for Reddit, use text for Twitter
        full_text = f"{row.get('title', '')} {row.get('text', '')}"
        
        # Analyze sentiment
        sentiment = processor.analyze_sentiment(full_text)
        
        # Extract mentioned tokens
        mentioned_tokens = processor.extract_crypto_entities(full_text)
        
        # Add processed data
        processed_row = row.to_dict()
        processed_row.update({
            'sentiment_compound': sentiment['compound'],
            'sentiment_positive': sentiment.get('positive', 0),
            'sentiment_negative': sentiment.get('negative', 0),
            'mentioned_tokens': mentioned_tokens,
            'processed_at': datetime.now()
        })
        
        processed_data.append(processed_row)
    
    return pd.DataFrame(processed_data)

import numpy as np
from sklearn.preprocessing import StandardScaler
from datetime import timedelta
import warnings
warnings.filterwarnings('ignore')

class DeFiSignalGenerator:
    def __init__(self, lookback_window=24):  # 24 hours
        self.lookback_window = lookback_window
        self.scaler = StandardScaler()
        
    def calculate_sentiment_metrics(self, sentiment_df):
        """Calculate aggregated sentiment metrics"""
        # Group by token and time windows
        current_time = datetime.now()
        recent_data = sentiment_df[
            sentiment_df['processed_at'] > current_time - timedelta(hours=self.lookback_window)
        ]
        
        metrics = {}
        
        for token in recent_data['mentioned_tokens'].explode().unique():
            if pd.isna(token):
                continue
                
            # Filter posts mentioning this token
            token_posts = recent_data[
                recent_data['mentioned_tokens'].apply(lambda x: token in x if isinstance(x, list) else False)
            ]
            
            if len(token_posts) == 0:
                continue
            
            # Calculate weighted sentiment (more followers = higher weight)
            weights = token_posts.get('followers', 1).fillna(1)
            weighted_sentiment = np.average(token_posts['sentiment_compound'], weights=weights)
            
            # Calculate sentiment velocity (change over time)
            token_posts_sorted = token_posts.sort_values('processed_at')
            if len(token_posts_sorted) > 1:
                recent_sentiment = token_posts_sorted.tail(10)['sentiment_compound'].mean()
                older_sentiment = token_posts_sorted.head(10)['sentiment_compound'].mean()
                sentiment_velocity = recent_sentiment - older_sentiment
            else:
                sentiment_velocity = 0
            
            # Calculate mention volume
            mention_volume = len(token_posts)
            mention_volume_change = self._calculate_volume_change(token_posts)
            
            metrics[token] = {
                'weighted_sentiment': weighted_sentiment,
                'sentiment_velocity': sentiment_velocity,
                'mention_volume': mention_volume,
                'volume_change': mention_volume_change,
                'post_count': len(token_posts),
                'avg_followers': weights.mean()
            }
        
        return metrics
    
    def _calculate_volume_change(self, token_posts):
        """Calculate percentage change in mention volume"""
        current_time = datetime.now()
        
        # Split into recent and previous periods
        mid_point = current_time - timedelta(hours=self.lookback_window/2)
        recent_posts = token_posts[token_posts['processed_at'] > mid_point]
        older_posts = token_posts[token_posts['processed_at'] <= mid_point]
        
        recent_count = len(recent_posts)
        older_count = len(older_posts)
        
        if older_count == 0:
            return 0
        
        return (recent_count - older_count) / older_count * 100
    
    def generate_trading_signals(self, sentiment_metrics):
        """Convert sentiment metrics to trading signals"""
        signals = {}
        
        for token, metrics in sentiment_metrics.items():
            # Signal strength components
            sentiment_score = metrics['weighted_sentiment']
            momentum_score = metrics['sentiment_velocity']
            volume_score = min(metrics['volume_change'] / 100, 2)  # Cap at 200%
            
            # Combine scores with weights
            combined_score = (
                0.4 * sentiment_score +      # 40% current sentiment
                0.3 * momentum_score +       # 30% sentiment change
                0.3 * volume_score           # 30% volume change
            )
            
            # Generate signal classification
            if combined_score > 0.5 and metrics['post_count'] >= 10:
                signal_type = 'STRONG_BUY'
                confidence = min(abs(combined_score), 1.0)
            elif combined_score > 0.2 and metrics['post_count'] >= 5:
                signal_type = 'BUY'
                confidence = min(abs(combined_score), 0.8)
            elif combined_score < -0.5 and metrics['post_count'] >= 10:
                signal_type = 'STRONG_SELL'
                confidence = min(abs(combined_score), 1.0)
            elif combined_score < -0.2 and metrics['post_count'] >= 5:
                signal_type = 'SELL'
                confidence = min(abs(combined_score), 0.8)
            else:
                signal_type = 'HOLD'
                confidence = 0.1
            
            signals[token] = {
                'signal': signal_type,
                'confidence': confidence,
                'combined_score': combined_score,
                'sentiment': sentiment_score,
                'momentum': momentum_score,
                'volume_change': metrics['volume_change'],
                'post_count': metrics['post_count'],
                'generated_at': datetime.now()
            }
        
        return signals

# Example usage
signal_generator = DeFiSignalGenerator()

# Process your sentiment data
processed_data = process_social_data(combined_social_data)

# Calculate sentiment metrics
sentiment_metrics = signal_generator.calculate_sentiment_metrics(processed_data)

# Generate trading signals
trading_signals = signal_generator.generate_trading_signals(sentiment_metrics)

# Display signals
for token, signal_data in trading_signals.items():
    print(f"{token}: {signal_data['signal']} (confidence: {signal_data['confidence']:.2f})")

import asyncio
import logging
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import json
import redis
from typing import Dict, Any

class DeFiSignalPipeline:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.redis_client = redis.Redis(
            host=config['redis']['host'],
            port=config['redis']['port'],
            decode_responses=True
        )
        
        # Initialize components
        self.collector = SocialMediaCollector(
            config['twitter_keys'], 
            config['reddit_keys']
        )
        self.processor = DeFiSentimentProcessor()
        self.signal_generator = DeFiSignalGenerator()
        
        # Setup logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
        
        # Setup scheduler
        self.scheduler = AsyncIOScheduler()
    
    async def run_signal_pipeline(self):
        """Execute complete signal generation pipeline"""
        try:
            self.logger.info("Starting signal generation pipeline")
            
            # Step 1: Collect social media data
            twitter_data = self.collector.collect_twitter_sentiment(
                self.config['defi_keywords']
            )
            reddit_data = self.collector.collect_reddit_sentiment(
                self.config['subreddits']
            )
            
            # Combine datasets
            combined_data = pd.concat([twitter_data, reddit_data], ignore_index=True)
            self.logger.info(f"Collected {len(combined_data)} social media posts")
            
            # Step 2: Process sentiment
            processed_data = process_social_data(combined_data)
            
            # Step 3: Generate signals
            sentiment_metrics = self.signal_generator.calculate_sentiment_metrics(processed_data)
            trading_signals = self.signal_generator.generate_trading_signals(sentiment_metrics)
            
            # Step 4: Store results in Redis
            await self.store_signals(trading_signals)
            
            # Step 5: Send notifications for strong signals
            await self.notify_strong_signals(trading_signals)
            
            self.logger.info(f"Generated {len(trading_signals)} trading signals")
            
        except Exception as e:
            self.logger.error(f"Pipeline error: {str(e)}")
            raise
    
    async def store_signals(self, signals: Dict[str, Any]):
        """Store trading signals in Redis for API access"""
        for token, signal_data in signals.items():
            key = f"signal:{token}"
            
            # Store signal data with 1-hour expiration
            self.redis_client.setex(
                key, 
                3600,  # 1 hour TTL
                json.dumps(signal_data, default=str)
            )
            
            # Store in historical data
            historical_key = f"history:{token}"
            self.redis_client.lpush(historical_key, json.dumps(signal_data, default=str))
            self.redis_client.ltrim(historical_key, 0, 100)  # Keep last 100 signals
    
    async def notify_strong_signals(self, signals: Dict[str, Any]):
        """Send notifications for high-confidence signals"""
        strong_signals = {
            token: signal for token, signal in signals.items()
            if signal['confidence'] > 0.7 and signal['signal'] in ['STRONG_BUY', 'STRONG_SELL']
        }
        
        if strong_signals:
            notification = {
                'timestamp': datetime.now().isoformat(),
                'type': 'strong_signals',
                'signals': strong_signals
            }
            
            # Store notification
            self.redis_client.lpush('notifications', json.dumps(notification, default=str))
            
            # Log strong signals
            for token, signal in strong_signals.items():
                self.logger.info(
                    f"STRONG SIGNAL: {token} - {signal['signal']} "
                    f"(confidence: {signal['confidence']:.2f})"
                )
    
    def start_scheduler(self):
        """Start automated signal generation"""
        # Run pipeline every 5 minutes
        self.scheduler.add_job(
            self.run_signal_pipeline,
            'interval',
            minutes=5,
            id='signal_pipeline'
        )
        
        self.scheduler.start()
        self.logger.info("Signal generation scheduler started")
    
    async def get_current_signals(self) -> Dict[str, Any]:
        """API endpoint to retrieve current signals"""
        signals = {}
        
        for key in self.redis_client.scan_iter(match="signal:*"):
            token = key.replace("signal:", "")
            signal_data = self.redis_client.get(key)
            
            if signal_data:
                signals[token] = json.loads(signal_data)
        
        return signals

# Deployment configuration
config = {
    'twitter_keys': {
        'consumer_key': 'your_twitter_consumer_key',
        'consumer_secret': 'your_twitter_consumer_secret',
        'access_token': 'your_twitter_access_token',
        'access_token_secret': 'your_twitter_access_token_secret'
    },
    'reddit_keys': {
        'client_id': 'your_reddit_client_id',
        'client_secret': 'your_reddit_client_secret',
        'user_agent': 'DeFiSentimentBot 1.0'
    },
    'redis': {
        'host': 'localhost',
        'port': 6379
    },
    'defi_keywords': ['#AAVE', '#Uniswap', '#Compound', '#MakerDAO', '#CurveDAO'],
    'subreddits': ['defi', 'aave_official', 'UniSwap', 'compound_protocol']
}

# Start the pipeline
pipeline = DeFiSignalPipeline(config)
pipeline.start_scheduler()

# Keep running
try:
    asyncio.get_event_loop().run_forever()
except KeyboardInterrupt:
    pipeline.logger.info("Shutting down signal pipeline")

class SignalPerformanceTracker:
    def __init__(self, redis_client):
        self.redis = redis_client
        
    def track_signal_outcome(self, token: str, signal_data: Dict, actual_outcome: str):
        """Record actual outcome vs predicted signal"""
        tracking_record = {
            'token': token,
            'predicted_signal': signal_data['signal'],
            'confidence': signal_data['confidence'],
            'actual_outcome': actual_outcome,
            'timestamp': datetime.now().isoformat()
        }
        
        self.redis.lpush('signal_outcomes', json.dumps(tracking_record))
    
    def calculate_accuracy_metrics(self, days_back: int = 7) -> Dict[str, float]:
        """Calculate performance metrics over time period"""
        outcomes = []
        
        # Retrieve recent outcomes
        raw_outcomes = self.redis.lrange('signal_outcomes', 0, -1)
        cutoff_date = datetime.now() - timedelta(days=days_back)
        
        for outcome_json in raw_outcomes:
            outcome = json.loads(outcome_json)
            outcome_time = datetime.fromisoformat(outcome['timestamp'])
            
            if outcome_time > cutoff_date:
                outcomes.append(outcome)
        
        if not outcomes:
            return {'accuracy': 0, 'total_signals': 0}
        
        # Calculate accuracy
        correct_predictions = sum(
            1 for o in outcomes 
            if self._is_prediction_correct(o['predicted_signal'], o['actual_outcome'])
        )
        
        total_predictions = len(outcomes)
        accuracy = correct_predictions / total_predictions
        
        # Calculate by confidence level
        high_confidence = [o for o in outcomes if o['confidence'] > 0.7]
        high_confidence_accuracy = sum(
            1 for o in high_confidence
            if self._is_prediction_correct(o['predicted_signal'], o['actual_outcome'])
        ) / len(high_confidence) if high_confidence else 0
        
        return {
            'accuracy': accuracy,
            'high_confidence_accuracy': high_confidence_accuracy,
            'total_signals': total_predictions,
            'high_confidence_signals': len(high_confidence)
        }
    
    def _is_prediction_correct(self, predicted: str, actual: str) -> bool:
        """Determine if prediction matches outcome"""
        positive_signals = ['BUY', 'STRONG_BUY']
        negative_signals = ['SELL', 'STRONG_SELL']
        positive_outcomes = ['PRICE_UP', 'STRONG_PRICE_UP']
        negative_outcomes = ['PRICE_DOWN', 'STRONG_PRICE_DOWN']
        
        if predicted in positive_signals and actual in positive_outcomes:
            return True
        elif predicted in negative_signals and actual in negative_outcomes:
            return True
        elif predicted == 'HOLD' and actual == 'NO_MOVEMENT':
            return True
        
        return False

from sklearn.ensemble import VotingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC

class EnsembleSentimentAnalyzer:
    def __init__(self):
        # Create ensemble of different sentiment models
        self.models = {
            'vader': SentimentIntensityAnalyzer(),
            'crypto_bert': self._load_crypto_bert(),
            'ensemble_classifier': self._create_ensemble_classifier()
        }
    
    def _create_ensemble_classifier(self):
        """Create ensemble of traditional ML models for sentiment"""
        # This would be trained on labeled crypto sentiment data
        estimators = [
            ('lr', LogisticRegression(random_state=42)),
            ('rf', RandomForestClassifier(n_estimators=100, random_state=42)),
            ('svm', SVC(probability=True, random_state=42))
        ]
        
        return VotingClassifier(estimators=estimators, voting='soft')
    
    def analyze_with_ensemble(self, text: str) -> Dict[str, float]:
        """Get consensus sentiment from multiple models"""
        results = {}
        
        # VADER analysis
        vader_scores = self.models['vader'].polarity_scores(text)
        results['vader'] = vader_scores['compound']
        
        # Crypto-BERT analysis
        bert_result = self.models['crypto_bert'](text)
        results['crypto_bert'] = self._extract_bert_sentiment(bert_result)
        
        # Traditional ML ensemble (if trained)
        # results['ensemble'] = self.models['ensemble_classifier'].predict_proba([text])[0][1]
        
        # Weighted average (adjust weights based on historical performance)
        weights = {'vader': 0.3, 'crypto_bert': 0.7}
        
        final_sentiment = sum(
            results[model] * weight 
            for model, weight in weights.items()
            if model in results
        )
        
        return {
            'final_sentiment': final_sentiment,
            'individual_scores': results,
            'confidence': self._calculate_consensus_confidence(results)
        }

import ccxt
import asyncio

class MarketAwareSentimentAnalyzer:
    def __init__(self, exchange='binance'):
        self.exchange = getattr(ccxt, exchange)()
        
    async def get_market_context(self, token: str) -> Dict[str, float]:
        """Get current market data for context"""
        try:
            # Get current price and volume
            ticker = self.exchange.fetch_ticker(f"{token}/USDT")
            
            # Get recent price changes
            ohlcv = self.exchange.fetch_ohlcv(f"{token}/USDT", '1h', limit=24)
            
            # Calculate price momentum
            current_price = ticker['last']
            price_24h_ago = ohlcv[-24][4] if len(ohlcv) >= 24 else current_price
            price_change_24h = (current_price - price_24h_ago) / price_24h_ago
            
            return {
                'current_price': current_price,
                'volume_24h': ticker['baseVolume'],
                'price_change_24h': price_change_24h,
                'volatility': self._calculate_volatility(ohlcv)
            }
            
        except Exception as e:
            return {'error': str(e)}
    
    def _calculate_volatility(self, ohlcv_data) -> float:
        """Calculate price volatility"""
        prices = [candle[4] for candle in ohlcv_data]  # Closing prices
        price_changes = [
            (prices[i] - prices[i-1]) / prices[i-1] 
            for i in range(1, len(prices))
        ]
        return np.std(price_changes)
    
    def adjust_signal_for_market(self, sentiment_signal: Dict, market_data: Dict) -> Dict:
        """Adjust sentiment signals based on market conditions"""
        if 'error' in market_data:
            return sentiment_signal
        
        adjusted_signal = sentiment_signal.copy()
        
        # Reduce signal strength in high volatility conditions
        if market_data['volatility'] > 0.05:  # 5% volatility threshold
            adjusted_signal['confidence'] *= 0.8
        
        # Boost signals that align with price momentum
        price_momentum = market_data['price_change_24h']
        sentiment_direction = 1 if sentiment_signal['combined_score'] > 0 else -1
        momentum_direction = 1 if price_momentum > 0 else -1
        
        if sentiment_direction == momentum_direction:
            adjusted_signal['confidence'] *= 1.2
        
        return adjusted_signal