Sentiment Analysis — Why VADER Fails on 'Mild' Reviews

# pip install vaderSentiment
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer

# VADER is stateless — one instance is all you need
analyzer = SentimentIntensityAnalyzer()

# A mix of review styles to show how VADER handles edge cases
review_samples = [
    "The delivery was SUPER fast and the packaging was perfect!",  # caps as intensifier
    "It's not bad at all, actually kind of useful.",               # negation + hedging
    "Worst. Product. Ever.",                                        # dramatic punctuation
    "Meh. Does what it says on the tin.",                          # neutral/slang
    "I can't believe how great this is!!! 😍",                    # emoji + exclamation
]

print(f"{'Review':<50} {'Negative':>9} {'Neutral':>8} {'Positive':>9} {'Compound':>9}")
print("-" * 90)

for review in review_samples:
    # polarity_scores returns a dict with neg, neu, pos, and compound
    # compound is the overall score: -1.0 (most negative) to +1.0 (most positive)
    scores = analyzer.polarity_scores(review)

    # Standard VADER thresholds: >= 0.05 positive, <= -0.05 negative, else neutral
    if scores["compound"] >= 0.05:
        label = "POSITIVE"
    elif scores["compound"] <= -0.05:
        label = "NEGATIVE"
    else:
        label = "NEUTRAL"

    # Truncate review for display neatness
    short_review = review[:47] + "..." if len(review) > 47 else review
    print(
        f"{short_review:<50} "
        f"{scores['neg']:>9.3f} "
        f"{scores['neu']:>8.3f} "
        f"{scores['pos']:>9.3f} "
        f"{scores['compound']:>9.3f}  → {label}"
    )

# pip install transformers torch
from transformers import pipeline

# This downloads ~260MB on first run and caches locally.
# distilbert is a 40%-smaller, 60%-faster distillation of BERT with ~97% of the accuracy.
# It's fine-tuned on SST-2 (Stanford Sentiment Treebank), a movie review dataset.
sentiment_pipeline = pipeline(
    task="sentiment-analysis",
    model="distilbert-base-uncased-finetuned-sst-2-english",
    truncation=True,   # silently truncates text longer than 512 tokens — important!
    max_length=512
)

# Cases that are genuinely hard for rule-based systems
tricky_sentences = [
    "This is sick — easily the best thing I've bought all year.",   # slang 'sick'
    "I expected to hate it, but somehow I love it.",                # expectation reversal
    "Not the worst thing I've ever used.",                          # double negation
    "For the price, I guess it's fine.",                            # hedged, muted
    "Absolutely flawless. Completely ruined my budget though.",     # mixed sentiment
]

results = sentiment_pipeline(tricky_sentences)

print(f"{'Sentence':<55} {'Label':<10} {'Confidence':>10}")
print("-" * 80)

for sentence, result in zip(tricky_sentences, results):
    short = sentence[:52] + "..." if len(sentence) > 52 else sentence
    # result is a dict: {"label": "POSITIVE", "score": 0.9998}
    confidence_pct = result["score"] * 100
    print(f"{short:<55} {result['label']:<10} {confidence_pct:>9.2f}%")

# pip install vaderSentiment pandas
import pandas as pd
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from collections import Counter

# --- STEP 1: Simulate loading product reviews (in real life: pd.read_csv or a DB query) ---
product_reviews = [
    {"review_id": 1, "reviewer": "alice",   "text": "Absolutely love this! Fast shipping and great quality."},
    {"review_id": 2, "reviewer": "bob",     "text": "Stopped working after 3 days. Total waste of money."},
    {"review_id": 3, "reviewer": "carol",   "text": "It's okay. Nothing special but does the job."},
    {"review_id": 4, "reviewer": "dan",     "text": "Unbelievably poor customer support. Never again."},
    {"review_id": 5, "reviewer": "eve",     "text": "Pretty good for the price! Would recommend to a friend."},
    {"review_id": 6, "reviewer": "frank",   "text": "Not bad but not great either. Delivery was slow."},
    {"review_id": 7, "reviewer": "grace",   "text": "Five stars. Changed my life, not exaggerating."},
    {"review_id": 8, "reviewer": "henry",   "text": "Cheap garbage. Broke on first use. DO NOT BUY."},
    {"review_id": 9, "reviewer": "iris",    "text": "Decent product. Instructions were a bit confusing."},
    {"review_id": 10, "reviewer": "james",  "text": "Exceeded expectations. Packaging was beautiful too!"},
]

# --- STEP 2: Score each review --- 
def score_reviews(reviews: list[dict], analyzer: SentimentIntensityAnalyzer) -> pd.DataFrame:
    """Run VADER over each review and return a DataFrame with scores + label."""
    scored = []
    for review in reviews:
        scores = analyzer.polarity_scores(review["text"])
        compound = scores["compound"]

        # Map compound score to human-readable label using standard VADER thresholds
        if compound >= 0.05:
            label = "POSITIVE"
        elif compound <= -0.05:
            label = "NEGATIVE"
        else:
            label = "NEUTRAL"

        scored.append({
            "review_id":  review["review_id"],
            "reviewer":   review["reviewer"],
            "text":       review["text"],
            "compound":   round(compound, 4),
            "label":      label,
        })
    return pd.DataFrame(scored)

