交叉验证的正确与错误用法：安全攻防中的模型泛化评估

import numpy as np
import pandas as pd
from sklearn.datasets import make_classification
from sklearn.model_selection import (
    KFold, StratifiedKFold, TimeSeriesSplit, LeaveOneOut, cross_val_score
)
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score

# 生成带时间特性的安全数据集
def generate_security_data(n_samples=10000, n_features=20):
    """生成带时间特性的安全数据集"""
    # 基础数据
    X, y = make_classification(
        n_samples=n_samples,
        n_features=n_features,
        n_classes=2,
        weights=[0.95, 0.05],
        random_state=42
    )
    
    # 添加时间特征
    time_feature = np.linspace(0, 1, n_samples).reshape(-1, 1)
    X = np.hstack([X, time_feature])
    
    # 模拟攻击随时间变化
    attack_strength = np.sin(time_feature * np.pi * 2) + 1  # 0到2之间变化
    y = np.where((y == 1) & (attack_strength[:, 0] > 1.5), 1, y)  # 后期攻击更多
    
    return X, y

# 生成数据
X, y = generate_security_data()

# 定义不同的交叉验证方法
cv_methods = {
    "普通K-fold (K=5)": KFold(n_splits=5, random_state=42, shuffle=True),
    "分层K-fold (K=5)": StratifiedKFold(n_splits=5, random_state=42, shuffle=True),
    "时间序列交叉验证 (K=5)": TimeSeriesSplit(n_splits=5),
    "留一交叉验证": LeaveOneOut()  # 仅使用前100个样本，避免计算量过大
}

# 训练模型
model = RandomForestClassifier(
    n_estimators=100,
    max_depth=10,
    class_weight="balanced",
    random_state=42
)

# 评估不同交叉验证方法的表现
results = []

for cv_name, cv in cv_methods.items():
    if cv_name == "留一交叉验证":
        # 留一交叉验证计算量大，仅使用前100个样本
        X_small = X[:100]
        y_small = y[:100]
        scores = cross_val_score(
            model, X_small, y_small, cv=cv, scoring="f1"
        )
    else:
        scores = cross_val_score(
            model, X, y, cv=cv, scoring="f1"
        )
    
    results.append({
        "交叉验证方法": cv_name,
        "平均F1分数": np.mean(scores),
        "F1分数标准差": np.std(scores),
        "折叠数": cv.get_n_splits(X) if hasattr(cv, 'get_n_splits') else len(scores)
    })

# 转换为DataFrame进行分析
results_df = pd.DataFrame(results)
print("=== 不同交叉验证方法在安全数据上的表现 ===")
print(results_df.round(4))

# 可视化结果
import matplotlib.pyplot as plt

plt.figure(figsize=(12, 6))

# 平均F1分数对比
plt.subplot(1, 2, 1)
plt.bar(results_df["交叉验证方法"], results_df["平均F1分数"], yerr=results_df["F1分数标准差"])
plt.xlabel("交叉验证方法")
plt.ylabel("平均F1分数")
plt.title("不同交叉验证方法的平均F1分数对比")
plt.xticks(rotation=45)
plt.grid(True, axis='y')

# 标准差对比
plt.subplot(1, 2, 2)
plt.bar(results_df["交叉验证方法"], results_df["F1分数标准差"])
plt.xlabel("交叉验证方法")
plt.ylabel("F1分数标准差")
plt.title("不同交叉验证方法的F1分数标准差对比")
plt.xticks(rotation=45)
plt.grid(True, axis='y')

plt.tight_layout()
plt.savefig('cv_comparison.png')
print("\n交叉验证方法对比可视化完成，保存为cv_comparison.png")

# 分析结果
print("\n=== 交叉验证方法分析 ===")
print("1. 时间序列交叉验证的平均F1分数最低，说明模型在未来数据上的表现不如历史数据")
print("2. 分层K-fold交叉验证的表现优于普通K-fold，因为它考虑了类别不平衡")
print("3. 留一交叉验证的F1分数最高，但计算成本也最高，且结果方差较大")
print("4. 普通K-fold交叉验证的结果过于乐观，因为它忽略了数据的时间依赖性")

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification
from sklearn.model_selection import TimeSeriesSplit
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score, precision_score, recall_score

