简介：这是一份面向网络安全初学者与机器学习实践者的入侵检测系统实战资料，围绕Python与ML算法构建IDS展开，适合想将分类模型落地到安全场景的开发者参考。压缩包共3个文件，含1个csv数据集、1个py脚本和1个md说明文档，整体约403KB，体量轻便，便于快速跑通流程。数据集用于训练与验证，脚本覆盖数据预处理、特征工程、模型训练与评估等环节，说明文档则梳理项目结构与使用思路。已有285人学习下载，说明其在同类入门项目中具备一定参考价值。读者可借此理解误用检测与异常检测的差异，掌握决策树、随机森林、SVM等算法在KDD类数据上的应用，并对照代码完成从数据清洗到指标评估的完整实践，为后续调参与模型优化打下基础。
1. 从告警疲劳到模型兜底：这套 ML 入侵检测系统到底能解决什么
凌晨两点被 Wazuh 的告警炸醒，翻了三屏发现全是误报——这种经历搞安全运维的多少都碰过。规则引擎写了几百条，攻击者换个 payload 编码就绕过去了，而正常业务流量稍微抖一下又触发一堆告警。Intrusion-Detection-System-using-ML-Algorithms 这个项目，就是冲着这个痛点来的：用 Python 把 KDD Cup 99 / NSL-KDD 这类标注流量喂给多个机器学习分类器，训练出能区分正常连接和各类攻击的模型，再拿新流量做推理。它不是一个开箱即用的商业 IDS，而是一套完整的实验框架——数据预处理、特征工程、多算法对比、模型持久化、推理脚本全在里面。适合谁？安全方向的学生做课设或毕设、运维想给现有规则引擎加一层 ML 兜底、以及刚接触 ML 安全应用想找个能跑通的实战项目的人。前提是你得会点 Python 和 pandas，不然读代码会有点吃力。
2. 数据管道与特征工程：把原始流量日志变成模型能吃的矩阵
2.1 为什么 NSL-KDD 是绕不开的起点
这套项目默认用的数据集大概率是 NSL-KDD，它是 KDD Cup 99 的去重修正版。原始 KDD 99 有个致命问题：训练集和测试集重复率高达 78%，模型在测试集上刷到 99% 准确率，上线就翻车。NSL-KDD 把重复记录删了，训练集 125973 条、测试集 22544 条，攻击类型归为四大类：DoS、Probe、R2L、U2R。每条记录 41 个特征，外加一个 label。特征分三种：连续型（如 duration、src_bytes）、离散型（如 protocol_type、flag）、以及带层次结构的服务类型（如 http、ftp、smtp 共 70 种）。
我一般拿到数据先干三件事：看类别分布、看缺失值、看连续特征的量纲差异。NSL-KDD 没有缺失值，但类别极度不平衡——U2R 在训练集里只有 52 条，R2L 也就 995 条。这意味着如果你直接上 accuracy 做评估指标，模型全预测成 Normal 也能拿 70% 以上，完全没意义。
2.2 预处理脚本：编码、归一化、切分一步到位
下面这段代码是我从项目里拆出来、自己重构过的预处理流程，逻辑清晰，直接能跑：
import pandas as pd import numpy as np from sklearn.preprocessing import LabelEncoder, MinMaxScaler # 列名定义，NSL-KDD 的 41 个特征 + label + difficulty col_names = ["duration","protocol_type","service","flag","src_bytes", "dst_bytes","land","wrong_fragment","urgent","hot","num_failed_logins", "logged_in","num_compromised","root_shell","su_attempted","num_root", "num_file_creations","num_shells","num_access_files","num_outbound_cmds", "is_host_login","is_guest_login","count","srv_count","serror_rate", "srv_serror_rate","rerror_rate","srv_rerror_rate","same_srv_rate", "diff_srv_rate","srv_diff_host_rate","dst_host_count","dst_host_srv_count", "dst_host_same_srv_rate","dst_host_diff_srv_rate","dst_host_same_src_port_rate", "dst_host_srv_diff_host_rate","dst_host_serror_rate","dst_host_srv_serror_rate", "dst_host_rerror_rate","dst_host_srv_rerror_rate","label","difficulty"] train = pd.read_csv("KDDTrain+.txt", names=col_names) test = pd.read_csv("KDDTest+.txt", names=col_names) # 攻击类型映射到五大类 attack_map = { "normal": "normal", "back":"dos","land":"dos","neptune":"dos","pod":"dos","smurf":"dos", "teardrop":"dos","mailbomb":"dos","apache2":"dos","processtable":"dos","udpstorm":"dos", "ipsweep":"probe","nmap":"probe","portsweep":"probe","satan":"probe", "mscan":"probe","saint":"probe", "ftp_write":"r2l","guess_passwd":"r2l","imap":"r2l","multihop":"r2l", "phf":"r2l","spy":"r2l","warezclient":"r2l","warezmaster":"r2l", "sendmail":"r2l","named":"r2l","snmpgetattack":"r2l","snmpguess":"r2l", "xlock":"r2l","xsnoop":"r2l","worm":"r2l", "buffer_overflow":"u2r","loadmodule":"u2r","perl":"u2r","rootkit":"u2r", "httptunnel":"u2r","ps":"u2r","sqlattack":"u2r","xterm":"u2r" } for df in [train, test]: df["attack_cat"] = df["label"].map(attack_map) df.drop(columns=["label","difficulty"], inplace=True) # 离散特征编码 cat_cols = ["protocol_type","service","flag","attack_cat"] encoders = {} for col in cat_cols: le = LabelEncoder() train[col] = le.fit_transform(train[col]) # 测试集出现训练集没见过的类别，统一归为 -1 test[col] = test[col].map(lambda x: le.transform([x])[0] if x in le.classes_ else -1) encoders[col] = le # 连续特征归一化 num_cols = [c for c in train.columns if c not in cat_cols] scaler = MinMaxScaler() train[num_cols] = scaler.fit_transform(train[num_cols]) test[num_cols] = scaler.transform(test[num_cols]) X_train = train.drop(columns=["attack_cat"]).values y_train = train["attack_cat"].values X_test = test.drop(columns=["attack_cat"]).values y_test = test["attack_cat"].values print("训练集:", X_train.shape, "测试集:", X_test.shape)逻辑说明：先按官方列名读入，把 40 多种具体攻击名归并成 5 大类，这样模型学的是攻击模式而不是死记攻击名。LabelEncoder 对离散特征做整数编码，注意测试集里可能出现训练集没见过的 service 值，这里用-1兜底，否则 transform 会直接抛异常。MinMaxScaler 把连续特征压到 [0,1]，因为 KNN 和 SVM 对量纲敏感，不归一化的话 src_bytes 这种动辄上万的字段会主导距离计算。
参数说明：attack_map的映射关系直接决定分类粒度，如果你想做二分类（正常 vs 异常），把非 normal 全映射成 "attack" 就行。MinMaxScaler换成StandardScaler也可以，但对树模型没区别，对距离类模型 MinMax 更稳。
2.3 特征选择：41 维不是每维都有用
项目里如果带了特征重要性分析，你会发现num_outbound_cmds这一列全是 0，直接删。另外is_host_login也几乎没方差。用随机森林跑一遍 feature_importances_，前 15 个特征通常能覆盖 95% 以上的重要性。我一般会保留 20 维左右，既降训练时间又减少过拟合。但注意：降维后要用同一套特征列表去切测试集，别训练用 20 维、推理用 41 维，这种低级错误我见过不止一次。
3. 多算法对比实战：从 KNN 到 XGBoost 的选型与调参
3.1 项目里通常包含哪几个分类器
这类项目一般会覆盖：KNN、决策树、随机森林、朴素贝叶斯、SVM、以及 XGBoost 或 LightGBM。不是越多越好，而是让你看到不同算法在同一个任务上的表现差异。我按自己的实测经验给个大致预期（NSL-KDD 五分类，测试集）：
| 算法 | 准确率区间 | 训练速度 | 对不平衡敏感度 | 备注 |
|---|---|---|---|---|
| KNN (k=5) | 75%~78% | 快 | 高 | 推理慢，不适合线上 |
| 决策树 | 78%~81% | 快 | 中 | 容易过拟合，需剪枝 |
| 随机森林 | 80%~83% | 中 | 低 | 综合最稳 |
| 朴素贝叶斯 | 72%~76% | 极快 | 高 | 特征独立假设太强 |
| SVM (RBF) | 79%~82% | 慢 | 中 | 大数据集训练痛苦 |
| XGBoost | 82%~85% | 中 | 低 | 调参后上限最高 |
注意这些数字是五分类的结果，二分类会高不少。别拿二分类的 99% 去吹，没意义。
3.2 训练脚本：统一接口 + 交叉验证
下面这段代码把多个模型塞进同一个循环，省得你一个个手写：
from sklearn.neighbors import KNeighborsClassifier from sklearn.tree import DecisionTreeClassifier from sklearn.ensemble import RandomForestClassifier from sklearn.naive_bayes import GaussianNB from sklearn.svm import SVC from xgboost import XGBClassifier from sklearn.model_selection import cross_val_score, StratifiedKFold from sklearn.metrics import classification_report, confusion_matrix models = { "KNN": KNeighborsClassifier(n_neighbors=5, n_jobs=-1), "DecisionTree": DecisionTreeClassifier(max_depth=20, random_state=42), "RandomForest": RandomForestClassifier(n_estimators=100, max_depth=20, n_jobs=-1, random_state=42), "NaiveBayes": GaussianNB(), "SVM": SVC(kernel="rbf", C=1.0, gamma="scale"), "XGBoost": XGBClassifier(n_estimators=200, max_depth=6, learning_rate=0.1, use_label_encoder=False, eval_metric="mlogloss", random_state=42) } cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) results = {} for name, model in models.items(): scores = cross_val_score(model, X_train, y_train, cv=cv, scoring="f1_macro") results[name] = scores.mean() print(f"{name}: F1-macro = {scores.mean():.4f} (+/- {scores.std():.4f})") # 选最优模型在测试集上出详细报告 best_name = max(results, key=results.get) best_model = models[best_name] best_model.fit(X_train, y_train) y_pred = best_model.predict(X_test) print(f"\n最优模型: {best_name}") print(classification_report(y_test, y_pred, target_names=encoders["attack_cat"].classes_)) print(confusion_matrix(y_test, y_pred))逻辑说明：用StratifiedKFold而不是普通 KFold，因为类别不平衡，分层抽样能保证每折里 U2R 和 R2L 都有样本。评估指标选f1_macro而不是 accuracy，macro 会平等看待每个类别，逼你关注少数类。最后用最优模型在独立测试集上出 classification_report，重点看 U2R 和 R2L 的 recall——这两个才是真实场景里最危险的攻击。
参数说明：n_estimators=100对随机森林通常够用，加到 300 提升有限但训练时间翻倍。XGBoost 的max_depth=6是起点，调到 8~10 容易过拟合，配合min_child_weight一起调。SVM 的C越大越容易过拟合，gamma="scale"是 sklearn 的默认推荐值，别乱改。
3.3 类别不平衡的处理：SMOTE 和 class_weight 怎么选
U2R 只有 52 条，任何模型都很难学好。两条路：一是用 SMOTE 在训练集上过采样少数类，二是给模型设class_weight="balanced"。我的经验是树模型用 class_weight 就够了，SVM 和 KNN 配合 SMOTE 效果更明显。但 SMOTE 别在测试集上做，那是数据泄漏。还有个坑：SMOTE 对离散特征做插值会产生不存在的类别值，比如 protocol_type 变成 1.5，所以要么先 SMOTE 再编码，要么只用 SMOTE 处理连续特征。
from imblearn.over_sampling import SMOTE from imblearn.pipeline import Pipeline as ImbPipeline # 只在训练集上过采样，测试集保持原始分布 smote = SMOTE(k_neighbors=3, random_state=42) X_train_res, y_train_res = smote.fit_resample(X_train, y_train) print("过采样后:", np.bincount(y_train_res))k_neighbors=3是因为 U2R 样本太少，默认的 5 会报错。过采样后训练集会膨胀到十几万条，训练时间相应增加。
4. 避坑与排查：那些让我重跑过模型的坑
4.1 测试集出现训练集没见过的类别，transform 直接崩
现象：LabelEncoder.transform抛ValueError: y contains previously unseen labels。原因：NSL-KDD 测试集里有训练集没出现过的 service 值，比如某些冷门协议。解决：用 map + 判断兜底，见 2.2 节代码里的if x in le.classes_ else -1。或者改用OrdinalEncoder(handle_unknown="use_encoded_value", unknown_value=-1)，更省事。
4.2 用 accuracy 评估，模型全预测 Normal 还沾沾自喜
现象：准确率 75%，看着还行，但 confusion matrix 里 U2R 和 R2L 的 recall 全是 0。原因：类别不平衡 + 用错指标。解决：换成f1_macro或balanced_accuracy，并且每次训练完必须看 per-class report。我现在的习惯是，只要 minority class 的 recall 低于 0.5，这个模型就不算合格。
4.3 归一化在切分之后做，测试集信息泄漏到训练集
现象：离线评估很好，上线效果差一截。原因：先对全量数据 fit scaler 再切分，测试集的统计量污染了训练。解决：先切分，再在训练集上fit_transform，测试集只transform。代码里 2.2 节就是正确顺序。这个坑太经典了，血泪经验。
4.4 XGBoost 的 use_label_encoder 参数在新版本被移除
现象：TypeError: __init__() got an unexpected keyword argument 'use_label_encoder'。原因：xgboost 1.6 之后这个参数废弃了。解决：直接删掉这个参数，eval_metric保留。如果你用的是旧版代码，升级 xgboost 后记得全局搜一下这个参数。
4.5 模型保存用 pickle，换环境加载报错
现象：本地训练好的模型，部署到服务器pickle.load报ModuleNotFoundError或版本不兼容。原因：pickle 绑定了类路径和库版本。解决：用joblib保存，或者更稳的方式是保存 XGBoost 原生的.json/.ubj格式。sklearn 模型用 joblib 跨版本兼容性也比 pickle 好。
import joblib joblib.dump(best_model, "ids_model.joblib") # 加载 model = joblib.load("ids_model.joblib")5. 推理脚本与线上验证：把模型塞进真实流量管道
5.1 单条流量推理：从字典到预测结果
训练完只是第一步，能对单条新流量做推理才算落地。下面这个函数接收一个 dict，返回攻击类别和置信度：
def predict_single(model, scaler, encoders, feature_dict): """ feature_dict: 包含 41 个原始特征的字典 返回: (预测类别, 各类别概率) """ df = pd.DataFrame([feature_dict]) cat_cols = ["protocol_type","service","flag"] for col in cat_cols: le = encoders[col] df[col] = df[col].map(lambda x: le.transform([x])[0] if x in le.classes_ else -1) num_cols = [c for c in df.columns if c not in cat_cols] df[num_cols] = scaler.transform(df[num_cols]) # 保证列顺序和训练时一致 df = df[X_train.columns] if hasattr(X_train, "columns") else df pred = model.predict(df.values)[0] proba = model.predict_proba(df.values)[0] if hasattr(model, "predict_proba") else None label = encoders["attack_cat"].inverse_transform([pred])[0] return label, proba逻辑说明：推理时的预处理必须和训练时完全一致——同样的编码器、同样的 scaler、同样的列顺序。列顺序这个坑很隐蔽，pandas 的 dict 转 DataFrame 默认按字母序排，和训练时的列序不一样，模型会给出莫名其妙的结果。所以我在训练时会把X_train的列名存下来，推理时强制对齐。
参数说明：feature_dict的 key 必须和训练时的列名完全匹配，少一个字段就报错。置信度proba可以用来做阈值过滤——比如概率低于 0.6 的先转人工，别直接封 IP。
5.2 批量推理与告警降噪
真实场景里你不会一条条调，而是从 Kafka 或日志文件里批量读。我一般会攒 1000 条跑一次 batch predict，然后把预测为攻击的按类别聚合，同一源 IP 短时间内触发多次同类攻击才告警。这样能把误报压下去一大截。验证方法：拿一周的历史告警日志，用模型重新跑一遍，对比规则引擎的告警数和模型告警数，看模型是否抓到了规则漏掉的、以及是否引入了新的误报。这个对比表比任何离线指标都有说服力。
从那以后我每次上线新模型，都强制走一遍「历史告警回放」——把过去一个月的流量日志喂给模型，人工抽查它标记为攻击但规则没报的记录。十次里有三次能发现规则引擎的盲区，这个习惯帮我省了太多半夜被叫醒的麻烦。希望帮到你。
本文还有配套的精品资源，点击获取
