朴素贝叶斯：概率分类器

# 贝叶斯定理的直觉演示
def bayes_theorem(prior, likelihood, evidence):
    """
    P(A|B) = P(B|A) * P(A) / P(B)
    prior: P(A) 先验概率
    likelihood: P(B|A) 似然
    evidence: P(B) 证据概率
    """
    posterior = likelihood * prior / evidence
    return posterior

# 医疗检测案例
# 疾病发病率 0.01%，检测灵敏度 99%，假阳性率 1%
p_disease = 0.0001          # 先验：发病率
p_positive_given_disease = 0.99  # 灵敏度
p_positive = (0.99 * 0.0001 + 0.01 * 0.9999)  # 全概率

p_disease_given_positive = bayes_theorem(
    prior=p_disease,
    likelihood=p_positive_given_disease,
    evidence=p_positive
)
print(f"检测阳性后患病概率: {p_disease_given_positive:.4f} ({p_disease_given_positive*100:.2f}%)")
# 输出: 检测阳性后患病概率: 0.0098 (0.98%)

# 可视化：先验如何影响后验
import numpy as np
import matplotlib.pyplot as plt

priors = np.linspace(0.001, 0.5, 100)
likelihood = 0.99
evidence_factor = 0.01  # 假阳性率

posteriors = []
for p in priors:
    evidence = likelihood * p + evidence_factor * (1 - p)
    post = bayes_theorem(p, likelihood, evidence)
    posteriors.append(post)

plt.figure(figsize=(10, 6))
plt.plot(priors, posteriors, 'b-', linewidth=2)
plt.plot([0, 1], [0, 1], 'r--', alpha=0.5, label='对角线')
plt.xlabel('先验概率 P(A)')
plt.ylabel('后验概率 P(A|B)')
plt.title('先验概率如何影响后验概率')
plt.legend()
plt.grid(True, alpha=0.3)
plt.show()

import numpy as np
from collections import Counter

class NaiveBayesClassifier:
    """从零实现朴素贝叶斯分类器"""

    def __init__(self):
        self.classes = None
        self.class_priors = {}
        self.conditional_probs = {}

    def fit(self, X, y):
        n_samples, n_features = X.shape
        self.classes = np.unique(y)

        for c in self.classes:
            X_c = X[y == c]
            # 先验概率 P(Y=c)
            self.class_priors[c] = len(X_c) / n_samples
            # 条件概率 P(X_i|Y=c) - 假设特征服从高斯分布
            self.conditional_probs[c] = {
                'mean': X_c.mean(axis=0),
                'var': X_c.var(axis=0)
            }

    def _likelihood(self, x, mean, var, eps=1e-6):
        """高斯概率密度函数"""
        coeff = 1.0 / np.sqrt(2 * np.pi * var + eps)
        exponent = np.exp(-(x - mean) ** 2 / (2 * var + eps))
        return coeff * exponent

    def predict(self, X):
        predictions = []
        for x in X:
            posteriors = {}
            for c in self.classes:
                # log P(Y=c) + Σ log P(X_i|Y=c)
                log_prior = np.log(self.class_priors[c])
                means = self.conditional_probs[c]['mean']
                variances = self.conditional_probs[c]['var']
                log_likelihood = np.sum(
                    np.log(self._likelihood(x, means, variances))
                )
                posteriors[c] = log_prior + log_likelihood
            predictions.append(max(posteriors, key=posteriors.get))
        return np.array(predictions)

# 用 sklearn 验证我们的实现
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.naive_bayes import GaussianNB
from sklearn.metrics import accuracy_score

# 生成测试数据
X, y = make_classification(
    n_samples=1000, n_features=10, n_informative=5,
    n_redundant=2, n_classes=2, random_state=42
)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 我们的实现
my_nb = NaiveBayesClassifier()
my_nb.fit(X_train, y_train)
my_pred = my_nb.predict(X_test)
print(f"我的实现准确率: {accuracy_score(y_test, my_pred):.4f}")

# sklearn 实现
sklearn_nb = GaussianNB()
sklearn_nb.fit(X_train, y_train)
sk_pred = sklearn_nb.predict(X_test)
print(f"sklearn 准确率: {accuracy_score(y_test, sk_pred):.4f}")