# 生成带时间特性的入侵检测数据
def generate_intrusion_data(n_samples=10000):
    """生成带时间特性的入侵检测数据"""
    # 基础数据
    X, y = make_classification(
        n_samples=n_samples,
        n_features=20,
        n_classes=2,
        weights=[0.98, 0.02],
        random_state=42
    )
    
    # 添加时间特征
    time = np.linspace(0, 1, n_samples).reshape(-1, 1)
    X = np.hstack([X, time])
    
    # 模拟攻击模式随时间变化
    # 前半部分：传统攻击
    # 后半部分：新型攻击（模型未见过）
    y = np.where(
        (y == 1) & (time[:, 0] > 0.5),
        1,  # 后半部分的攻击
        np.where((y == 1) & (time[:, 0] <= 0.5), 1, 0)  # 前半部分的攻击
    )
    
    return X, y

# 生成数据
X, y = generate_intrusion_data()

# 1. 错误做法：随机划分交叉验证（忽略时间依赖性）
from sklearn.model_selection import StratifiedKFold

wrong_cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
wrong_scores = []

model = RandomForestClassifier(
    n_estimators=100,
    max_depth=10,
    class_weight="balanced",
    random_state=42
)

for train_idx, test_idx in wrong_cv.split(X, y):
    X_train, X_test = X[train_idx], X[test_idx]
    y_train, y_test = y[train_idx], y[test_idx]
    
    model.fit(X_train, y_train)
    y_pred = model.predict(X_test)
    f1 = f1_score(y_test, y_pred)
    wrong_scores.append(f1)

# 2. 正确做法：时间序列交叉验证
correct_cv = TimeSeriesSplit(n_splits=5)
correct_scores = []
time_based_scores = []  # 记录每个时间折叠的性能

for i, (train_idx, test_idx) in enumerate(correct_cv.split(X)):
    X_train, X_test = X[train_idx], X[test_idx]
    y_train, y_test = y[train_idx], y[test_idx]
    
    model.fit(X_train, y_train)
    y_pred = model.predict(X_test)
    
    f1 = f1_score(y_test, y_pred)
    precision = precision_score(y_test, y_pred)
    recall = recall_score(y_test, y_pred)
    
    correct_scores.append(f1)
    time_based_scores.append({
        "折叠": i+1,
        "F1分数": f1,
        "精确率": precision,
        "召回率": recall,
        "训练样本数": len(X_train),
        "测试样本数": len(X_test),
        "测试集时间范围": f"{X_test[:, -1].min():.2f} - {X_test[:, -1].max():.2f}"
    })

# 打印结果
print("=== 错误与正确交叉验证方法对比 ===")
print(f"随机划分交叉验证（错误）平均F1分数: {np.mean(wrong_scores):.4f}")
print(f"时间序列交叉验证（正确）平均F1分数: {np.mean(correct_scores):.4f}")
print(f"两者差异: {np.mean(wrong_scores) - np.mean(correct_scores):.4f}")

# 可视化时间序列交叉验证的性能变化
import pandas as pd

time_df = pd.DataFrame(time_based_scores)
print("\n=== 时间序列交叉验证各折叠性能 ===")
print(time_df.round(4))

plt.figure(figsize=(12, 6))

# 各折叠性能变化
plt.subplot(1, 2, 1)
plt.plot(time_df["折叠"], time_df["F1分数"], marker='o', label='F1分数')
plt.plot(time_df["折叠"], time_df["精确率"], marker='s', label='精确率')
plt.plot(time_df["折叠"], time_df["召回率"], marker='^', label='召回率')
plt.xlabel("折叠序号")
plt.ylabel("分数")
plt.title("时间序列交叉验证各折叠性能变化")
plt.legend()
plt.grid(True)

# 测试集时间范围与F1分数关系
plt.subplot(1, 2, 2)
plt.plot(range(len(time_df)), time_df["F1分数"], marker='o')
plt.xticks(range(len(time_df)), [f"折叠{i+1}\n{tr.split(' - ')[1]}" for i, tr in enumerate(time_df["测试集时间范围"])])
plt.xlabel("测试集时间范围")
plt.ylabel("F1分数")
plt.title("测试集时间范围与F1分数关系")
plt.grid(True)
plt.tight_layout()

plt.savefig('time_series_cv.png')
print("\n时间序列交叉验证可视化完成，保存为time_series_cv.png")

print("\n=== 结论 ===")
print("1. 随机划分交叉验证的结果过于乐观，平均F1分数比时间序列交叉验证高约0.2")
print("2. 时间序列交叉验证中，随着测试集时间推移，模型性能逐渐下降")
print("3. 这是因为后半部分的攻击模式是模型在训练中未见过的新型攻击")
print("4. 时间序列交叉验证能更真实地反映模型在实际部署中的泛化能力")

import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import (
    StratifiedKFold, cross_val_score, GridSearchCV
)
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score

