Naive Bayes - 35% False Positive from Imbalanced Priors

# bayes_theorem_walkthrough.py
# Let's verify Bayes' theorem manually before using any library.
# We'll use a medical test scenario: does a patient have a rare disease?

# ---- Setup: prior knowledge from medical literature ----
prob_has_disease = 0.01          # 1% of the population has this disease (prior)
prob_no_disease = 1 - prob_has_disease  # 99% do not

# The test is 95% accurate:
# If you HAVE the disease, it correctly says 'positive' 95% of the time
prob_positive_given_disease = 0.95

# If you DON'T have the disease, it incorrectly says 'positive' 5% of the time
prob_positive_given_no_disease = 0.05

# ---- Step 1: Calculate the total probability of testing positive ----
# This accounts for BOTH true positives and false positives
prob_positive = (
    prob_positive_given_disease * prob_has_disease
    + prob_positive_given_no_disease * prob_no_disease
)

# ---- Step 2: Apply Bayes' Theorem ----
# P(disease | positive test) = P(positive | disease) * P(disease) / P(positive)
prob_disease_given_positive = (
    prob_positive_given_disease * prob_has_disease
) / prob_positive

print(f"Probability of testing positive overall:       {prob_positive:.4f} ({prob_positive*100:.2f}%)")
print(f"Probability of ACTUALLY having the disease")
print(f"  AFTER a positive test result:               {prob_disease_given_positive:.4f} ({prob_disease_given_positive*100:.2f}%)")
print()
print("Key insight: Even with a 95%-accurate test,")
print(f"a positive result only means {prob_disease_given_positive*100:.1f}% chance of having the disease.")
print("The low prior (1% prevalence) dominates the math.")
print("This is why base rates matter enormously in Naive Bayes.")

# naive_bayes_from_scratch.py
# A fully hand-rolled Naive Bayes spam classifier.
# Every probability is computed manually — no sklearn magic here.

import math
from collections import defaultdict

# ---- Training data: (message, label) pairs ----
training_emails = [
    ("free money click here now",           "spam"),
    ("win a free iphone congratulations",   "spam"),
    ("cheap pills free offer limited time", "spam"),
    ("click here to claim your prize free", "spam"),
    ("you won congratulations claim now",   "spam"),
    ("meeting at 3pm in the boardroom",     "ham"),
    ("can you review my pull request",      "ham"),
    ("lunch tomorrow works for me",         "ham"),
    ("please send the quarterly report",    "ham"),
    ("the deployment is scheduled friday",  "ham"),
]

# ---- Step 1: Count word frequencies per class ----
word_counts = {"spam": defaultdict(int), "ham": defaultdict(int)}
class_doc_counts = {"spam": 0, "ham": 0}
vocabulary = set()

for message, label in training_emails:
    class_doc_counts[label] += 1
    for word in message.split():
        word_counts[label][word] += 1
        vocabulary.add(word)          # build the full vocabulary

total_docs = sum(class_doc_counts.values())
vocab_size = len(vocabulary)

print(f"Vocabulary size:  {vocab_size} unique words")
print(f"Spam messages:    {class_doc_counts['spam']}")
print(f"Ham messages:     {class_doc_counts['ham']}")
print()

# ---- Step 2: Calculate prior log-probabilities for each class ----
# log() turns multiplication into addition — avoids floating-point underflow
log_prior = {
    label: math.log(count / total_docs)
    for label, count in class_doc_counts.items()
}

# ---- Step 3: Define prediction function with Laplace smoothing ----
def classify_message(message: str) -> tuple[str, dict]:
    """
    Classify a message as 'spam' or 'ham'.
    Returns the predicted label and the log-probability scores for both classes.
    """
    words = message.lower().split()
    log_scores = {}

    for label in ["spam", "ham"]:
        # Start with the prior probability for this class
        score = log_prior[label]

        # Total words seen in this class (for denominator)
        total_words_in_class = sum(word_counts[label].values())

        for word in words:
            # Laplace smoothing: add 1 to numerator, vocab_size to denominator
            # This prevents any word from having zero probability
            word_count_in_class = word_counts[label].get(word, 0)
            smoothed_probability = (
                (word_count_in_class + 1)
                / (total_words_in_class + vocab_size)
            )
            # Add log-probability instead of multiplying raw probability
            score += math.log(smoothed_probability)

        log_scores[label] = score

    predicted_label = max(log_scores, key=log_scores.get)
    return predicted_label, log_scores

# ---- Step 4: Test on new, unseen messages ----
test_messages = [
    "free offer click here win prize",
    "can we reschedule the meeting to friday",
    "congratulations you won a free phone",
    "the report is ready for your review",
]

print("=" * 55)
print(f"{'Message':<38} {'Prediction':>10}")
print("=" * 55)

