简介:基于CNN与LSTM相结合的网络流量检测系统Python源码,专为高校课程设计、毕业设计及期末大作业场景打造,也适合希望掌握深度学习流量识别方法的开发者参考。项目整体难度适中,代码均经过本地编译验证可运行,评审得分达98分,内容经助教老师审定,符合学习与项目实践要求。压缩包共6个文件,以5个Python脚本与1个Markdown文档构成,脚本分别负责数据加载与预处理、模型结构定义、训练与测试流程及主程序入口,说明文档则梳理了项目背景与使用方式;资源包仅6KB,结构精简,便于快速阅读和修改。目前已有132人学习下载,可直接作为课程设计、毕设或期末大作业的完整参考,也可帮助初学者理解CNN与LSTM在网络流量分类中的搭配思路。
1. 基于CNN+LSTM做网络流量检测:为什么这个组合是课设的稳妥选择
做网络流量检测的课设或毕设时,为了区分恶意流量和正常流量,大多数人第一反应是提取统计特征后丢给随机森林或XGBoost。这套方案做起来确实快,但特征要手工设计,效果上限往往卡在特征工程上。另一个常见思路是把流量当作文本,用CNN或RNN单独去做分类,可单独用CNN会丢掉时间维度上的前后依赖,单独用LSTM又容易在长序列上收敛慢、局部模式抓不牢。CNN+LSTM的组合恰好补上了彼此的短板:CNN用卷积核抓流量的局部空间特征,LSTM负责建模时序依赖,最后接一个全连接层做分类。这套结构不仅适合做二分类的恶意流量识别,也能平滑迁移到入侵检测、异常流量发现等场景,是课设答辩时很好讲清楚、同时代码工作量可控的方案。这篇文章把整个链路拆开讲清楚:数据怎么整理、模型怎么写、参数怎么调、哪些坑我踩过,照着做就能跑通。
2. CNN与LSTM如何分工:网络结构设计与输入形态选择
2.1 为什么CNN在前、LSTM在后:局部特征与时序依赖的组合逻辑
网络流量检测的关键在于从原始数据中找到能区分正常和异常的模式。正常流量和攻击流量之间,往往存在两类差异:一类是短时间窗口内的局部特征,比如某个数据包长度异常、标志位组合特殊、单位时间内突发的小包数量增多;另一类是跨时间步的长程依赖,比如慢速端口扫描会持续几十秒,C2通信会呈现周期性心跳。CNN擅长前者,卷积核在一个局部窗口内做特征提取,能得到类似“N-gram”的局部模式;LSTM擅长后者,通过门控机制记住前面若干个时间步的状态。把CNN放在前面,先对每个时间步的原始特征做局部抽象和降维,再把抽象后的序列喂给LSTM,LSTM学到的就是更高层的时序规律,最后用全连接层把隐藏状态映射成类别概率。
我一般会明确的告诉来问我的同学:这个组合的收益来源不是“两个模型叠加”,而是“两种归纳偏置互补”。如果你把CNN的输出直接拉平做分类,丢弃了时间顺序,就退化成单纯的局部特征模型;如果你把原始高维特征直接送进LSTM,LSTM一方面要处理维度灾难,另一方面低层局部模式没有被抽象出来,训练时会慢很多。一个经过CNN抽象的低维序列,对LSTM来说既减少了输入维度,还隐式做了噪声过滤。
2.2 流量输入怎么组织:特征矩阵的两种形态与选择
模型结构定了,输入长什么样同样重要。常见做法是把一个网络流(flow)表示成一个二维矩阵,形状是T × F,T 是时间步数,F 是每个时间步的特征维度。特征怎么选,决定了矩阵里填什么数。
- 形态A:按固定时间窗口聚合原始包统计特征。比如每个窗口内计算包数量、字节数均值、包长方差、活跃端口数等,作为一行特征。适合在pcap文件或NetFlow数据上做,特征含义清晰,容易解释。
- 形态B:直接使用原始包字节序列的前N个字节,做离散化编码后作为输入。端到端学习能力更强,但数据量需求大,训练更慢,课设阶段如果数据规模不够,容易欠拟合。
我的建议是选形态A。因为课设的数据集通常只有几万到十几万个样本,形态B需要大量数据才能学出有效的字节级特征,形态A的统计特征和CNN+LSTM的组合配合得更稳定,调参路径也更短。每个流的样本组织方式如下:先按五元组(源IP、目的IP、源端口、目的端口、协议)把原始流量聚成一条条流,然后对每条流按时间顺序切窗口,得到序列样本。
2.3 基线网络结构:卷积核、隐藏层、分类头的参数设定
设计具体的网络结构时,不需要从零摸索。对于流量检测这类中等规模表格型序列数据,基线结构基本收敛到一套稳定配置:输入特征矩阵先进入一维CNN层,卷积核大小取3或5,卷积核数量取64到128,然后跟一个最大池化;池化后的序列输入单层或双层LSTM,隐藏单元数取64到128;最后一个时间步的隐藏状态接全连接层和sigmoid输出。这个结构简洁、训练快、不太依赖GPU,大部分CPU也能跑完。
参数上有几个值得注意的地方。卷积核大小决定局部感受野,流量检测里取3或5通常足够,取大了会把本来独立的包间隔模式混在一起。LSTM层数超过2层时收益有限,数据量不够还容易过拟合,课设场景下单层或双层都合理,dropout建议设0.3到0.5。另外,class权重不要忽略:恶意流量在数据集中占比往往低于正常流量,不处理类别不平衡,模型会偏向预测多数类,后面的评估章节会详细写。
下面是这个基线结构用PyTorch实现的核心代码:
import torch import torch.nn as nn class CNNLSTM(nn.Module): def __init__(self, feature_dim=20, hidden_dim=128, num_layers=2, num_classes=1, dropout=0.4): super(CNNLSTM, self).__init__() # 一维CNN:输入形状为 (batch, feature_dim, seq_len) # 这里把特征维度放在通道维,卷积核在时间维度上滑动 self.conv1 = nn.Conv1d(in_channels=feature_dim, out_channels=64, kernel_size=3, padding=1) self.relu = nn.ReLU() self.pool = nn.MaxPool1d(kernel_size=2, stride=2) self.conv2 = nn.Conv1d(in_channels=64, out_channels=128, kernel_size=3, padding=1) self.lstm = nn.LSTM( input_size=128, hidden_size=hidden_dim, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0.0, bidirectional=False ) self.dropout = nn.Dropout(dropout) self.fc = nn.Linear(hidden_dim, num_classes) def forward(self, x): # x 形状: (batch, seq_len, feature_dim) # 转成 (batch, feature_dim, seq_len) 以适应Conv1d x = x.permute(0, 2, 1) x = self.conv1(x) x = self.relu(x) x = self.pool(x) x = self.conv2(x) x = self.relu(x) x = self.pool(x) # 转回 (batch, seq_len, channels) x = x.permute(0, 2, 1) lstm_out, (hn, cn) = self.lstm(x) # 取最后一个时间步的隐藏状态 out = hn[-1] # 形状: (batch, hidden_dim) out = self.dropout(out) out = self.fc(out) return out逻辑说明:第一层卷积把20维特征映射成64个通道,经过ReLU和池化后序列长度减半;第二层卷积升到128个通道,再池化一次。这样每个时间步的特征从20维变成128维的抽象表示,再输入LSTM。hn[-1]取最后一层LSTM最后一个时间步的隐藏状态,作为整个序列的汇总向量。参数说明:feature_dim要与数据预处理时输出的特征列数一致,hidden_dim控制LSTM记忆容量(128是甜点值,大模型容易过拟合),num_layers=2时dropout才会在层间生效,kernel_size=3加padding=1保证序列长度在卷积后不缩短。
3. 数据预处理与数据集构建:从原始流量到可训练的序列样本
3.1 流特征提取与标签构造
模型代码写完只是第一步,更影响最终分数的是数据质量。原始pcap文件不能直接送进网络,需要先把数据包聚合成流。我常用的流程是:读pcap → 按五元组哈希分组 → 同一方向的数据包按时间排序 → 切分时间窗口 → 计算每个窗口的特征向量 → 生成标签。如果找的是公开数据集,很多已经做成了CSV格式,一行一个样本,字段包括流持续时间、包数、字节数、各种统计值,这类数据能直接进入特征处理环节。唯一要注意的是,公开数据集里的标签列通常叫Label或Class,要把恶意类标成1,正常类标成0,别搞反。
构造特征矩阵时,seq_len(时间步数)和feature_dim(每步特征维度)这两个值直接决定模型输入形状。我一般把seq_len设为50,即每个样本包含50个连续时间窗口。feature_dim设为20左右,特征组合推荐:该窗口内包数量、平均包长、包长标准差、平均到达间隔、TCP标志位计数、上下行流量比、活跃端口数量等。这些特征都是统计量,计算简单,且能和网络里的卷积核尺寸匹配。
3.2 滑动窗口切分与序列对齐
聚合完流以后,一条长流可能包含几千个包,不能整条塞进模型。标准做法是滑动窗口切分:窗口大小固定,步长可以等于窗口大小(不重叠)或小于窗口大小(重叠采样)。我用不重叠切分做基线,数据量不够时再考虑重叠采样做数据增强。每切出一个窗口序列,就生成一条训练样本,标签取该流整体的类别。这里有一个很容易忽略的点:一条流内窗口级别可能有噪声,比如某个窗口内恰好没有恶意行为,但整条流被标记为恶意。实践下来,只要窗口数量足够多、标签服从多数,模型能容忍这种噪声,不需要做窗口级别的精细标注。
切窗口时的另一个注意点是丢尾。窗口序列长度不足seq_len时,我直接丢弃该样本。填充零的做法会把大量全零行喂给LSTM,让模型学到“末尾全是空”的虚假模式,测试时反而降低泛化能力。如果你的数据集中短流比例很高,可以把seq_len调小到30甚至20,不要硬凑。
3.3 归一化的时机和训练集划分:不做标准化,loss很难收敛
特征之间的量纲差异很大:包数量可能是几百,包长方差可能是几万,到达间隔可能是小数。LSTM内部是sigmoid和tanh激活,输入数值范围太宽会让梯度计算不稳定。所以标准化这一步必须做。常用的方式是Z-score归一化,但归一化有一个隐藏坑:必须先用训练集计算均值和标准差,再用同一组参数去归一化验证集和测试集。如果对整个数据集归一化再划分,验证集的信息已经污染了训练过程,属于数据泄漏,评估指标会虚高,答辩时被问到容易翻车。
划分数据集时,训练、验证、测试按8:1:1或7:1.5:1.5比较合适。验证集用于早停和调参,测试集只在最终评估时碰一次。代码实现上,用train_test_split做一次划分后用StandardScaler拟合训练集的scaler,再把三个集合分别transform。
import pandas as pd import numpy as np from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler # 假设df是处理好的特征文件,label列为0/1,其余为数值特征 feature_cols = [c for c in df.columns if c != 'label'] X = df[feature_cols].values y = df['label'].values # 先划分,后标准化,避免数据泄漏 X_train, X_temp, y_train, y_temp = train_test_split(X, y, test_size=0.2, stratify=y, random_state=42) X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=0.5, stratify=y_temp, random_state=42) scaler = StandardScaler() X_train = scaler.fit_transform(X_train) # 只有训练集参与fit X_val = scaler.transform(X_val) X_test = scaler.transform(X_test) # 把特征矩阵重排成 (样本数, seq_len, feature_dim) # 这里假设每个样本的原始特征行数是 seq_len 的整数倍 seq_len = 50 feature_dim = X_train.shape[1] // seq_len X_train = X_train.reshape(-1, seq_len, feature_dim) X_val = X_val.reshape(-1, seq_len, feature_dim) X_test = X_test.reshape(-1, seq_len, feature_dim) print(f"训练集形状: {X_train.shape}, 验证集形状: {X_val.shape}")逻辑说明:先用stratify=y保证训练集和临时集里的正负样本比例一致,避免随机划分导致验证集里某一类占比过高。StandardScaler只调用fit_transform一次,后续验证集和测试集只做transform。reshape的假设前提是特征文件里的每一行已经是按时间窗口顺序排好的窗口特征,如果原始数据顺序不对,要先按流ID和时间排序再导出特征文件。参数说明:random_state固定后,每次运行划分结果一致,便于复现;seq_len必须整除特征总数,否则说明窗口个数与seq_len不匹配。
4. 用PyTorch实现CNN+LSTM模型:完整训练代码与参数配置
4.1 训练循环实现:早停、学习率衰减与模型保存
模型定义好了,数据也整理成(样本数, seq_len, feature_dim)的形状,接下来就是写训练循环。这个阶段最影响成败的三个机制:早停、学习率衰减、模型保存。早停的作用是在验证集loss连续多个epoch不下降时终止训练,防止过拟合;学习率衰减用ReduceLROnPlateau实现,验证loss进入平台期就降低学习率;模型保存只保留验证集上表现最好的那一版权重,等训练结束后加载。下面是一份可以直接跑的训练代码:
import torch import torch.nn as nn from torch.utils.data import TensorDataset, DataLoader device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = CNNLSTM(feature_dim=feature_dim, hidden_dim=128, num_layers=2).to(device) # 类别不平衡时,给少数类更高的权重 pos_count = int(y_train.sum()) neg_count = len(y_train) - pos_count pos_weight = torch.tensor([neg_count / max(pos_count, 1)], dtype=torch.float32).to(device) criterion = nn.BCEWithLogitsLoss(pos_weight=pos_weight) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode="min", factor=0.5, patience=3) train_dataset = TensorDataset(torch.FloatTensor(X_train), torch.FloatTensor(y_train)) val_dataset = TensorDataset(torch.FloatTensor(X_val), torch.FloatTensor(y_val)) train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=128, shuffle=False) epochs = 30 best_f1 = 0.0 best_path = "best_model.pt" patience_counter = 0 max_patience = 6 for epoch in range(epochs): model.train() train_loss = 0.0 for batch_x, batch_y in train_loader: batch_x, batch_y = batch_x.to(device), batch_y.to(device) logits = model(batch_x).squeeze(1) loss = criterion(logits, batch_y) optimizer.zero_grad() loss.backward() nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() train_loss += loss.item() * batch_x.size(0) # 验证集评估 model.eval() val_preds, val_labels = [], [] val_loss = 0.0 with torch.no_grad(): for batch_x, batch_y in val_loader: batch_x, batch_y = batch_x.to(device), batch_y.to(device) logits = model(batch_x).squeeze(1) loss = criterion(logits, batch_y) val_loss += loss.item() * batch_x.size(0) probs = torch.sigmoid(logits) val_preds.extend((probs > 0.5).cpu().numpy()) val_labels.extend(batch_y.cpu().numpy()) from sklearn.metrics import f1_score val_f1 = f1_score(val_labels, val_preds, zero_division=0) val_loss_avg = val_loss / len(val_dataset) train_loss_avg = train_loss / len(train_dataset) print(f"Epoch {epoch+1}/{epochs} | train_loss={train_loss_avg:.4f} | val_loss={val_loss_avg:.4f} | val_f1={val_f1:.4f}") scheduler.step(val_loss_avg) if val_f1 > best_f1: best_f1 = val_f1 torch.save(model.state_dict(), best_path) patience_counter = 0 else: patience_counter += 1 if patience_counter >= max_patience: print(f"早停触发:连续{max_patience}个epoch验证F1未提升") break逻辑说明:BCEWithLogitsLoss直接在logits上计算损失,不需要在模型里先过sigmoid,数值稳定性更好。pos_weight根据训练集正负样本比例设置,让模型对少数类误判付出更大代价。clip_grad_norm_把梯度范数限制在1.0以内,防止LSTM训练中出现梯度爆炸导致loss变成NaN。ReduceLROnPlateau检测验证loss两三个epoch不降就把学习率减半,配合早停双重防过拟合。参数说明:batch_size取64到128都行,太小时BN统计不稳定,太大时收敛慢;weight_decay=1e-5做L2正则,配合dropout一起控制过拟合;早停的max_patience设6个epoch,避免因为验证集小波动就过早停止。
4.2 关键超参数速查表与调整思路
训练过程中,超参数之间是联动的,但有几个规律可以遵循。学习率从1e-3起步,如果训练loss前几个epoch不下降,降到3e-4;如果loss下降但验证F1波动剧烈,也考虑降学习率。LSTM隐藏层大小从128开始,验证集F1上不去时试试256,但要注意过拟合风险。seq_len从50开始,数据中长流多可以上调,短流多就下调。dropout太低会过拟合,太高会让LSTM学不到时序依赖,0.3到0.5之间调。卷积核数量翻倍(从64到128)能提升表示能力,但训练时间也接近翻倍。
| 超参数 | 推荐初始值 | 调整方向 |
|---|---|---|
| 学习率 | 1e-3 | loss不降则降为3e-4,振荡则降为5e-4 |
| batch_size | 64 | 显存不足或收敛慢时降到32 |
| seq_len | 50 | 短流多为30,长流为主可到80 |
| 卷积核大小 | 3 | 时间局部模式弱时改5 |
| LSTM隐藏层 | 128 | F1不够时试256,注意过拟合 |
| dropout | 0.4 | 过拟合调大,欠拟合调小 |
| 早停epoch数 | 6 | 训练波动大时放宽到8 |
4.3 训练后需要保存哪些产物
训练结束后,除了最佳权重文件,还需要把标准化器保存下来。原因很直接:部署或测试时,新样本必须用训练集同款均值和标准差做归一化,直接加载模型权重但忘记scaler,模型输出的概率会是垃圾数值。用joblib.dump把scaler保存成.pkl文件,和模型权重放同一目录。另外,训练时的验证集预测结果和混淆矩阵最好也导出成CSV或图片,做课设报告时直接引用,答辩现场不用重新跑模型。
5. 训练与评估中的避坑清单:5个真实踩坑记录
5.1 归一化泄漏导致验证集指标虚高
现象:训练时验证集F1稳定在0.95以上,测试集F1却只有0.7左右,线上线下差距很大。原因:某些教程里直接对所有数据算fit_transform再划分训练集和测试集,验证集和测试集的分布信息提前暴露给了标准化器,相当于模型偷偷见过测试数据的统计特征。解决:严格按训练集→fit、验证集和测试集→transform的顺序处理。我在某个模拟项目的流量检测实验里,仅修正这一个bug,测试集F1就从0.74降到0.71的真实水平——之前的0.74是含水分的。
5.2 类别不平衡下准确率陷阱
现象:恶意样本只占5%,模型把所有样本都预测成正常类,准确率95%,看起来模型很强,实际F1是0。原因:没有设置pos_weight,模型发现全预测多数类就能把loss压得很低。解决:训练时给BCEWithLogitsLoss传pos_weight,评估时不要看准确率,主看F1和召回率。恶意流量检测场景里,漏报的代价远高于误报,所以召回率比精确率更值得关注。
5.3 LSTM梯度爆炸导致loss为NaN
现象:训练到第几个epoch时loss突然变成NaN,之后无法恢复。原因:LSTM反向传播路径长,梯度范数可能指数增长,学习率稍高就会触发。解决:在optimizer.step()前加clip_grad_norm_(model.parameters(), max_norm=1.0)。我在双LSTM层配置下遇到过两次,加上梯度裁剪后没有再出现过。如果裁剪后仍然NaN,检查输入数据里是否存在NaN值,np.isnan(X).sum()跑一下。
5.4 序列窗口切分错位导致标签错配
现象:训练loss正常下降,但验证F1忽高忽低,而且不同随机种子结果差异巨大。原因:特征文件导出时流ID没有排序,同一个流的窗口被拆散,reshape成(样本数, seq_len, feature_dim)后,一个样本里混入了不同流的数据,标签也跟着错位。解决:切窗口之前先按流ID和时间戳排序,确认每个样本的seq_len个窗口都来自同一条流。检查方法很简单:导出一个样本的窗口,打印每个窗口对应的流ID,应该全部相等。
5.5 验证集标签泄漏:重复样本导致评估过于乐观
现象:验证集F1比测试集高一大截,且怎么调参都拉不平。原因:特征文件里存在完全重复的行,划分数据时同一个样本同时出现在训练集和验证集里。原因一般是数据处理阶段滑动窗口步长小于窗口大小,重叠窗口被原样保留,没有去重。解决:切窗口前对特征行去重,或者在train_test_split之前检查df.duplicated().sum(),把重复行删掉。数据量允许时,重叠采样生成的样本也可以全部去重。
6. 模型验证与进一步优化:用混淆矩阵做评估,向注意力机制扩展
6.1 混淆矩阵与能展示的评估指标
训练完成、权重保存好了,最后一步是把测试集跑一遍,输出完整的评估结果。测试集在训练过程中一次都不能碰,只在最后评估时使用。我习惯在notebook里同时打印准确率、精确率、召回率、F1和混淆矩阵,再画ROC曲线。对课堂答辩来说,混淆矩阵的展示价值最高:你可以直观指出模型把多少恶意流量判成了正常,把多少正常流量误报为恶意。
import numpy as np from sklearn.metrics import confusion_matrix, classification_report, roc_auc_score model.load_state_dict(torch.load(best_path, map_location=device)) model.eval() test_preds, test_probs = [], [] with torch.no_grad(): for batch_x in torch.FloatTensor(X_test).split(128): batch_x = batch_x.to(device) logits = model(batch_x).squeeze(1) probs = torch.sigmoid(logits) test_probs.extend(probs.cpu().numpy()) test_preds.extend((probs > 0.5).cpu().numpy()) print(classification_report(y_test, test_preds)) print("混淆矩阵:\n", confusion_matrix(y_test, test_preds)) print("ROC-AUC:", roc_auc_score(y_test, test_probs))关于判定阈值,默认0.5并不是最优选择。在恶意流量检测场景下,我会在验证集上尝试0.3到0.7范围内的不同阈值,挑召回率更高且F1不崩溃的点作为最终阈值,而不是无脑用0.5。这一步在答辩时很加分,因为它说明你理解“分类阈值是业务决策参数”这一层。ROC-AUC是阈值无关的综合指标,用来横向比较不同模型配置的稳定性。
6.2 进阶方向:注意力机制、多分类与部署
CNN+LSTM基线做完以后,如果想在报告里加“扩展讨论”,有三条路性价比比较高。第一,在LSTM输出后加一个注意力池化层,替代“取最后一个时间步隐状态”的做法,让模型自己决定哪些时间步对分类更重要。注意力的实现代码很短,大概二十行左右,但F1通常能在基线上再涨一到两个点。第二,从二分类扩展到多分类,把恶意流量细分出扫描、爆破、DDoS等类型,损失函数换成CrossEntropyLoss,最后一层输出维度改成类别数。第三,部署层面,把模型导出为ONNX格式,用ONNX Runtime做CPU推理,单条样本预测时间能压到毫秒级,适合在报告里写“实时检测可行性分析”。
我自己做流量检测实验的习惯是:先把基线跑通并保存所有产物,再逐项尝试优化,每一项都单独记录对F1的影响,这样实验记录自然成为报告的实验章节。最后建议你在答辩前做一次全流程复现检查:从原始特征文件开始,跑完整套脚本,确认F1能复现到训练记录的水平。这一步能避免现场演示时因为环境问题翻车。希望这篇笔记能帮你把课设做得扎实,少走我之前走过的弯路。
本文还有配套的精品资源,点击获取