Lab — Classify and Cluster

from datasets import load_dataset  # uv add datasets

# Load IMDB dataset
dataset = load_dataset("imdb")

# Shuffle — IMDB is sorted by label
dataset["train"] = dataset["train"].shuffle(seed=42)
dataset["test"] = dataset["test"].shuffle(seed=42)

# Use subsets for faster iteration
train_texts = dataset["train"]["text"][:5000]
train_labels = dataset["train"]["label"][:5000]
test_texts = dataset["test"]["text"][:2000]
test_labels = dataset["test"]["label"][:2000]

print(f"Training: {len(train_texts):,} reviews")
print(f"Test:     {len(test_texts):,} reviews")
print(f"Balance:  {sum(train_labels)} positive, {len(train_labels) - sum(train_labels)} negative")

import time
from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.linear_model import LogisticRegression
from sklearn.svm import LinearSVC
from sklearn.metrics import accuracy_score, f1_score

classifiers = {
    "Naive Bayes": MultinomialNB(),
    "Logistic Regression": LogisticRegression(max_iter=1000),
    "SVM (Linear)": LinearSVC(max_iter=1000),
}

results = {}

for name, clf in classifiers.items():
    pipe = Pipeline([
        ("tfidf", TfidfVectorizer(max_features=20000, ngram_range=(1, 2))),
        ("clf", clf),
    ])

    start = time.time()
    pipe.fit(train_texts, train_labels)
    train_time = time.time() - start

    preds = pipe.predict(test_texts)
    acc = accuracy_score(test_labels, preds)
    f1 = f1_score(test_labels, preds, average="weighted")

    results[name] = {
        "pipeline": pipe,
        "predictions": preds,
        "accuracy": acc,
        "f1": f1,
        "train_time": train_time,
    }
    print(f"{name:25s}  accuracy={acc:.3f}  F1={f1:.3f}  time={train_time:.2f}s")

import matplotlib.pyplot as plt
import numpy as np

names = list(results.keys())
accs = [results[n]["accuracy"] for n in names]
f1s = [results[n]["f1"] for n in names]
times = [results[n]["train_time"] for n in names]

fig, axes = plt.subplots(1, 2, figsize=(10, 4))

# Accuracy and F1
x = np.arange(len(names))
width = 0.35
axes[0].bar(x - width / 2, accs, width, label="Accuracy", color="#4C72B0")
axes[0].bar(x + width / 2, f1s, width, label="F1 (weighted)", color="#55A868")
axes[0].set_xticks(x)
axes[0].set_xticklabels(names, fontsize=9)
axes[0].set_ylim(0.8, 0.92)
axes[0].set_ylabel("Score")
axes[0].set_title("Classification Performance")
axes[0].legend()

# Training time
axes[1].bar(names, times, color="#C44E52")
axes[1].set_ylabel("Seconds")
axes[1].set_title("Training Time")

plt.tight_layout()
plt.show()

import pandas as pd

# Get the Logistic Regression pipeline
lr_pipe = results["Logistic Regression"]["pipeline"]
features = lr_pipe["tfidf"].get_feature_names_out()
weights = lr_pipe["clf"].coef_[0]

weight_df = pd.DataFrame({"feature": features, "weight": weights})
weight_df = weight_df.sort_values("weight", ascending=False)

print("Top 10 features for POSITIVE sentiment:")
print(weight_df.head(10).to_string(index=False))
print("\nTop 10 features for NEGATIVE sentiment:")
print(weight_df.tail(10).to_string(index=False))

# Use the SVM predictions (our best performer)
best_preds = results["SVM (Linear)"]["predictions"]
label_names = ["negative", "positive"]

errors = []
for text, true_l, pred_l in zip(test_texts, test_labels, best_preds):
    if true_l != pred_l:
        errors.append({"text": text, "true": label_names[true_l], "predicted": label_names[pred_l]})

print(f"Errors: {len(errors)} out of {len(test_texts)} ({len(errors) / len(test_texts):.1%})")

# False positives: predicted positive, actually negative
false_positives = [e for e in errors if e["true"] == "negative"]
print(f"=== False Positives ({len(false_positives)} total) ===")
print(f"(Predicted POSITIVE, actually NEGATIVE)\n")
for e in false_positives[:3]:
    print(f"  {e['text'][:200]}...")
    print()

