简介一份基于Python的城市交通流量数据可视化分析系统完整项目实例面向具备Python基础的数据分析师、Web后端或GUI开发人员及交通专业高校师生。项目采用分层架构覆盖数据采集、清洗、存储、多维度统计与交互式可视化并演示基于Matplotlib和Plotly的图表及简单预测建模适合作为数据科学综合应用的教学案例或交通管理决策支持工具。资源包共1个文件为docx文档压缩包大小116KB内含系统架构、MySQL数据库设计、完整程序代码与代码详解目录涵盖项目背景、模型架构、关键代码示例及常见问题解决思路等。已有115人学习下载便于对照文档动手搭建从数据预处理到后端API与前端GUI的完整流程深入掌握模块化设计与前后端交互核心要点。1. 城市交通流量可视化系统难点不在画图而在数据进门很多接触过交通数据的人都有同感一条路口的车流量、速度、时间、路段编号看起来不过三五个字段真正拿去做分析时才发现全是坑——不同来源的数据格式不一样、时间戳有的按分钟有的按十五分钟、缺失值和异常值混在几百万行记录里。这套基于 Python 的城市交通流量数据可视化分析系统就是把数据进门这件事从头到尾做通了从模拟数据生成、pandas 清洗、MySQL 分层入库到聚合统计、Matplotlib 静态图、Plotly 交互图、FastAPI 接口再到 Tkinter 桌面 GUI全链路在一个工程里串起来。它适合两类人一类是想把交通数据落地成可视化分析工具的数据分析师或研发另一类是用一个完整系统案例来学 Python 数据科学和系统设计的学生。2. 数据入口用 pandas 清洗多源异构流量数据落到 MySQL 五张表2.1 模拟数据生成没有交管数据时先自造真实交通数据往往拿不到或者拿到的样本里混杂了各种设备采集问题。这个项目里带了数据生成脚本核心逻辑是模拟一台路侧检测器在一个监测点上按固定频率采集车流量同时按概率注入缺失值和异常值。生成结果是一份 CSV字段和实际业务表对齐方便后续走完整个管道。import pandas as pd import numpy as np from datetime import datetime, timedelta # 生成 30 天、每 5 分钟一条的模拟流量数据 np.random.seed(42) monitor_id M001 time_list [] volume_list [] start datetime(2024, 3, 1, 0, 0, 0) for i in range(30 * 24 * 12): # 30天 * 24小时 * 每5分钟12条 ts start timedelta(minutes5 * i) hour ts.hour # 早晚高峰抬升夜间压低 if 7 hour 9: base 120 elif 17 hour 19: base 110 elif 0 hour 5: base 10 else: base 60 volume int(base np.random.normal(0, 15)) volume max(0, volume) time_list.append(ts) volume_list.append(volume) df pd.DataFrame({monitor_id: monitor_id, record_time: time_list, volume: volume_list}) # 注入约 3% 缺失值置为 NaN missing_idx np.random.choice(df.index, sizeint(len(df) * 0.03), replaceFalse) df.loc[missing_idx, volume] np.nan # 注入约 1% 异常值放大 3~5 倍 anomaly_idx np.random.choice(df.index, sizeint(len(df) * 0.01), replaceFalse) df.loc[anomaly_idx, volume] df.loc[anomaly_idx, volume] * np.random.uniform(3, 5, len(anomaly_idx)) df.to_csv(raw_traffic_m001.csv, indexFalse)这段脚本的逻辑是按 5 分钟频率生成时间序列用不同时段的基础流量值模拟早晚高峰形态再用正态分布加噪声让数据看起来更接近真实。关键参数是start生成起始时间和5 * i采样间隔5 分钟30 天下来一共 8640 条记录足够测试清洗、聚合、可视化全流程。如果你手上已经有真实数据可以跳过这步直接读 CSV字段对齐就行。这里注入的缺失率 3% 和异常率 1% 是参考实际交通数据质量评估报告的常见水平不是拍脑袋定的。2.2 数据清洗管道时间戳对齐、缺失值插值、异常值剔除清洗是这个项目里权重最高的一块。交通流量数据常见的脏数据有三类时间戳格式不统一、缺失记录、数值异常比如瞬时流量是均值的 5 倍。清洗模块用 pandas 按统一流程处理结果落到另一张表。def clean_traffic_data(raw_path, out_path): df pd.read_csv(raw_path) df[record_time] pd.to_datetime(df[record_time]) df df.sort_values([monitor_id, record_time]) # 1. 去重复完全相同的记录保留第一条 df df.drop_duplicates(subset[monitor_id, record_time]) # 2. 缺失值按监测点分组用前后时间点做线性插值限制连续缺失不超过 3 条 df[volume] df[volume].interpolate(methodlinear, limit3) # 3. 异常值超过该监测点均值 3 倍标准差时视为异常用滚动中位数替换 stats df.groupby(monitor_id)[volume].transform(lambda x: x.rolling(12, min_periods1).median()) mean df.groupby(monitor_id)[volume].transform(mean) std df.groupby(monitor_id)[volume].transform(std) abnormal (df[volume] - mean).abs() 3 * std df.loc[abnormal, volume] stats[abnormal] df.to_csv(out_path, indexFalse) return df清洗逻辑分三步去重、插值、替换异常。interpolate(methodlinear, limit3)表示对连续缺失最多 3 条的区间做线性插值超过 3 条就直接保留 NaN交给后续聚合层处理——连续长时间缺失说明设备可能断电了插值反而会让数据失真。异常值检测用 3 倍标准差是因为流量数据大体服从偏正态分布3σ 之外基本可以判定为设备抖动或信号错误替换值用rolling(12)的滚动中位数而不是均值因为中位数对离群点更稳健不会出现用一个被污染的值替换另一个污染值的问题。提示滚动窗口 12 对应 5 分钟频率下 1 小时的数据窗口。如果换成了 15 分钟采样频率窗口数要改成 4否则平滑力度会过强。2.3 MySQL 表结构设计原始层、清洗层、聚合层分离这个项目的数据库设计采用分层存储这也是我比较认可的一点。项目设计了五张核心表交通监测点基础信息表、原始交通流量数据表、清洗后的标准交通流量数据表、聚合统计结果表日级或小时级、用户与角色权限管理表。原始表只做插入不做修改保证追溯清洗表存处理后的明细数据聚合表存预计算结果供可视化直接查。表名主要字段用途monitor_infomonitor_id, road_name, road_level, district, longitude, latitude监测点基础信息traffic_rawid, monitor_id, record_time, volume, speed, source原始采集数据traffic_cleanedid, monitor_id, record_time, volume, speed, is_imputed清洗后明细数据traffic_hourly_aggmonitor_id, hour_start, avg_volume, peak_flag, road_level小时级聚合结果sys_useruser_id, username, password_hash, role, status登录与权限字段类型上要特别留意record_time用DATETIME而不是VARCHAR否则后面按小时分组时会遇到意外的排序问题这个坑在第 5 章详细说。is_imputed字段用来标记这条记录是不是插值补出来的做数据质量报告时非常有用可以直接统计插值占比评估该监测点设备健康度。聚合层的peak_flag字段在入库时由业务层计算可视化层查询时不再重新算性能上有明显的收益。3. 聚合统计与高峰识别把千万级流量记录压成可交互图表3.1 resample 频率选择的实际考量原始数据是 5 分钟一条一个城市几十个监测点跑一个月就是几十万条一年上千万条。这种量级直接渲染图表必然卡顿所以聚合层的设计很关键。项目里默认把 5 分钟数据重采样成 15 分钟和 1 小时两个粒度分别服务于看趋势和看规律。选 15 分钟是因为城市交通管理的常规统计口径就是 15 分钟和信号配时、路段饱和度评估的周期一致。df pd.read_csv(traffic_cleaned.csv, parse_dates[record_time]) df df.set_index(record_time) # 按 15 分钟重采样多个监测点按同一时间窗求均值 hourly_agg ( df.groupby(monitor_id)[volume] .resample(15min) .mean() .reset_index() ) hourly_agg.columns [monitor_id, time_window, avg_volume]重采样的核心在于groupby(monitor_id).resample(15min)的组合先按监测点分组再在每个组内按 15 分钟窗口聚合聚合函数用mean()是为了消除随机波动突出趋势。如果你的分析目标是定位拥堵峰值建议用max()而不是mean()因为堵车是一瞬间发生的均值会把峰值抹平。这个细节是很多人的习惯性坑拿到聚合结果后发现高峰消失了多半就是把聚合函数用错了。3.2 道路等级与时段维度组合光有监测点的聚合还不够交通分析更关注的是城市快速路早高峰怎么变化、主干道周末和工作日差多少这类跨维度问题。聚合层需要把道路等级维度和时间维度组合在一起靠监测点基础信息表 JOIN 过来再做一次分组。agg pd.read_sql(SELECT monitor_id, time_window, avg_volume FROM traffic_hourly_agg, engine) info pd.read_sql(SELECT monitor_id, road_level FROM monitor_info, engine) merged agg.merge(info, onmonitor_id, howleft) # 道路等级 × 小时 的多维透视 pivot merged.pivot_table( indexmerged[time_window].dt.hour, columnsroad_level, valuesavg_volume, aggfuncmean )pivot_table的index取小时、columns取道路等级、values取平均流量这样生成的就是一张横轴小时、纵轴道路等级的交叉表。做这个透视之前先把time_window的小时提取出来比直接在 SQL 里GROUP BY HOUR(time_window)更灵活因为 pandas 侧可以继续叠加工作日/周末等自定义维度。类似地想对比工作日和周末就再加一列is_workday按布尔值分组即可。3.3 高峰期识别与阈值设定高峰期的识别不能用一个拍脑袋的固定值因为不同道路等级的流量基数差很多——城市快速路平时就是一百多支路高峰也就几十。项目里采用的是按道路等级分组以该等级历史流量的 85 分位数作为高峰阈值的做法超过阈值就标记peak_flag。# 按道路等级分组求 85 分位数 threshold ( merged.groupby(road_level)[avg_volume] .quantile(0.85) .to_dict() ) merged[peak_flag] merged.apply( lambda row: 1 if row[avg_volume] threshold.get(row[road_level], 9999) else 0, axis1 )用分位数而不是均值的原因在于交通流量的分布是右偏的早晚高峰把均值拉得很高用均值当阈值会把非高峰时段也误判成高峰。85 分位数的含义是只有历史上 15% 的时间流量超过这个值这些时刻通常就是真实的拥堵时段。这个阈值建议每季度重新计算一次因为城市路网在变半年前的 85 分位数可能已经不适合当前状态。3.4 性能策略预聚合表 按需查询我在看这个项目时的第一反应是为什么不直接在可视化时按条件实时聚合原因很简单——千万级记录做一次GROUP BY在 MySQL 里可能要一两秒用户拖一下时间轴就触发一次查询体验会非常差。项目的做法是每天凌晨跑一次批处理把前一天的数据聚合成小时级、日级结果写入聚合表。可视化默认只查聚合表只有用户明确选择查看原始明细时才走明细表接口。这套预聚合 按需查明细的双轨策略是这个系统能保持图表流畅的主要原因。4. 可视化双层方案Matplotlib 静态图与 Plotly 交互式的切换技巧4.1 Matplotlib 静态图适合报告和论文出图项目里用 Matplotlib 做的第一类图是纯静态时间序列适合放进周报、论文或者 PPT。代码不长但有几个参数值得说。import matplotlib.pyplot as plt import matplotlib.dates as mdates fig, ax plt.subplots(figsize(12, 5)) ax.plot(df[record_time], df[volume], linewidth0.8, color#2c7fb8) # 只显示每天 0 点刻度避免 x 轴标签挤成一团 ax.xaxis.set_major_locator(mdates.DayLocator(interval1)) ax.xaxis.set_major_formatter(mdates.DateFormatter(%m-%d)) ax.set_xlabel(日期) ax.set_ylabel(车流量辆/5分钟) ax.set_title(监测点 M001 连续 30 天流量趋势) plt.xticks(rotation30) plt.tight_layout() plt.show()这里的关键是DayLocator和DateFormatter配对使用只让 x 轴显示每天的零点刻度格式化成月-日否则 Matplotlib 默认的自动刻度会把标签密密麻麻排出来整张图没法看。linewidth0.8是调出来的经验值——交通流量数据本身毛刺多线太粗会掩盖细节太细又看不清整体趋势。这个参数因人而异但默认值绝不是 Matplotlib 的 1.5。4.2 Plotly 交互式图服务动态筛选和悬停查看静态图满足不了拖动时间轴看变化这类需求所以项目又用 Plotly 做了一层交互式可视化。Plotly 的图可以直接嵌入 FastAPI 返回的 HTML 页面也可以在 Jupyter 里实时查看适用面很广。import plotly.graph_objects as go fig go.Figure() fig.add_trace( go.Scatter( xdf[record_time], ydf[volume], modelines, name实际流量, linedict(width1, color#e4572e), hovertemplate%{x|%m-%d %H:%M}br车流量%{y}extra/extra ) ) fig.update_layout( title监测点流量交互视图, xaxis_title时间, yaxis_title车流量, hovermodex unified, templateplotly_white, height450 ) fig.show()hovertemplate里%{x|%m-%d %H:%M}是 Plotly 的时间格式化语法控制悬停时显示的精度如果不写这个模板默认会显示完整时间戳一长串数字非常干扰看图。hovermodex unified表示悬停到某个时间点时同时间的所有曲线一起显示这对多监测点对比场景很好用。如果你的数据量超过几万条Plotly 也会卡项目里的解法是前端只请求当前时间窗内经过聚合的数据而不是一次性把所有点都传给前端渲染。4.3 Tkinter 嵌入 Matplotlib桌面 GUI 的集成方式这个项目没有用 Web 前端而是用 Tkinter 做桌面 GUI把 Matplotlib 图表嵌进窗体。这个组合在毕设和内部工具里很常见核心是把绘图区域塞进 Tkinter 的容器里。from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg def draw_trend(parent_frame, df): fig, ax plt.subplots(figsize(8, 4)) ax.plot(df[record_time], df[volume], color#1a6b8a) canvas FigureCanvasTkAgg(fig, masterparent_frame) canvas.draw() canvas.get_tk_widget().pack(fillboth, expandTrue)FigureCanvasTkAgg的作用是把 Matplotlib 的 Figure 对象转成 Tkinter 能渲染的组件master指定它嵌在哪个父容器里pack(fillboth, expandTrue)让图表随窗口缩放。要注意每次调用draw_trend时都新建了 Figure窗口反复切换时如果不主动关闭旧 Figure内存会缓慢增长GUI 用久了会变卡。项目里的处理方式是在绘制新图前先plt.close(fig)这个细节做桌面工具时建议保留。5. 避坑从数据库字段类型到 GUI 嵌入的六条翻车记录5.1 时间字段存成 VARCHAR导致排序和聚合错乱现象凌晨 0 点的数据排到了晚上 10 点之后ORDER BY record_time完全失控按小时分组统计的结果一片混乱。原因建表时图省事把record_time定义成了VARCHAR(20)文本排序是按字符逐位比较的00:00 大于 22:00 的首字符 2所以凌晨的记录反被排到了最后。MySQL 里DATETIME和VARCHAR的索引策略完全不同也能用到索引做范围查询。解决重新定义表结构把record_time改成DATETIME类型。已经存进去的脏数据只能用STR_TO_DATE()刷一遍。从那以后我建表时对时间字段一律先确认类型再谈其他。5.2 pandas 读到整列缺失时间戳后直接丢弃现象某天的数据整体消失清洗前后记录数对不上最后发现那一整天的record_time都是 NaT。原因数据源里的时间戳格式不统一有的行是2024-03-01 08:00:00有的是2024/03/01 08:00pd.to_datetime默认errorsraise会报错改成errorscoerce后无法解析的变成了 NaT后面排序时 NaT 被排到末尾或被丢弃。这个情况多出现在多源数据合并阶段。解决先用pd.to_datetime(..., errorscoerce)做第一遍解析然后单独统计 NaT 占比超过一定阈值就直接把该数据源标记为不可用而不是静默丢弃。常见做法是加一道校验df[df[record_time].isna()].groupby(source).size()快速定位哪个源格式有问题。5.3 resample 后 fillna 把夜间零值填成了均值现象夜间 0 点的车流量变成了 60 多而实际上夜间很多路口车流量就是接近 0 的。原因重采样到小时粒度后部分小时窗口没数据直接调用df.fillna(df.mean())于是全表均值被填进了夜间时段。这个方法用在夜间零值是真实零值的场景下是完全错误的把交通流量的天然低谷扭曲成了平稳下降。解决正确做法是先区分缺失和真零。夜间可能是没车也可能是设备离线前者是有效记录后者需要插值。处理方式是只对连续性缺失做插值零值不动或者按监测点分组后提取该监测点同时段历史均值来填充而不是用全表均值。我一般会多生成一列is_missing把填充痕迹留在数据里方便后来人判断。5.4 Tkinter 嵌入图表中文显示成方块现象GUI 里所有标题和坐标轴中文全部显示成空心方块英文数字正常。原因Matplotlib 默认的字体库不包含中文字体Tkinter 环境里又不像 Web 端有系统字体自动回退机制所以中文直接变成了 tofu 方块。解决在绘制前显式指定中文字体。Windows 下常见做法是plt.rcParams[font.sans-serif] [SimHei]macOS 则改为[PingFang SC]同时设置plt.rcParams[axes.unicode_minus] False防止负号显示异常。这个配置放到项目入口文件里统一设置一次避免每个绘制函数都重复写。5.5 图表一次性加载全量数据窗口拖拽卡死现象选择监测点后绘图要等 3 秒拖拽缩放更是卡到怀疑人生几万条线直接糊成一团。原因可视化从明细表把全时间范围的记录都查出来渲染数据量上去后 Matplotlib 的渲染管线根本扛不住。这是预聚合 按需查询策略没执行到位导致的。解决回调函数里只查询当前选中的时间窗口比如默认只加载最近 7 天的小时聚合结果用户拖动时间范围时重新请求接口。数据量实在压不下来时再叠加降采样比如从 5 分钟粒度降到小时粒度视觉趋势基本一致但点数量直接下降 12 倍。这个项目给了一个很好的默认值窗口默认 7 天、粒度 1 小时显示效果和性能平衡得比较好。5.6 FastAPI 返回 numpy 类型导致 JSON 序列化失败现象接口单独跑测试一切正常前端图表抽数据时报Object of type int64 is not JSON serializable页面白屏。原因pandas 的聚合结果里avg_volume是numpy.int64类型FastAPI 的 JSON 序列化器不认识这个类型直接抛异常。而这个异常发生在前端已经拿到接口地址、请求发出之后后端日志里却只有一行500 Internal Server Error定位很费劲。解决在 API 层做统一的数据转换把 DataFrame 的列强制转成 Python 原生类型常见做法是df[avg_volume] df[avg_volume].astype(float)时间字段转成str。或者用orientrecords转成 dict 列表再返回。需要注意astype(float)会改变数值精度但流量数据本身不需要整数精度影响可忽略。6. 预测接口与端到端验证让系统从看历史到看趋势这章的落点是给系统嫁接一个轻量预测能力以及验证全部链路是否真的通了。项目里没有上复杂的 LSTM 或 Prophet而是用滚动平均加线性回归做基线预测好处是逻辑透明、训练快、部署时不需要额外依赖。6.1 轻量预测模型封装from sklearn.linear_model import LinearRegression def predict_next_window(series, window_size12, future_steps6): # 构建滑动窗口特征用最近 window_size 个点的均值 时间索引斜率做预测 idx np.arange(len(series)).reshape(-1, 1) model LinearRegression() model.fit(idx, series.values) last_idx len(series) pred_idx np.arange(last_idx, last_idx future_steps).reshape(-1, 1) return model.predict(pred_idx)这段代码的思路是用时间序号做自变量、流量做因变量拟合一条直线来推未来几个时间窗的走势。window_size12对应 5 分钟频率下 1 小时的历史窗口future_steps6表示预测未来半小时。线性回归在交通流量上只能做短期基线早晚高峰的突变它预测不出来但作为系统的内置基线已经够用了——它的价值在于用户能直观看到预测曲线和实际曲线的偏差进而理解为什么需要更复杂的模型。6.2 端到端验证清单项目部署完成后建议按这张表逐项过一遍能走通说明系统闭环没问题。验证项操作预期结果数据入库运行数据生成脚本写入原始表记录数与生成条数一致清洗管道执行清洗流程缺失值占比降到 1% 以下聚合回填运行小时聚合任务聚合表记录数 监测点数量 × 24 × 天数API 连通调用/api/v1/traffic/agg接口返回 JSON 数组字段名与前端约定一致可视化渲染GUI 选择监测点切换时间窗口折线图 1 秒内渲染完成预测接口传入最近 1 小时序列返回未来 6 个时间窗预测值这套流程我跑过很多遍之后形成了一个习惯每次改完数据库表结构哪怕只是加一个字段也会强制跑一遍生成→清洗→聚合→查接口的完整链路而不是只测当前改动的那个模块。层与层之间靠字段名和类型隐式耦合一处没对齐前端拿到的就是空数组或报错。项目把这条链路讲得很清楚照着走基本不会卡住。希望这篇文章能帮你少踩几个我踩过的坑也让你把时间花在真正值得研究的数据规律上。本文还有配套的精品资源点击获取
