Sentiment Analysis NLP for Reddit: Complete Technical Guide

from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
import pandas as pd

class VADERAnalyzer:
    """VADER-based sentiment analyzer for Reddit text."""

    def __init__(self):
        self.analyzer = SentimentIntensityAnalyzer()
        # Add Reddit-specific lexicon updates
        self._add_reddit_lexicon()

    def _add_reddit_lexicon(self):
        """Add Reddit-specific terms to lexicon."""
        reddit_terms = {
            'bullish': 2.0,
            'bearish': -2.0,
            'diamond hands': 1.5,
            'paper hands': -1.5,
            'to the moon': 2.5,
            'ape': 1.0,  # Positive in WSB context
            'hodl': 1.5,
            'fud': -1.5,
            'shill': -2.0,
            'based': 1.5,
            'cringe': -1.5,
            '/s': 0.0,  # Sarcasm marker - neutralize
        }
        self.analyzer.lexicon.update(reddit_terms)

    def analyze(self, text: str) -> dict:
        """Analyze sentiment of text."""
        scores = self.analyzer.polarity_scores(text)

        # Classify based on compound score
        if scores['compound'] >= 0.05:
            label = 'positive'
        elif scores['compound'] <= -0.05:
            label = 'negative'
        else:
            label = 'neutral'

        return {
            'label': label,
            'compound': scores['compound'],
            'positive': scores['pos'],
            'negative': scores['neg'],
            'neutral': scores['neu']
        }

    def analyze_batch(self, texts: list) -> pd.DataFrame:
        """Analyze multiple texts efficiently."""
        results = [self.analyze(text) for text in texts]
        return pd.DataFrame(results)

# Usage example
analyzer = VADERAnalyzer()
result = analyzer.analyze("This product is amazing! Best purchase ever.")
print(result)
# {'label': 'positive', 'compound': 0.8516, 'positive': 0.514, ...}

import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from typing import List, Dict
import numpy as np

class TransformerSentimentAnalyzer:
    """Transformer-based sentiment analyzer with batching support."""

    def __init__(
        self,
        model_name: str = "cardiffnlp/twitter-roberta-base-sentiment-latest",
        device: str = None
    ):
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.to(self.device)
        self.model.eval()

        # Label mapping for this model
        self.labels = ['negative', 'neutral', 'positive']

    def preprocess(self, text: str) -> str:
        """Clean and preprocess Reddit text."""
        # Handle Reddit-specific patterns
        text = text.replace('\n', ' ')
        text = text.replace('&', '&')

        # Truncate to model max length
        if len(text) > 500:
            text = text[:500]

        return text

    @torch.no_grad()
    def analyze(self, text: str) -> Dict:
        """Analyze sentiment of single text."""
        text = self.preprocess(text)

        inputs = self.tokenizer(
            text,
            return_tensors="pt",
            truncation=True,
            max_length=512
        ).to(self.device)

        outputs = self.model(**inputs)
        probs = torch.softmax(outputs.logits, dim=-1).cpu().numpy()[0]

        predicted_idx = np.argmax(probs)

        return {
            'label': self.labels[predicted_idx],
            'confidence': float(probs[predicted_idx]),
            'scores': {
                label: float(prob)
                for label, prob in zip(self.labels, probs)
            }
        }

    @torch.no_grad()
    def analyze_batch(self, texts: List[str], batch_size: int = 32) -> List[Dict]:
        """Analyze multiple texts with batching for efficiency."""
        results = []

        for i in range(0, len(texts), batch_size):
            batch_texts = [self.preprocess(t) for t in texts[i:i+batch_size]]

            inputs = self.tokenizer(
                batch_texts,
                return_tensors="pt",
                truncation=True,
                padding=True,
                max_length=512
            ).to(self.device)

            outputs = self.model(**inputs)
            probs = torch.softmax(outputs.logits, dim=-1).cpu().numpy()

            for prob in probs:
                predicted_idx = np.argmax(prob)
                results.append({
                    'label': self.labels[predicted_idx],
                    'confidence': float(prob[predicted_idx]),
                    'scores': {
                        label: float(p)
                        for label, p in zip(self.labels, prob)
                    }
                })

        return results

# Usage
analyzer = TransformerSentimentAnalyzer()
result = analyzer.analyze("This is actually pretty good!")
print(result)
# {'label': 'positive', 'confidence': 0.923, 'scores': {...}}