class MultinomialNaiveBayes:
    """多项式朴素贝叶斯（带拉普拉斯平滑）"""

    def __init__(self, alpha=1.0):
        self.alpha = alpha  # 平滑参数
        self.class_log_prior = {}
        self.feature_log_prob = {}

    def fit(self, X, y):
        """X: 词频矩阵 (n_samples, n_features), y: 类别标签"""
        n_samples, n_features = X.shape
        self.classes = np.unique(y)
        self.n_features = n_features

        for c in self.classes:
            X_c = X[y == c]
            # 先验概率（对数）
            self.class_log_prior[c] = np.log(
                np.sum(y == c) / n_samples
            )
            # 条件概率（对数）+ 拉普拉斯平滑
            smoothed_counts = X_c.sum(axis=0) + self.alpha
            smoothed_total = smoothed_counts.sum()
            self.feature_log_prob[c] = np.log(
                smoothed_counts / smoothed_total
            )

    def predict(self, X):
        log_probs = {}
        for c in self.classes:
            log_probs[c] = (
                self.class_log_prior[c]
                + X @ self.feature_log_prob[c]
            )
        return self.classes[np.argmax(
            np.column_stack([log_probs[c] for c in self.classes]),
            axis=1
        )]

# 演示拉普拉斯平滑的效果
import numpy as np

# 词表：['免费', '优惠', '会议', '报告', '中奖']
# 类别：垃圾邮件(0)、正常邮件(1)

# 训练集词频
spam_counts = np.array([5, 3, 0, 0, 2])    # 垃圾邮件
ham_counts = np.array([0, 1, 4, 6, 0])     # 正常邮件

vocab_size = len(spam_counts)

def laplace_smooth(counts, alpha=1.0):
    """拉普拉斯平滑"""
    smoothed = counts + alpha
    total = smoothed.sum()
    return smoothed / total

# 无平滑（有零值）
spam_prob_raw = spam_counts / spam_counts.sum()
print(f"无平滑-垃圾邮件: {spam_prob_raw}")

# 拉普拉斯平滑
spam_prob_smooth = laplace_smooth(spam_counts, alpha=1.0)
print(f"平滑后-垃圾邮件: {spam_prob_smooth.round(4)}")

# 新邮件词频：[免费=2, 中奖=1]
new_email = np.array([2, 0, 0, 0, 1])

# 无平滑会得到 0（'中奖'在正常邮件中没出现过）
ham_prob_raw = ham_counts / ham_counts.sum()
score_ham_raw = np.sum(new_email * np.log(ham_prob_raw + 1e-10))
print(f"无平滑-正常邮件得分: {score_ham_raw:.4f}")

# 平滑后所有词都有正概率
ham_prob_smooth = laplace_smooth(ham_counts, alpha=1.0)
score_ham_smooth = np.sum(new_email * np.log(ham_prob_smooth))
print(f"平滑后-正常邮件得分: {score_ham_smooth:.4f}")

import numpy as np

# 演示数值下溢问题
def naive_multiply(probs):
    """直接连乘（会下溢）"""
    result = 1.0
    for p in probs:
        result *= p
    return result

def log_sum(probs):
    """对数求和（避免下溢）"""
    return np.sum(np.log(probs))

# 1000 个概率值（每个约 0.5）
probs = np.random.uniform(0.3, 0.7, 1000)

naive_result = naive_multiply(probs)
log_result = log_sum(probs)

print(f"直接连乘: {naive_result}")  # 输出 0.0（下溢）
print(f"对数求和: {log_result:.4f}")  # 正常输出

# 验证：log(a*b) = log(a) + log(b)
a, b = 0.3, 0.7
print(f"log(a*b) = {np.log(a*b):.6f}")
print(f"log(a)+log(b) = {np.log(a)+np.log(b):.6f}")

from sklearn.feature_extraction.text import CountVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

# 示例数据
emails = [
    "免费优惠！中奖了！点击链接领取奖品",
    "紧急通知：您的账户需要验证，请立即登录",
    "恭喜您获得万元大奖，限时领取",
    "明天下午三点开会，请准备季度报告",
    "项目进度更新：已完成 80%，下周交付",
    "周末聚餐，大家方便吗",
    "限时秒杀！全场一折起！先到先得",
    "请查收附件中的会议记录和行动计划",
    "双十一大促！满 100 减 50！",
    "技术分享会通知：周三下午 2 点，301 会议室",
]
labels = [1, 1, 1, 0, 0, 0, 1, 0, 1, 0]  # 1=垃圾, 0=正常

