基于CICIDS2017数据集实现SOTA准确率的实战指南：从数据预处理到模型优化

1次阅读

没有评论

共计 2130 个字符，预计需要花费 6 分钟才能阅读完成。

CICIDS2017 是网络安全领域广泛使用的入侵检测基准数据集，包含正常流量和常见攻击类型（如 DDoS、Brute Force 等）。其核心价值在于：

真实网络环境采集，覆盖完整攻击链
精细标注的流量特征（78 个网络层 / 传输层特征）
适合研究新型检测算法

但该数据集存在两大挑战：

严重类别不平衡（正常流量占比 83.07%，Web 攻击仅 0.03%）
高维度特征中存在大量相关性冗余（如 Flow Duration 与Total Fwd Packets强相关）

优点：抗过拟合、特征重要性直观
缺点：对类别不平衡敏感

优点：内置权重调节、并行计算快
缺点：调参复杂度高

优点：自动特征提取
缺点：需要大量数据增强

推荐方案：XGBoost + 集成采样（后文详细说明）

import pandas as pd
from sklearn.preprocessing import StandardScaler

# 处理缺失值（用特征中位数填充）df = pd.read_csv('CICIDS2017.csv')
df.fillna(df.median(), inplace=True)

# 标准化数值特征（除离散型如 Protocol Type）numeric_cols = df.select_dtypes(include=['float64']).columns
scaler = StandardScaler()
df[numeric_cols] = scaler.fit_transform(df[numeric_cols])

# 特征选择（基于 IV 值过滤）from sklearn.feature_selection import SelectKBest, f_classif
X, y = df.drop('Label', axis=1), df['Label']
selector = SelectKBest(f_classif, k=30)
X_new = selector.fit_transform(X, y)

from imblearn.over_sampling import SMOTE

# 原始样本分布：{0:284332, 1:54, 2:173,...}
sm = SMOTE(sampling_strategy='minority', random_state=42)
X_res, y_res = sm.fit_resample(X_new, y)

# 验证采样后分布：{0:284332, 1:284332,...}
print(pd.Series(y_res).value_counts())

import xgboost as xgb
from sklearn.model_selection import train_test_split

# 划分数据集
X_train, X_test, y_train, y_test = train_test_split(X_res, y_res, test_size=0.2)

# 配置类别权重（按逆频率设置）weights = dict(1 / y_train.value_counts(normalize=True))

# 关键参数设置
model = xgb.XGBClassifier(
    scale_pos_weight=weights,
    max_depth=6,
    learning_rate=0.1,
    subsample=0.8,
    colsample_bytree=0.8,
    eval_metric='aucpr'  # 使用 PR-AUC 应对不平衡数据
)
model.fit(X_train, y_train)

避免使用 Accuracy（被多数类主导）
优先看 PR 曲线和 F1-score

方法	F1-micro	Recall（少数类）
原始数据	0.92	0.15
SMOTE	0.96	0.89
ADASYN	0.95	0.87

数据泄露：在采样前拆分训练 / 测试集
指标误用：多分类问题避免直接套用 binary AUC
特征冗余：定期检查特征相关性矩阵
过拟合：使用 Early Stopping 和交叉验证
部署延迟：测试推理速度（XGBoost 需 <10ms/request）

import pickle
from fastapi import FastAPI

app = FastAPI()
model = pickle.load(open('xgb_model.pkl', 'rb'))

@app.post('/detect')
def predict(features: dict):
    df = pd.DataFrame([features])
    return {'prediction': int(model.predict(df)[0])}