from transformers import (
    AutoTokenizer,
    AutoModelForSequenceClassification,
    TrainingArguments,
    Trainer
)
from datasets import Dataset
import pandas as pd

class RedditSentimentTrainer:
    """Fine-tune transformer models on Reddit sentiment data."""

    def __init__(self, base_model: str = "distilbert-base-uncased"):
        self.tokenizer = AutoTokenizer.from_pretrained(base_model)
        self.model = AutoModelForSequenceClassification.from_pretrained(
            base_model,
            num_labels=3  # negative, neutral, positive
        )
        self.label_map = {'negative': 0, 'neutral': 1, 'positive': 2}

    def prepare_dataset(self, df: pd.DataFrame) -> Dataset:
        """Convert DataFrame to Hugging Face Dataset."""

        def tokenize(examples):
            return self.tokenizer(
                examples['text'],
                truncation=True,
                padding='max_length',
                max_length=256
            )

        # Convert labels to integers
        df['label'] = df['sentiment'].map(self.label_map)

        dataset = Dataset.from_pandas(df[['text', 'label']])
        dataset = dataset.map(tokenize, batched=True)

        return dataset

    def train(
        self,
        train_df: pd.DataFrame,
        val_df: pd.DataFrame,
        output_dir: str = "./reddit-sentiment-model"
    ):
        """Fine-tune the model."""

        train_dataset = self.prepare_dataset(train_df)
        val_dataset = self.prepare_dataset(val_df)

        training_args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=3,
            per_device_train_batch_size=16,
            per_device_eval_batch_size=32,
            warmup_steps=500,
            weight_decay=0.01,
            logging_dir="./logs",
            logging_steps=100,
            evaluation_strategy="steps",
            eval_steps=500,
            save_steps=1000,
            load_best_model_at_end=True,
        )

        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=train_dataset,
            eval_dataset=val_dataset,
        )

        trainer.train()
        trainer.save_model(output_dir)
        self.tokenizer.save_pretrained(output_dir)

        return trainer

# Training example
trainer = RedditSentimentTrainer()
trainer.train(train_df, val_df)

from transformers import pipeline
import re

class AspectSentimentAnalyzer:
    """Extract sentiment for specific aspects within text."""

    def __init__(self):
        self.sentiment_pipeline = pipeline(
            "sentiment-analysis",
            model="cardiffnlp/twitter-roberta-base-sentiment-latest"
        )

        # Common aspects for product reviews
        self.aspects = [
            'price', 'quality', 'performance', 'design',
            'support', 'battery', 'camera', 'screen',
            'speed', 'ui', 'features'
        ]

    def extract_aspect_sentences(self, text: str) -> dict:
        """Extract sentences mentioning each aspect."""
        sentences = re.split(r'[.!?]+', text)
        aspect_sentences = {}

        for aspect in self.aspects:
            matching = [
                s.strip() for s in sentences
                if aspect.lower() in s.lower() and s.strip()
            ]
            if matching:
                aspect_sentences[aspect] = matching

        return aspect_sentences

    def analyze(self, text: str) -> dict:
        """Analyze sentiment for each aspect found in text."""
        aspect_sentences = self.extract_aspect_sentences(text)

        results = {
            'overall': self.sentiment_pipeline(text[:512])[0],
            'aspects': {}
        }

        for aspect, sentences in aspect_sentences.items():
            # Analyze each sentence mentioning the aspect
            sentiments = self.sentiment_pipeline(sentences)

            # Aggregate sentiment for this aspect
            pos_count = sum(1 for s in sentiments if s['label'] == 'positive')
            neg_count = sum(1 for s in sentiments if s['label'] == 'negative')

            if pos_count > neg_count:
                aspect_sentiment = 'positive'
            elif neg_count > pos_count:
                aspect_sentiment = 'negative'
            else:
                aspect_sentiment = 'neutral'

            results['aspects'][aspect] = {
                'sentiment': aspect_sentiment,
                'sentences': sentences,
                'details': sentiments
            }

        return results

# Usage
analyzer = AspectSentimentAnalyzer()
text = "The camera quality is amazing but the battery life is terrible. Price seems fair."
result = analyzer.analyze(text)
print(result['aspects'])
# {'camera': {'sentiment': 'positive', ...}, 'battery': {'sentiment': 'negative', ...}}

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict
import pandas as pd
from datetime import datetime

