几十万条车辆轨迹数据摆在面前你能从中看出什么这是我做这个 Python 交通数据分析项目时最真实的开场。数据表里混着时间戳、车速、经纬度、设备编号乱七八糟什么都有但唯独没有现成的结论。我花了差不多两周时间把采集、清洗、统计、可视化、简单部署整套流程全部跑通核心工具就是 Python。这篇文章是完整复盘重点讲清楚每个环节为什么这么做以及我会避开哪些坑。适合正在学 Python 数据分析的入门者也适合已经接触过一些数据工具、想快速搭一套交通数据报表的人。Python 在交通数据分析这个领域有个很实在的优势生态太全了。pandas 处理表格numpy 做数值计算matplotlib 和 plotly 画图scikit-learn 做预测folium 做地图可视化一条链路全包。你不需要在 Excel、数据库、BI 工具之间来回折腾。我这次做的东西也不复杂就是拿真实的路口流量、路段速度数据算清楚几个关键指标再用图表把“什么时候堵、哪里堵、是什么程度的堵”直观展示出来。下面我从项目整体设计开始按实际开发顺序把每一步遇到的问题和解决办法都聊透。1. 需求拆解与方案选型这个项目到底要解决什么问题1.1 交通数据分析的典型场景很多人一想到交通数据分析就下意识觉得要做“人工智能预测交通流”其实大部分实际需求远没有这么玄乎。拿我这次的场景来说甲方给了一批路口监测数据核心问题就三个全天流量分布长什么样高峰期出现在几点哪几个方向最堵。这三个问题拆开后本质是统计聚合加排序连机器学习都不一定需要用上。我梳理出交通数据分析里最常见的四类场景你可以对号入座交通流量统计统计某个断面、路口、路段在单位时间内通过的车辆数这是最基础的需求。常用指标有小时流量、日流量、车种比例。运行速度分析利用浮动车数据或路侧设备数据计算路段平均行程速度、平均行驶速度用来判断交通拥堵程度。拥堵识别与排名通过速度阈值或拥堵持续时间把路段状态划分成畅通、缓行、拥堵、严重拥堵等级再做排名。设施规划支撑分析某片区历史流量变化趋势为新建立交、调整信号配时提供数据依据。对新手来说最容易犯的错是一上来就钻进建模想着搞个十层八层的深度学习。真到落地时你会发现第一步永远是数据整理。数据不干净模型再花哨也是空中楼阁。我建议先用最简单的分组聚合把业务指标跑出来让数据先“说话”再去评估是否真的需要预测模型。这样分析链路短、容易验证给业务方看也直白。1.2 为什么选 Python 而不是其他工具先说结论如果你的目标是快速交付一套可复用的交通数据分析应用Python 是目前综合成本最低的选择。我知道有人会提 R 语言统计功能确实强但工程化能力偏弱写接口、做定时任务、对接数据库都比较别扭。也有人提 Power BI、Tableau 这类 BI 工具拖拽确实爽可一旦数据量大、逻辑复杂交互式操作反而成了瓶颈而且不好做自动化更新。Python 的优势在于“代码即文档”分析过程可以完全复现。你改了参数重新跑一遍结果差异在哪一眼就能看出来这在交通这种对准确性敏感的场景里非常关键。再说性能问题很多人的顾虑是“Python 慢”。实际上在交通数据分析这个量级大部分单表数据在几十万到几百万行pandas 用向量化操作处理这类数据完全在几秒钟就能完成。真遇到上亿条轨迹需要做分布式也有 PySpark 这样成熟方案可以切换。你不需要一开始就规划大数据平台往往一台普通电脑加 Python 就够了先把核心逻辑跑通更重要。1.3 整体功能架构我这个应用的整体数据流其实很清晰数据接入 → 数据清洗 → 指标计算 → 可视化 → 结果输出。用代码结构表达的话大致是这样的模块划分模块主要职责需要用到的库数据接入层读取 CSV、数据库或 API 数据pandas, requests数据清洗层处理缺失值、重复值、异常值pandas, numpy指标计算层流量汇总、车速计算、拥堵指数pandas, numpy可视化层图表与地图展示matplotlib, folium服务输出层简单 Web 接口供前端调用Flask这个结构不是拍脑袋定的而是按“每一层只干一件事”的原则拆的。比如数据清洗层我单独抽成函数后面接 API 数据或者接爬虫数据时输入格式变但只要逻辑不变就能复用。做项目最忌讳的是把所有代码堆在一个脚本里改一个地方牵一发动全身后续维护成本会高到你不想再打开它。2. 数据从哪来、怎么洗干净完整数据预处理流程2.1 数据源的获取方式交通数据的来源五花八门。我这次用到的数据来自三个方向你可以根据自己的实际情况选择官方开放数据平台。很多城市会把交通流量、公交线路、出租车 GPS 轨迹数据脱敏后放到公共数据平台格式一般有 CSV、JSON 或者提供 API 接口。这类数据质量相对高字段完整适合做项目起点。但缺点是指标口径经常变你最好下载后第一时间把字段说明文档也存一份。自有设备批量导出。如果你有条件接触路侧设备、信号机后台导出来的通常是比较原始的报文记录。这类数据字段繁杂量很大但分析价值也最直接。我当时拿到的是每 30 秒一条的设备心跳记录里面包含车牌识别结果、车道号、时间和速度信息。爬虫采集公开页面数据。用 requests 写爬虫从网页获取堵车指数、路段拥堵排行这类公开数据也可以。比如某些地图平台的公开接口返回 JSON 格式数据很规整。但要注意爬取频率、robots 协议和使用边界别给人家服务器造成压力更不能用采集的数据做违反规则的事情。如果一时找不到合适数据我建议你直接用代码生成一份模拟数据先练手。模拟数据不算“假练”因为清洗和分析逻辑完全一致。你可以用随机数生成时间、车速、经纬度再人为掺入缺失值、重复值这样反而能更好地测试自己的清洗逻辑是否健壮。等逻辑调通再换真实数据基本能无缝切换。2.2 第一步读取数据并做基础探查拿到数据文件后我习惯先跑一段简短的探查代码搞清楚表结构长什么样而不是直接开始清洗。探查一般包含三个动作看前几行、看列的信息、统计缺失情况。import pandas as pd df pd.read_csv(traffic_records.csv, parse_dates[timestamp]) print(df.head()) print(df.info()) print(df.isnull().sum())这里的parse_dates很关键。交通数据的时间戳几乎都是字符串不转成 datetime 类型后面按小时、按天聚合就会非常痛苦。转换之后pandas 会把时间列识别成专门的日期时间类型可以直接.dt.hour、.dt.dayofweek这样提取时间特征。info()可以看到每列的非空个数和数据类型如果某列显示 object 但你心里预期是数值那多半是里面混了脏数据比如空字符串、字母单位这种。这时候不能强行转换得先把脏值找出来处理掉。探查完数据我最大的感受是真实的交通数据从来都不是干净的。你会碰到时间戳格式不统一、同一辆车同一个时刻被识别两次、车速出现负数、经纬度跑到海里等等问题。这些都是正常现象也是分析流程里必须处理的环节。2.3 清洗三件套重复、缺失、异常整个清洗过程中我重点做了三类处理去重。交通数据里最容易出现重复记录原因可能是设备重复上报或者数据合并时叠加。去重的逻辑要看场景。如果同一时刻同一设备同一辆车出现了两条一模一样的数据直接删除多余记录。但如果只是部分字段重复就要谨慎比如同一辆车在多个断面被拍到这不算严格意义上的重复不能删。df df.drop_duplicates(subset[device_id, timestamp, plate_no])这行代码的意思是以设备号、时间、车牌三个字段作为联合判断依据保留第一条。实际使用时要根据你的业务逻辑定义“什么算重复”不能机械操作。缺失值。缺失值的策略分两种。如果缺失比例很小比如某列缺失不到 5%而且该字段不是关键字段可以直接删除缺失行。但对于时间序列里的数值字段比如车速、流量直接删除会造成序列断裂更推荐用插值法。pandas 的interpolate()方法就很好用它会根据缺失点前后值进行线性插值对于短时间窗口的数据缺失效果非常自然。df[speed] df[speed].interpolate(methodlinear, limit_directionboth)如果某个设备连续几小时不回传数据插值就没有意义了说明设备本身可能离线这段数据应该从分析中剔除否则会把一个“无数据”的时段误判成“零流量”时段导致后面高峰识别完全偏掉。异常值。交通数据里最常见的异常值就是不合理车速。速度小于 0直接是数据错误速度大于一个物理上限比如超过 180 公里每小时大部分场景下也是异常。我用的是极值过滤加业务规则两层方式df df[(df[speed] 1) (df[speed] 150)]这里有个细节过滤即删除所以执行前最好先打印一下被过滤掉的数据量做到心里有数。如果删掉的比例超过 10%说明数据质量很差或者你的阈值设置得不合理要回头检查。2.4 时间序列重采样与特征衍生清洗后的数据是一秒一条还是三十秒一条往往不统一。做流量分析时我通常会统一重采样到固定时间粒度比如 5 分钟或 15 分钟。这个动作相当于把散乱记录“规整化”后面所有分析都能对齐。df[time_bin] df[timestamp].dt.floor(5min) flow df.groupby([time_bin, direction]).size().reset_index(namecount).dt.floor(5min)是个很常用的小技巧它把每条记录自动落到最近的 5 分钟区间然后我按区间分组统计车辆数冲掉原始零散时间带来的噪声。除了时间分箱我还习惯额外生成几个特征列小时、星期几、是否工作日、是否高峰时段。这些特征在后面做拥堵分析时直接分组就能用不用到时候再重复筛选df[hour] df[timestamp].dt.hour df[dayofweek] df[timestamp].dt.dayofweek df[is_workday] df[dayofweek] 5 df[is_peak] df[hour].isin([7, 8, 9, 17, 18, 19])这些衍生字段看着简单但对业务方展示时非常直观。对方问“早高峰到底有多堵”你直接按is_peak分组求平均速度一句话就能解答。3. 核心分析模块流量、车速、拥堵指数怎么算3.1 我使用的核心指标体系数据清洗完分析才有意义。我这次定义了三类核心指标也是交通行业里最通用的三类指标流量。流量是指单位时间内通过某一断面的车辆数。最小时粒度通常取 5 分钟或 15 分钟再聚合到小时。流量能回答“这条路忙不忙”但没法回答“堵不堵”——有时候流量低但车速也非常慢说明已经堵到车都进不来了。平均车速。平均车速有两种算法一种是简单的算术平均把所有车速度加起来除以车辆数另一种是更科学的时间平均速度把每辆车通过某段路的行程时间加权平均。如果手里有原始车牌识别数据我推荐按旅行时间算因为更能反映驾驶人实际感受。拥堵指数。拥堵指数我采用行业里常用的方法把实际平均车速与自由流速度畅通时的参考车速比如 60 km/h做比值用一个百分数表达。值越大代表越接近自由流越小代表越拥堵。free_flow_speed 60 df[congestion_ratio] df[avg_speed] / free_flow_speed df[congestion_level] pd.cut( df[congestion_ratio], bins[0, 0.3, 0.5, 0.8, 1.0], labels[严重拥堵, 拥堵, 缓行, 畅通], )这里用pd.cut把连续比例切成等级四个档位对应不同颜色后面可视化阶段直接映射成红色、橙色、黄色、绿色展示效果非常清楚。阈值可以结合实际城市快速路情况调整没有一个放之四海而皆准的标准。3.2 早晚高峰识别实战高峰识别是交通分析里最有“获得感”的一块。我把清洗后的数据按小时汇总流量然后截取早上六点到晚上十点的曲线用简单的pivot_table就能做出来flow_by_hour df.pivot_table( indexhour, columnsdirection, valuesrecord_id, aggfunccount, )结果是一个表格行为小时列为方向值为车流量。直接打印出来哪几个小时数值明显高早晚高峰一眼就出来了。想要更自动化可以设置阈值流量超过全天平均水平一定倍数的小时视为高峰。但实际业务中人工看图确认往往更高效。有一个细节要特别提醒高峰识别一定要分方向。同一个路段早高峰进城方向流量大晚高峰出城方向流量大混在一起统计会把结论完全盖住。所以分组聚合时方向字段必须参与分组。3.3 拥堵路段排名与区域分析排名的思路也很直白。把每条路段在每天高峰时段内的平均车速求出来再算拥堵指数排序取倒数前十就得到“最堵路段榜单”。peak_df df[df[is_peak]] result ( peak_df.groupby([road_name, direction]) .agg(avg_speed(speed, mean), volume(record_id, count)) .reset_index() ) result[congestion_ratio] result[avg_speed] / free_flow_speed result result.sort_values(congestion_ratio) print(result.head(10))这里agg同时计算多个聚合字段类似于 Excel 的多重透视。这条路平均速度是多少、流量多大两条信息放同一个表里分析效率很高。按拥堵指数升序排列后最堵的十条路就在最上面。区域分析则更偏空间维度。我把地图按网格切块比如 500 米乘 500 米一个格子把经纬度落进格子统计每个格子高峰时段的车流速度。速度最低的格子就是拥堵热点后续做热力地图时数据基础就来自于这个网格聚合。3.4 做一个小型车速预测模型虽然我前面说不要一上来就做预测但当我拿到两周完整数据后还是忍不住加了一个简单的预测模块用的是 scikit-learn 里的随机森林回归。特征只用几个和时间、空间有关的字段小时、星期几、是否工作日、方向编号。目标值是预测下一个小时该路段的平均车速。from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split features [hour, dayofweek, is_workday, direction_code] X df[features] y df[avg_speed] X_train, X_test, y_train, y_test train_test_split(X, y, test_size0.2, random_state42) model RandomForestRegressor(n_estimators200, max_depth10, random_state42) model.fit(X_train, y_train) print(R2 score:, model.score(X_test, y_test))这个模型的 R2 我实测在 0.75 左右。对于一个只用时间特征、不用上下游实时数据的模型来说已经很有解释力了。拿不到 0.9 以上很正常交通系统本身随机性强天气、事故这些因素没进来。我建议不要为了刷指标强行加复杂模型能用简单模型解释 70% 的规律就值得先上线。4. 让结果看得见可视化与轻量落地4.1 Matplotlib 绘制流量时间曲线可视化这部分我做的最多是画时间曲线。横轴是时间纵轴是流量或平均车速多方向多颜色叠加直观展示全天变化。import matplotlib.pyplot as plt plt.figure(figsize(12, 5)) for direction, group in flow_by_hour.items(): plt.plot(group.index, group.values, labeldirection) plt.xlabel(小时) plt.ylabel(车流量) plt.legend() plt.grid(alpha0.4) plt.tight_layout() plt.savefig(hourly_flow.png, dpi150)经验之谈出图时dpi一定要调高至少 150。默认的 72 dpi 在屏幕上看着还行放进报告里一放大就糊得没法看。另外中文显示是个老坑Windows 下要设置字体否则坐标轴会出现方块字。我一般用这样的方式临时设置plt.rcParams[font.sans-serif] [SimHei] plt.rcParams[axes.unicode_minus] False4.2 Folium 制作拥堵热力地图地图可视化我用 Folium 库它底层调用 Leaflet生成的是 HTML 文件浏览器直接打开就能交互缩放。我先把网格聚合后的数据转成热力点再用 HeatMap 插件叠加。import folium from folium.plugins import HeatMap center [lat_mean, lng_mean] map_obj folium.Map(locationcenter, zoom_start12) heat_data df[[lat, lng, avg_speed]].values.tolist() HeatMap(heat_data, radius15, blur10, min_opacity0.4).add_to(map_obj) map_obj.save(traffic_heatmap.html)生成 HTML 地图这个方案最大的好处是不需要额外部署服务直接把文件发给同事双击打开就能用。我做项目汇报时经常先导出地图文件再配合几张大图效果比口头描述一百遍都强。如果数据量很大热力点的数量建议做降采样不然 HTML 文件会很大浏览器打开卡得厉害。4.3 用 Flask 搭一个最简数据查询接口项目最后我顺手用 Flask 搭了一个极简查询接口实现“输入路段名和日期范围返回该路段平均车速和拥堵等级”的功能。接口本身不长核心代码几十行就能写完。from flask import Flask, jsonify, request app Flask(__name__) app.route(/api/congestion, methods[GET]) def get_congestion(): road request.args.get(road) date request.args.get(date) subset df[(df[road_name] road) (df[date] date)] avg_speed subset[speed].mean() level 畅通 if avg_speed 40 else 拥堵 return jsonify({road: road, date: date, avg_speed: avg_speed, level: level}) if __name__ __main__: app.run(host0.0.0.0, port8000)这里host0.0.0.0表示允许局域网内其他机器访问自己测试时用默认的 127.0.0.1 就够了。接口跑起来之后前端或者报表工具通过 URL 直接调用比如访问http://localhost:8000/api/congestion?road人民路date2025-11-18就能拿到 JSON 结果。整个应用就从“分析脚本”升级成了“小型分析服务”。5. 实战中踩过的坑与排查清单5.1 时间戳与时区的坑交通数据里的时间戳最容易出问题我再强调一遍。第一次分析时我直接对字符串时间排序、聚合结果发现排序是字母序而不是时间序天然少了一个小时。原因就是没转 datetime。时区问题更隐蔽。设备端记录的是北京时间但服务器存储用了 UTC直接混用导致我画出来的流量曲线整体偏移了 8 个小时早高峰跑到了下午。排查半天才发现是时区没有统一。我的教训是项目第一天就全局约定统一使用北京时间字符串内部处理统一用 pandas datetime到输出报表时再格式化成目标时区。时区问题一旦混入后期找 bug 的成本极高。5.2 内存与性能优化几十万行的交通数据pandas 处理起来其实毫无压力。但如果你做网格聚合把经纬度、时间段全部笛卡尔积展开数据量瞬间膨胀到几千万行内存就开始吃紧。我的解决套路有几个读数据时用usecols只选取需要的列别把没用的超大字符串列读进来中间变量及时用del删除或者用gc.collect()手动触发垃圾回收如果连续做多次聚合先把数据按关键字段排序再用groupby性能会比反复过滤高不少。实在顶不住用dtype参数把数字列指定成int32或float32内存占用直接减半。这个优化技巧很土但真的很有效。5.3 经纬度出界的排查地图热力图上我发现有几个高亮热点落在海里明显不合理。排查后发现是设备上报的经纬度偶尔会出现零值或者错误值。这种异常我用一个简单的范围过滤就能拦住df df[(df[lat] 20) (df[lat] 50)] df df[(df[lng] 100) (df[lng] 125)]按国内绝大多数城市的经纬度范围收紧区间经验值直接写进去比用“是否在海陆边界”的复杂判断更省事。这类过滤逻辑建议放在清洗模块里跟车速过滤并列这样每次重跑数据都能自动生效。5.4 环境配置问题最后说一个容易被忽视的坑Python 环境。很多同事在自己电脑上跑这些代码时报错都出在环境上而代码本身没问题。常见的有几个Python 装到了带中文或空格的路径下导致一些底层依赖编译失败直接全局装包版本冲突后项目跑不起来用 VS Code 写代码但没选中正确解释器。我的建议是项目一开始就用python -m venv venv创建独立虚拟环境然后pip install pandas matplotlib scikit-learn folium flask所有依赖锁进一个 requirements.txt。别嫌这个步骤啰嗦交通数据分析这种项目数据量一大依赖版本稍微不一致结果都对不上排查起来才是最麻烦的。虚拟环境能帮你把问题范围缩到最小。这份代码之外我更想分享的一点体会项目做完之后我复盘了一下真正让我觉得有价值的不是画出了好看的图也不是接口能跑通而是我建立了一套“先清洗、再指标、后可视化”的分析习惯。交通数据的特点是脏、乱、但是量大、规律性强只要前 70% 的数据整理功夫做扎实后面出活非常快。我强烈建议你在自己电脑上把整套流程跑一遍哪怕先用模拟数据。刚开始可能觉得清洗很枯燥但等你拿到真实数据发现清洗逻辑能直接复用的时候你会回来感谢自己当初没有跳过这一步。如果做的时候碰到奇怪的问题欢迎按我上面列的那些坑去排查大概率你遇到的就是其中之一。
Python交通数据分析：从数据清洗到可视化的完整实践
几十万条车辆轨迹数据摆在面前你能从中看出什么这是我做这个 Python 交通数据分析项目时最真实的开场。数据表里混着时间戳、车速、经纬度、设备编号乱七八糟什么都有但唯独没有现成的结论。我花了差不多两周时间把采集、清洗、统计、可视化、简单部署整套流程全部跑通核心工具就是 Python。这篇文章是完整复盘重点讲清楚每个环节为什么这么做以及我会避开哪些坑。适合正在学 Python 数据分析的入门者也适合已经接触过一些数据工具、想快速搭一套交通数据报表的人。Python 在交通数据分析这个领域有个很实在的优势生态太全了。pandas 处理表格numpy 做数值计算matplotlib 和 plotly 画图scikit-learn 做预测folium 做地图可视化一条链路全包。你不需要在 Excel、数据库、BI 工具之间来回折腾。我这次做的东西也不复杂就是拿真实的路口流量、路段速度数据算清楚几个关键指标再用图表把“什么时候堵、哪里堵、是什么程度的堵”直观展示出来。下面我从项目整体设计开始按实际开发顺序把每一步遇到的问题和解决办法都聊透。1. 需求拆解与方案选型这个项目到底要解决什么问题1.1 交通数据分析的典型场景很多人一想到交通数据分析就下意识觉得要做“人工智能预测交通流”其实大部分实际需求远没有这么玄乎。拿我这次的场景来说甲方给了一批路口监测数据核心问题就三个全天流量分布长什么样高峰期出现在几点哪几个方向最堵。这三个问题拆开后本质是统计聚合加排序连机器学习都不一定需要用上。我梳理出交通数据分析里最常见的四类场景你可以对号入座交通流量统计统计某个断面、路口、路段在单位时间内通过的车辆数这是最基础的需求。常用指标有小时流量、日流量、车种比例。运行速度分析利用浮动车数据或路侧设备数据计算路段平均行程速度、平均行驶速度用来判断交通拥堵程度。拥堵识别与排名通过速度阈值或拥堵持续时间把路段状态划分成畅通、缓行、拥堵、严重拥堵等级再做排名。设施规划支撑分析某片区历史流量变化趋势为新建立交、调整信号配时提供数据依据。对新手来说最容易犯的错是一上来就钻进建模想着搞个十层八层的深度学习。真到落地时你会发现第一步永远是数据整理。数据不干净模型再花哨也是空中楼阁。我建议先用最简单的分组聚合把业务指标跑出来让数据先“说话”再去评估是否真的需要预测模型。这样分析链路短、容易验证给业务方看也直白。1.2 为什么选 Python 而不是其他工具先说结论如果你的目标是快速交付一套可复用的交通数据分析应用Python 是目前综合成本最低的选择。我知道有人会提 R 语言统计功能确实强但工程化能力偏弱写接口、做定时任务、对接数据库都比较别扭。也有人提 Power BI、Tableau 这类 BI 工具拖拽确实爽可一旦数据量大、逻辑复杂交互式操作反而成了瓶颈而且不好做自动化更新。Python 的优势在于“代码即文档”分析过程可以完全复现。你改了参数重新跑一遍结果差异在哪一眼就能看出来这在交通这种对准确性敏感的场景里非常关键。再说性能问题很多人的顾虑是“Python 慢”。实际上在交通数据分析这个量级大部分单表数据在几十万到几百万行pandas 用向量化操作处理这类数据完全在几秒钟就能完成。真遇到上亿条轨迹需要做分布式也有 PySpark 这样成熟方案可以切换。你不需要一开始就规划大数据平台往往一台普通电脑加 Python 就够了先把核心逻辑跑通更重要。1.3 整体功能架构我这个应用的整体数据流其实很清晰数据接入 → 数据清洗 → 指标计算 → 可视化 → 结果输出。用代码结构表达的话大致是这样的模块划分模块主要职责需要用到的库数据接入层读取 CSV、数据库或 API 数据pandas, requests数据清洗层处理缺失值、重复值、异常值pandas, numpy指标计算层流量汇总、车速计算、拥堵指数pandas, numpy可视化层图表与地图展示matplotlib, folium服务输出层简单 Web 接口供前端调用Flask这个结构不是拍脑袋定的而是按“每一层只干一件事”的原则拆的。比如数据清洗层我单独抽成函数后面接 API 数据或者接爬虫数据时输入格式变但只要逻辑不变就能复用。做项目最忌讳的是把所有代码堆在一个脚本里改一个地方牵一发动全身后续维护成本会高到你不想再打开它。2. 数据从哪来、怎么洗干净完整数据预处理流程2.1 数据源的获取方式交通数据的来源五花八门。我这次用到的数据来自三个方向你可以根据自己的实际情况选择官方开放数据平台。很多城市会把交通流量、公交线路、出租车 GPS 轨迹数据脱敏后放到公共数据平台格式一般有 CSV、JSON 或者提供 API 接口。这类数据质量相对高字段完整适合做项目起点。但缺点是指标口径经常变你最好下载后第一时间把字段说明文档也存一份。自有设备批量导出。如果你有条件接触路侧设备、信号机后台导出来的通常是比较原始的报文记录。这类数据字段繁杂量很大但分析价值也最直接。我当时拿到的是每 30 秒一条的设备心跳记录里面包含车牌识别结果、车道号、时间和速度信息。爬虫采集公开页面数据。用 requests 写爬虫从网页获取堵车指数、路段拥堵排行这类公开数据也可以。比如某些地图平台的公开接口返回 JSON 格式数据很规整。但要注意爬取频率、robots 协议和使用边界别给人家服务器造成压力更不能用采集的数据做违反规则的事情。如果一时找不到合适数据我建议你直接用代码生成一份模拟数据先练手。模拟数据不算“假练”因为清洗和分析逻辑完全一致。你可以用随机数生成时间、车速、经纬度再人为掺入缺失值、重复值这样反而能更好地测试自己的清洗逻辑是否健壮。等逻辑调通再换真实数据基本能无缝切换。2.2 第一步读取数据并做基础探查拿到数据文件后我习惯先跑一段简短的探查代码搞清楚表结构长什么样而不是直接开始清洗。探查一般包含三个动作看前几行、看列的信息、统计缺失情况。import pandas as pd df pd.read_csv(traffic_records.csv, parse_dates[timestamp]) print(df.head()) print(df.info()) print(df.isnull().sum())这里的parse_dates很关键。交通数据的时间戳几乎都是字符串不转成 datetime 类型后面按小时、按天聚合就会非常痛苦。转换之后pandas 会把时间列识别成专门的日期时间类型可以直接.dt.hour、.dt.dayofweek这样提取时间特征。info()可以看到每列的非空个数和数据类型如果某列显示 object 但你心里预期是数值那多半是里面混了脏数据比如空字符串、字母单位这种。这时候不能强行转换得先把脏值找出来处理掉。探查完数据我最大的感受是真实的交通数据从来都不是干净的。你会碰到时间戳格式不统一、同一辆车同一个时刻被识别两次、车速出现负数、经纬度跑到海里等等问题。这些都是正常现象也是分析流程里必须处理的环节。2.3 清洗三件套重复、缺失、异常整个清洗过程中我重点做了三类处理去重。交通数据里最容易出现重复记录原因可能是设备重复上报或者数据合并时叠加。去重的逻辑要看场景。如果同一时刻同一设备同一辆车出现了两条一模一样的数据直接删除多余记录。但如果只是部分字段重复就要谨慎比如同一辆车在多个断面被拍到这不算严格意义上的重复不能删。df df.drop_duplicates(subset[device_id, timestamp, plate_no])这行代码的意思是以设备号、时间、车牌三个字段作为联合判断依据保留第一条。实际使用时要根据你的业务逻辑定义“什么算重复”不能机械操作。缺失值。缺失值的策略分两种。如果缺失比例很小比如某列缺失不到 5%而且该字段不是关键字段可以直接删除缺失行。但对于时间序列里的数值字段比如车速、流量直接删除会造成序列断裂更推荐用插值法。pandas 的interpolate()方法就很好用它会根据缺失点前后值进行线性插值对于短时间窗口的数据缺失效果非常自然。df[speed] df[speed].interpolate(methodlinear, limit_directionboth)如果某个设备连续几小时不回传数据插值就没有意义了说明设备本身可能离线这段数据应该从分析中剔除否则会把一个“无数据”的时段误判成“零流量”时段导致后面高峰识别完全偏掉。异常值。交通数据里最常见的异常值就是不合理车速。速度小于 0直接是数据错误速度大于一个物理上限比如超过 180 公里每小时大部分场景下也是异常。我用的是极值过滤加业务规则两层方式df df[(df[speed] 1) (df[speed] 150)]这里有个细节过滤即删除所以执行前最好先打印一下被过滤掉的数据量做到心里有数。如果删掉的比例超过 10%说明数据质量很差或者你的阈值设置得不合理要回头检查。2.4 时间序列重采样与特征衍生清洗后的数据是一秒一条还是三十秒一条往往不统一。做流量分析时我通常会统一重采样到固定时间粒度比如 5 分钟或 15 分钟。这个动作相当于把散乱记录“规整化”后面所有分析都能对齐。df[time_bin] df[timestamp].dt.floor(5min) flow df.groupby([time_bin, direction]).size().reset_index(namecount).dt.floor(5min)是个很常用的小技巧它把每条记录自动落到最近的 5 分钟区间然后我按区间分组统计车辆数冲掉原始零散时间带来的噪声。除了时间分箱我还习惯额外生成几个特征列小时、星期几、是否工作日、是否高峰时段。这些特征在后面做拥堵分析时直接分组就能用不用到时候再重复筛选df[hour] df[timestamp].dt.hour df[dayofweek] df[timestamp].dt.dayofweek df[is_workday] df[dayofweek] 5 df[is_peak] df[hour].isin([7, 8, 9, 17, 18, 19])这些衍生字段看着简单但对业务方展示时非常直观。对方问“早高峰到底有多堵”你直接按is_peak分组求平均速度一句话就能解答。3. 核心分析模块流量、车速、拥堵指数怎么算3.1 我使用的核心指标体系数据清洗完分析才有意义。我这次定义了三类核心指标也是交通行业里最通用的三类指标流量。流量是指单位时间内通过某一断面的车辆数。最小时粒度通常取 5 分钟或 15 分钟再聚合到小时。流量能回答“这条路忙不忙”但没法回答“堵不堵”——有时候流量低但车速也非常慢说明已经堵到车都进不来了。平均车速。平均车速有两种算法一种是简单的算术平均把所有车速度加起来除以车辆数另一种是更科学的时间平均速度把每辆车通过某段路的行程时间加权平均。如果手里有原始车牌识别数据我推荐按旅行时间算因为更能反映驾驶人实际感受。拥堵指数。拥堵指数我采用行业里常用的方法把实际平均车速与自由流速度畅通时的参考车速比如 60 km/h做比值用一个百分数表达。值越大代表越接近自由流越小代表越拥堵。free_flow_speed 60 df[congestion_ratio] df[avg_speed] / free_flow_speed df[congestion_level] pd.cut( df[congestion_ratio], bins[0, 0.3, 0.5, 0.8, 1.0], labels[严重拥堵, 拥堵, 缓行, 畅通], )这里用pd.cut把连续比例切成等级四个档位对应不同颜色后面可视化阶段直接映射成红色、橙色、黄色、绿色展示效果非常清楚。阈值可以结合实际城市快速路情况调整没有一个放之四海而皆准的标准。3.2 早晚高峰识别实战高峰识别是交通分析里最有“获得感”的一块。我把清洗后的数据按小时汇总流量然后截取早上六点到晚上十点的曲线用简单的pivot_table就能做出来flow_by_hour df.pivot_table( indexhour, columnsdirection, valuesrecord_id, aggfunccount, )结果是一个表格行为小时列为方向值为车流量。直接打印出来哪几个小时数值明显高早晚高峰一眼就出来了。想要更自动化可以设置阈值流量超过全天平均水平一定倍数的小时视为高峰。但实际业务中人工看图确认往往更高效。有一个细节要特别提醒高峰识别一定要分方向。同一个路段早高峰进城方向流量大晚高峰出城方向流量大混在一起统计会把结论完全盖住。所以分组聚合时方向字段必须参与分组。3.3 拥堵路段排名与区域分析排名的思路也很直白。把每条路段在每天高峰时段内的平均车速求出来再算拥堵指数排序取倒数前十就得到“最堵路段榜单”。peak_df df[df[is_peak]] result ( peak_df.groupby([road_name, direction]) .agg(avg_speed(speed, mean), volume(record_id, count)) .reset_index() ) result[congestion_ratio] result[avg_speed] / free_flow_speed result result.sort_values(congestion_ratio) print(result.head(10))这里agg同时计算多个聚合字段类似于 Excel 的多重透视。这条路平均速度是多少、流量多大两条信息放同一个表里分析效率很高。按拥堵指数升序排列后最堵的十条路就在最上面。区域分析则更偏空间维度。我把地图按网格切块比如 500 米乘 500 米一个格子把经纬度落进格子统计每个格子高峰时段的车流速度。速度最低的格子就是拥堵热点后续做热力地图时数据基础就来自于这个网格聚合。3.4 做一个小型车速预测模型虽然我前面说不要一上来就做预测但当我拿到两周完整数据后还是忍不住加了一个简单的预测模块用的是 scikit-learn 里的随机森林回归。特征只用几个和时间、空间有关的字段小时、星期几、是否工作日、方向编号。目标值是预测下一个小时该路段的平均车速。from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split features [hour, dayofweek, is_workday, direction_code] X df[features] y df[avg_speed] X_train, X_test, y_train, y_test train_test_split(X, y, test_size0.2, random_state42) model RandomForestRegressor(n_estimators200, max_depth10, random_state42) model.fit(X_train, y_train) print(R2 score:, model.score(X_test, y_test))这个模型的 R2 我实测在 0.75 左右。对于一个只用时间特征、不用上下游实时数据的模型来说已经很有解释力了。拿不到 0.9 以上很正常交通系统本身随机性强天气、事故这些因素没进来。我建议不要为了刷指标强行加复杂模型能用简单模型解释 70% 的规律就值得先上线。4. 让结果看得见可视化与轻量落地4.1 Matplotlib 绘制流量时间曲线可视化这部分我做的最多是画时间曲线。横轴是时间纵轴是流量或平均车速多方向多颜色叠加直观展示全天变化。import matplotlib.pyplot as plt plt.figure(figsize(12, 5)) for direction, group in flow_by_hour.items(): plt.plot(group.index, group.values, labeldirection) plt.xlabel(小时) plt.ylabel(车流量) plt.legend() plt.grid(alpha0.4) plt.tight_layout() plt.savefig(hourly_flow.png, dpi150)经验之谈出图时dpi一定要调高至少 150。默认的 72 dpi 在屏幕上看着还行放进报告里一放大就糊得没法看。另外中文显示是个老坑Windows 下要设置字体否则坐标轴会出现方块字。我一般用这样的方式临时设置plt.rcParams[font.sans-serif] [SimHei] plt.rcParams[axes.unicode_minus] False4.2 Folium 制作拥堵热力地图地图可视化我用 Folium 库它底层调用 Leaflet生成的是 HTML 文件浏览器直接打开就能交互缩放。我先把网格聚合后的数据转成热力点再用 HeatMap 插件叠加。import folium from folium.plugins import HeatMap center [lat_mean, lng_mean] map_obj folium.Map(locationcenter, zoom_start12) heat_data df[[lat, lng, avg_speed]].values.tolist() HeatMap(heat_data, radius15, blur10, min_opacity0.4).add_to(map_obj) map_obj.save(traffic_heatmap.html)生成 HTML 地图这个方案最大的好处是不需要额外部署服务直接把文件发给同事双击打开就能用。我做项目汇报时经常先导出地图文件再配合几张大图效果比口头描述一百遍都强。如果数据量很大热力点的数量建议做降采样不然 HTML 文件会很大浏览器打开卡得厉害。4.3 用 Flask 搭一个最简数据查询接口项目最后我顺手用 Flask 搭了一个极简查询接口实现“输入路段名和日期范围返回该路段平均车速和拥堵等级”的功能。接口本身不长核心代码几十行就能写完。from flask import Flask, jsonify, request app Flask(__name__) app.route(/api/congestion, methods[GET]) def get_congestion(): road request.args.get(road) date request.args.get(date) subset df[(df[road_name] road) (df[date] date)] avg_speed subset[speed].mean() level 畅通 if avg_speed 40 else 拥堵 return jsonify({road: road, date: date, avg_speed: avg_speed, level: level}) if __name__ __main__: app.run(host0.0.0.0, port8000)这里host0.0.0.0表示允许局域网内其他机器访问自己测试时用默认的 127.0.0.1 就够了。接口跑起来之后前端或者报表工具通过 URL 直接调用比如访问http://localhost:8000/api/congestion?road人民路date2025-11-18就能拿到 JSON 结果。整个应用就从“分析脚本”升级成了“小型分析服务”。5. 实战中踩过的坑与排查清单5.1 时间戳与时区的坑交通数据里的时间戳最容易出问题我再强调一遍。第一次分析时我直接对字符串时间排序、聚合结果发现排序是字母序而不是时间序天然少了一个小时。原因就是没转 datetime。时区问题更隐蔽。设备端记录的是北京时间但服务器存储用了 UTC直接混用导致我画出来的流量曲线整体偏移了 8 个小时早高峰跑到了下午。排查半天才发现是时区没有统一。我的教训是项目第一天就全局约定统一使用北京时间字符串内部处理统一用 pandas datetime到输出报表时再格式化成目标时区。时区问题一旦混入后期找 bug 的成本极高。5.2 内存与性能优化几十万行的交通数据pandas 处理起来其实毫无压力。但如果你做网格聚合把经纬度、时间段全部笛卡尔积展开数据量瞬间膨胀到几千万行内存就开始吃紧。我的解决套路有几个读数据时用usecols只选取需要的列别把没用的超大字符串列读进来中间变量及时用del删除或者用gc.collect()手动触发垃圾回收如果连续做多次聚合先把数据按关键字段排序再用groupby性能会比反复过滤高不少。实在顶不住用dtype参数把数字列指定成int32或float32内存占用直接减半。这个优化技巧很土但真的很有效。5.3 经纬度出界的排查地图热力图上我发现有几个高亮热点落在海里明显不合理。排查后发现是设备上报的经纬度偶尔会出现零值或者错误值。这种异常我用一个简单的范围过滤就能拦住df df[(df[lat] 20) (df[lat] 50)] df df[(df[lng] 100) (df[lng] 125)]按国内绝大多数城市的经纬度范围收紧区间经验值直接写进去比用“是否在海陆边界”的复杂判断更省事。这类过滤逻辑建议放在清洗模块里跟车速过滤并列这样每次重跑数据都能自动生效。5.4 环境配置问题最后说一个容易被忽视的坑Python 环境。很多同事在自己电脑上跑这些代码时报错都出在环境上而代码本身没问题。常见的有几个Python 装到了带中文或空格的路径下导致一些底层依赖编译失败直接全局装包版本冲突后项目跑不起来用 VS Code 写代码但没选中正确解释器。我的建议是项目一开始就用python -m venv venv创建独立虚拟环境然后pip install pandas matplotlib scikit-learn folium flask所有依赖锁进一个 requirements.txt。别嫌这个步骤啰嗦交通数据分析这种项目数据量一大依赖版本稍微不一致结果都对不上排查起来才是最麻烦的。虚拟环境能帮你把问题范围缩到最小。这份代码之外我更想分享的一点体会项目做完之后我复盘了一下真正让我觉得有价值的不是画出了好看的图也不是接口能跑通而是我建立了一套“先清洗、再指标、后可视化”的分析习惯。交通数据的特点是脏、乱、但是量大、规律性强只要前 70% 的数据整理功夫做扎实后面出活非常快。我强烈建议你在自己电脑上把整套流程跑一遍哪怕先用模拟数据。刚开始可能觉得清洗很枯燥但等你拿到真实数据发现清洗逻辑能直接复用的时候你会回来感谢自己当初没有跳过这一步。如果做的时候碰到奇怪的问题欢迎按我上面列的那些坑去排查大概率你遇到的就是其中之一。
