类别不平衡处理：SMOTE, 代价敏感学习

import numpy as np
from sklearn.datasets import make_classification

X, y = make_classification(
    n_samples=10000, n_features=20,
    weights=[0.98, 0.02],  # 2% 正样本
    flip_y=0.05, random_state=42
)
print(f"类别分布: {np.bincount(y)}")  # [9800, 200]
ratio = np.bincount(y)[0] / np.bincount(y)[1]
print(f"不平衡比: {ratio:.1f}:1")  # 49.0:1

from sklearn.model_selection import cross_val_score
from sklearn.ensemble import RandomForestClassifier

clf = RandomForestClassifier(random_state=42)
# 直接训练会偏向多数类
scores = cross_val_score(clf, X, y, cv=5, scoring="accuracy")
print(f"准确率: {scores.mean():.4f}")  # ~0.98，但少数类全错

from sklearn.metrics import classification_report
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
clf.fit(X_train, y_train)
print(classification_report(y_test, clf.predict(X_test)))

from sklearn.datasets import load_breast_cancer
from imblearn.under_sampling import RandomUnderSampler

X, y = load_breast_cancer(return_X_y=True)
# 人为制造不平衡
y_imbalanced = np.where(y == 0, 0, np.random.choice([0, 1], size=len(y), p=[0.85, 0.15]))
print(f"原始分布: {np.bincount(y_imbalanced)}")

# 随机欠采样
rus = RandomUnderSampler(random_state=42)
X_rus, y_rus = rus.fit_resample(X, y_imbalanced)
print(f"欠采样后: {np.bincount(y_rus)}")  # 两类数量相等

from imblearn.over_sampling import RandomOverSampler

# 随机过采样
ros = RandomOverSampler(random_state=42)
X_ros, y_ros = ros.fit_resample(X, y_imbalanced)
print(f"过采样后: {np.bincount(y_ros)}")

# 关键：先 split 再采样
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(
    X, y_imbalanced, stratify=y_imbalanced, random_state=42
)
# 只对训练集采样
X_train_res, y_train_res = ros.fit_resample(X_train, y_train)
print(f"测试集保持原始: {np.bincount(y_test)}")

from imblearn.over_sampling import SMOTE
from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(
    X, y_imbalanced, stratify=y_imbalanced, random_state=42
)

# 标准 SMOTE
smote = SMOTE(k_neighbors=5, random_state=42)
X_smote, y_smote = smote.fit_resample(X_train, y_train)
print(f"SMOTE 后分布: {np.bincount(y_smote)}")

# 验证合成样本质量
from sklearn.neighbors import NearestNeighbors
nn = NearestNeighbors(n_neighbors=3).fit(X_smote[y_smote == 1])
distances, _ = nn.kneighbors()
print(f"平均合成样本距离: {distances.mean():.4f}")

from imblearn.over_sampling import BorderlineSMOTE, ADASYN

# Borderline-SMOTE: 只对边界样本插值
bsmote = BorderlineSMOTE(
    k_neighbors=5, m_neighbors=10,
    kind="borderline-1", random_state=42
)
X_bsmote, y_bsmote = bsmote.fit_resample(X_train, y_train)
print(f"Borderline-SMOTE: {np.bincount(y_bsmote)}")

# ADASYN: 自适应密度加权合成
adasyn = ADASYN(n_neighbors=5, random_state=42)
X_adasyn, y_adasyn = adasyn.fit_resample(X_train, y_train)
print(f"ADASYN 后分布: {np.bincount(y_adasyn)}")

# ADASYN 生成的样本数与局部密度成反比
print(f"ADASYN 自适应比例: {adasyn.nns_:_.shape[0]} 个样本参与密度计算")

from sklearn.svm import SVC
from sklearn.linear_model import LogisticRegression

# 自动平衡权重
clf_lr = LogisticRegression(
    class_weight="balanced", random_state=42, max_iter=1000
)
clf_lr.fit(X_train, y_train)
print(f"自动权重: {clf_lr.class_weight_}")

# 自定义权重
clf_custom = LogisticRegression(
    class_weight={0: 1, 1: 50},  # 少数类权重 50 倍
    random_state=42, max_iter=1000
)
clf_custom.fit(X_train, y_train)
print(f"自定义权重下的召回率: {recall_score(y_test, clf_custom.predict(X_test)):.4f}")

from sklearn.metrics import recall_score, precision_score, fbeta_score

# 比较不同策略
strategies = {
    "baseline": LogisticRegression(random_state=42, max_iter=1000),
    "balanced": LogisticRegression(class_weight="balanced", random_state=42, max_iter=1000),
    "custom_10": LogisticRegression(class_weight={0:1, 1:10}, random_state=42, max_iter=1000),
    "custom_50": LogisticRegression(class_weight={0:1, 1:50}, random_state=42, max_iter=1000),
}

for name, clf in strategies.items():
    clf.fit(X_train, y_train)
    y_pred = clf.predict(X_test)
    rec = recall_score(y_test, y_pred)
    prec = precision_score(y_test, y_pred)
    f2 = fbeta_score(y_test, y_pred, beta=2)
    print(f"{name:12s} | Recall={rec:.3f} Prec={prec:.3f} F2={f2:.3f}")

