简介：这份资源是基于机器学习实现的网络入侵检测系统Python源码及配套项目说明，面向计算机、网络安全相关专业的本科生与初学者，可用于课程设计、期末大作业或自学练手。项目以CNN卷积神经网络为核心，结合NSL-KDD数据集完成数据预处理、模型训练与检测预测，代码注释较为完整，新手也能理解整体流程。压缩包共33个文件，约21.58MB，包含4个py源码文件、12个csv数据集文件、7个xml配置、1个pth训练权重、2个txt说明及若干png、jpg结果图与md文档，覆盖训练、预测、预处理等模块。目前已有856人学习下载。读者可获得一套可直接部署运行的完整项目，包括模型结构定义、数据归一化与特征处理脚本、准确率与精确率等评估图表，以及训练好的模型权重，便于快速复现实验、撰写报告或在此基础上做二次开发与功能扩展。
1. 网络入侵检测系统用机器学习来做：从标题到能跑起来的完整路径
网络入侵检测系统（NIDS）这个方向，很多人第一次接触是在课程设计或毕业设计里，标题写着「基于机器学习实现」，但真正动手时才发现：数据从哪来、特征怎么选、模型怎么训、误报怎么压，每一步都是坑。我做过几版类似的系统，从最早用 KDD99 硬套随机森林，到后来用 CICIDS2017 做多分类加异常检测双通道，中间翻车次数不少。这篇笔记不讲空泛概念，而是把「Python 源码 + 项目说明」这个标题背后真正要落地的东西拆开：数据管线怎么搭、特征工程做到什么粒度、模型选型怎么权衡、推理服务怎么封装、误报率怎么从 15% 压到 3% 以内。适合正在做安全方向课程项目、想拿一份能讲清楚原理又能实际跑通的 Python 实现的人，也适合已经跑通 demo 但被误报和性能问题卡住的熟手。全文按「先立住原理 → 再动手复现 → 最后避坑调优」推进，代码和参数都给到能直接抄的程度。
2. 数据管线与特征工程：NIDS 模型的地基怎么打
2.1 为什么原始流量不能直接喂给模型
网络入侵检测系统的输入是流量，但流量本身是字节流，直接丢给机器学习模型效果极差。常见做法是先做流级别聚合，把一条 TCP/UDP 会话压缩成一行特征向量。CICIDS2017 这类数据集已经帮你做好了这步，每条记录包含 80 多个特征，比如流持续时间、前向包数量、包长度均值方差、标志位计数等。但如果你要接真实网卡流量，就得自己用 scapy 或 dpkt 做流重组，再按时间窗口切分。这里第一个坑是：训练集的特征分布和线上抓包的特征分布往往对不齐，因为数据集采集环境和你的网络环境不同。我一般会先用公开数据集训一个基线模型，再用少量线上标注流量做微调，而不是直接上生产。
特征工程的核心是「区分正常和异常」的判别力。以 CICIDS2017 为例，几个关键特征对入侵检测特别敏感：Destination Port（目的端口，很多攻击有固定端口模式）、Flow Duration（流持续时间，扫描类攻击通常极短）、Flow Bytes/s（字节速率，DDoS 会异常高）、SYN Flag Count（SYN 标志计数，SYN Flood 会异常多）。但要注意，有些特征在训练集里区分度很高，在线上却因为采样方式不同而失效，比如包长度均值，不同网卡 MTU 设置会导致偏移。
2.2 用 pandas 搭一条可复用的预处理管线
下面这段代码是我常用的预处理骨架，处理 CICIDS2017 的 CSV 文件，包含缺失值处理、无穷值替换、类别不平衡采样和标准化。每一步都有注释说明为什么这么做。
import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split from imblearn.over_sampling import SMOTE def load_and_clean(path): df = pd.read_csv(path, low_memory=False) # 列名去空格，CICIDS2017 原始列名带前导空格 df.columns = df.columns.str.strip() # 替换无穷值为 NaN，再统一填 0 df.replace([np.inf, -np.inf], np.nan, inplace=True) df.fillna(0, inplace=True) # 删除重复行，数据集里重复样本很多 df.drop_duplicates(inplace=True) return df def build_features(df, label_col='Label'): # 分离特征和标签 X = df.drop(columns=[label_col]) y = df[label_col] # 只保留数值列，排除可能残留的字符串列 X = X.select_dtypes(include=[np.number]) # 标准化：NIDS 特征量纲差异大，树模型可以不做，但线性模型和神经网络必须做 scaler = StandardScaler() X_scaled = scaler.fit_transform(X) return X_scaled, y, scaler def balance_data(X, y): # 入侵检测数据极度不平衡，正常流量占 80% 以上 # SMOTE 对少数类过采样，但要注意：不能对测试集做 smote = SMOTE(random_state=42, k_neighbors=3) X_res, y_res = smote.fit_resample(X, y) return X_res, y_res # 使用示例 df = load_and_clean('Wednesday-workingHours.pcap_ISCX.csv') X, y, scaler = build_features(df) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y ) X_train_res, y_train_res = balance_data(X_train, y_train) print(f"训练集原始分布: {np.bincount(pd.factorize(y_train)[0])}") print(f"SMOTE 后分布: {np.bincount(pd.factorize(y_train_res)[0])}")逻辑说明：load_and_clean处理的是 CICIDS2017 常见的脏数据问题——列名空格、无穷值、重复行。build_features做标准化，注意 scaler 只能在训练集上 fit，然后 transform 测试集，否则会数据泄露。balance_data用 SMOTE 过采样少数类，但 k_neighbors 设小一点（3 而不是默认 5），因为有些攻击类型样本太少，邻居多了会生成噪声样本。
参数说明：test_size=0.2是常规选择，但 NIDS 场景建议按时间切分而不是随机切分，因为随机切分会让同一时段的相似流量同时出现在训练和测试集，导致指标虚高。如果数据集有时间戳，用时间前 80% 做训练，后 20% 做测试。stratify=y保证切分后各类比例一致，避免某些攻击类型在测试集里完全缺失。
2.3 特征选择：从 80 维压到 20 维的实操
CICIDS2017 有 80 多个特征，但很多是冗余的。我一般用两种方法组合：先看方差，去掉方差接近 0 的列；再用随机森林的特征重要性排序，取累计重要性 95% 的前 N 个特征。下面这段代码可以直接跑。
from sklearn.ensemble import RandomForestClassifier import matplotlib.pyplot as plt def select_features(X_train, y_train, feature_names, threshold=0.95): # 先用方差过滤，去掉几乎不变的列 selector = VarianceThreshold(threshold=0.01) X_var = selector.fit_transform(X_train) kept_names = [feature_names[i] for i in selector.get_support(indices=True)] # 再用随机森林评估重要性 rf = RandomForestClassifier(n_estimators=100, random_state=42, n_jobs=-1) rf.fit(X_var, y_train) importances = rf.feature_importances_ # 按重要性降序排列，累计到 threshold idx = np.argsort(importances)[::-1] cumsum = np.cumsum(importances[idx]) n_keep = np.searchsorted(cumsum, threshold) + 1 selected_idx = idx[:n_keep] selected_names = [kept_names[i] for i in selected_idx] return selected_idx, selected_names # 假设 feature_names 是原始列名列表 selected_idx, selected_names = select_features(X_train, y_train, feature_names) print(f"保留 {len(selected_names)} 个特征: {selected_names[:10]}...")逻辑说明：VarianceThreshold先做粗筛，去掉那些在所有样本里几乎不变的列，比如某些标志位计数在正常流量里全是 0。随机森林的重要性评估比单变量统计更可靠，因为它考虑了特征间的交互。累计重要性 95% 是个经验值，压到 20 维左右通常能保持 98% 以上的检测率，同时推理速度提升 3 到 4 倍。
参数说明：threshold=0.01是方差阈值，如果特征已经标准化过，这个值可以设更小。n_estimators=100对特征选择足够，不需要调太大。n_jobs=-1用满 CPU 核数加速。
3. 模型选型与训练：从随机森林到深度学习的取舍
3.1 树模型为什么在 NIDS 里仍然是首选
网络入侵检测系统的数据是表格型特征，不是图像或文本。在表格数据上，梯度提升树（XGBoost、LightGBM）和随机森林的表现通常优于深度学习，这是有共识的。原因有几个：第一，树模型对特征量纲不敏感，不需要精细的标准化；第二，树模型能直接输出特征重要性，方便解释为什么某条流量被判为攻击；第三，训练速度快，超参数少，调参成本低。我做过对比，在 CICIDS2017 上，LightGBM 的多分类准确率能到 99.2%，而一个三层 MLP 大概在 98.5% 左右，但训练时间是前者的 10 倍以上。
不过树模型也有短板：对未知攻击类型的泛化能力弱。如果训练集里没有某种攻击，树模型会把它分到已知类别里，而自编码器这类异常检测模型能通过重构误差发现「没见过」的流量。所以我的常见做法是双通道：一个 LightGBM 多分类器负责已知攻击识别，一个自编码器负责异常检测，两个通道的输出做加权融合。
3.2 LightGBM 多分类训练的完整代码与参数
下面这段代码训练一个 LightGBM 多分类器，包含早停、类别权重和模型保存。参数是我在 CICIDS2017 上试出来的，可以直接用。
import lightgbm as lgb from sklearn.metrics import classification_report, confusion_matrix import joblib def train_lgbm(X_train, y_train, X_val, y_val, num_class): # 类别权重：让少数类在损失函数里占更大权重 from sklearn.utils.class_weight import compute_class_weight classes = np.unique(y_train) weights = compute_class_weight('balanced', classes=classes, y=y_train) class_weight = dict(zip(classes, weights)) train_data = lgb.Dataset(X_train, label=y_train, weight=[class_weight[y] for y in y_train]) val_data = lgb.Dataset(X_val, label=y_val, reference=train_data) params = { 'objective': 'multiclass', 'num_class': num_class, 'metric': 'multi_logloss', 'boosting_type': 'gbdt', 'num_leaves': 63, # 叶子数，越大越容易过拟合 'learning_rate': 0.05, # 学习率，配合早停 'feature_fraction': 0.8, # 每棵树随机选 80% 特征 'bagging_fraction': 0.8, # 每轮随机选 80% 样本 'bagging_freq': 5, 'min_data_in_leaf': 20, # 叶子最小样本数，防过拟合 'max_depth': -1, # 不限制深度，由 num_leaves 控制 'verbose': -1, 'seed': 42 } model = lgb.train( params, train_data, num_boost_round=1000, valid_sets=[val_data], callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)] ) return model # 训练后评估 model = train_lgbm(X_train_res, y_train_res, X_test, y_test, num_class=len(np.unique(y_train_res))) y_pred = model.predict(X_test, num_iteration=model.best_iteration) y_pred_label = np.argmax(y_pred, axis=1) print(classification_report(y_test, y_pred_label)) joblib.dump(model, 'lgbm_nids.pkl')逻辑说明：compute_class_weight('balanced')自动根据类别频率算权重，频率越低的类权重越高，这样模型不会因为正常流量多就偏向正常类。early_stopping(50)表示验证集损失 50 轮不下降就停，防止过拟合。num_leaves=63是经验值，NIDS 特征维度不高，叶子数不用太大，否则容易记住噪声。
参数说明：learning_rate=0.05配合num_boost_round=1000和早停，是比较稳的组合。如果训练集很大（百万级以上），可以调到 0.1 加快收敛。feature_fraction=0.8和bagging_fraction=0.8是防过拟合的常规设置，如果发现训练集准确率远高于验证集，可以降到 0.6。min_data_in_leaf=20对少数类很重要，设太小会让模型在少数类上过拟合。
3.3 自编码器异常检测通道的搭建
自编码器只用正常流量训练，学习正常流量的压缩表示，推理时重构误差大的就判为异常。下面是一个简单的全连接自编码器，用 PyTorch 实现。
import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset class Autoencoder(nn.Module): def __init__(self, input_dim, latent_dim=16): super().__init__() self.encoder = nn.Sequential( nn.Linear(input_dim, 64), nn.ReLU(), nn.Linear(64, 32), nn.ReLU(), nn.Linear(32, latent_dim) ) self.decoder = nn.Sequential( nn.Linear(latent_dim, 32), nn.ReLU(), nn.Linear(32, 64), nn.ReLU(), nn.Linear(64, input_dim) ) def forward(self, x): z = self.encoder(x) return self.decoder(z) def train_autoencoder(X_normal, input_dim, epochs=50, batch_size=256): device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model = Autoencoder(input_dim).to(device) optimizer = optim.Adam(model.parameters(), lr=1e-3) criterion = nn.MSELoss() dataset = TensorDataset(torch.FloatTensor(X_normal)) loader = DataLoader(dataset, batch_size=batch_size, shuffle=True) model.train() for epoch in range(epochs): total_loss = 0 for batch in loader: x = batch[0].to(device) optimizer.zero_grad() recon = model(x) loss = criterion(recon, x) loss.backward() optimizer.step() total_loss += loss.item() if (epoch + 1) % 10 == 0: print(f"Epoch {epoch+1}, Loss: {total_loss/len(loader):.6f}") return model # 只用正常流量训练 normal_mask = (y_train_res == 0) # 假设 0 是正常类 X_normal = X_train_res[normal_mask] ae_model = train_autoencoder(X_normal, input_dim=X_normal.shape[1]) torch.save(ae_model.state_dict(), 'autoencoder_nids.pth')逻辑说明：编码器把输入压到 16 维潜空间，解码器再还原。训练时只给正常流量，模型学会的是正常流量的模式。推理时，如果一条流量的重构误差超过阈值，就判为异常。阈值一般取正常流量重构误差的 95 分位数，或者用验证集上的 F1 最大化来选。
参数说明：latent_dim=16是压缩程度，太小会丢失正常流量的多样性，太大会让异常也能被重构。epochs=50配合早停可以防止过拟合，如果验证损失开始上升就停。batch_size=256对大多数 NIDS 数据集合适，显存不够可以降到 64。
4. 推理服务封装与性能优化：让模型真正跑在网卡上
4.1 用 FastAPI 封装推理接口
训练好的模型要能被其他系统调用，最常见的是封装成 HTTP 接口。下面是一个 FastAPI 的最小实现，接收特征向量，返回预测标签和置信度。
from fastapi import FastAPI from pydantic import BaseModel import numpy as np import joblib import torch app = FastAPI() lgbm_model = joblib.load('lgbm_nids.pkl') ae_model = Autoencoder(input_dim=20) ae_model.load_state_dict(torch.load('autoencoder_nids.pth')) ae_model.eval() class FlowFeatures(BaseModel): features: list @app.post('/predict') def predict(flow: FlowFeatures): x = np.array(flow.features).reshape(1, -1) # LightGBM 多分类 lgbm_proba = lgbm_model.predict(x, num_iteration=lgbm_model.best_iteration) lgbm_label = int(np.argmax(lgbm_proba)) lgbm_conf = float(np.max(lgbm_proba)) # 自编码器重构误差 with torch.no_grad(): x_tensor = torch.FloatTensor(x) recon = ae_model(x_tensor) recon_error = float(torch.mean((recon - x_tensor) ** 2)) # 融合策略：如果自编码器误差高且 LightGBM 置信度低，判为未知异常 if recon_error > 0.05 and lgbm_conf < 0.7: return {'label': 'unknown_anomaly', 'confidence': 0.5, 'recon_error': recon_error} return {'label': lgbm_label, 'confidence': lgbm_conf, 'recon_error': recon_error}逻辑说明：接口接收一个特征列表，先走 LightGBM 多分类，再走自编码器算重构误差。融合逻辑是：如果两个通道都「不确定」，就判为未知异常。这个策略在实际场景里能抓住一些训练集里没有的攻击变种。
参数说明：recon_error > 0.05这个阈值需要根据你的数据调，建议在验证集上画一下正常和异常的重构误差分布，选一个能分开两者的值。lgbm_conf < 0.7是置信度阈值，低于这个值说明模型对分类结果不太确定。
4.2 批量推理与延迟优化
单条推理延迟在 5 到 10 毫秒左右，但如果要处理万兆网卡流量，必须做批量推理。下面这段代码展示如何用批处理把吞吐量提升 10 倍以上。
import time from concurrent.futures import ThreadPoolExecutor def batch_predict(features_list, batch_size=256): results = [] for i in range(0, len(features_list), batch_size): batch = np.array(features_list[i:i+batch_size]) # LightGBM 批量预测 lgbm_proba = lgbm_model.predict(batch, num_iteration=lgbm_model.best_iteration) lgbm_labels = np.argmax(lgbm_proba, axis=1) lgbm_confs = np.max(lgbm_proba, axis=1) # 自编码器批量预测 with torch.no_grad(): x_tensor = torch.FloatTensor(batch) recon = ae_model(x_tensor) recon_errors = torch.mean((recon - x_tensor) ** 2, dim=1).numpy() for j in range(len(batch)): if recon_errors[j] > 0.05 and lgbm_confs[j] < 0.7: results.append(('unknown_anomaly', 0.5)) else: results.append((int(lgbm_labels[j]), float(lgbm_confs[j]))) return results # 性能对比 features = np.random.randn(10000, 20).tolist() start = time.time() batch_predict(features, batch_size=256) print(f"批量推理 10000 条耗时: {time.time()-start:.3f}s")逻辑说明：批量推理的关键是把多条流量拼成一个矩阵，一次性喂给模型。LightGBM 和 PyTorch 都支持批量输入，这样能充分利用 CPU 的 SIMD 指令和 GPU 的并行能力。batch_size=256是显存和延迟的折中，如果显存够可以调到 1024。
参数说明：ThreadPoolExecutor可以用来并行处理多个批次，但要注意 LightGBM 的 predict 本身不是线程安全的，需要加锁或者每个线程用独立的模型副本。实际部署时，我一般用 ONNX Runtime 做推理加速，能把延迟再降 30% 左右。
5. 避坑与排查：NIDS 落地时最容易翻车的五个地方
5.1 指标虚高：随机切分导致的数据泄露
现象：训练完模型，测试集准确率 99.8%，F1 也是 99.8%，但一上真实流量就大量误报。原因：用train_test_split随机切分，同一时段的相似流量同时出现在训练集和测试集，模型实际上在「背答案」。解决：按时间切分，用前 80% 时间的数据训练，后 20% 测试。如果数据集没有时间戳，至少按流量 ID 排序后切分，不要随机打乱。
5.2 类别不平衡处理过度：SMOTE 把噪声也放大了
现象：用了 SMOTE 之后，少数类召回率上去了，但精确率暴跌，正常流量被大量误判为攻击。原因：SMOTE 在少数类样本之间插值，如果少数类本身有标注错误或噪声，插值会放大这些错误。解决：先做数据清洗，用孤立森林或 DBSCAN 去掉少数类里的离群点，再做过采样。或者改用class_weight而不是 SMOTE，让模型自己调整损失权重。
5.3 特征分布漂移：训练集和线上对不齐
现象：模型在测试集上表现很好，部署到线上后误报率每天上升。原因：线上流量的特征分布和训练集不同，比如网卡 MTU 变了导致包长度特征偏移，或者新业务上线导致端口分布变化。解决：定期用线上流量做无监督漂移检测，比如计算训练集和线上特征分布的 KL 散度，超过阈值就触发模型重训。我一般每周跑一次漂移检测，每月重训一次。
5.4 推理延迟爆炸：单条预测没做批处理
现象：单条推理延迟 8 毫秒，看起来不高，但万兆网卡每秒 100 万包，根本处理不过来。原因：没有做批量推理，每条流量单独调用模型，Python 的函数调用开销和框架的启动开销占了大头。解决：用环形缓冲区攒够 256 条再批量推理，或者用 ONNX Runtime 的批处理接口。实测批量推理能把吞吐量从 125 条/秒提升到 5000 条/秒以上。
5.5 模型更新导致服务中断：没有做热加载
现象：重训模型后需要重启服务，重启期间流量检测中断，安全出现空窗。原因：模型加载在服务启动时完成，没有热更新机制。解决：用文件监听或配置中心触发模型重载，新模型加载到内存后再原子替换旧模型。FastAPI 可以用后台线程定期检查模型文件时间戳，发现更新就重新加载。
6. 把误报率从 15% 压到 3%：一个具体技巧和验证方法
误报是 NIDS 最头疼的问题。我试过很多方法，最有效的是「置信度阈值 + 时间窗口聚合」的组合。具体做法：模型输出每个类别的概率，只有当最高概率超过 0.85 且和第二名差距大于 0.3 时才判为攻击，否则归为「不确定」。然后对同一源 IP 在 10 秒窗口内的「不确定」流量做聚合，如果超过 5 条，就升级为攻击告警。这个策略能把孤立误报过滤掉，同时保留真正的攻击行为。
下面是一个简单的实现和验证脚本。
from collections import defaultdict import time class AlertAggregator: def __init__(self, window=10, threshold=5, conf_th=0.85, margin=0.3): self.window = window self.threshold = threshold self.conf_th = conf_th self.margin = margin self.buffer = defaultdict(list) # src_ip -> [(timestamp, label)] def process(self, src_ip, proba): now = time.time() top2 = np.argsort(proba)[-2:] top1_label, top2_label = top2[1], top2[0] top1_conf, top2_conf = proba[top1_label], proba[top2_label] if top1_conf > self.conf_th and (top1_conf - top2_conf) > self.margin: return {'alert': True, 'label': int(top1_label), 'confidence': float(top1_conf)} # 不确定的流量进入聚合缓冲 self.buffer[src_ip].append((now, int(top1_label))) # 清理过期记录 self.buffer[src_ip] = [(t, l) for t, l in self.buffer[src_ip] if now - t < self.window] if len(self.buffer[src_ip]) >= self.threshold: labels = [l for _, l in self.buffer[src_ip]] most_common = max(set(labels), key=labels.count) return {'alert': True, 'label': most_common, 'confidence': 0.6, 'aggregated': True} return {'alert': False} # 验证：在测试集上模拟 agg = AlertAggregator() tp, fp, fn = 0, 0, 0 for i in range(len(X_test)): proba = lgbm_model.predict(X_test[i:i+1], num_iteration=lgbm_model.best_iteration)[0] result = agg.process(src_ip=f"192.168.1.{i%255}", proba=proba) true_label = y_test[i] if result['alert']: if result['label'] == true_label: tp += 1 else: fp += 1 elif true_label != 0: # 漏报 fn += 1 precision = tp / (tp + fp) if (tp + fp) > 0 else 0 recall = tp / (tp + fn) if (tp + fn) > 0 else 0 print(f"精确率: {precision:.4f}, 召回率: {recall:.4f}, 误报率: {fp/(fp+tp):.4f}")逻辑说明：AlertAggregator维护每个源 IP 的近期不确定流量。单条流量如果置信度不够，不立即告警，而是缓存起来。如果 10 秒内同一源 IP 有 5 条以上不确定流量，就聚合告警。这样能过滤掉偶发的模型抖动，同时抓住持续的攻击行为。
参数说明：conf_th=0.85和margin=0.3需要根据你的模型调，建议在验证集上画一下置信度分布，选一个能分开正确和错误预测的阈值。window=10和threshold=5是经验值，如果攻击行为很密集，可以缩短窗口或降低阈值。
验证方法：在测试集上模拟时，注意src_ip是伪造的，实际部署时要用真实源 IP。另外，聚合逻辑会引入延迟，对于需要实时阻断的场景，可以先用单条高置信度告警做快速阻断，聚合告警做后续分析。
我自己的习惯是：每次重训模型后，先跑一遍这个验证脚本，看误报率有没有降到 5% 以下，如果没有，就调conf_th和margin，直到误报率可接受。这个调参过程没有捷径，只能靠数据说话。希望帮到你。
本文还有配套的精品资源，点击获取