for msg in test_messages:
    prediction, scores = classify_message(msg)
    print(f"{msg[:37]:<38} {prediction:>10}")
    print(f"  spam score: {scores['spam']:.3f}  |  ham score: {scores['ham']:.3f}")
    print()

# spam_classifier_sklearn.py
# Production-grade spam classifier using sklearn.
# Includes pipeline, evaluation metrics, and probability calibration.

from sklearn.naive_bayes import MultinomialNB
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import classification_report, confusion_matrix
from sklearn.calibration import CalibratedClassifierCV
import numpy as np

# ---- Dataset: realistic spam/ham examples ----
email_messages = [
    # SPAM
    "WINNER! You have been selected. Claim your FREE prize now!",
    "Cheap Viagra! Buy online, no prescription needed.",
    "Make money fast from home — $5000/week guaranteed!",
    "Your account has been suspended click here immediately",
    "Congratulations! You won our lottery drawing. Reply now.",
    "FREE iPhone 15 Pro — limited time offer, click to claim",
    "URGENT: Your bank account needs verification now",
    "Hot singles in your area — click to meet them tonight",
    "Earn passive income with this one weird trick",
    "You have been pre-approved for a $50,000 loan no credit check",
    # HAM
    "Can you send me the updated project timeline?",
    "The sprint retrospective is moved to Thursday 2pm.",
    "I reviewed your PR — a few comments on the auth module.",
    "Quarterly revenue figures are attached. Let me know your thoughts.",
    "Are you joining the team lunch on Friday?",
    "The deployment went smoothly. All services are green.",
    "Could you review the new API documentation draft?",
    "Reminder: performance reviews are due by end of month.",
    "Thanks for the feedback on the design mockups.",
    "The client approved the proposal. Kickoff is next Monday.",
]

labels = (
    ["spam"] * 10  # first 10 are spam
    + ["ham"] * 10  # last 10 are ham
)

# ---- Split data ----
messages_train, messages_test, labels_train, labels_test = train_test_split(
    email_messages, labels,
    test_size=0.25,
    random_state=42,
    stratify=labels   # keeps class ratio balanced across train/test
)

# ---- Build a Pipeline: TF-IDF vectorisation + Naive Bayes ----
# TF-IDF is better than raw counts — it downweights common words like 'the'
spam_pipeline = Pipeline([
    (
        "tfidf_vectorizer",
        TfidfVectorizer(
            ngram_range=(1, 2),    # use single words AND two-word phrases
            min_df=1,              # include words appearing at least once
            stop_words="english",  # ignore 'the', 'is', 'and', etc.
            sublinear_tf=True,     # apply log scaling to term frequency
        )
    ),
    (
        "naive_bayes_classifier",
        MultinomialNB(alpha=1.0)   # alpha=1.0 is standard Laplace smoothing
    ),
])

# ---- Cross-validation score on training data ----
cv_scores = cross_val_score(
    spam_pipeline, messages_train, labels_train,
    cv=3, scoring="f1_macro"
)
print(f"Cross-validation F1 scores:  {cv_scores.round(3)}")
print(f"Mean CV F1:                  {cv_scores.mean():.3f}")
print()

# ---- Train and evaluate on test set ----
spam_pipeline.fit(messages_train, labels_train)
predictions = spam_pipeline.predict(messages_test)

print("Classification Report:")
print(classification_report(labels_test, predictions, target_names=["ham", "spam"]))

print("Confusion Matrix (rows=actual, cols=predicted):")
print(f"              Predicted Ham  Predicted Spam")
cm = confusion_matrix(labels_test, predictions, labels=["ham", "spam"])
print(f"Actual Ham          {cm[0][0]}              {cm[0][1]}")
print(f"Actual Spam         {cm[1][0]}              {cm[1][1]}")
print()

# ---- Show confidence scores for new messages ----
new_emails = [
    "You have won a free holiday package. Call now!",
    "Please review the attached contract before signing.",
]

probabilities = spam_pipeline.predict_proba(new_emails)
class_labels = spam_pipeline.classes_

print("Probability Breakdown for New Emails:")
print("-" * 55)
for email, prob_row in zip(new_emails, probabilities):
    ham_prob = prob_row[list(class_labels).index("ham")]
    spam_prob = prob_row[list(class_labels).index("spam")]
    verdict = "SPAM" if spam_prob > ham_prob else "HAM"
    print(f"Email: '{email[:45]}...'")
    print(f"  Ham probability:  {ham_prob:.3f}")
    print(f"  Spam probability: {spam_prob:.3f}  →  Verdict: {verdict}")
    print()

# naive_bayes_incremental_learning.py
# Demonstrates partial_fit() — training Naive Bayes incrementally.
# Perfect for scenarios where data arrives in batches (streaming, live systems).