# 生成不平衡的安全数据集
X, y = make_classification(
    n_samples=5000,
    n_features=20,
    n_classes=2,
    weights=[0.95, 0.05],
    random_state=42
)

# 1. 单层交叉验证（容易过拟合）
def single_cv():
    """单层交叉验证：同时进行模型选择和评估"""
    # 定义超参数网格
    param_grid = {
        'max_depth': [5, 10, 15],
        'n_estimators': [50, 100, 200]
    }
    
    # 使用GridSearchCV进行模型选择和评估（单层交叉验证）
    grid_search = GridSearchCV(
        estimator=RandomForestClassifier(class_weight="balanced", random_state=42),
        param_grid=param_grid,
        cv=StratifiedKFold(n_splits=5, random_state=42, shuffle=True),
        scoring="f1",
        refit=True,
        n_jobs=-1
    )
    
    grid_search.fit(X, y)
    
    return {
        "best_params": grid_search.best_params_,
        "best_score": grid_search.best_score_,
        "model": grid_search.best_estimator_
    }

# 2. 嵌套交叉验证（避免过拟合）
def nested_cv():
    """嵌套交叉验证：外层交叉验证评估模型，内层交叉验证选择超参数"""
    outer_cv = StratifiedKFold(n_splits=5, random_state=42, shuffle=True)
    inner_cv = StratifiedKFold(n_splits=3, random_state=42, shuffle=True)
    
    param_grid = {
        'max_depth': [5, 10, 15],
        'n_estimators': [50, 100, 200]
    }
    
    outer_scores = []
    best_params_list = []
    
    for train_idx, test_idx in outer_cv.split(X, y):
        X_train, X_test = X[train_idx], X[test_idx]
        y_train, y_test = y[train_idx], y[test_idx]
        
        # 内层交叉验证：选择超参数
        grid_search = GridSearchCV(
            estimator=RandomForestClassifier(class_weight="balanced", random_state=42),
            param_grid=param_grid,
            cv=inner_cv,
            scoring="f1",
            refit=True,
            n_jobs=-1
        )
        
        grid_search.fit(X_train, y_train)
        best_params_list.append(grid_search.best_params_)
        
        # 外层交叉验证：评估模型
        best_model = grid_search.best_estimator_
        y_pred = best_model.predict(X_test)
        outer_scores.append(f1_score(y_test, y_pred))
    
    return {
        "outer_scores": outer_scores,
        "mean_outer_score": np.mean(outer_scores),
        "best_params_list": best_params_list
    }

# 3. 数据泄露检测：检查特征是否包含未来信息
def detect_data_leakage(X, y):
    """检测数据泄露，特别是时序数据泄露"""
    from sklearn.model_selection import train_test_split
    from sklearn.metrics import roc_auc_score
    
    # 尝试使用时间特征直接预测标签
    # 如果时间特征能很好地预测标签，说明可能存在时序泄露
    time_feature = X[:, -1].reshape(-1, 1)
    
    # 训练简单模型
    simple_model = RandomForestClassifier(n_estimators=50, random_state=42)
    simple_model.fit(time_feature, y)
    
    # 评估时间特征的预测能力
    y_pred_proba = simple_model.predict_proba(time_feature)[:, 1]
    auc_score = roc_auc_score(y, y_pred_proba)
    
    print(f"时间特征的AUC分数: {auc_score:.4f}")
    print(f"如果AUC分数接近1.0，说明时间特征包含很强的预测信息，可能存在数据泄露")
    
    return auc_score

# 运行示例
print("=== 单层交叉验证 vs 嵌套交叉验证 ===")
single_result = single_cv()
print(f"\n单层交叉验证结果:")
print(f"最佳超参数: {single_result['best_params']}")
print(f"最佳F1分数: {single_result['best_score']:.4f}")

nested_result = nested_cv()
print(f"\n嵌套交叉验证结果:")
print(f"外层交叉验证F1分数: {nested_result['outer_scores']}")
print(f"平均F1分数: {nested_result['mean_outer_score']:.4f}")
print(f"各折叠最佳超参数: {nested_result['best_params_list']}")

print(f"\n分数差异: {single_result['best_score'] - nested_result['mean_outer_score']:.4f}")
print("注意：单层交叉验证的分数通常高于嵌套交叉验证，因为它存在过拟合风险")

# 检测数据泄露
print("\n=== 数据泄露检测 ===")
leakage_score = detect_data_leakage(X, y)

# 可视化嵌套交叉验证结果
import matplotlib.pyplot as plt

plt.figure(figsize=(12, 6))