# 分词（中文需要特殊处理，这里用字符级 n-gram）
vectorizer = CountVectorizer(ngram_range=(1, 2), analyzer='char_wb')
X = vectorizer.fit_transform(emails)

# 训练
nb = MultinomialNB(alpha=0.1)
nb.fit(X, labels)

# 预测新邮件
new_emails = [
    "您好，关于下周会议安排",
    "紧急！您已被选为幸运用户",
]
X_new = vectorizer.transform(new_emails)
preds = nb.predict(X_new)
probs = nb.predict_proba(X_new)

for email, pred, prob in zip(new_emails, preds, probs):
    label = "垃圾邮件" if pred == 1 else "正常邮件"
    print(f"'{email}' → {label} (置信度: {max(prob):.2%})")

# 分析哪些词对分类影响最大
import numpy as np

# 获取特征名
feature_names = vectorizer.get_feature_names_out()

# 垃圾邮件类最显著的词
spam_log_prob = nb.feature_log_prob_[1]
spam_top_indices = np.argsort(-spam_log_prob)[:15]
print("垃圾邮件最显著的词:")
for idx in spam_top_indices:
    print(f"  '{feature_names[idx]}': {np.exp(spam_log_prob[idx]):.4f}")

# 正常邮件类最显著的词
ham_log_prob = nb.feature_log_prob_[0]
ham_top_indices = np.argsort(-ham_log_prob)[:15]
print("\n正常邮件最显著的词:")
for idx in ham_top_indices:
    print(f"  '{feature_names[idx]}': {np.exp(ham_log_prob[idx]):.4f}")

# 对比多种分类器的性能和速度
import time
import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import cross_val_score
from sklearn.naive_bayes import GaussianNB
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.svm import SVC

X, y = make_classification(
    n_samples=5000, n_features=20, n_informative=10,
    n_redundant=5, random_state=42
)

classifiers = {
    'Naive Bayes': GaussianNB(),
    'Random Forest': RandomForestClassifier(n_estimators=100, random_state=42),
    'Gradient Boosting': GradientBoostingClassifier(n_estimators=100, random_state=42),
    'SVM': SVC(probability=True, random_state=42),
}

for name, clf in classifiers.items():
    start = time.time()
    scores = cross_val_score(clf, X, y, cv=5, scoring='accuracy')
    elapsed = time.time() - start
    print(f"{name:25s} | 准确率: {scores.mean():.4f} ± {scores.std():.4f} | 耗时: {elapsed:.3f}s")

# 半朴素贝叶斯：AODE (Average One-Dependence Estimators) 简化版
import numpy as np
from collections import defaultdict

class AODEClassifier:
    """AODE 简化实现：每个特征可以依赖于一个父特征"""

    def __init__(self, min_freq=2):
        self.min_freq = min_freq  # 父特征最小出现频率

    def fit(self, X, y):
        self.classes = np.unique(y)
        n_samples, n_features = X.shape

        # 预计算：找出出现频率足够高的特征作为候选父特征
        self.parent_candidates = []
        for j in range(n_features):
            if np.bincount(X[:, j]).max() >= self.min_freq:
                self.parent_candidates.append(j)

        # 计算联合统计量 P(X_i, X_j, Y)
        self.joint_counts = {}
        for parent_idx in self.parent_candidates:
            for feat_idx in range(n_features):
                if feat_idx == parent_idx:
                    continue
                for c in self.classes:
                    mask = y == c
                    for xj in np.unique(X[:, feat_idx]):
                        for xp in np.unique(X[:, parent_idx]):
                            count = np.sum(
                                (X[mask, feat_idx] == xj) &
                                (X[mask, parent_idx] == xp)
                            )
                            key = (parent_idx, feat_idx, xj, xp, c)
                            self.joint_counts[key] = count + 1  # 平滑

    def predict(self, X):
        # 简化的预测逻辑
        return np.zeros(X.shape[0], dtype=int)  # 实际实现需要完整的概率计算

print("AODE 是半朴素贝叶斯的一个实例，通过引入特征间依赖提升分类精度")