# False negatives: predicted negative, actually positive
false_negatives = [e for e in errors if e["true"] == "positive"]
print(f"=== False Negatives ({len(false_negatives)} total) ===")
print(f"(Predicted NEGATIVE, actually POSITIVE)\n")
for e in false_negatives[:3]:
    print(f"  {e['text'][:200]}...")
    print()

# Let's check for some telltale patterns in misclassified reviews
negation_words = {"not", "no", "never", "neither", "barely", "hardly", "doesn't", "didn't", "wasn't", "won't", "can't", "couldn't", "shouldn't"}
contrast_words = {"but", "however", "although", "though", "yet", "despite", "except"}

for category, keywords in [("Negation", negation_words), ("Contrast/mixed", contrast_words)]:
    count = 0
    for e in errors:
        words = set(e["text"].lower().split())
        if words & keywords:
            count += 1
    print(f"Errors containing {category.lower()} words: {count}/{len(errors)} ({count / len(errors):.0%})")

from sklearn.metrics import classification_report, confusion_matrix, ConfusionMatrixDisplay

print(classification_report(test_labels, best_preds, target_names=label_names))

cm = confusion_matrix(test_labels, best_preds)
disp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=label_names)
disp.plot(cmap="Blues")
plt.title("SVM Sentiment Classifier — Confusion Matrix")
plt.show()

from sklearn.datasets import fetch_20newsgroups

categories = ["rec.sport.baseball", "sci.space", "comp.graphics", "talk.politics.guns"]

newsgroups = fetch_20newsgroups(
    subset="train",
    categories=categories,
    remove=("headers", "footers", "quotes"),  # Strip metadata to focus on content
)

print(f"Loaded {len(newsgroups.data):,} documents")
print(f"Categories: {newsgroups.target_names}")
print(f"\nSample document (first 200 chars):")
print(f"  {newsgroups.data[0][:200]}...")
print(f"  Category: {newsgroups.target_names[newsgroups.target[0]]}")

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans

# Vectorize
vectorizer = TfidfVectorizer(max_features=5000, stop_words="english")
X_tfidf = vectorizer.fit_transform(newsgroups.data)
print(f"TF-IDF matrix: {X_tfidf.shape[0]} documents × {X_tfidf.shape[1]} features")

# Cluster
km = KMeans(n_clusters=4, random_state=42, n_init=10)
km.fit(X_tfidf)

# What are the clusters about? Look at the top terms near each centroid
feature_names = vectorizer.get_feature_names_out()

print("\n--- Top terms per cluster ---")
for i in range(4):
    top_indices = km.cluster_centers_[i].argsort()[-10:][::-1]
    top_terms = [feature_names[j] for j in top_indices]
    print(f"  Cluster {i}: {', '.join(top_terms)}")

import numpy as np
from collections import Counter

print("--- Cluster composition ---")
for i in range(4):
    cluster_mask = km.labels_ == i
    true_labels = np.array(newsgroups.target)[cluster_mask]
    label_counts = Counter(true_labels)
    total = sum(label_counts.values())
    print(f"\nCluster {i} ({total} docs):")
    for label_id, count in label_counts.most_common():
        pct = count / total * 100
        print(f"  {newsgroups.target_names[label_id]:30s} {count:4d} ({pct:.0f}%)")

inertias = []
K_range = range(2, 9)

for k in K_range:
    km_temp = KMeans(n_clusters=k, random_state=42, n_init=10)
    km_temp.fit(X_tfidf)
    inertias.append(km_temp.inertia_)

plt.figure(figsize=(7, 4))
plt.plot(list(K_range), inertias, "o-", color="#4C72B0")
plt.xlabel("Number of clusters (K)")
plt.ylabel("Inertia")
plt.title("Elbow Method for K-Means")
plt.xticks(list(K_range))
plt.grid(alpha=0.3)
plt.show()

from sklearn.feature_extraction.text import CountVectorizer
from sklearn.decomposition import LatentDirichletAllocation

# LDA works with raw counts, not TF-IDF
count_vec = CountVectorizer(max_features=5000, stop_words="english")
X_counts = count_vec.fit_transform(newsgroups.data)

# Fit LDA with 4 topics
lda = LatentDirichletAllocation(n_components=4, random_state=42, max_iter=10)
lda.fit(X_counts)

# Show top words per topic
lda_feature_names = count_vec.get_feature_names_out()