# --- STEP 3: Aggregate results into a summary --- 
def generate_summary(df: pd.DataFrame) -> None:
    """Print a business-readable summary of the sentiment distribution."""
    label_counts = Counter(df["label"])
    total = len(df)

    print("\n📊 SENTIMENT SUMMARY")
    print("=" * 40)
    for label in ["POSITIVE", "NEUTRAL", "NEGATIVE"]:
        count = label_counts.get(label, 0)
        pct = (count / total) * 100
        bar = "█" * int(pct / 5)  # simple ASCII bar chart
        print(f"{label:<10} {count:>3} reviews  ({pct:>5.1f}%)  {bar}")

    avg_compound = df["compound"].mean()
    print(f"\nAverage compound score: {avg_compound:.4f}")
    print(f"Overall sentiment: {'😊 Positive' if avg_compound > 0.05 else '😐 Neutral' if avg_compound > -0.05 else '😠 Negative'}")

# --- STEP 4: Flag reviews that need human attention ---
def flag_negative_reviews(df: pd.DataFrame, threshold: float = -0.3) -> None:
    """Surface the most negative reviews — the ones a human should read first."""
    flagged = df[df["compound"] <= threshold].sort_values("compound")
    print("\n🚩 REVIEWS FLAGGED FOR HUMAN REVIEW (compound ≤ {threshold})")
    print("=" * 40)
    if flagged.empty:
        print("No severely negative reviews found.")
        return
    for _, row in flagged.iterrows():
        print(f"[{row['reviewer']:>6}] score={row['compound']:>7.4f} | {row['text']}")

# --- MAIN ---
analyzer = SentimentIntensityAnalyzer()
reviews_df = score_reviews(product_reviews, analyzer)

print("\n📋 FULL REVIEW SCORES")
print(reviews_df[["reviewer", "compound", "label", "text"]].to_string(index=False))

generate_summary(reviews_df)
flag_negative_reviews(reviews_df)

# pip install transformers torch scikit-learn pandas
import pandas as pd
import torch
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
from sklearn.metrics import classification_report, confusion_matrix, ConfusionMatrixDisplay
import matplotlib.pyplot as plt

# --- Load model and tokeniser (replace with your fine-tuned model if applicable) ---
model_name = "distilbert-base-uncased-finetuned-sst-2-english"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(model_name)
sentiment_pipeline = pipeline("sentiment-analysis", model=model, tokenizer=tokenizer)

# --- Ground truth labels (0 = negative, 1 = positive) ---
test_texts = [
    "This is terrible, broke immediately.",
    "Love it! Perfect for my needs.",
    "Doesn't work as described.",
    "Excellent quality and fast shipping.",
    "Meh, it's okay I guess.",
]
# Manually labelled: 0=neg, 1=pos
y_true = [0, 1, 0, 1, 1]  # note: 'Meh' is positive? Let's keep it as neutral/positive for demo
# In reality, you'd have hundreds of labelled examples.

# --- Get predictions ---
predictions = sentiment_pipeline(test_texts)
y_pred = [1 if p['label'] == 'POSITIVE' else 0 for p in predictions]

# --- Classification report ---
print("Classification Report:")
print(classification_report(y_true, y_pred, target_names=['negative','positive']))

# --- Confusion matrix ---
cm = confusion_matrix(y_true, y_pred)
disp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=['negative','positive'])
disp.plot()
plt.title("Confusion Matrix")
plt.show()

# --- Identify misclassified examples ---
print("\nMisclassified examples:")
for i, (text, true, pred) in enumerate(zip(test_texts, y_true, y_pred)):
    if true != pred:
        print(f"  Text: {text}")
        print(f"  True: {'positive' if true else 'negative'}, Pred: {'positive' if pred else 'negative'}")
        print(f"  Confidence: {predictions[i]['score']:.3f}\n")

# pip install fastapi uvicorn transformers torch
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import pipeline
import torch
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Sentiment API", version="1.0")

# Load model once at startup
# Use GPU if available, else CPU
if torch.cuda.is_available():
    sentiment_pipeline = pipeline(
        "sentiment-analysis",
        model="distilbert-base-uncased-finetuned-sst-2-english",
        truncation=True,
        max_length=512,
        device=0  # GPU
    )
else:
    sentiment_pipeline = pipeline(
        "sentiment-analysis",
        model="distilbert-base-uncased-finetuned-sst-2-english",
        truncation=True,
        max_length=512
    )

class TextInput(BaseModel):
    text: str

class BatchInput(BaseModel):
    texts: list[str]

@app.get("/health")
def health():
    return {"status": "ok"}

@app.post("/predict")
def predict_single(input: TextInput):
    try:
        result = sentiment_pipeline(input.text)[0]
        score = result['score']
        label = result['label']
        logger.info(f"Predicted {label} with confidence {score:.3f} for text: {input.text[:50]}...")
        return {
            "text": input.text,
            "sentiment": label,
            "confidence": score
        }
    except Exception as e:
        logger.error(f"Prediction failed: {e}")
        raise HTTPException(status_code=500, detail="Prediction error")

@app.post("/predict_batch")
def predict_batch(input: BatchInput):
    """Batch predict for throughput — use this endpoint for bulk processing."""
    try:
        results = sentiment_pipeline(input.texts, batch_size=32)
        outputs = []
        for text, res in zip(input.texts, results):
            outputs.append({
                "text": text,
                "sentiment": res['label'],
                "confidence": res['score']
            })
        return outputs
    except Exception as e:
        logger.error(f"Batch prediction failed: {e}")
        raise HTTPException(status_code=500, detail="Batch prediction error")

# Run with: uvicorn deploy_sentiment_api:app --host 0.0.0.0 --port 8000