简介:这套基于CNN+LSTM实现的网络流量检测系统Python源码压缩包,属于课设项目并附带文档说明,面向计算机、人工智能、通信工程等专业在校生,可满足课程设计、毕业设计、初期立项演示等场景,也能作为入门深度学习时序建模的练习材料。包内共6个文件,包括5个Python脚本和1个txt文档;脚本分别承担数据预处理、模型定义、训练与测试、主流程入口等职责,文档补充项目使用说明与必要背景,压缩包整体仅6KB,结构紧凑、便于快速阅读和二次修改。代码已在本地编译运行通过,评审得分保持在95分以上,难度适中,能从数据加载、特征提取到CNN与LSTM结合完成流量分类的完整过程展示出来。已有296人学习下载,适合需要复现课设结果、理解网络流量检测原理或在此基础上扩展功能的学习者。
1. 网络流量检测系统的CNN+LSTM落地:三行命令跑通,难点全在数据怎么喂进模型
网络流量检测系统听起来是个工程活,但这套基于CNN+LSTM的python课设源码,把检测流程做成了能跑通、能答辩的完整体。传统检测靠端口和固定规则,加密流量和伪装端口一多就会失效。思路是把网络流拆成多个时间窗口,窗口内字节reshape成二维矩阵交给CNN,连续窗口再交给LSTM,最后判断会话是否异常。源码在traffic_identification-main目录下,data_preprocess转数据、data_load按batch喂数据、model定义模型、train_and_test训练测试,main是入口。评审95分主要靠每块独立可运行。适合有Python和PyTorch基础、准备人工智能方向课设的人,也适合想把深度学习写进简历的求职者。
2. 数据预处理与加载链路:从pcap到DataLoader的shape之战
这个项目的代码量不算大,但结构清晰得有点课设模板的意思。我拿到的压缩包里,traffic_identification-main目录下五个py文件加一份文档,每个文件对应一个环节。要快速理解整份源码,正确顺序是先跟着数据走一遍,而不是先看模型。
2.1 文件结构与各模块职责
| 文件 | 职责 | 关键输出 |
|---|---|---|
| data_preprocess.py | 读原始流量,按五元组聚合会话,切时间窗口,转张量 | features.npy / labels.npy |
| data_load.py | 封装Dataset和DataLoader,完成训练集测试集切分 | train_loader / test_loader |
| model.py | CNN+LSTM模型定义 | CNNLSTM实例 |
| train_and_test.py | 训练、评估、保存权重 | best_model.pth、指标日志 |
| main.py | 命令行入口,串起前四个模块 | 一条命令跑完整流程 |
| 文档.txt | 设计思路、参数说明、答辩要点 | 给评审看的文字材料 |
文件的组织结构直接决定了答辩的讲述顺序。评审老师很少一上来就问准确率,更多是沿着“数据怎么处理→模型怎么设计→训练怎么验证”往下问。所以这个项目里每个脚本名都对着一个环节,注释里也分别写着各自的输入输出约定。如果你要把它改成自己的仓库,建议保留这个拆分,不要图省事把四个脚本合成一个,否则改动时每动一行都得重新梳理整段逻辑。
2.2 data_preprocess.py:从pcap到可训练张量
网络流量不是天生就能进卷积的。一张pcap里有成千上万个包,包长、到达时间、协议头各不相同。常见做法是先按五元组(源IP、目的IP、协议、源端口、目的端口)把原始包聚成会话,再在时间轴上切窗口,每个窗口取固定字节数,不够补零,多了截断,最后reshape成二维矩阵。
# data_preprocess.py 的简化流程 import numpy as np def pcap_to_session_features(pcap_path, seq_len=10, window_bytes=784): flows = read_flows_from_pcap(pcap_path) # 按五元组聚合 feat_list = [] for flow in flows: flow.sort_by_time() # 按到达时间排序 frames = split_by_time_window(flow, window_seconds=5) if len(frames) < seq_len: # 窗口不足的会话补零 frames = frames + [np.zeros(window_bytes)] * (seq_len - len(frames)) windows = np.array([np.resize(f, window_bytes) for f in frames[:seq_len]]) feat_list.append(windows.reshape(seq_len, 1, 28, 28)) labels = np.array([get_label(flow) for flow in flows]) return np.array(feat_list), labels这里几个参数是全局设计,改任何一个都会影响后面模型输入尺寸:
seq_len=10:每条会话样本由10个连续时间窗口组成,这是给LSTM的时间维度。window_bytes=784:每个窗口截取784字节,正好reshape成28×28。这个尺寸刻意对齐MNIST图像惯例,卷积层不用改padding和核大小就能直接吃。np.resize帮你做了两件事:长度不足自动补零,长度超了自动截断。这样每个样本形状完全一致,才能堆成同一个batch训练。
写这个函数时有个很容易被忽视的细节:补零的位置。我一般只在会话末尾补全零窗口,这对应“这条流后面的确没有流量了”;不要在单个窗口内部补零,否则会把一个真实包的字节拆散,CNN学到的空间特征就失真了。如果你手里的数据不是pcap而是公开数据集的CSV,处理路径也类似:把特征向量补零到784维,再reshape成28×28,后续模型部分一行不用改。
预处理环节我建议在函数入口和出口各加一行打印shape的代码,比如print(features.shape, labels.shape)。这个项目能在评审拿高分,和每一步数据形状可验证有直接关系。答辩时老师问“为什么输入是五维”,你能随手写出(样本数, seq_len, 1, 28, 28),这个问题就过了。
2.3 data_load.py:Dataset和DataLoader,以及一个常见翻车点
预处理完的数据通常会整体load进内存,课设数据量级在这个量级下没问题。data_load.py的核心是把已经转成numpy的特征封装成PyTorch的Dataset,再交给DataLoader按batch切分,同时把shuffle逻辑放在加载层。
# data_load.py import torch from torch.utils.data import Dataset, DataLoader class TrafficDataset(Dataset): def __init__(self, features, labels): self.features = torch.FloatTensor(features) self.labels = torch.LongTensor(labels) def __len__(self): return len(self.labels) def __getitem__(self, idx): return self.features[idx], self.labels[idx] def build_loaders(features, labels, batch_size=64, train_ratio=0.8): split = int(len(features) * train_ratio) train_ds = TrafficDataset(features[:split], labels[:split]) test_ds = TrafficDataset(features[split:], labels[split:]) train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True, num_workers=0) test_loader = DataLoader(test_ds, batch_size=batch_size, shuffle=False, num_workers=0) return train_loader, test_loader默认写法里train_ratio=0.8按顺序切分,这在流量检测场景里有隐患。攻击流量往往按时间聚集,如果前80%几乎全是正常会话,后20%集中了攻击会话,测试出来的F1会很丢人,但准确率可能很高。更稳妥的做法是换用分层切分,让训练集和测试集里的正常/攻击比例都接近总体比例。
from sklearn.model_selection import train_test_split X_train, X_test, y_train, y_test = train_test_split( features, labels, test_size=0.2, random_state=42, stratify=labels)stratify=labels表示按标签比例分层抽样,比手写[:split]更稳。另一个容易踩的是Windows下num_workers设太大,DataLoader会反复报错或者卡住,这个项目把num_workers=0是安全的选择;我一般只在Linux服务器上才开到4。到这里,数据和模型之间就差一层对接了,下一节看model.py怎么把五维张量吃进去。
3. model.py的CNN+LSTM结构:空间特征与时间依赖怎么叠
模型结构是答辩时讨论最多的地方。CNN和LSTM各负责一段,要能讲清楚为什么必须两个一起用。
3.1 CNN部分:窗口内的空间特征
单个时间窗口内的字节排列并不随机。一个正常的HTTP GET请求包和一个扫描探测包,头部字段分布、负载字节的统计特征都会在28×28的矩阵里留下不同的局部模式。卷积核在这里做的事,就是扫描这些局部模式:3×3的核滑过矩阵,某一种字节组合反复出现,对应位置的卷积响应就被激活。这和图像识别里检测边缘、纹理的逻辑一致,只不过输入从像素换成了字节。
这个项目里CNN采用了两层卷积加池化的组合。第一层32个卷积核负责捕捉细粒度字节组合,第二层64个卷积核在前一层基础上组合出更抽象的流量模式。窗口内的字节如果全是补零,卷积后特征图也趋近于零值,这个信号对后面的LSTM同样有用:它告诉模型这个时间窗内没有有效流量。
3.2 LSTM部分:跨窗口的时序依赖
网络攻击很少在一个时间窗口内完成。慢速扫描把探测动作分散到多个窗口,攻击载荷被拆成小包穿插在正常流量里,单看任何一个窗口都看不出问题。LSTM在序列维度上把这些窗口串起来,用学习到的门控机制决定保留哪些信息、丢弃哪些信息。
和普通RNN相比,LSTM的输入门、遗忘门、输出门让梯度能沿时间维度传得更远,模型能记住十来个窗口之前的可疑动作。答辩时如果被问到“为什么不用RNN”,可以从这里切入:RNN在长序列上有梯度消失问题,LSTM用门控缓解了它,CNN和RNN提取的是不同侧面的特征,这里按需叠加。
有一点值得注意:这个项目用的是单向LSTM而不是双向。原因是流量检测到某个时刻才能做判断,双向LSTM会引入未来信息,实时检测场景下不成立。课设里提这一点,比单纯说“我用了LSTM”更能体现工程判断。
3.3 模型实现与张量维度流转
model.py里,CNN和LSTM不是前后并联,而是CNN先对每个窗口独立提取特征,再把整段特征序列交给LSTM,这个顺序也决定了forward里维度要先摊平再重组。一个典型实现如下:
# model.py import torch.nn as nn class CNNLSTM(nn.Module): def __init__(self, seq_len=10, lstm_hidden=64, num_classes=2, dropout=0.5): super().__init__() self.cnn = nn.Sequential( nn.Conv2d(1, 32, kernel_size=3, padding=1), nn.ReLU(), nn.MaxPool2d(2), nn.Conv2d(32, 64, kernel_size=3, padding=1), nn.ReLU(), nn.MaxPool2d(2) ) self.lstm = nn.LSTM( input_size=64 * 7 * 7, hidden_size=lstm_hidden, num_layers=2, batch_first=True, dropout=dropout ) self.classifier = nn.Sequential( nn.Linear(lstm_hidden, 32), nn.ReLU(), nn.Dropout(dropout), nn.Linear(32, num_classes) ) def forward(self, x): # 输入: (batch, seq_len, 1, 28, 28) batch, seq, c, h, w = x.size() x = x.view(batch * seq, c, h, w) # 每个窗口独立过CNN cnn_out = self.cnn(x) # (batch*seq, 64, 7, 7) cnn_out = cnn_out.view(batch, seq, -1) # (batch, seq, 64*7*7) lstm_out, _ = self.lstm(cnn_out) # (batch, seq, lstm_hidden) logits = self.classifier(lstm_out[:, -1, :]) return logits维度流转是整个模型能不能跑起来的命门:
- 输入x是五维张量
(batch, seq_len, 1, 28, 28),batch是同时训练的样本数,seq_len是时间窗口数,后三位是单通道窗口图。 x.view(batch * seq, c, h, w)把10个窗口摊成一个大batch过CNN。好处是不写循环、训练快,所有窗口共享同一套卷积权重。- 池化两次后特征图从28×28降到7×7,通道数从1涨到64,每个窗口被压缩成64×7×7=3136维的特征向量。
- LSTM看到的序列长度是seq_len,每步输入3136维。
batch_first=True保证输出维度是(batch, seq, hidden),取lstm_out[:, -1, :]就是最后一个时间步的隐状态,作为整段会话的摘要去做分类。
实际运行中,CNN阶段的参数量主要来自卷积核,LSTM阶段则集中在四个门控矩阵。整体参数量在百万级别,CPU训练一个epoch要几分钟,GPU几十秒。如果训练时间太长,优先把seq_len从10降到8,效果损失通常不大;如果显存溢出,把batch_size从64降到32。注意nn.LSTM里的dropout只在num_layers>1时生效,这里设了2层所以没问题,但如果改成单层LSTM,这个参数会被忽略。
4. train_and_test.py的调参与验证:跑通流程,更要避开假准确率
模型定下来后,训练脚本决定能不能交付一份好看的指标。这个项目的train_and_test.py把训练、验证、保存权重放在一个入口里,结构上没什么花活,但细节不少。
4.1 训练循环的基本骨架
# train_and_test.py 骨架 import torch import torch.nn as nn from torch.optim import Adam from data_load import build_loaders from model import CNNLSTM def train_one_epoch(model, loader, optimizer, criterion, device): model.train() total_loss, correct, total = 0, 0, 0 for x, y in loader: x, y = x.to(device), y.to(device) optimizer.zero_grad() logits = model(x) loss = criterion(logits, y) loss.backward() optimizer.step() total_loss += loss.item() * y.size(0) correct += (logits.argmax(dim=1) == y).sum().item() total += y.size(0) return total_loss / total, correct / total def evaluate(model, loader, criterion, device): model.eval() total_loss, correct, total = 0, 0, 0 preds, trues = [], [] with torch.no_grad(): for x, y in loader: x, y = x.to(device), y.to(device) logits = model(x) total_loss += criterion(logits, y).item() * y.size(0) correct += (logits.argmax(dim=1) == y).sum().item() preds.extend(logits.argmax(dim=1).cpu().tolist()) trues.extend(y.cpu().tolist()) total += y.size(0) return total_loss / total, correct / total, preds, trues训练循环里三个位置最容易忽略:
model.train()和model.eval()必须成对切换。dropout在训练时随机丢弃神经元,评估时关闭,忘记切会让测试指标忽高忽低。optimizer.zero_grad()要放在每个batch计算前。放在loss.backward()之后,上一次梯度和本次梯度会叠加,损失曲线出现明显震荡。- 评估要包在
torch.no_grad()里,省显存,也避免PyTorch误记录计算图。
主流程里一般还会做早停:每个epoch后在验证集上计算F1,连续5个epoch不提升就停掉训练,并用验证F1最高的那一轮权重作为最终结果权重。这个习惯比“固定训练30轮”稳得多,尤其在小数据集上。
best_f1, bad_epochs = 0, 0 patience = 5 for epoch in range(epochs): train_loss, train_acc = train_one_epoch(model, train_loader, optimizer, criterion, device) _, _, preds, trues = evaluate(model, val_loader, criterion, device) val_f1 = f1_score(trues, preds, zero_division=0) if val_f1 > best_f1: best_f1 = val_f1 torch.save(model.state_dict(), "best_model.pth") bad_epochs = 0 else: bad_epochs += 1 if bad_epochs >= patience: break4.2 超参数设定与调整范围
项目里可调的超参数集中在build_loaders和CNNLSTM的构造参数上。下面是我建议的起点值和调整方向:
| 参数 | 建议值 | 调整方向 |
|---|---|---|
| batch_size | 64 | 内存不足降32,显存充足可以升128 |
| learning_rate | 1e-3 | 损失震荡降到3e-4或1e-4 |
| epochs | 30 | 配合早停,看验证曲线 |
| seq_len | 10 | 慢速攻击场景考虑12~16 |
| window_bytes | 784 | 负载内容少可以减到256或512 |
| lstm_hidden | 64 | 数据量大考虑128 |
| lstm_layers | 2 | 1层更快,3层过拟合风险明显 |
| dropout | 0.5 | 过拟合升0.6,欠拟合降0.3 |
这些参数之间不是独立的。window_bytes变了,CNN后面的展开维度就变,LSTM的input_size要同步更新;seq_len变了,输入样本的时间维度变,预处理脚本里的切窗逻辑也要对应改。我一般建议先把seq_len和window_bytes定死,因为它们牵动预处理,训练中只调学习率和dropout。
4.3 评估指标不要只看Accuracy
课设里最容易出现的假象是准确率很高但模型实际没什么用。当正常流量占90%时,一个把全部样本判成正常的模型也有90%准确率。攻击检测里漏报的代价远高于误报,所以汇报时要以F1和召回率为主,而不是只念准确率。
from sklearn.metrics import confusion_matrix, f1_score, recall_score import torch def print_metrics(trues, preds): acc = (torch.tensor(preds) == torch.tensor(trues)).float().mean().item() recall = recall_score(trues, preds, zero_division=0) f1 = f1_score(trues, preds, zero_division=0) print(f"Accuracy={acc:.4f} Recall={recall:.4f} F1={f1:.4f}") print(confusion_matrix(trues, preds))每次训练完我都会打印一次混淆矩阵,看攻击类样本具体被分错到哪个方向。如果漏报集中在某一类攻击,回到预处理去看这类攻击的窗口字节是否被截断太多,通常是窗口开太小,攻击特征被np.resize截掉了。
4.4 main.py:把全流程串成一条命令
main.py的作用是把上面的脚本按顺序跑一遍。通常它会先调pcap_to_session_features生成数据,再调build_loaders,然后实例化模型开训练,最后保存权重。把它做成能用python main.py --epochs 30这类命令行参数调用的入口,在答辩演示时很加分。
5. 避坑指南:这个课设最容易翻车的五个点
5.1 输入张量维度对不上
现象:跑第一个epoch直接报错,提示Expected 5D input, got 4D,或者模型能跑但loss完全不下降。
原因:预处理输出少了一个维度,常见是切完窗口后忘了把(seq_len, 1, 28, 28)组合成完整样本;或者DataLoader把四维数据stack成了五维再传进来,和forward里expect的五维对不上。
解决:在build_loaders之前强制打印features.shape,标准答案是(样本数, seq_len, 1, 28, 28)。缺哪个维度就回预处理对应行去找。如果不想靠眼睛数维度,在训练前加一行x = next(iter(train_loader))[0]; print(x.shape),比任何文档都直接。这个报错看着吓人,但其实是最好修的一类,因为它把问题精确指到了维度上。
5.2 shuffle设置反了
现象:训练损失正常下降,测试F1却低得离谱;或者测试时指标和训练时差距过大。
原因:训练集没shuffle,模型按时间顺序看到相邻样本,学到的是样本之间的连续性关系,而不是分类边界。测试集shuffle=True虽然不影响准确率计算,但会让人对样本顺序产生误判,做可视化时极容易对错标签。
解决:训练loader固定shuffle=True,测试loader固定shuffle=False。这个规则在思路上也想清楚:训练要打乱,才能让模型看到多样本组合;测试要保持原始顺序,后续画曲线和抽样例才靠谱。
5.3 正负样本不均衡,准确率假象
现象:测试准确率99%,但F1只有0.3,混淆矩阵显示模型几乎把攻击样本都判成正常。
原因:训练数据里攻击样本占比太低,交叉熵损失被多数类主导,模型发现“全判正常”损失也不高。流量数据集里正常流量本来就占大头,这个问题几乎一定会遇到。
解决:优先用分层切分保证训练集和测试集的类别比例一致,然后给少数类加权。PyTorch里最简单的方式是在CrossEntropyLoss里传class_weight:
import torch import torch.nn as nn class_weights = torch.FloatTensor([1.0, 5.0]).to(device) # 攻击类给更高权重 criterion = nn.CrossEntropyLoss(weight=class_weights)如果还不行,就把build_loaders里的采样方式换成WeightedRandomSampler,让攻击样本以更高概率被抽进每个batch。权重怎么定,可以直接看训练集里两类样本数量比值,不是拍脑袋。
5.4 用全量数据做归一化,信息泄露
现象:离线测试指标很好看,换一批新流量重测,效果肉眼可见地下降。
原因:预处理时用了整个数据集(包括测试集)的均值方差做归一化,模型在训练阶段已经侧面接触了测试集分布。流量数据的统计特性随时间变化,测试集本来就是要模拟没见过的数据。
解决:先切分训练集和测试集,再各自做归一化,且只统计训练集的mean和std:
from sklearn.preprocessing import StandardScaler scaler = StandardScaler().fit(X_train) X_train = scaler.transform(X_train) X_test = scaler.transform(X_test)这个习惯从数据侧杜绝了作弊,答辩时老师很容易问“你归一化是在切数据之前还是之后”,答“之后”就是加分项。如果用的是MinMaxScaler,逻辑完全一样,绝不能用全量数据去fit。
5.5 权重保存和加载格式不一致
现象:训练跑完了,把所有attack流量都判成normal,检查脚本逻辑没问题,就是结果不对。
原因:保存时用了torch.save(model, "best_model.pth"),推理时用model.load_state_dict(torch.load(...))读取;或者训练在GPU上、推理在CPU上,张量设备不匹配报错。这类错误在换环境跑时最容易出现。
解决:训练端只保存model.state_dict(),推理端配合模型结构加载:
torch.save(model.state_dict(), "best_model.pth") # 推理端 model = CNNLSTM() model.load_state_dict(torch.load("best_model.pth", map_location="cpu")) model.eval()map_location="cpu"可以让你在没GPU的机器上也能跑通演示,也是答辩演示机最常见的坑。如果你在训练脚本里顺手把优化器状态也存了,加载时记得只取model_state_dict对应的键。
6. 单条流量预测与可视化验证:把训练Demo变成能答辩的检测工具
课设答辩最怕老师现场说“那你拿一条没见过的流量预测一下”。所以拿到这个项目后第一件事就是补一个单条推理入口,把训练好的权重固化下来。这能让程序从训练Demo变成检测工具,也让前面跑出来的准确率有了实际落点。
6.1 单条新流量预测
import torch def predict_pcap(model, pcap_path, seq_len=10, window_bytes=784): model.eval() features, _ = pcap_to_session_features(pcap_path) # 复用预处理函数 x = torch.FloatTensor(features).unsqueeze(0) # (1, seq_len, 1, 28, 28) with torch.no_grad(): prob = torch.softmax(model(x), dim=1) label = "attack" if prob.argmax().item() == 1 else "normal" print(f"预测: {label}, 置信度: {prob.max().item():.3f}")推理时unsqueeze(0)加的是batch维度。训练时DataLoader自动加batch,单条预测时常常有人忘记这一步,一跑就报错。我习惯把它写进函数第一行,形成条件反射。
6.2 验证模型学到的到底是什么
光跑一遍训练就交付,风险太高。我的做法是训练结束后从测试集里随机抽10条样本,逐条预测并打印真实标签和置信度。如果某条attack样本置信度低于0.6,回到预处理看原始窗口是否被截断过多、补零占比是否过高,这比盯着总指标判断更有价值。也可以顺手提取几层卷积特征图存成图片,看一眼模型在流量窗口上关注的字节区域,答辩展示时比口述有说服力。
从那以后我每次做这类课设都强制走一遍:先打印shape,再盯训练损失,再查混淆矩阵,最后做单条预测。任何一个环节不对劲,立刻回到data_preprocess去看数据本身,而不是急着改模型结构。神经网络在流量检测上不是玄学,数据喂端正了,模型才公平地执行它该做的事。这套经验以后拿去做入侵检测、恶意流量分类都能复用,希望帮到你。
本文还有配套的精品资源,点击获取