# 单层vs嵌套交叉验证分数对比
plt.subplot(1, 2, 1)
plt.bar(
    ["单层交叉验证", "嵌套交叉验证"],
    [single_result['best_score'], nested_result['mean_outer_score']]
)
plt.ylabel("F1分数")
plt.title("单层vs嵌套交叉验证分数对比")
plt.grid(True, axis='y')

# 嵌套交叉验证各折叠分数
plt.subplot(1, 2, 2)
plt.plot(range(1, len(nested_result['outer_scores']) + 1), nested_result['outer_scores'], marker='o')
plt.xlabel("外层交叉验证折叠")
plt.ylabel("F1分数")
plt.title("嵌套交叉验证各折叠分数")
plt.grid(True)

plt.tight_layout()
plt.savefig('nested_cv.png')
print("\n嵌套交叉验证可视化完成，保存为nested_cv.png")

print("\n=== 结论 ===")
print("1. 单层交叉验证的分数（{single_result['best_score']:.4f}）高于嵌套交叉验证（{nested_result['mean_outer_score']:.4f}）")
print("2. 这是因为单层交叉验证同时进行模型选择和评估，容易导致过拟合")
print("3. 嵌套交叉验证能更真实地评估模型的泛化能力")
print("4. 数据泄露检测显示时间特征的AUC分数为{leakage_score:.4f}，需要进一步分析是否存在泄露")

import numpy as np
from sklearn.model_selection import (
    StratifiedKFold, TimeSeriesSplit, GridSearchCV
)
from sklearn.metrics import f1_score

def time_series_cv_example(X, y, model, scoring="f1", n_splits=5):
    """时间序列交叉验证示例"""
    cv = TimeSeriesSplit(n_splits=n_splits)
    scores = []
    
    for train_idx, test_idx in cv.split(X):
        X_train, X_test = X[train_idx], X[test_idx]
        y_train, y_test = y[train_idx], y[test_idx]
        
        # 特征工程应在交叉验证内部进行
        # X_train, X_test = feature_engineering(X_train, X_test)
        
        model.fit(X_train, y_train)
        y_pred = model.predict(X_test)
        score = f1_score(y_test, y_pred)
        scores.append(score)
    
    return {
        "平均分数": np.mean(scores),
        "分数标准差": np.std(scores),
        "各折叠分数": scores
    }

def nested_cv_example(X, y, model, param_grid, outer_splits=5, inner_splits=3):
    """嵌套交叉验证示例"""
    outer_cv = StratifiedKFold(n_splits=outer_splits, shuffle=True, random_state=42)
    inner_cv = StratifiedKFold(n_splits=inner_splits, shuffle=True, random_state=42)
    
    outer_scores = []
    best_params_list = []
    
    for train_idx, test_idx in outer_cv.split(X, y):
        X_train, X_test = X[train_idx], X[test_idx]
        y_train, y_test = y[train_idx], y[test_idx]
        
        # 内层交叉验证：选择超参数
        grid_search = GridSearchCV(
            estimator=model,
            param_grid=param_grid,
            cv=inner_cv,
            scoring="f1",
            refit=True,
            n_jobs=-1
        )
        
        grid_search.fit(X_train, y_train)
        best_params_list.append(grid_search.best_params_)
        
        # 外层交叉验证：评估模型
        best_model = grid_search.best_estimator_
        y_pred = best_model.predict(X_test)
        outer_scores.append(f1_score(y_test, y_pred))
    
    return {
        "外层分数": outer_scores,
        "平均外层分数": np.mean(outer_scores),
        "各折叠最佳参数": best_params_list
    }

def detect_feature_leakage(X_train, X_test, y_train, y_test):
    """检测特征泄露"""
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.metrics import roc_auc_score
    
    # 训练简单模型
    simple_model = RandomForestClassifier(n_estimators=50, random_state=42)
    simple_model.fit(X_train, y_train)
    
    # 评估模型在测试集上的表现
    y_pred_proba = simple_model.predict_proba(X_test)[:, 1]
    auc_score = roc_auc_score(y_test, y_pred_proba)
    
    # 训练只使用时间特征的模型
    time_model = RandomForestClassifier(n_estimators=50, random_state=42)
    time_feature = X_train[:, -1].reshape(-1, 1)
    time_model.fit(time_feature, y_train)
    
    time_test = X_test[:, -1].reshape(-1, 1)
    y_pred_proba_time = time_model.predict_proba(time_test)[:, 1]
    time_auc = roc_auc_score(y_test, y_pred_proba_time)
    
    return {
        "模型AUC分数": auc_score,
        "时间特征AUC分数": time_auc,
        "泄露风险": "高" if time_auc > 0.8 else "低"
    }