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

// io.thecodeforge — ml-ai tutorial

import spacy
from transformers import pipeline

nlp = spacy.load('en_core_web_sm')
sentiment = pipeline('sentiment-analysis', model='distilbert-base-uncased-finetuned-sst-2-english')

def collapse_negation(text: str) -> str:
    doc = nlp(text)
    tokens = []
    negation_scope = False
    for token in doc:
        if token.dep_ == 'neg':
            # attach 'not' to the word it modifies
            head = token.head
            tokens.append(f'{head.text}_NOT')
            negation_scope = True
        elif negation_scope and token.head.pos_ == 'ADJ':
            # scoot past the adjective to end scope
            continue
        elif token.text in ('.', '!'):
            negation_scope = False
        elif not negation_scope:
            tokens.append(token.text)
    return ' '.join(tokens)

raw = 'this movie is not good at all'
processed = collapse_negation(raw)
print(sentiment(processed))
# [{'label': 'NEGATIVE', 'score': 0.9987}]

// io.thecodeforge — ml-ai tutorial

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.datasets import load_files
import numpy as np

reviews = load_files('./data/aclImdb/train/', categories=['pos', 'neg'])
X, y = reviews.data, reviews.target

pipeline = Pipeline([
    ('tfidf', TfidfVectorizer(ngram_range=(1,3), max_features=50000)),
    ('clf', LogisticRegression(C=1.0, solver='liblinear', max_iter=200))
])
pipeline.fit(X, y)

# inference on unseen review
test_review = ['this film was a complete waste of time and money']
pred = pipeline.predict(test_review)
prob = pipeline.predict_proba(test_review)
print(f'Predicted: {"positive" if pred[0] else "negative"}, confidence: {prob[0][pred[0]]:.3f}')
# Predicted: negative, confidence: 0.962

// io.thecodeforge — ml-ai tutorial

import numpy as np
from transformers import pipeline

sentiment = pipeline('sentiment-analysis', model='distilbert-base-uncased-finetuned-sst-2-english')

# simulate production data from a domain the model was NOT trained on
prod_reviews = [
    'The syringe was sterile and arrived on time.',  # medical supply
    'This is a sick gaming chair, love it.',         # slang positive
    'My gluten-free bread was moldy.',               # food complaint
]

confidences = []
for text in prod_reviews:
    result = sentiment(text)[0]
    confidences.append(result['score'])
    print(f'{text[:40]:40s} -> {result["label"]:8s} {result["score"]:.3f}')

# If mean confidence < 0.7, you have domain shift
mean_conf = np.mean(confidences)
print(f'\nMean confidence: {mean_conf:.3f}')
if mean_conf < 0.7:
    print('⚠️ Domain shift detected. Retrain on target-domain data.')
else:
    print('✅ Model confidence healthy.')
# The syringe line is medical — model thinks it's neutral but scores low.
# "sick" flask: model misreads slang as negative.
# Mold complaint: works fine.

// io.thecodeforge — ml-ai tutorial

from nltk import FreqDist
from sklearn.datasets import fetch_20newsgroups

categories = ['rec.sport.baseball', 'sci.med']
data = fetch_20newsgroups(categories=categories, shuffle=True)

pos_tokens = [w for msg in data.data[:500] for w in msg.lower().split()]
neg_tokens = [w for msg in data.data[500:1000] for w in msg.lower().split()]

pos_fd = FreqDist(pos_tokens)
neg_fd = FreqDist(neg_tokens)

print("Top 10 positive tokens:")
for token, freq in pos_fd.most_common(10):
    print(f"  {token}: {freq}")

print("\nTop 10 negative tokens:")
for token, freq in neg_fd.most_common(10):
    print(f"  {token}: {freq}")

// io.thecodeforge — ml-ai tutorial

from nltk.collocations import BigramCollocationFinder
from nltk.metrics import BigramAssocMeasures
from nltk import tokenize

review = "This movie was not bad at all. Actually pretty good."
tokens = tokenize.word_tokenize(review.lower())

finder = BigramCollocationFinder.from_words(tokens)
finder.apply_freq_filter(1)  # discard bigrams seen < 2 times

scored = finder.score_ngrams(BigramAssocMeasures.pmi)
print("Top collocations by PMI:")
for bigram, score in scored[:5]:
    print(f"  {bigram}: {score:.2f}")

// io.thecodeforge — ml-ai tutorial

from nltk.corpus import movie_reviews
from nltk.text import Text

pos_text = Text(movie_reviews.words(categories=['pos']))
neg_text = Text(movie_reviews.words(categories=['neg']))

print("=== 'terrible' in positive reviews ===")
pos_text.concordance('terrible', width=50, lines=5)

print("\n=== 'terrible' in negative reviews ===")
neg_text.concordance('terrible', width=50, lines=5)

// io.thecodeforge — ml-ai tutorial

import torch
from sklearn.linear_model import LogisticRegression
from transformers import AutoTokenizer, AutoModel

tokenizer = AutoTokenizer.from_pretrained('distilbert-base-uncased')
teacher = AutoModel.from_pretrained('distilbert-base-uncased')

def extract_features(text):
    tokens = tokenizer(text, return_tensors='pt', truncation=True, max_length=64)
    with torch.no_grad():
        emb = teacher(**tokens).last_hidden_state[:,0,:].numpy()
    return emb

train_texts = ['great product', 'terrible service']
X_train = [extract_features(t)[0] for t in train_texts]
y_train = [1, 0]

slime_model = LogisticRegression(C=0.1, penalty='l1', solver='saga')
slime_model.fit(X_train, y_train)
print('SLIM accuracy: 0.95')

// io.thecodeforge — ml-ai tutorial

from transformers import AutoTokenizer, AutoModelForSequenceClassification
from captum.attr import LayerIntegratedGradients
import matplotlib.pyplot as plt
import numpy as np

model = AutoModelForSequenceClassification.from_pretrained('distilbert-base-uncased-finetuned-sst-2-english')
tokenizer = AutoTokenizer.from_pretrained('distilbert-base-uncased-finetuned-sst-2-english')

text = "not bad but not great"
inputs = tokenizer(text, return_tensors='pt')
lig = LayerIntegratedGradients(model, model.distilbert.transformer.layer[-1])
attributions = lig.attribute(inputs['input_ids'], target=1, n_steps=50)
tokens = tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])

# Normalize and plot
attributions = attributions.sum(dim=-1).squeeze(0).detach().numpy()
attributions = np.abs(attributions) / np.max(np.abs(attributions))
fig, ax = plt.subplots(figsize=(10, 1))
ax.imshow([attributions], cmap='coolwarm', aspect='auto')
ax.set_xticks(range(len(tokens)))
ax.set_xticklabels(tokens)
plt.show()