from imblearn.ensemble import EasyEnsembleClassifier, BalancedBaggingClassifier

# EasyEnsemble: 多次抽样 + 集成
eec = EasyEnsembleClassifier(
    n_estimators=10,
    sampling_strategy="auto",
    random_state=42
)
eec.fit(X_train, y_train)
y_pred = eec.predict(X_test)
print(f"EasyEnsemble 结果:")
print(f"  Recall: {recall_score(y_test, y_pred):.4f}")
print(f"  Precision: {precision_score(y_test, y_pred):.4f}")

# BalancedBagging: 每个基学习器使用平衡采样
bbc = BalancedBaggingClassifier(
    n_estimators=50,
    sampling_strategy="auto",
    replacement=False,
    random_state=42
)
bbc.fit(X_train, y_train)

from imblearn.ensemble import RUSBoostClassifier
from sklearn.ensemble import GradientBoostingClassifier

# RUSBoost: 欠采样 + AdaBoost
rusboost = RUSBoostClassifier(
    n_estimators=200,
    random_state=42,
    learning_rate=0.1
)
rusboost.fit(X_train, y_train)

# 特征重要性分析（集成方法的优势）
importances = rusboost.feature_importances_
top_features = np.argsort(importances)[-5:]
for feat_idx in top_features:
    print(f"特征 {feat_idx}: {importances[feat_idx]:.4f}")

# 概率校准
from sklearn.calibration import CalibratedClassifierCV
calibrated = CalibratedClassifierCV(rusboost, cv=5, method="isotonic")
calibrated.fit(X_train, y_train)

from sklearn.metrics import (
    roc_auc_score, average_precision_score,
    precision_recall_curve, fbeta_score
)

# 获取预测概率
y_proba = clf.predict_proba(X_test)[:, 1]

# ROC-AUC（在不平衡数据上可能误导）
roc_auc = roc_auc_score(y_test, y_proba)
print(f"ROC-AUC: {roc_auc:.4f}")

# AUC-PR（更适合不平衡数据）
ap_score = average_precision_score(y_test, y_proba)
print(f"AUC-PR: {ap_score:.4f}")

# F2-Score（重视召回率）
y_pred = clf.predict(X_test)
f2 = fbeta_score(y_test, y_pred, beta=2)
print(f"F2-Score: {f2:.4f}")

import matplotlib.pyplot as plt

# 绘制 PR 曲线
precision, recall, thresholds = precision_recall_curve(y_test, y_proba)
fig, ax = plt.subplots(figsize=(8, 6))
ax.plot(recall, precision, label=f"AP={ap_score:.3f}", linewidth=2)
ax.set_xlabel("Recall", fontsize=12)
ax.set_ylabel("Precision", fontsize=12)
ax.set_title("Precision-Recall Curve", fontsize=14)
ax.legend()
ax.grid(True, alpha=0.3)
plt.show()

# 选择最佳阈值（最大化 F2）
f2_scores = []
for t in thresholds:
    y_pred_t = (y_proba >= t).astype(int)
    f2_t = fbeta_score(y_test, y_pred_t, beta=2, zero_division=0)
    f2_scores.append(f2_t)

best_threshold = thresholds[np.argmax(f2_scores)]
print(f"最佳阈值 (F2 最大): {best_threshold:.4f}")
print(f"最佳 F2: {max(f2_scores):.4f}")

from imblearn.pipeline import Pipeline
from imblearn.over_sampling import SMOTE
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier

# 构建完整 Pipeline（避免数据泄露）
pipeline = Pipeline([
    ("scaler", StandardScaler()),
    ("smote", SMOTE(random_state=42)),
    ("clf", RandomForestClassifier(n_estimators=100, random_state=42))
])

# 交叉验证（Pipeline 确保每折独立采样）
from sklearn.model_selection import cross_validate
from sklearn.metrics import make_scorer

scoring = {
    "ap": "average_precision",
    "recall": "recall",
    "f2": make_scorer(fbeta_score, beta=2)
}
results = cross_validate(pipeline, X, y_imbalanced, cv=5, scoring=scoring)
for metric, scores in results.items():
    if metric.startswith("test_"):
        print(f"{metric}: {scores.mean():.4f} (+/- {scores.std():.4f})")

from imblearn.model_selection import StratifiedShuffleSplit
from sklearn.model_selection import GridSearchCV

# 网格搜索最佳参数
param_grid = {
    "smote__k_neighbors": [3, 5, 7],
    "clf__n_estimators": [50, 100, 200],
    "clf__max_depth": [None, 5, 10],
}

cv = StratifiedShuffleSplit(n_splits=3, random_state=42)
grid = GridSearchCV(
    pipeline, param_grid, cv=cv,
    scoring="average_precision", n_jobs=-1
)
grid.fit(X_train, y_train)
print(f"最佳参数: {grid.best_params_}")
print(f"最佳 AUC-PR: {grid.best_score_:.4f}")

# 最终评估
best_model = grid.best_estimator_
y_proba = best_model.predict_proba(X_test)[:, 1]
print(f"测试集 AUC-PR: {average_precision_score(y_test, y_proba):.4f}")