from sklearn.naive_bayes import MultinomialNB
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.metrics import accuracy_score

# HashingVectorizer doesn't need to 'see' all data upfront — perfect for streaming
# It hashes words to a fixed-size feature vector without storing a vocabulary
vectorizer = HashingVectorizer(
    n_features=2**14,      # 16,384 feature buckets — enough for most text tasks
    alternate_sign=False,  # MultinomialNB requires non-negative values
    norm=None,             # raw counts, not normalised (MultinomialNB prefers this)
    stop_words="english",
)

# MultinomialNB with partial_fit — must declare all classes upfront
online_classifier = MultinomialNB(alpha=1.0)
all_classes = ["spam", "ham"]

# ---- Simulate three batches of incoming emails ----
batch_1 = [
    ("free money win cash prize now",        "spam"),
    ("meeting scheduled for 10am tomorrow",  "ham"),
    ("click here claim your free reward",    "spam"),
    ("please approve the budget proposal",   "ham"),
]

batch_2 = [
    ("urgent bank account suspended verify", "spam"),
    ("team offsite is confirmed for June",   "ham"),
    ("you won congratulations call now",     "spam"),
    ("deployment pipeline updated",          "ham"),
]

batch_3 = [
    ("cheap pills no prescription needed",   "spam"),
    ("client feedback received review docs", "ham"),
    ("earn money from home guaranteed",      "spam"),
    ("sprint planning at 9am Monday",        "ham"),
]

def train_on_batch(batch, batch_number):
    """Vectorise one batch and update the classifier using partial_fit."""
    texts, batch_labels = zip(*batch)  # unzip into separate lists

    # Transform text into feature vectors
    feature_matrix = vectorizer.transform(texts)

    # partial_fit updates the model WITHOUT forgetting what it learned before
    online_classifier.partial_fit(
        feature_matrix, batch_labels,
        classes=all_classes  # required on the FIRST call; harmless on subsequent calls
    )
    print(f"Batch {batch_number} processed — {len(batch)} messages ingested.")


def evaluate_on_held_out():
    """Test on a fixed set to see how accuracy improves with each batch."""
    test_messages = [
        "win a free iPhone click here",    # spam
        "can you review the pull request", # ham
        "guaranteed passive income online", # spam
        "the invoice is attached for Q2",  # ham
    ]
    true_labels = ["spam", "ham", "spam", "ham"]

    test_features = vectorizer.transform(test_messages)
    predictions = online_classifier.predict(test_features)
    accuracy = accuracy_score(true_labels, predictions)

    for msg, true, pred in zip(test_messages, true_labels, predictions):
        status = "✓" if true == pred else "✗"
        print(f"  {status} [{true:>4}] predicted [{pred:>4}]: '{msg[:40]}'")
    print(f"  Accuracy after this batch: {accuracy:.0%}")
    print()


# ---- Train incrementally, evaluate after each batch ----
for batch_num, batch_data in enumerate([batch_1, batch_2, batch_3], start=1):
    train_on_batch(batch_data, batch_num)
    print(f"Model state after Batch {batch_num}:")
    evaluate_on_held_out()

# calibrate_naive_bayes.py
# Demonstrate probability calibration for Naive Bayes

from sklearn.naive_bayes import MultinomialNB
from sklearn.calibration import CalibratedClassifierCV
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
import numpy as np

# Simulate a small spam dataset
emails = [
    "free money win cash now",
    "meeting at 3pm",
    "congratulations you won prize",
    "review pull request please",
    "click here claim your free",
    "quarterly report attached",
    "urgent account suspended",
    "lunch tomorrow works for me",
] * 10  # 80 samples total
labels = (["spam"] * 4 + ["ham"] * 4) * 10

X_train, X_val, y_train, y_val = train_test_split(
    emails, labels, test_size=0.25, random_state=42, stratify=labels
)

# Uncalibrated pipeline
pipeline_uncalibrated = Pipeline([
    ("vectorizer", TfidfVectorizer(stop_words="english")),
    ("nb", MultinomialNB(alpha=1.0))
])
pipeline_uncalibrated.fit(X_train, y_train)
raw_probs = pipeline_uncalibrated.predict_proba(X_val)

# Calibrated using Platt scaling (method='sigmoid') with 5-fold CV
calibrated = CalibratedClassifierCV(
    estimator=MultinomialNB(alpha=1.0),
    method='sigmoid',   # Platt scaling
    cv=5
)
pipeline_calibrated = Pipeline([
    ("vectorizer", TfidfVectorizer(stop_words="english")),
    ("calibrated_nb", calibrated)
])
pipeline_calibrated.fit(X_train, y_train)
calib_probs = pipeline_calibrated.predict_proba(X_val)

