12 KiB
12 KiB
In [ ]:
from pathlib import Path
import numpy as np
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
accuracy_score,
confusion_matrix,
f1_score,
precision_score,
recall_score,
)
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
DATA_PATH = Path("data/SMSSpamCollection")
RANDOM_STATE = 42
THRESHOLDS = np.arange(0.20, 0.81, 0.05)
if not DATA_PATH.exists():
raise FileNotFoundError(
"Missing data/SMSSpamCollection. Follow data/README.md to download the official dataset."
)
In [ ]:
df = pd.read_csv(DATA_PATH, sep="\t", header=None, names=["label", "message"])
df["label"] = df["label"].astype("string").str.strip().str.lower()
df["message"] = df["message"].astype("string").str.strip()
df = df.dropna(subset=["label", "message"])
df = df[df["label"].isin(["ham", "spam"]) & df["message"].ne("")]
conflict_counts = df.groupby("message")["label"].nunique()
df = df[~df["message"].isin(conflict_counts[conflict_counts.gt(1)].index)]
df = df.drop_duplicates(subset=["label", "message"]).reset_index(drop=True)
X = df["message"]
y = df["label"].map({"ham": 0, "spam": 1}).astype(int)
In [ ]:
X_train, X_temp, y_train, y_temp = train_test_split(
X,
y,
test_size=0.40,
random_state=RANDOM_STATE,
stratify=y,
)
X_valid, X_test, y_valid, y_test = train_test_split(
X_temp,
y_temp,
test_size=0.50,
random_state=RANDOM_STATE,
stratify=y_temp,
)
print("Train / validation / test:", len(X_train), len(X_valid), len(X_test))
print("Spam rates:", f"{y_train.mean():.2%}", f"{y_valid.mean():.2%}", f"{y_test.mean():.2%}")
In [ ]:
lr_pipeline = Pipeline([
("tfidf", TfidfVectorizer(
lowercase=True,
ngram_range=(1, 2),
min_df=1,
sublinear_tf=True,
)),
("model", LogisticRegression(
max_iter=2000,
random_state=RANDOM_STATE,
)),
])
lr_pipeline.fit(X_train, y_train)
valid_probabilities = lr_pipeline.predict_proba(X_valid)[:, 1]
print("First probabilities:", np.round(valid_probabilities[:10], 3))
In [ ]:
rows = []
for threshold in THRESHOLDS:
predictions = (valid_probabilities >= threshold).astype(int)
cm = confusion_matrix(y_valid, predictions, labels=[0, 1])
tn, fp, fn, tp = cm.ravel()
rows.append({
"threshold": threshold,
"spam_precision": precision_score(y_valid, predictions, pos_label=1, zero_division=0),
"spam_recall": recall_score(y_valid, predictions, pos_label=1, zero_division=0),
"spam_f1": f1_score(y_valid, predictions, pos_label=1, zero_division=0),
"false_positives": int(fp),
"false_negatives": int(fn),
})
threshold_table = pd.DataFrame(rows).set_index("threshold")
eligible = threshold_table[threshold_table["spam_recall"].ge(0.80)]
if eligible.empty:
selected_threshold = float(threshold_table["spam_f1"].idxmax())
else:
selected_threshold = float(
eligible.sort_values(["spam_precision", "spam_f1"], ascending=False).index[0]
)
print(threshold_table.round(4))
print("Selected threshold:", selected_threshold)
In [ ]:
test_probabilities = lr_pipeline.predict_proba(X_test)[:, 1]
test_predictions = (test_probabilities >= selected_threshold).astype(int)
test_cm = confusion_matrix(y_test, test_predictions, labels=[0, 1])
tn, fp, fn, tp = test_cm.ravel()
test_metrics = {
"threshold": selected_threshold,
"accuracy": accuracy_score(y_test, test_predictions),
"spam_precision": precision_score(y_test, test_predictions, pos_label=1, zero_division=0),
"spam_recall": recall_score(y_test, test_predictions, pos_label=1, zero_division=0),
"spam_f1": f1_score(y_test, test_predictions, pos_label=1, zero_division=0),
"false_positives": int(fp),
"false_negatives": int(fn),
}
print(pd.Series(test_metrics).round(4))
print(test_cm)
In [ ]:
feature_names = lr_pipeline.named_steps["tfidf"].get_feature_names_out()
coefficients = lr_pipeline.named_steps["model"].coef_[0]
coefficient_series = pd.Series(coefficients, index=feature_names)
top_ham = coefficient_series.nsmallest(15).rename("coefficient").reset_index()
top_ham.columns = ["term", "coefficient"]
top_ham["direction"] = "ham"
top_spam = coefficient_series.nlargest(15).rename("coefficient").reset_index()
top_spam.columns = ["term", "coefficient"]
top_spam["direction"] = "spam"
top_features = pd.concat([top_spam, top_ham], ignore_index=True)
top_features
In [ ]:
policy_actions = pd.cut(
test_probabilities,
bins=[-np.inf, 0.30, 0.70, np.inf],
labels=["allow", "analyst_review", "quarantine"],
right=False,
)
action_summary = pd.crosstab(
pd.Series(y_test.to_numpy(), name="actual"),
pd.Series(policy_actions, name="action"),
).rename(index={0: "ham", 1: "spam"})
action_summary
In [ ]:
assert len(X_train) + len(X_valid) + len(X_test) == len(df)
assert set(lr_pipeline.named_steps) == {"tfidf", "model"}
assert np.all((valid_probabilities >= 0) & (valid_probabilities <= 1))
assert threshold_table.index.is_monotonic_increasing
assert selected_threshold in threshold_table.index
assert len(test_predictions) == len(y_test)
assert test_cm.shape == (2, 2)
assert len(top_features) == 30
assert set(top_features["direction"]) == {"ham", "spam"}
assert set(policy_actions.astype("string").unique()).issubset({"allow", "analyst_review", "quarantine"})
print("✓ Lab 5 completed")