简介面向计算机专业课程设计、期末大作业与项目实战练习的恶意流量检测完整工程采用深度Q网络强化学习驱动机器学习建模解决恶意流量识别与分类问题。资源包含环境交互、智能体训练、检测验证等核心模块便于学习者理解强化学习与安全检测的结合方式。压缩包内共18个文件包括7个Python源码模块、8个CSV数据集、1个模型文件以及使用说明与依赖清单整体大小7.04MB结构清晰解压即可直接运行调试。当前已有252人学习使用。对于需要快速搭建恶意流量检测实验或深入理解深度强化学习在安全领域应用的学生和开发者可借助这套源码、数据集与文档组合完成课程报告、项目答辩并为进一步改进检测模型打下基础。1. 用DQN自动搜索恶意流量特征子集把调模型变成一场强化学习游戏一个常被忽略的事实流量检测模型的效果上限更多由特征决定而不是分类器。但拿到CICIDS2017这类数据集时几十个流量特征先筛哪几个、保留哪几个很多人靠correlation矩阵和直觉拍板。基于DQN强化学习生成恶意流量检测模型的思路是把特征选择重新建模成序贯决策agent在每一步决定要不要往特征集里加入某个特征下游分类器给出F1作为奖励最终用一个最优特征子集训练出可上线的检测模型。这条路径把调参过程变成马尔可夫决策过程适合特征维度高、正负样本不均衡、想自动化特征工程的场景。会用Python和PyTorch就能复现不需要先读懂深度强化学习的全部理论。2. 把恶意流量检测中的特征选择写成MDPDQN为什么比网格搜索划算先厘清一个关键认知这里的DQN是优化器不是检测器。它不直接读原始流量抓恶意样本它的工作是找到“哪些特征喂给随机森林或逻辑回归效果最好”。这也是标题里“生成模型”的含义——用强化学习自动生成一个可用的机器学习检测模型配置。这一步想通了后面所有代码都不容易跑偏。2.1 状态、动作、奖励的三元组如何映射流量特征工程常见做法是把特征选择写成如下MDP四元组这也是开源项目里大多数强化学习特征选择方案的底层设计状态s_t当前选中的特征子集掩码向量mask ∈ {0,1}^nn是原始特征总数再加上上一轮下游分类器在验证集上的F1值。把标量F1也拼进状态agent能感知“当前配置已经有多好”而不是盲目试探。动作a_t一共n1个离散动作。前n个动作分别对应“把第i个特征加入已选集合”最后一个动作是“停止选择输出最终子集”。这里不用连续动作因为特征选择天然是离散组合优化。奖励r_t执行动作后用当前特征子集训练一个轻量分类器在固定验证集上算F1增量。增量奖励比绝对F1更稳定能避免agent停在某个局部高水位。为什么增量奖励更稳因为绝对F1在特征数少时通常偏低agent会学到“多选特征总是好”的惰性策略而增量奖励配合下面的特征成本惩罚能让它在“多加一个特征带来的提升”和“多维护一个特征的代价”之间做真正的权衡。2.2 计算代价怎么控制奖励函数里的折扣与特征成本百万级流量数据上在每个episode里跑几十次完整交叉验证非常昂贵训练一轮可能就要数小时。两个手段是这类项目的标配第一个手段是评估用小批量数据加逻辑回归。每步奖励计算用采样后的2万条样本、2000次迭代的逻辑回归跑一次只要几百毫秒最终选定特征后再用全量随机森林或LightGBM复测。逻辑回归只做搜索期打分器不做上线模型这是刻意为之。第二个手段是给特征数量加惩罚项。奖励函数写成r_t (F1_now - F1_prev) * 100 - cost_per_feature终止时额外给一个终止奖励r_terminal F1_final * 100 - 0.5 * n_selected_features参数说明F1乘100是为了把0.7到0.9这个区间的差异放大到20分避免奖励值过小导致Q值无法区分动作好坏cost_per_feature取0.1对应“一个特征至少要带来0.001的F1提升才值得保留”。终止奖励里减去0.5乘特征数是让agent在搜索后期有动力收手。2.3 为什么DQN而不是随机搜索或遗传算法随机搜索每次从零开始完全浪费了“上一轮已经知道某些组合很差”的信息。遗传算法有记忆但交叉和变异算子作用在二进制掩码上会破坏特征的组合语义且评估每个个体同样要训练分类器总开销并不低。DQN核心优势是经验回放和目标网络历史转移样本被反复利用Q值能跨episode积累同样的评估次数下能学到“哪些特征组合持续带来正奖励”的长期规律。下表是三种方案在同等评估预算下的差异方案是否复用历史评估对离散掩码的适配调参难度随机搜索否天然适配低遗传算法部分种群内需设计算子中DQN是经验回放动作离散化即可中高调参难度是DQN的短板但换来的是搜索效率上限更高。实际项目中如果你的特征数在30个以下随机搜索加早停可能够用超过50个特征DQN的优势才会显现。3. 数据准备从原始CSV到可喂给DQN的特征掩码很多强化学习项目最终跑不出结果问题不在网络结构而在数据预处理阶段就埋了雷。这一章讲实际操作路径。3.1 用公开的二分类数据集起步恶意流量领域公开数据集常见的是CICIDS2017、UNSW-NB15和ISCX VPN。CICIDS2017的CSV按天分多个文件需要先合并标签列会同时出现良性、暴力破解、DDoS、端口扫描等多个值做特征选择实验时建议先统一成二分类Benign为0其余攻击类别全部为1。反直觉的一点是不要在最开始就做多分类任务。特征选择阶段的奖励来自F1类别一多F1对特征组合的区分度会下降二分类更容易观察到“某个特征是否有用”的因果效应。UNSW-NB15特征语义差异较大但文件更规整适合用来做验证集验证选出的特征是否跨数据集泛化。如果你本地没有这两个数据集用sklearn的fetch_openml拉一份网络入侵相关数据也可以关键是数据里必须包含足够多的原始特征列才能体现特征选择的必要。3.2 预处理流水线抽样、归一化与掩码生成数据清洗步骤顺序很重要我的固定顺序是去空值、去全零列、类型转换、样本均衡抽样、划分训练验证集、归一化。其中去全零列经常被漏掉流量数据集里某些端口相关特征在特定抓包阶段全为0留着只会增加动作空间维度。import pandas as pd import numpy as np from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler raw pd.read_csv(cicids2017_merged.csv, low_memoryFalse) # 1. 删掉全空或全零列 raw raw.dropna(axis1, howall) raw raw.loc[:, (raw ! 0).any(axis0)] # 2. 标签二值化良性0其余攻击1 y (raw[Label] Benign).astype(int).values # 3. 去掉标签列和ID列 X raw.drop(columns[Label, Flow ID, Timestamp]).select_dtypes(include[np.number]) # 4. 分层抽样到10万条保持正负比例 X_small, _, y_small, _ train_test_split( X, y, train_size100000, stratifyy, random_state42 ) # 5. 归一化并保留scaler供后续测试集使用 scaler StandardScaler().fit(X_small) X_scaled scaler.transform(X_small) np.save(X_scaled.npy, X_scaled) np.save(y_scaled.npy, y_small) np.save(feature_names.npy, np.array(X.columns))逻辑说明抽样到10万条不是为了省内存而是为了控制奖励计算的时间。逻辑回归在两万条子样本上加1000次迭代收敛单次评估稳定在0.2秒以内一个episode跑60步评估也只要12秒。stratifyy保证了抽样后恶意样本比例与原始数据一致避免奖励信号被类别分布波动污染。这一步如果只写train_test_split(X, y, train_size...)不指定stratify在攻击样本占比较低的数据集上很容易抽出一个类别比例失真的子集导致搜索期F1虚高。3.3 归一化与掩码状态之间的顺序关系这里有个容易踩的坑先归一化再选特征还是先选特征再归一化答案是训练集上先归一化再选特征。因为逻辑回归的收敛速度依赖特征尺度一致如果每次动作改变特征子集后重新做归一化状态空间里同一个掩码对应的数据分布会漂移DQN永远学不到稳定规律。正确做法如上节代码所示scaler在全部特征上fit之后无论选中哪几个特征都直接从X_scaled取对应列。代价是标准化时混入了最终被丢弃的特征统计量但在特征选择MDP里影响很小换来的是状态转移的一致性。验证集和测试集只用scaler.transform绝不在数据划分前对整个数据集做fit这是防止数据泄漏的最低要求。4. 用PyTorch实现FeatureSelectionEnv和DQN训练循环这一章给出核心可执行代码。整体结构分三块环境类、Q网络、训练循环三者可以分别独立调试。4.1 环境类掩码状态、增量奖励、终止条件import numpy as np from sklearn.linear_model import LogisticRegression from sklearn.metrics import f1_score class FeatureSelectionEnv: def __init__(self, X, y, max_steps30, cost_per_feature0.1, eval_size20000): self.X X self.y y self.max_steps max_steps self.cost_per_feature cost_per_feature self.eval_size eval_size self.n_features X.shape[1] self.action_dim self.n_features 1 # 最后一个动作是停止 def _evaluate_f1(self, mask): idx np.where(mask 1)[0] if len(idx) 0: return 0.0 # 每次评估从训练数据里采样保证不同step的奖励可比 sample_idx np.random.choice(self.X.shape[0], self.eval_size, replaceFalse) X_sub self.X[sample_idx][:, idx] y_sub self.y[sample_idx] clf LogisticRegression(max_iter500, n_jobs-1) clf.fit(X_sub, y_sub) return f1_score(y_sub, clf.predict(X_sub)) def reset(self): self.mask np.zeros(self.n_features, dtypenp.int8) self.steps 0 self.prev_f1 0.0 return np.concatenate([self.mask, [self.prev_f1]]).astype(np.float32) def step(self, action): if action self.n_features or self.steps self.max_steps: # 终止动作用全量训练集算最终F1 idx np.where(self.mask 1)[0] if len(idx) 0: reward, done -1.0, True return None, reward, done final_f1 self._evaluate_f1(self.mask) reward final_f1 * 100 - 0.5 * len(idx) return None, reward, True self.mask[action] 1 self.steps 1 cur_f1 self._evaluate_f1(self.mask) reward (cur_f1 - self.prev_f1) * 100 - self.cost_per_feature self.prev_f1 cur_f1 next_state np.concatenate([self.mask, [cur_f1]]).astype(np.float32) return next_state, reward, False逻辑说明reset返回的状态是特征掩码拼接上一轮F1掩码本身是离散0/1但拼上F1后整体作为连续向量输入网络。step里前n个动作只把对应位置置1不提供移除动作这是刻意的——恶意流量特征选择任务里一个特征选了再移除会让状态空间膨胀一倍而增量奖励机制已经能抑制无效特征加入。终止动作返回的最终奖励使用全量训练集评估避免搜索期采样子集带来的噪声影响最终决策。参数说明max_steps30限制单个episode最多选30个特征防止mask全满原始数据集如果80个特征动作空间就是81维。eval_size20000控制奖励计算速度机器配置低可以降到10000但F1波动会增大。4.2 Q网络结构与目标网络import torch import torch.nn as nn import torch.optim as optim class QNet(nn.Module): def __init__(self, state_dim, action_dim, hidden_dim128): super().__init__() self.net nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, action_dim) ) def forward(self, x): return self.net(x) state_dim env.n_features 1 online_net QNet(state_dim, env.action_dim) target_net QNet(state_dim, env.action_dim) target_net.load_state_dict(online_net.state_dict()) optimizer optim.Adam(online_net.parameters(), lr1e-3)结构说明这个Q网络只有3层MLP没有用RNN。原因是状态里已经带了完整的特征掩码向量历史选择信息都被压缩在当前mask里MDP的状态表示满足马尔可夫性不需要额外记忆。hidden_dim128对80维特征输入足够特征超过200个时可以加到256。参数说明lr1e-3是Adam在中小型Q网络上的安全起点目标网络每200步同步一次在线网络权重等待一会儿给出训练循环里的具体写法。4.3 训练主循环与epsilon贪心衰减from collections import deque import random buffer deque(maxlen50000) epsilon 1.0 epsilon_min 0.05 epsilon_decay 0.995 gamma 0.99 batch_size 64 target_update_steps 200 episodes 300 for episode in range(episodes): state env.reset() total_reward 0 done False while not done: if random.random() epsilon: action random.randint(0, env.action_dim - 1) else: with torch.no_grad(): q_values online_net(torch.tensor(state, dtypetorch.float32)) action int(q_values.argmax().item()) next_state, reward, done env.step(action) if next_state is not None: buffer.append((state, action, reward, next_state, done)) else: buffer.append((state, action, reward, state, done)) total_reward reward state next_state if next_state is not None else state if len(buffer) batch_size: batch random.sample(buffer, batch_size) states torch.tensor([b[0] for b in batch], dtypetorch.float32) actions torch.tensor([b[1] for b in batch], dtypetorch.long) rewards torch.tensor([b[2] for b in batch], dtypetorch.float32) next_states torch.tensor([b[3] for b in batch], dtypetorch.float32) dones torch.tensor([b[4] for b in batch], dtypetorch.float32) q_values online_net(states).gather(1, actions.unsqueeze(1)).squeeze(1) with torch.no_grad(): next_q target_net(next_states).max(1)[0] target rewards gamma * next_q * (1 - dones) loss nn.MSELoss()(q_values, target) optimizer.zero_grad() loss.backward() optimizer.step() if episode * 50 % target_update_steps 0: target_net.load_state_dict(online_net.state_dict()) epsilon max(epsilon_min, epsilon * epsilon_decay) torch.save(online_net.state_dict(), dqn_feature_selector.pth)逻辑说明衰减系数0.995意味着300个episode后epsilon约为0.05前期探索充分后期利用充分。目标网络同步条件写了episode * 50 % target_update_steps 0这里的50是每个episode的平均步数估算值保险起见应该在每步累加一个step_counter每满200步同步一次实操时把target_update_steps用全局步数判断更稳健。参数说明经验回放池50000条deque(maxlen50000)超出后自动丢弃旧样本batch_size64对Q网络来说够用。终止状态时next_state为None这里用当前state填充并让done1后续目标值计算中(1-dones)会把它清零。5. 训练结束后的模型评估与检测脚本落地训练完的dqn_feature_selector.pth只是“特征选择策略”不是检测模型。这一章讲怎么把它变成真正能上线用的恶意流量检测模型。5.1 从训练日志里找出最优特征子集训练过程中每个episode终止时都会得到一个最终mask和对应F1。把这两者记录到日志里训练结束后直接取F1最高的那条记录import pandas as pd log pd.read_csv(training_log.csv) # 每行: episode, mask, f1, feature_count best_row log.loc[log[f1].idxmax()] best_mask np.array([int(x) for x in best_row[mask].split(,)]) selected_idx np.where(best_mask 1)[0] print(best f1:, best_row[f1], selected features:, selected_idx)注意不要取最后几个episode的结果而是取历史最优。epsilon衰减后agent可能收敛到局部最优历史最优能避开这种波动。如果发现最优mask特征数超过20个说明奖励里的特征成本系数设得太小下次训练把cost_per_feature从0.1调到0.3。5.2 用更强分类器复测并保存上线模型搜索期用逻辑回归确定特征后进行线上测试时换成随机森林。逻辑回归在特征选择时给出的排序结论通常能迁移到树模型但F1绝对值会不同必须重新测。from sklearn.ensemble import RandomForestClassifier import joblib X_selected X_scaled[:, selected_idx] clf RandomForestClassifier(n_estimators200, max_depth12, n_jobs-1, random_state42) clf.fit(X_selected, y_small) joblib.dump(clf, production_detector.joblib) joblib.dump(selected_idx, selected_features.joblib) joblib.dump(scaler, feature_scaler.joblib)参数说明max_depth12限制树深度恶意流量检测场景下特征数通常不超过50深度太深容易过拟合到训练集的流量模式泛化到新攻击类型时反而掉F1。random_state42必须固定否则模型评估结果每次不同后续排查问题时分不清是代码bug还是随机性。5.3 一条命令完成新流量预测上线的检测脚本读入原始CSV先过scaler再选特征最后走随机森林import numpy as np import pandas as pd import joblib def predict_csv(input_path, output_path): scaler joblib.load(feature_scaler.joblib) selected_idx joblib.load(selected_features.joblib) clf joblib.load(production_detector.joblib) df pd.read_csv(input_path, low_memoryFalse) X df.select_dtypes(include[np.number]).fillna(0) X_scaled scaler.transform(X) y_pred clf.predict(X_scaled[:, selected_idx]) proba clf.predict_proba(X_scaled[:, selected_idx])[:, 1] df[prediction] y_pred df[malicious_prob] proba df.to_csv(output_path, indexFalse) if __name__ __main__: predict_csv(new_traffic.csv, predicted.csv)这里列名必须和训练时的特征列完全一致。常见做法是在训练时把feature_names.npy也存一份预测前做一次列对齐检查如果新数据少了某个特征scaler.transform会直接报错这个错误比静默预测更值得欢迎。6. 边界与调试技巧奖励放缩、可复现性与特征冗余最后给出几个实战中必须注意的技巧也是排查问题时的优先检查项。第一个技巧是奖励放缩。逻辑回归在恶意流量数据上F1通常在0.85到0.95之间差值只有0.1直接作为奖励会被Q值的估计噪声淹没。乘100变成85到95分的区间后TD误差更容易驱动网络参数更新。如果训练日志里平均奖励长期不涨先检查单步增量奖励是否大多数是负值如果是把成本系数降到0.05或直接去掉只保留终止奖励。第二个技巧是可复现性。PyTorch的CUDA卷积在同一个seed下仍可能产生微小差异但这里Q网络是全连接层CPU训练即可。固定三个seed即可复现完整实验env里的np.random.seed(42)PyTorch的torch.manual_seed(42)以及训练循环里的random.seed(42)。不要GPU训练这个任务全连接层小网络在CPU上300个episode通常不超过1小时GPU反而引入非确定性。第三个技巧是特征冗余检查。DQN选出的特征子集里可能同时存在高度相关的两个特征比如平均包大小和最大包大小。增量奖励机制会让其中一个特征的正向收益被另一个吸收最终导致选出的特征数偏多。训练结束后对选中的特征做一次相关性扫描相关系数超过0.9的只保留一个重新训练分类器通常F1不降且模型体积更小。最后一个值得记住的规律如果DQN搜索出的最优特征子集在上线后的新流量上掉点严重大概率不是特征选择问题而是训练数据分布与线上分布不一致。此时应该回到数据准备阶段检查采样策略而不是调DQN的超参数强化学习优化的是给定数据分布下的特征组合它无法修正上游数据代表性不足的问题。压测时用另一份时间窗口的流量数据做时间维度的前向验证比随机划分更能暴露这类问题。本文还有配套的精品资源点击获取
DQN强化学习特征选择：恶意流量检测模型自动调优实战
简介面向计算机专业课程设计、期末大作业与项目实战练习的恶意流量检测完整工程采用深度Q网络强化学习驱动机器学习建模解决恶意流量识别与分类问题。资源包含环境交互、智能体训练、检测验证等核心模块便于学习者理解强化学习与安全检测的结合方式。压缩包内共18个文件包括7个Python源码模块、8个CSV数据集、1个模型文件以及使用说明与依赖清单整体大小7.04MB结构清晰解压即可直接运行调试。当前已有252人学习使用。对于需要快速搭建恶意流量检测实验或深入理解深度强化学习在安全领域应用的学生和开发者可借助这套源码、数据集与文档组合完成课程报告、项目答辩并为进一步改进检测模型打下基础。1. 用DQN自动搜索恶意流量特征子集把调模型变成一场强化学习游戏一个常被忽略的事实流量检测模型的效果上限更多由特征决定而不是分类器。但拿到CICIDS2017这类数据集时几十个流量特征先筛哪几个、保留哪几个很多人靠correlation矩阵和直觉拍板。基于DQN强化学习生成恶意流量检测模型的思路是把特征选择重新建模成序贯决策agent在每一步决定要不要往特征集里加入某个特征下游分类器给出F1作为奖励最终用一个最优特征子集训练出可上线的检测模型。这条路径把调参过程变成马尔可夫决策过程适合特征维度高、正负样本不均衡、想自动化特征工程的场景。会用Python和PyTorch就能复现不需要先读懂深度强化学习的全部理论。2. 把恶意流量检测中的特征选择写成MDPDQN为什么比网格搜索划算先厘清一个关键认知这里的DQN是优化器不是检测器。它不直接读原始流量抓恶意样本它的工作是找到“哪些特征喂给随机森林或逻辑回归效果最好”。这也是标题里“生成模型”的含义——用强化学习自动生成一个可用的机器学习检测模型配置。这一步想通了后面所有代码都不容易跑偏。2.1 状态、动作、奖励的三元组如何映射流量特征工程常见做法是把特征选择写成如下MDP四元组这也是开源项目里大多数强化学习特征选择方案的底层设计状态s_t当前选中的特征子集掩码向量mask ∈ {0,1}^nn是原始特征总数再加上上一轮下游分类器在验证集上的F1值。把标量F1也拼进状态agent能感知“当前配置已经有多好”而不是盲目试探。动作a_t一共n1个离散动作。前n个动作分别对应“把第i个特征加入已选集合”最后一个动作是“停止选择输出最终子集”。这里不用连续动作因为特征选择天然是离散组合优化。奖励r_t执行动作后用当前特征子集训练一个轻量分类器在固定验证集上算F1增量。增量奖励比绝对F1更稳定能避免agent停在某个局部高水位。为什么增量奖励更稳因为绝对F1在特征数少时通常偏低agent会学到“多选特征总是好”的惰性策略而增量奖励配合下面的特征成本惩罚能让它在“多加一个特征带来的提升”和“多维护一个特征的代价”之间做真正的权衡。2.2 计算代价怎么控制奖励函数里的折扣与特征成本百万级流量数据上在每个episode里跑几十次完整交叉验证非常昂贵训练一轮可能就要数小时。两个手段是这类项目的标配第一个手段是评估用小批量数据加逻辑回归。每步奖励计算用采样后的2万条样本、2000次迭代的逻辑回归跑一次只要几百毫秒最终选定特征后再用全量随机森林或LightGBM复测。逻辑回归只做搜索期打分器不做上线模型这是刻意为之。第二个手段是给特征数量加惩罚项。奖励函数写成r_t (F1_now - F1_prev) * 100 - cost_per_feature终止时额外给一个终止奖励r_terminal F1_final * 100 - 0.5 * n_selected_features参数说明F1乘100是为了把0.7到0.9这个区间的差异放大到20分避免奖励值过小导致Q值无法区分动作好坏cost_per_feature取0.1对应“一个特征至少要带来0.001的F1提升才值得保留”。终止奖励里减去0.5乘特征数是让agent在搜索后期有动力收手。2.3 为什么DQN而不是随机搜索或遗传算法随机搜索每次从零开始完全浪费了“上一轮已经知道某些组合很差”的信息。遗传算法有记忆但交叉和变异算子作用在二进制掩码上会破坏特征的组合语义且评估每个个体同样要训练分类器总开销并不低。DQN核心优势是经验回放和目标网络历史转移样本被反复利用Q值能跨episode积累同样的评估次数下能学到“哪些特征组合持续带来正奖励”的长期规律。下表是三种方案在同等评估预算下的差异方案是否复用历史评估对离散掩码的适配调参难度随机搜索否天然适配低遗传算法部分种群内需设计算子中DQN是经验回放动作离散化即可中高调参难度是DQN的短板但换来的是搜索效率上限更高。实际项目中如果你的特征数在30个以下随机搜索加早停可能够用超过50个特征DQN的优势才会显现。3. 数据准备从原始CSV到可喂给DQN的特征掩码很多强化学习项目最终跑不出结果问题不在网络结构而在数据预处理阶段就埋了雷。这一章讲实际操作路径。3.1 用公开的二分类数据集起步恶意流量领域公开数据集常见的是CICIDS2017、UNSW-NB15和ISCX VPN。CICIDS2017的CSV按天分多个文件需要先合并标签列会同时出现良性、暴力破解、DDoS、端口扫描等多个值做特征选择实验时建议先统一成二分类Benign为0其余攻击类别全部为1。反直觉的一点是不要在最开始就做多分类任务。特征选择阶段的奖励来自F1类别一多F1对特征组合的区分度会下降二分类更容易观察到“某个特征是否有用”的因果效应。UNSW-NB15特征语义差异较大但文件更规整适合用来做验证集验证选出的特征是否跨数据集泛化。如果你本地没有这两个数据集用sklearn的fetch_openml拉一份网络入侵相关数据也可以关键是数据里必须包含足够多的原始特征列才能体现特征选择的必要。3.2 预处理流水线抽样、归一化与掩码生成数据清洗步骤顺序很重要我的固定顺序是去空值、去全零列、类型转换、样本均衡抽样、划分训练验证集、归一化。其中去全零列经常被漏掉流量数据集里某些端口相关特征在特定抓包阶段全为0留着只会增加动作空间维度。import pandas as pd import numpy as np from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler raw pd.read_csv(cicids2017_merged.csv, low_memoryFalse) # 1. 删掉全空或全零列 raw raw.dropna(axis1, howall) raw raw.loc[:, (raw ! 0).any(axis0)] # 2. 标签二值化良性0其余攻击1 y (raw[Label] Benign).astype(int).values # 3. 去掉标签列和ID列 X raw.drop(columns[Label, Flow ID, Timestamp]).select_dtypes(include[np.number]) # 4. 分层抽样到10万条保持正负比例 X_small, _, y_small, _ train_test_split( X, y, train_size100000, stratifyy, random_state42 ) # 5. 归一化并保留scaler供后续测试集使用 scaler StandardScaler().fit(X_small) X_scaled scaler.transform(X_small) np.save(X_scaled.npy, X_scaled) np.save(y_scaled.npy, y_small) np.save(feature_names.npy, np.array(X.columns))逻辑说明抽样到10万条不是为了省内存而是为了控制奖励计算的时间。逻辑回归在两万条子样本上加1000次迭代收敛单次评估稳定在0.2秒以内一个episode跑60步评估也只要12秒。stratifyy保证了抽样后恶意样本比例与原始数据一致避免奖励信号被类别分布波动污染。这一步如果只写train_test_split(X, y, train_size...)不指定stratify在攻击样本占比较低的数据集上很容易抽出一个类别比例失真的子集导致搜索期F1虚高。3.3 归一化与掩码状态之间的顺序关系这里有个容易踩的坑先归一化再选特征还是先选特征再归一化答案是训练集上先归一化再选特征。因为逻辑回归的收敛速度依赖特征尺度一致如果每次动作改变特征子集后重新做归一化状态空间里同一个掩码对应的数据分布会漂移DQN永远学不到稳定规律。正确做法如上节代码所示scaler在全部特征上fit之后无论选中哪几个特征都直接从X_scaled取对应列。代价是标准化时混入了最终被丢弃的特征统计量但在特征选择MDP里影响很小换来的是状态转移的一致性。验证集和测试集只用scaler.transform绝不在数据划分前对整个数据集做fit这是防止数据泄漏的最低要求。4. 用PyTorch实现FeatureSelectionEnv和DQN训练循环这一章给出核心可执行代码。整体结构分三块环境类、Q网络、训练循环三者可以分别独立调试。4.1 环境类掩码状态、增量奖励、终止条件import numpy as np from sklearn.linear_model import LogisticRegression from sklearn.metrics import f1_score class FeatureSelectionEnv: def __init__(self, X, y, max_steps30, cost_per_feature0.1, eval_size20000): self.X X self.y y self.max_steps max_steps self.cost_per_feature cost_per_feature self.eval_size eval_size self.n_features X.shape[1] self.action_dim self.n_features 1 # 最后一个动作是停止 def _evaluate_f1(self, mask): idx np.where(mask 1)[0] if len(idx) 0: return 0.0 # 每次评估从训练数据里采样保证不同step的奖励可比 sample_idx np.random.choice(self.X.shape[0], self.eval_size, replaceFalse) X_sub self.X[sample_idx][:, idx] y_sub self.y[sample_idx] clf LogisticRegression(max_iter500, n_jobs-1) clf.fit(X_sub, y_sub) return f1_score(y_sub, clf.predict(X_sub)) def reset(self): self.mask np.zeros(self.n_features, dtypenp.int8) self.steps 0 self.prev_f1 0.0 return np.concatenate([self.mask, [self.prev_f1]]).astype(np.float32) def step(self, action): if action self.n_features or self.steps self.max_steps: # 终止动作用全量训练集算最终F1 idx np.where(self.mask 1)[0] if len(idx) 0: reward, done -1.0, True return None, reward, done final_f1 self._evaluate_f1(self.mask) reward final_f1 * 100 - 0.5 * len(idx) return None, reward, True self.mask[action] 1 self.steps 1 cur_f1 self._evaluate_f1(self.mask) reward (cur_f1 - self.prev_f1) * 100 - self.cost_per_feature self.prev_f1 cur_f1 next_state np.concatenate([self.mask, [cur_f1]]).astype(np.float32) return next_state, reward, False逻辑说明reset返回的状态是特征掩码拼接上一轮F1掩码本身是离散0/1但拼上F1后整体作为连续向量输入网络。step里前n个动作只把对应位置置1不提供移除动作这是刻意的——恶意流量特征选择任务里一个特征选了再移除会让状态空间膨胀一倍而增量奖励机制已经能抑制无效特征加入。终止动作返回的最终奖励使用全量训练集评估避免搜索期采样子集带来的噪声影响最终决策。参数说明max_steps30限制单个episode最多选30个特征防止mask全满原始数据集如果80个特征动作空间就是81维。eval_size20000控制奖励计算速度机器配置低可以降到10000但F1波动会增大。4.2 Q网络结构与目标网络import torch import torch.nn as nn import torch.optim as optim class QNet(nn.Module): def __init__(self, state_dim, action_dim, hidden_dim128): super().__init__() self.net nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, action_dim) ) def forward(self, x): return self.net(x) state_dim env.n_features 1 online_net QNet(state_dim, env.action_dim) target_net QNet(state_dim, env.action_dim) target_net.load_state_dict(online_net.state_dict()) optimizer optim.Adam(online_net.parameters(), lr1e-3)结构说明这个Q网络只有3层MLP没有用RNN。原因是状态里已经带了完整的特征掩码向量历史选择信息都被压缩在当前mask里MDP的状态表示满足马尔可夫性不需要额外记忆。hidden_dim128对80维特征输入足够特征超过200个时可以加到256。参数说明lr1e-3是Adam在中小型Q网络上的安全起点目标网络每200步同步一次在线网络权重等待一会儿给出训练循环里的具体写法。4.3 训练主循环与epsilon贪心衰减from collections import deque import random buffer deque(maxlen50000) epsilon 1.0 epsilon_min 0.05 epsilon_decay 0.995 gamma 0.99 batch_size 64 target_update_steps 200 episodes 300 for episode in range(episodes): state env.reset() total_reward 0 done False while not done: if random.random() epsilon: action random.randint(0, env.action_dim - 1) else: with torch.no_grad(): q_values online_net(torch.tensor(state, dtypetorch.float32)) action int(q_values.argmax().item()) next_state, reward, done env.step(action) if next_state is not None: buffer.append((state, action, reward, next_state, done)) else: buffer.append((state, action, reward, state, done)) total_reward reward state next_state if next_state is not None else state if len(buffer) batch_size: batch random.sample(buffer, batch_size) states torch.tensor([b[0] for b in batch], dtypetorch.float32) actions torch.tensor([b[1] for b in batch], dtypetorch.long) rewards torch.tensor([b[2] for b in batch], dtypetorch.float32) next_states torch.tensor([b[3] for b in batch], dtypetorch.float32) dones torch.tensor([b[4] for b in batch], dtypetorch.float32) q_values online_net(states).gather(1, actions.unsqueeze(1)).squeeze(1) with torch.no_grad(): next_q target_net(next_states).max(1)[0] target rewards gamma * next_q * (1 - dones) loss nn.MSELoss()(q_values, target) optimizer.zero_grad() loss.backward() optimizer.step() if episode * 50 % target_update_steps 0: target_net.load_state_dict(online_net.state_dict()) epsilon max(epsilon_min, epsilon * epsilon_decay) torch.save(online_net.state_dict(), dqn_feature_selector.pth)逻辑说明衰减系数0.995意味着300个episode后epsilon约为0.05前期探索充分后期利用充分。目标网络同步条件写了episode * 50 % target_update_steps 0这里的50是每个episode的平均步数估算值保险起见应该在每步累加一个step_counter每满200步同步一次实操时把target_update_steps用全局步数判断更稳健。参数说明经验回放池50000条deque(maxlen50000)超出后自动丢弃旧样本batch_size64对Q网络来说够用。终止状态时next_state为None这里用当前state填充并让done1后续目标值计算中(1-dones)会把它清零。5. 训练结束后的模型评估与检测脚本落地训练完的dqn_feature_selector.pth只是“特征选择策略”不是检测模型。这一章讲怎么把它变成真正能上线用的恶意流量检测模型。5.1 从训练日志里找出最优特征子集训练过程中每个episode终止时都会得到一个最终mask和对应F1。把这两者记录到日志里训练结束后直接取F1最高的那条记录import pandas as pd log pd.read_csv(training_log.csv) # 每行: episode, mask, f1, feature_count best_row log.loc[log[f1].idxmax()] best_mask np.array([int(x) for x in best_row[mask].split(,)]) selected_idx np.where(best_mask 1)[0] print(best f1:, best_row[f1], selected features:, selected_idx)注意不要取最后几个episode的结果而是取历史最优。epsilon衰减后agent可能收敛到局部最优历史最优能避开这种波动。如果发现最优mask特征数超过20个说明奖励里的特征成本系数设得太小下次训练把cost_per_feature从0.1调到0.3。5.2 用更强分类器复测并保存上线模型搜索期用逻辑回归确定特征后进行线上测试时换成随机森林。逻辑回归在特征选择时给出的排序结论通常能迁移到树模型但F1绝对值会不同必须重新测。from sklearn.ensemble import RandomForestClassifier import joblib X_selected X_scaled[:, selected_idx] clf RandomForestClassifier(n_estimators200, max_depth12, n_jobs-1, random_state42) clf.fit(X_selected, y_small) joblib.dump(clf, production_detector.joblib) joblib.dump(selected_idx, selected_features.joblib) joblib.dump(scaler, feature_scaler.joblib)参数说明max_depth12限制树深度恶意流量检测场景下特征数通常不超过50深度太深容易过拟合到训练集的流量模式泛化到新攻击类型时反而掉F1。random_state42必须固定否则模型评估结果每次不同后续排查问题时分不清是代码bug还是随机性。5.3 一条命令完成新流量预测上线的检测脚本读入原始CSV先过scaler再选特征最后走随机森林import numpy as np import pandas as pd import joblib def predict_csv(input_path, output_path): scaler joblib.load(feature_scaler.joblib) selected_idx joblib.load(selected_features.joblib) clf joblib.load(production_detector.joblib) df pd.read_csv(input_path, low_memoryFalse) X df.select_dtypes(include[np.number]).fillna(0) X_scaled scaler.transform(X) y_pred clf.predict(X_scaled[:, selected_idx]) proba clf.predict_proba(X_scaled[:, selected_idx])[:, 1] df[prediction] y_pred df[malicious_prob] proba df.to_csv(output_path, indexFalse) if __name__ __main__: predict_csv(new_traffic.csv, predicted.csv)这里列名必须和训练时的特征列完全一致。常见做法是在训练时把feature_names.npy也存一份预测前做一次列对齐检查如果新数据少了某个特征scaler.transform会直接报错这个错误比静默预测更值得欢迎。6. 边界与调试技巧奖励放缩、可复现性与特征冗余最后给出几个实战中必须注意的技巧也是排查问题时的优先检查项。第一个技巧是奖励放缩。逻辑回归在恶意流量数据上F1通常在0.85到0.95之间差值只有0.1直接作为奖励会被Q值的估计噪声淹没。乘100变成85到95分的区间后TD误差更容易驱动网络参数更新。如果训练日志里平均奖励长期不涨先检查单步增量奖励是否大多数是负值如果是把成本系数降到0.05或直接去掉只保留终止奖励。第二个技巧是可复现性。PyTorch的CUDA卷积在同一个seed下仍可能产生微小差异但这里Q网络是全连接层CPU训练即可。固定三个seed即可复现完整实验env里的np.random.seed(42)PyTorch的torch.manual_seed(42)以及训练循环里的random.seed(42)。不要GPU训练这个任务全连接层小网络在CPU上300个episode通常不超过1小时GPU反而引入非确定性。第三个技巧是特征冗余检查。DQN选出的特征子集里可能同时存在高度相关的两个特征比如平均包大小和最大包大小。增量奖励机制会让其中一个特征的正向收益被另一个吸收最终导致选出的特征数偏多。训练结束后对选中的特征做一次相关性扫描相关系数超过0.9的只保留一个重新训练分类器通常F1不降且模型体积更小。最后一个值得记住的规律如果DQN搜索出的最优特征子集在上线后的新流量上掉点严重大概率不是特征选择问题而是训练数据分布与线上分布不一致。此时应该回到数据准备阶段检查采样策略而不是调DQN的超参数强化学习优化的是给定数据分布下的特征组合它无法修正上游数据代表性不足的问题。压测时用另一份时间窗口的流量数据做时间维度的前向验证比随机划分更能暴露这类问题。本文还有配套的精品资源点击获取