☰
基于Python的交通流量预测:随机森林模型与数据预处理实战
2026/10/12 16:52:56 网站建设 项目流程

简介:针对城市交通拥堵预测的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时刻,格式HHMM,0000到2355
directionNORTH_BOUND方向,南/北/东/西
typeFREEWAY道路类型
linkIDWI-MNT_XML_V001-21012传感器ID,也就是最终输出要用的ID
length1268路段长度(英尺)
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对应0,LIGHT_CONGESTION对应1,MEDIUM_CONGESTION对应2,HEAVY_CONGESTION对应3。这个映射要在预处理脚本里一次性完成,避免训练和测试时映射不一致。常见做法是写一个字典统一处理,而不是在多个脚本里各写一份。

2.2 filtrate_sensor.py 过滤逻辑:为什么按天补齐288条

855个传感器不是每个都有完整4天数据。有些传感器在某些时段掉线,或者某天完全没上报。如果直接拿这些残缺序列训练,滑动窗口构造样本时前后时间点会对不齐,模型会学到虚假的时间关联。我一般先跑过滤脚本,把“每天记录数不足288条”的传感器剔除,保留4天数据都完整的子集。

import pandas as pd # 读取原始训练集,字段顺序按数据说明固定 df = pd.read_csv('traffic_train.txt', header=None, names=['date','time','direction','type','linkID', 'length','travelTime','volumn','speed', 'occupancy','congestionLevel']) # 按传感器+日期统计每天记录数 counts = df.groupby(['linkID', 'date']).size().reset_index(name='cnt') # 保留每天都有完整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', index=False)

这段脚本的关键判断条件是cnt == 288。因为数据每5分钟一条,24小时正好288条,任何少于288条的情况都说明这个传感器当天存在缺失。过滤后还要打印原始数量和过滤后数量,方便判断剔除比例是否合理。如果一下剔掉200多个传感器,说明原始数据质量较差,那就需要考虑用插值而不是直接剔除。

过滤这个环节容易被新手跳过,觉得缺失几行不影响大局。实际上交通数据的时间序列特性决定了,缺失点会造成样本构造错位:第n个窗口和第n+1个窗口之间出现时间跳跃,模型会误以为“凌晨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, window=6, horizon=6): # 按时间排序,只保留建模需要的特征列 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, window=6, horizon=6) 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', X=X, y=y)

这里的核心逻辑是滑动窗口。window=6表示每个样本包含6个连续时间步,每个时间步有5个特征(volumn、travelTime、speed、occupancy、label),所以每个样本的特征维度是6×5=30。horizon=6表示要预测未来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-6+1≈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_estimators=200, max_depth=15, min_samples_leaf=5, n_jobs=-1, random_state=42 ) ) 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], average='weighted') print(f'第{step+1}个时间步 ACC={acc:.4f} F1={f1:.4f}') # 保存模型 import joblib joblib.dump(model, 'traffic_model.pkl')

RandomForestClassifier的关键参数里,n_estimators=200是树的数量,一般200到500之间准确率提升开始放缓;max_depth=15限制树的深度,防止过拟合,因为特征维度只有30,深度15已经足够;min_samples_leaf=5保证每个叶子节点至少5个样本,避免模型记住个别异常样本;n_jobs=-1让所有CPU核并行训练,这一步能省大量时间。

切分策略这里要放重点:我按样本顺序前75%训练、后25%验证,因为原始样本是按时间顺序构造的(每个传感器内按时间推进),所以这个切分等价于前3天训练、第4天验证。绝不能用train_test_split的默认shuffle=True,那会导致训练集里混入第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:i+6].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_weight='balanced_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字段转成分钟数或周期编码后放进样本。

从那以后,我每次拿到交通类时间序列数据,第一件事不是建模,而是先画出每个传感器的速度-时间曲线,直观确认数据有没有断档、有没有明显噪声,再做过滤和样本构造。这套流程走得多了,翻车的概率就越来越低。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询