class RedditSentimentPipeline:
    """Production sentiment analysis pipeline for Reddit data."""

    def __init__(self, model_path: str = None):
        if model_path:
            self.analyzer = TransformerSentimentAnalyzer(model_path)
        else:
            self.analyzer = TransformerSentimentAnalyzer()

        self.executor = ThreadPoolExecutor(max_workers=4)

    def process_posts(self, posts: List[Dict]) -> pd.DataFrame:
        """Process Reddit posts and return sentiment analysis."""

        # Extract text from posts
        texts = []
        for post in posts:
            # Combine title and body for analysis
            text = post.get('title', '')
            if post.get('selftext'):
                text += ' ' + post['selftext']
            texts.append(text)

        # Batch analyze
        sentiments = self.analyzer.analyze_batch(texts)

        # Combine results with post metadata
        results = []
        for post, sentiment in zip(posts, sentiments):
            results.append({
                'id': post.get('id'),
                'title': post.get('title'),
                'subreddit': post.get('subreddit'),
                'score': post.get('score'),
                'created_utc': post.get('created_utc'),
                'sentiment': sentiment['label'],
                'confidence': sentiment['confidence'],
                'sentiment_scores': sentiment['scores']
            })

        return pd.DataFrame(results)

    def aggregate_sentiment(self, df: pd.DataFrame) -> Dict:
        """Aggregate sentiment metrics."""

        total = len(df)

        return {
            'total_posts': total,
            'sentiment_distribution': df['sentiment'].value_counts().to_dict(),
            'sentiment_percentages': {
                label: count / total * 100
                for label, count in df['sentiment'].value_counts().items()
            },
            'average_confidence': df['confidence'].mean(),
            'high_confidence_count': len(df[df['confidence'] > 0.9]),
            'weighted_sentiment': self._calculate_weighted_sentiment(df),
            'analyzed_at': datetime.utcnow().isoformat()
        }

    def _calculate_weighted_sentiment(self, df: pd.DataFrame) -> float:
        """Calculate weighted sentiment score (-1 to 1)."""
        sentiment_map = {'negative': -1, 'neutral': 0, 'positive': 1}

        # Weight by post score (engagement)
        df['sentiment_value'] = df['sentiment'].map(sentiment_map)
        weights = df['score'].clip(lower=1)  # Minimum weight of 1

        weighted_sum = (df['sentiment_value'] * weights).sum()
        total_weight = weights.sum()

        return weighted_sum / total_weight if total_weight > 0 else 0

# Usage
pipeline = RedditSentimentPipeline()
df = pipeline.process_posts(reddit_posts)
summary = pipeline.aggregate_sentiment(df)
print(summary)

class RobustSentimentAnalyzer:
    """Sentiment analyzer with comprehensive error handling."""

    def analyze_safe(self, text: str) -> Dict:
        """Analyze with error handling and edge case management."""

        # Handle empty/None input
        if not text or not text.strip():
            return {
                'label': 'unknown',
                'confidence': 0.0,
                'error': 'empty_input'
            }

        # Handle very short text
        if len(text.split()) < 3:
            return {
                'label': 'unknown',
                'confidence': 0.0,
                'error': 'text_too_short'
            }

        # Handle non-English text
        if self._is_non_english(text):
            return {
                'label': 'unknown',
                'confidence': 0.0,
                'error': 'non_english'
            }

        try:
            result = self.analyzer.analyze(text)

            # Flag low confidence predictions
            if result['confidence'] < 0.6:
                result['warning'] = 'low_confidence'

            # Flag potential sarcasm
            if '/s' in text.lower() or 'sarcasm' in text.lower():
                result['warning'] = 'potential_sarcasm'

            return result

        except Exception as e:
            return {
                'label': 'unknown',
                'confidence': 0.0,
                'error': str(e)
            }

    def _is_non_english(self, text: str) -> bool:
        """Simple heuristic for non-English detection."""
        # Count ASCII letters vs total characters
        ascii_count = sum(1 for c in text if c.isascii() and c.isalpha())
        alpha_count = sum(1 for c in text if c.isalpha())

        if alpha_count == 0:
            return False

        return ascii_count / alpha_count < 0.8