print("--- LDA Topics ---")
for i, topic in enumerate(lda.components_):
    top_indices = topic.argsort()[-10:][::-1]
    top_terms = [lda_feature_names[j] for j in top_indices]
    print(f"  Topic {i}: {', '.join(top_terms)}")

# Transform a few documents to see their topic mixtures
topic_distributions = lda.transform(X_counts[:5])

for i in range(5):
    print(f"Document {i} (true: {newsgroups.target_names[newsgroups.target[i]]})")
    for j in range(4):
        bar = "█" * int(topic_distributions[i][j] * 30)
        print(f"  Topic {j}: {topic_distributions[i][j]:.2f} {bar}")
    print()

import spacy

nlp = spacy.load("en_core_web_sm")

domain_texts = {
    "News": (
        "President Biden met with Chancellor Scholz in Berlin on Monday "
        "to discuss a $50 billion NATO defense package."
    ),
    "Medical": (
        "Patient was prescribed Metformin 500mg and Lisinopril 10mg. "
        "Dr. Sarah Chen at Mayo Clinic recommends follow-up in March."
    ),
    "Social media": (
        "omg just saw Beyonce at Coachella in Indio CA!!! "
        "she was wearing Gucci and arrived in a Tesla lol #iconic"
    ),
    "Legal": (
        "The defendant, John Smith, is charged under Section 18 U.S.C. 1343. "
        "Judge Martinez of the Southern District of New York will preside."
    ),
    "Scientific": (
        "Researchers at Stanford University used the CRISPR-Cas9 system "
        "to modify the BRCA1 gene in a study published in Nature Genetics."
    ),
}

for domain, text in domain_texts.items():
    doc = nlp(text)
    ents = [(ent.text, ent.label_) for ent in doc.ents]
    print(f"{domain}:")
    for ent_text, ent_label in ents:
        print(f"  {ent_text:30s} → {ent_label}")
    if not ents:
        print("  (no entities detected)")
    print()

# Gold annotations: (text, expected_entities)
gold_annotations = {
    "News": [
        ("Biden", "PERSON"), ("Scholz", "PERSON"), ("Berlin", "GPE"),
        ("Monday", "DATE"), ("$50 billion", "MONEY"), ("NATO", "ORG"),
    ],
    "Medical": [
        ("Metformin", "PRODUCT"), ("Lisinopril", "PRODUCT"),
        ("Sarah Chen", "PERSON"), ("Mayo Clinic", "ORG"), ("March", "DATE"),
    ],
    "Social media": [
        ("Beyonce", "PERSON"), ("Coachella", "EVENT"), ("Indio", "GPE"),
        ("CA", "GPE"), ("Gucci", "ORG"), ("Tesla", "ORG"),
    ],
    "Legal": [
        ("John Smith", "PERSON"), ("Martinez", "PERSON"),
        ("Southern District of New York", "ORG"),
    ],
    "Scientific": [
        ("Stanford University", "ORG"), ("Nature Genetics", "ORG"),
    ],
}

error_types = {"boundary": 0, "type_confusion": 0, "missed": 0, "spurious": 0}

for domain, text in domain_texts.items():
    doc = nlp(text)
    pred = {(ent.text, ent.label_) for ent in doc.ents}
    gold = set(gold_annotations[domain])

    pred_texts = {ent_text for ent_text, _ in pred}
    gold_texts = {ent_text for ent_text, _ in gold}

    for ent_text, ent_label in gold:
        if (ent_text, ent_label) in pred:
            continue  # Correct
        elif ent_text in pred_texts:
            error_types["type_confusion"] += 1
        else:
            # Check for partial overlap (boundary error)
            found_partial = False
            for p_text, _ in pred:
                if ent_text in p_text or p_text in ent_text:
                    error_types["boundary"] += 1
                    found_partial = True
                    break
            if not found_partial:
                error_types["missed"] += 1

    for ent_text, ent_label in pred:
        if ent_text not in gold_texts:
            # Check if it's a boundary variant
            found_partial = False
            for g_text, _ in gold:
                if ent_text in g_text or g_text in ent_text:
                    found_partial = True
                    break
            if not found_partial:
                error_types["spurious"] += 1

print("--- NER Error Summary Across Domains ---")
for err_type, count in error_types.items():
    print(f"  {err_type:20s}: {count}")