print("Sample probability comparison:")
for i in range(min(5, len(X_val))):
    raw = raw_probs[i]
    cal = calib_probs[i]
    print(f"  '{X_val[i][:30]}' -> raw: {raw.max():.3f} | calibrated: {cal.max():.3f}")
print()
print("After calibration, probabilities spread across the range more realistically.")

// io.thecodeforge — ml-ai tutorial

from sklearn.feature_selection import mutual_info_classif
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
import numpy as np

# Assume X_train is a sparse matrix of token counts
# Mutual info between each feature and target
mi_scores = mutual_info_classif(X_train, y_train)

# Keep only features with MI > 0.01 (tune this)
keep = np.where(mi_scores > 0.01)[0]
X_filtered = X_train[:, keep]

# Also drop features that are pairwise correlated > 0.9
from scipy.sparse import csr_matrix
corr_matrix = np.corrcoef(X_filtered.toarray().T)
high_corr_pairs = np.where(np.triu(np.abs(corr_matrix) > 0.9, k=1))
drop_idx = set(high_corr_pairs[1])  # drop one from each pair

final_keep = [i for i in range(X_filtered.shape[1]) if i not in drop_idx]
X_clean = X_filtered[:, final_keep]

model = MultinomialNB()
model.fit(X_clean, y_train)
print(f"Features before: {X_train.shape[1]}, after: {X_clean.shape[1]}")

// io.thecodeforge — ml-ai tutorial

from sklearn.naive_bayes import MultinomialNB
from sklearn.model_selection import GridSearchCV
from sklearn.metrics import log_loss
import numpy as np

# X_train, y_train from your pipeline
param_grid = {'alpha': [0.01, 0.1, 0.5, 1.0, 2.0, 5.0]}
nb = MultinomialNB()

grid = GridSearchCV(nb, param_grid, scoring='neg_log_loss', cv=5)
grid.fit(X_train, y_train)

best_alpha = grid.best_params_['alpha']
print(f"Best alpha: {best_alpha}")

# Fit final model with best alpha
final_model = MultinomialNB(alpha=best_alpha)
final_model.fit(X_train, y_train)
y_pred_proba = final_model.predict_proba(X_test)
print(f"Test log-loss: {log_loss(y_test, y_pred_proba):.4f}")

// io.thecodeforge — ml-ai tutorial

import numpy as np
from sklearn.naive_bayes import MultinomialNB

class SafeMultinomialNB(MultinomialNB):
    def predict_log_proba(self, X):
        # Override to cap log-probabilities
        log_proba = super().predict_log_proba(X)
        # Clip to avoid underflow: -700 is safe for float64
        log_proba = np.clip(log_proba, -700, 0)
        return log_proba

    def predict_proba(self, X):
        log_proba = self.predict_log_proba(X)
        # Exponentiate safely
        proba = np.exp(log_proba)
        # Renormalize to sum to 1
        proba /= proba.sum(axis=1, keepdims=True)
        return proba

# Usage identical to standard MultinomialNB
model = SafeMultinomialNB(alpha=0.1)
model.fit(X_train, y_train)
y_proba = model.predict_proba(X_test)
print(f"Min probability: {y_proba.min():.2e}, Max: {y_proba.max():.2e}")

// io.thecodeforge — ml-ai tutorial

import numpy as np

# Prior: P(spam) from your training data
prior_spam = 0.01  # 1% of emails are spam
prior_ham = 0.99

# Likelihoods: P("free" | class)
like_free_given_spam = 0.60  # 60% of spam has "free"
like_free_given_ham = 0.02   # 2% of ham has "free"

# Evidence: P("free") = sum of joint probabilities
evidence = (prior_spam * like_free_given_spam) + (prior_ham * like_free_given_ham)

# Posterior: P(spam | "free")
posterior_spam = (prior_spam * like_free_given_spam) / evidence
print(f"P(spam | 'free') = {posterior_spam:.4f}")

// io.thecodeforge — ml-ai tutorial

import numpy as np
from sklearn.naive_bayes import MultinomialNB

# Simulated counts: vocab size = 1000, seen tokens = 500
# New token "syndicate" never appears in training
vocab_size = 1000
seen_tokens = 500  # unique tokens in training

# Without Laplace: P("syndicate"|spam) = 0/500 = 0.0 → posterior = 0
# With Laplace alpha=1: adds 1 to each count, 2 to denominator
smoothed_prob = (0 + 1) / (500 + vocab_size)  # alpha=1

model = MultinomialNB(alpha=1.0)
# model.fit(X_train, y_train)

# Prediction: unseen token gets non-zero but tiny probability
print(f"Smoothed P('syndicate'|spam): {smoothed_prob:.6f}")
print("Score stays alive — but bias creeps in for all tokens")