简介针对城市交通拥堵预测的Python实践项目包面向计算机相关专业学生、毕业设计开发者及机器学习初学者用于解决道路时间段内车辆流量与拥堵状态预测问题。项目基于GCM走廊真实主干道传感器数据每5分钟记录一次交通流特征速度、流量、占有率等并标注拥堵等级训练集提供4天数据测试集评估未来30分钟内所有传感器的拥堵状态。包内涵盖完整项目说明文档、训练集与测试集数据、多份Python脚本数据过滤、预处理、模型训练可支撑课程设计、毕业设计、算法对比实验。资源共19个文件包括6个py脚本、6个zip数据包、5个txt说明文件及2个csv数据集总大小32KB轻量易部署。已有1190人浏览学习适合在现有代码基础上修改扩展快速搭建从数据处理到结果输出的完整预测流程项目说明对数据集格式、预测任务和评估方式做了清晰阐述输出结果按传感器ID拥堵等级序列组织便于验证和汇报。1. 交通拥堵预测基于Python的车辆流量预测项目到底做了什么交通拥堵预测最常见的一个误区是以为拿到车流量数据就能直接训练模型。实际上这份基于Python的车辆流量预测项目处理的是一套完整的传感器时间序列——GCM走廊16座城镇之间部署了855个传感器每5分钟回传一条交通状态记录一天就是288条每条含日期、时间、方向、linkID、通行时间、流量、速度、占有率、拥堵等级等11个字段。任务是用前4天数据训练模型预测第5天接下来30分钟内所有传感器的拥堵状态输出格式是传感器ID加6个数字0通畅、1轻微、2中度、3重度。它适合做毕设、课程设计也适合想搞懂时间序列怎么落地成分类预测的在职人员。2. 数据预处理从原始交通流到可训练的样本集2.1 原始数据字段拆解11列各自代表什么拿到训练集文件后第一件事不是跑模型而是逐字段确认含义。原始数据是逗号分隔的文本每条记录对应一次传感器上报字段顺序固定字段样例值含义date707自增ID或日期序号time0000时刻格式HHMM0000到2355directionNORTH_BOUND方向南/北/东/西typeFREEWAY道路类型linkIDWI-MNT_XML_V001-21012传感器ID也就是最终输出要用的IDlength1268路段长度英尺travelTime40通过该路段的通行时间秒volumn218车流量辆speed31.292915平均速度occupancy2.4道路占有率百分比congestionLevelNON_CONGESTION拥堵等级样例“707,0000,NORTH_BOUND,FREEWAY,WI-MNT_XML_V001-21012,1268,40,218,31.292915,2.4,NON_CONGESTION”里最需要关注的是linkID、travelTime、volumn、speed、occupancy这五列。travelTime比volumn更能直接反映拥堵堵车时通行时间会明显拉长occupancy表示传感器覆盖路段被车辆占用的比例数值越高越接近拥堵。这三个字段配合历史拥堵等级是预测未来状态的核心信号。拥堵等级原始值是英文枚举建模前必须映射成0到3的数字NON_CONGESTION对应0LIGHT_CONGESTION对应1MEDIUM_CONGESTION对应2HEAVY_CONGESTION对应3。这个映射要在预处理脚本里一次性完成避免训练和测试时映射不一致。常见做法是写一个字典统一处理而不是在多个脚本里各写一份。2.2 filtrate_sensor.py 过滤逻辑为什么按天补齐288条855个传感器不是每个都有完整4天数据。有些传感器在某些时段掉线或者某天完全没上报。如果直接拿这些残缺序列训练滑动窗口构造样本时前后时间点会对不齐模型会学到虚假的时间关联。我一般先跑过滤脚本把“每天记录数不足288条”的传感器剔除保留4天数据都完整的子集。import pandas as pd # 读取原始训练集字段顺序按数据说明固定 df pd.read_csv(traffic_train.txt, headerNone, names[date,time,direction,type,linkID, length,travelTime,volumn,speed, occupancy,congestionLevel]) # 按传感器日期统计每天记录数 counts df.groupby([linkID, date]).size().reset_index(namecnt) # 保留每天都有完整288条记录的传感器 valid counts[counts[cnt] 288][linkID].unique() df_filtered df[df[linkID].isin(valid)].copy() print(f原始传感器数量: {df[linkID].nunique()}) print(f过滤后传感器数量: {df_filtered[linkID].nunique()}) df_filtered.to_csv(traffic_filtered.csv, indexFalse)这段脚本的关键判断条件是cnt 288。因为数据每5分钟一条24小时正好288条任何少于288条的情况都说明这个传感器当天存在缺失。过滤后还要打印原始数量和过滤后数量方便判断剔除比例是否合理。如果一下剔掉200多个传感器说明原始数据质量较差那就需要考虑用插值而不是直接剔除。过滤这个环节容易被新手跳过觉得缺失几行不影响大局。实际上交通数据的时间序列特性决定了缺失点会造成样本构造错位第n个窗口和第n1个窗口之间出现时间跳跃模型会误以为“凌晨2点和凌晨3点的数据是连续相邻的”。这种做法在最终预测时不会报错但结果会很离谱。2.3 process.py 滑动窗口构造样本30分钟预测窗口怎么对齐过滤完传感器后需要把时间序列转成监督学习的样本。这里的关键设计是用过去N个时间步的特征预测未来M个时间步的拥堵等级。题目要求预测未来30分钟拥堵状态也就是6个时间点每5分钟一个所以horizon设为6。过去窗口我一般取6或12取6代表用过去30分钟预测未来30分钟取12代表用过去1小时预测未来30分钟。在数据量只有4天的情况下窗口取6比较均衡样本量损失小特征也够用。import numpy as np import pandas as pd df pd.read_csv(traffic_filtered.csv) # 拼接日期和时间转成datetime保证排序正确 df[datetime] pd.to_datetime( df[date].astype(str) df[time].astype(str).str.zfill(4), format%Y%m%d%H%M ) # 拥堵等级映射为0-3 CONG_MAP { NON_CONGESTION: 0, LIGHT_CONGESTION: 1, MEDIUM_CONGESTION: 2, HEAVY_CONGESTION: 3 } df[label] df[congestionLevel].map(CONG_MAP) def build_samples(sensor_df, window6, horizon6): # 按时间排序只保留建模需要的特征列 sensor_df sensor_df.sort_values(datetime) features sensor_df[[volumn, travelTime, speed, occupancy, label]].values X, y [], [] for i in range(len(features) - window - horizon 1): x features[i:i window] y_seq features[i window:i window horizon, 4] X.append(x.reshape(-1)) y.append(y_seq) return np.array(X), np.array(y) X_all, y_all [], [] for linkID, g in df.groupby(linkID): X, y build_samples(g, window6, horizon6) X_all.append(X) y_all.append(y) X np.concatenate(X_all) y np.concatenate(y_all) print(f样本总量: {X.shape[0]}, 特征维度: {X.shape[1]}) np.savez(train_samples.npz, XX, yy)这里的核心逻辑是滑动窗口。window6表示每个样本包含6个连续时间步每个时间步有5个特征volumn、travelTime、speed、occupancy、label所以每个样本的特征维度是6×530。horizon6表示要预测未来6个时间步的label值y的shape是(样本数, 6)每个元素都是0到3的整数。需要注意特征列里把历史label也放进去是有意的。拥堵状态有很强的自相关性过去30分钟重度拥堵未来大概率不会立刻恢复通畅。把历史label作为特征等于直接告诉模型当前拥堵状态这对短期预测非常有效。但这里有个边界只能放当前窗口内的历史label不能把未来label放进来否则就是数据泄漏。构造完样本后可以打印一下样本总量。4天数据、每天288条、几百个传感器过滤后样本量通常在几十万量级这个数量级跑随机森林完全没问题。3. 模型训练与评估树模型在时序预测上的落地3.1 为什么先跑树模型而不是一上来就LSTM很多人看到时间序列就条件反射地想到LSTM但在这种数据条件下我建议先用树模型打基线。原因很直接第一过滤后每个传感器虽然有上千条记录但独立时间序列长度只有4天对LSTM来说序列太短难以学到长程依赖第二树模型不需要归一化volumn是几百、occupancy是几点几、travelTime是几十不同量纲直接喂给随机森林完全没问题省掉一大截预处理第三树模型天然支持多输出。数据量方面也要说清楚。过滤后如果保留600个传感器每个传感器构造出的样本约288×4-6-61≈1141个总样本量接近68万个。这个规模下随机森林训练几分钟能完成而LSTM即使有GPU调参和训练的时间成本也高出几倍最终准确率未必比树模型好。做毕设或课程设计时先把基线跑通远比直接追求深度学习模型更稳妥。3.2 train.py 训练流程窗口、多输出、切分策略训练脚本的核心是把6个时间步的预测作为一个多输出分类问题处理。sklearn的MultiOutputClassifier会对每个输出单独训练一个基分类器互不干扰。这意味着第1个时间步的预测模型和第6个时间步的预测模型是独立的前者依赖的规律更偏近期趋势后者更依赖长时间窗口的惯性。import numpy as np from sklearn.ensemble import RandomForestClassifier from sklearn.multioutput import MultiOutputClassifier from sklearn.metrics import accuracy_score, f1_score data np.load(train_samples.npz) X, y data[X], data[y] # 按时间切分前3天训练第4天验证 # 不能随机打乱必须保持时间顺序 split int(len(X) * 0.75) X_train, y_train X[:split], y[:split] X_val, y_val X[split:], y[split:] model MultiOutputClassifier( RandomForestClassifier( n_estimators200, max_depth15, min_samples_leaf5, n_jobs-1, random_state42 ) ) model.fit(X_train, y_train) # 验证集评估 y_pred model.predict(X_val) for step in range(6): acc accuracy_score(y_val[:, step], y_pred[:, step]) f1 f1_score(y_val[:, step], y_pred[:, step], averageweighted) print(f第{step1}个时间步 ACC{acc:.4f} F1{f1:.4f}) # 保存模型 import joblib joblib.dump(model, traffic_model.pkl)RandomForestClassifier的关键参数里n_estimators200是树的数量一般200到500之间准确率提升开始放缓max_depth15限制树的深度防止过拟合因为特征维度只有30深度15已经足够min_samples_leaf5保证每个叶子节点至少5个样本避免模型记住个别异常样本n_jobs-1让所有CPU核并行训练这一步能省大量时间。切分策略这里要放重点我按样本顺序前75%训练、后25%验证因为原始样本是按时间顺序构造的每个传感器内按时间推进所以这个切分等价于前3天训练、第4天验证。绝不能用train_test_split的默认shuffleTrue那会导致训练集里混入第4天数据验证集里出现第1天数据相当于用未来预测过去指标虚高。3.3 输出格式还原让预测结果匹配提交规格模型预测出的y_pred是(样本数, 6)的数组每个值0到3。但提交要求是“传感器ID:0,1,2,3,3,2”这种逐传感器输出的格式。这里需要把样本对应回传感器ID和时间点。我在样本构造时记录过每个样本对应的linkID和起始datetime这里再取一次拼成提交格式。import pandas as pd # 清洗测试集构造样本 df_test pd.read_csv(test.csv) df_test[datetime] pd.to_datetime( df_test[date].astype(str) df_test[time].astype(str).str.zfill(4), format%Y%m%d%H%M ) df_test[label] df_test[congestionLevel].map(CONG_MAP) # 取测试集最后6个样本的起始时间作为预测点 # 每个传感器取最后一个完整窗口的位置 submissions [] for linkID, g in df_test.groupby(linkID): g g.sort_values(datetime) features g[[volumn,travelTime,speed,occupancy,label]].values last_windows [] for i in range(len(features) - 6 - 6 1): last_windows.append(features[i:i6].reshape(-1)) X_test np.array(last_windows) pred model.predict(X_test) # 取最后一个窗口的预测结果作为未来30分钟预测 final_pred pred[-1] pred_str ,.join(str(int(x)) for x in final_pred) submissions.append(f{linkID}:{pred_str}) with open(submission.txt, w) as f: f.write(\n.join(submissions))这里要理解一个关键点模型是滚动预测的对于测试集的每一个时刻都可以用前面30分钟的数据预测后面30分钟。提交时如果要求预测“接下来30分钟内”的状态通常取测试集最后一个已知窗口之后的6个时间点。上面的代码取的是每个传感器最后一个可用窗口的预测结果这与题目要求一致。如果题目要求预测某一整天的多个时段那就需要按起始时间筛选把对应的窗口样本挑出来单独预测。我在实际做的时候会把起始时间也存下来方便后面按任意时刻切片。4. 避坑指南交通流量预测最容易翻车的五个环节4.1 现象模型准确率95%以上验证集F1却只有0.3第一次跑通时准确率漂亮得不像话但看F1就露馅了。原因是训练集中“通畅”状态占比超过70%模型只需要全部预测为0就能拿到高准确率但重度拥堵一个都预测不出来。解决不要只看accuracy重点关注每个类别的召回率。如果类别严重不平衡在RandomForestClassifier里设置class_weightbalanced_subsample或者对少量类别做上采样。实际项目里我还会把混淆矩阵打印出来逐类看预测情况。4.2 现象验证集指标比训练集还高出现这种现象百分之百是数据泄漏。常见原因有两个一是构造样本时把未来label混进了特征列二是切分训练集验证集时用了随机打乱同一传感器的相邻窗口被分到两边验证集里出现了训练集样本的“下半截”。解决先检查特征列是否包含窗口范围之外的数据再确认切分代码里没有shuffle。我一般会在构造样本前把数据按datetime严格排序并按“传感器分组、时间前进”的原则构造窗口切分时直接按顺序切。4.3 现象某些传感器预测结果全是0但实际是重拥堵这是过滤脚本没跑干净的典型症状。如果某个传感器数据断断续续滑动窗口样本的前后时间点不连续模型看到的特征序列是“凌晨、中午、傍晚”混在一起根本学不到拥堵的延续规律。解决过滤时不仅要求每天288条还要检查相邻两条记录的时间间隔是否为5分钟。间隔过大的传感器要么剔除要么用前后时刻的线性插值补上。插值时优先补volumn和occupancy这类连续值label字段不要插值用前向填充。4.4 现象训练集和测试集的拥堵等级编码不一致原始训练集里写的是NON_CONGESTION这种字符串但测试集文件里某些批次可能已经转成了0/1/2/3数字。如果预处理脚本直接调用map函数映射遇到已经编码过的数据就会得到NaN。解决写映射前先打印两个文件的取值分布。pd.unique()看一眼congestionLevel这列到底有哪些值。统一做法是写一个兼容函数检测字段是字符串还是数字分别处理def normalize_label(x): if isinstance(x, str): return CONG_MAP.get(x, 0) return int(x)4.5 现象提交格式不对导致评估报错题目要求“传感器ID:0,1,2,3,3,2”一条一行但实际输出时容易多出引号、空格或缺失冒号。尤其是用DataFrame的to_csv直接导出时字符串可能被加引号。解决手动拼字符串写入文件不用to_csv。写完先看文件头五行再统计行数和传感器数量是否一致。这个问题最蠢但最容易犯每次提交前我都强制检查一遍。5. 进阶验证从准确率到真正可用的拥堵预测模型树模型跑通后如果想往上提指标最先做的不是换深度学习而是检查验证流程是否真实反映预测能力。我现在的习惯是在最终训练前做一次“按天分组”的时序交叉验证4天数据每次拿3天训练、1天验证轮换4次取指标平均值。这与前面按75%/25%切分不同能看出模型在不同星期几、不同天气条件下的稳定性。验证指标上用加权F1而不是准确率。交通拥堵预测里“通畅”类占比太高准确率被整体拉高的同时中度和重度拥堵的预测能力可能在退化。我还会单独统计预测结果的类别分布如果预测集里重度拥堵占比明显低于训练集说明模型在收缩输出空间这时候可以调整分类阈值对每个时间步的输出拿到概率后按自定义阈值决策而不是直接取最大概率类。一个常用的进阶技巧是给不同预测时间步分配不同权重。第1个时间步离当前最近预测置信度最高第6个时间步最难模型可能退化成直接复用历史label。因此可以给6个时间步的损失或加权F1设不同系数例如[1.0, 0.9, 0.8, 0.7, 0.6, 0.5]引导模型在前几步尽力、后几步稳守。可视化验证也很重要。我会随机挑3个传感器把连续48小时的真实拥堵等级和预测结果画在同一张图上。肉眼能看出的规律是拥堵通常出现在早晚高峰如果模型在平峰时段频繁报中度拥堵说明它没学到时间分布特征这时考虑加入“时刻”作为特征把time字段转成分钟数或周期编码后放进样本。从那以后我每次拿到交通类时间序列数据第一件事不是建模而是先画出每个传感器的速度-时间曲线直观确认数据有没有断档、有没有明显噪声再做过滤和样本构造。这套流程走得多了翻车的概率就越来越低。希望帮到你。本文还有配套的精品资源点击获取