简介这是一份面向Python初学者与城市数据分析爱好者的轻量级交通分析脚本资源聚焦于利用公开数据源计算并呈现城市交通拥堵指数解决实际场景中交通态势量化评估的学习需求。压缩包共2个文件1个核心Python脚本1个配套HTML页面总大小仅2KB结构简洁主脚本涵盖requests网络请求、pandas数据清洗、NumPy数学运算及自定义拥堵指数公式实现HTML文件用于本地可视化结果展示便于快速运行与调试。已有1815人学习下载体现了其在入门级数据分析项目中的实用热度。读者可直接复用该脚本框架接入真实交通API或本地CSV数据掌握从数据获取、指标建模到结果输出的完整闭环同时获得含异常处理、控制台交互与基础绘图逻辑的可扩展代码模板。1. 这不是“算个平均值”python统计城市交通拥堵指数.py是一套可落地的交通态势感知脚本专为交管部门、智慧城市项目组和交通研究者设计你手头有一份 GPS 轨迹数据比如出租车/网约车 5 秒采样点或者某市公开的浮动车速度 CSV 文件但不知道怎么从原始坐标、时间戳、瞬时速度里挖出真正能汇报、能对比、能驱动信号优化的「拥堵指数」——不是简单求个路段平均速度而是要还原真实通行效率衰减程度、识别早晚高峰结构性堵点、支撑周环比分析。python统计城市交通拥堵指数.py就是这样一个轻量级但生产可用的统计脚本它不依赖 ArcGIS 或商业平台用标准 Python 生态pandas geopandas numpy完成「路网匹配 → 分时段分路段聚合 → 拥堵指数计算 → 空间可视化」全链路。它解决的是「数据有、工具全、但结果没人信」这个典型痛点——指数定义符合《GB/T 33171-2016 城市道路交通运行评价指标》中自由流速度比法TTR核心逻辑同时兼容实际业务中常见的非标准数据格式如缺失经纬度、无路名字段、时间戳乱序。适合刚接手交通数据的工程师快速验证数据质量也适合已有 GIS 平台的团队补位做离线批量计算。2. 从原始轨迹到拥堵指数四步不可跳过的数据处理链2.1 路网匹配为什么不用高德/百度 API本地化匹配才是可控关键很多团队第一反应是调用地图 SDK 做轨迹纠偏但实际落地时会卡在三个硬伤API 调用量限制导致批量失败、返回结果无路网拓扑关系、无法复现历史计算今天匹配结果和昨天不一致。我们采用基于 RTree Shapely 的本地路网空间匹配核心是把道路中心线转为 GeoDataFrame构建空间索引后对每个 GPS 点做最近线段查找。import geopandas as gpd from shapely.geometry import Point, LineString import rtree # 加载预处理好的路网GeoJSON 或 Shapefile必须含 road_id, length_m 字段 roads gpd.read_file(beijing_road_network.geojson) roads roads.to_crs(epsg4326) # 统一 WGS84 roads[geometry] roads[geometry].buffer(0) # 修复无效几何 # 构建 RTree 索引加速空间查询 idx rtree.index.Index() for i, geom in enumerate(roads.geometry): if geom.is_valid: idx.insert(i, geom.bounds) # 对单个 GPS 点匹配最近道路示例函数 def match_point_to_road(point: Point, roads_gdf: gpd.GeoDataFrame, rtree_idx: rtree.index.Index, max_dist100): # 先用 RTree 快速筛选候选道路bounding box 内 candidates list(rtree_idx.intersection(point.bounds)) if not candidates: return None # 计算点到每条候选道路的垂直距离Shapely distances [] for idx_i in candidates: road_geom roads_gdf.iloc[idx_i].geometry if road_geom.is_valid: dist point.distance(road_geom) if dist max_dist: distances.append((dist, idx_i)) if not distances: return None _, best_idx min(distances, keylambda x: x[0]) return roads_gdf.iloc[best_idx][road_id]提示max_dist100是关键参数——设太小如 30 米会漏匹配高架桥下或立交匝道设太大如 500 米会导致主干道误匹配到平行支路。北京五环内建议 80–120 米深圳南山科技园建议 60–90 米需结合当地路网密度实测校准。2.2 分时段分路段聚合用pandas.Grouper抓住通勤节律而非简单切片拥堵不是静态值是时间维度上的动态衰减过程。直接按小时切片会抹平早高峰7:30–8:45这种尖峰特征。我们采用滑动窗口 自适应分组时间粒度默认 15 分钟足够捕捉潮汐变化又避免噪声放大分组逻辑pd.Grouper(keytimestamp, freq15T, closedleft, labelleft)关键增强对每个(road_id, time_bin)组不仅计算平均速度还保留std_speed标准差、pct_under_10kmh低于 10 km/h 车辆占比、sample_count有效样本数import pandas as pd from datetime import datetime, timedelta # 假设 df_gps 已含 road_id, speed_kmh, timestampdatetime 类型 df_gps[timestamp] pd.to_datetime(df_gps[timestamp]) df_gps df_gps.sort_values([road_id, timestamp]) # 按 15 分钟分组左闭右开区间 grouped df_gps.groupby([ road_id, pd.Grouper(keytimestamp, freq15T, closedleft, labelleft) ]) # 聚合指标注意speed_kmh 用 median 防异常值污染不用 mean agg_result grouped.agg( avg_speed(speed_kmh, median), std_speed(speed_kmh, std), sample_count(speed_kmh, count), pct_slow(speed_kmh, lambda x: (x 10).mean() * 100), max_speed(speed_kmh, max) ).reset_index() # 过滤低置信度组样本数 5 或标准差 30 km/h 视为数据异常 agg_result agg_result[ (agg_result[sample_count] 5) (agg_result[std_speed].fillna(0) 30) ]参数说明freq15T中的T表示 minuteclosedleft意味着[07:00, 07:15)归入 07:00 组labelleft让时间列显示为区间的左端点便于后续 join 和排序。若你的数据源是固定整点上报如每 5 分钟一次可改用freq5T但需同步调整max_dist和sample_count下限。2.3 拥堵指数计算TTR 法本地化实现拒绝黑匣子公式国家标准 GB/T 33171-2016 推荐的 TTRTravel Time Ratio法本质是拥堵指数 实际行程时间 / 自由流行程时间 × 100%但自由流速度不能靠查表——我们用历史同期分位数法动态生成取过去 7 天同一时段±15 分钟该路段所有速度样本的 85% 分位数作为自由流速度基准。# 假设已有过去 7 天的聚合结果 df_historical含 road_id, time_bin, avg_speed # 按 road_id time_bin 分组计算 85% 分位数 free_flow_speed df_historical.groupby([road_id, time_bin])[avg_speed].quantile(0.85).reset_index(nameff_speed_85p) # 当前日数据 df_current 与自由流基准 merge df_merged pd.merge(df_current, free_flow_speed, on[road_id, time_bin], howleft) # 计算 TTR 指数防除零ff_speed_85p 5 km/h 时设为 5 df_merged[ff_speed_safe] df_merged[ff_speed_85p].clip(lower5) df_merged[ttr_index] (df_merged[ff_speed_safe] / df_merged[avg_speed]).round(2) # 指数分级国标推荐≤1.2 为畅通1.2–1.5 为缓行1.5 为拥堵 df_merged[congestion_level] pd.cut( df_merged[ttr_index], bins[0, 1.2, 1.5, float(inf)], labels[畅通, 缓行, 拥堵] )为什么不用固定自由流速度北京长安街早高峰自由流是 45 km/h而深圳深南大道同路段是 60 km/h同一路段雨天自由流可能下降 20%。用历史分位数法指数天然带时空自适应性避免“常年报拥堵”的假阳性。3. 避坑这 4 类错误让指数失真90% 的初学者都踩过3.1 现象早高峰指数普遍偏低1.0甚至出现 0.8原因GPS 采样点未剔除静止车辆停车、等红灯、临时停靠导致avg_speed被严重拉低而自由流基准ff_speed_85p又因包含大量静止样本被压低最终 TTR 值反常偏小。解决在轨迹清洗阶段强制过滤speed_kmh 1且持续时间 60 秒的连续点段对单次采样speed_kmh 0的点用前后 5 秒均值插补而非直接丢弃。3.2 现象某条主干道全天指数恒为 1.0无波动原因路网匹配时该路段 geometry 为空.is_valid False所有 GPS 点匹配失败后road_id为None后续分组被自动 drop最终该路段无任何输出。解决在gpd.read_file()后立即执行roads roads[roads[geometry].notna() roads[geometry].apply(lambda g: g.is_valid)]并用roads[~roads[geometry].apply(lambda g: g.is_valid)]单独导出问题路段人工修复。3.3 现象周五下午指数突增但实地无拥堵原因数据源中周五下午存在大量网约车空驶轨迹司机接单前绕行其速度分布偏离真实通行特征拉低了avg_speed但自由流基准ff_speed_85p未排除空驶样本导致 TTR 虚高。解决引入载客状态标识如有若无用speed_kmh 60 km/h 且持续 30 秒判定为空驶段聚合前过滤或对pct_slow 80% 的组额外检查max_speed是否 50 km/h是则标记为“疑似空驶干扰”。3.4 现象跨日期计算时pd.Grouper把 23:59 和次日 00:00 归入不同组原因freq15T默认以自然日为界23:59:59属于当日最后一组00:00:00属于次日第一组导致早高峰起始段被割裂。解决改用pd.date_range手动构造连续时间桶再pd.cut映射# 构造覆盖全周期的 15 分钟桶从首条记录前 1 小时到末条后 1 小时 full_range pd.date_range( startdf_gps[timestamp].min() - pd.Timedelta(1H), enddf_gps[timestamp].max() pd.Timedelta(1H), freq15T ) df_gps[time_bin] pd.cut(df_gps[timestamp], binsfull_range, rightFalse, labelsfull_range[:-1])4. 输出与验证不只是生成 CSV而是构建可追溯的拥堵证据链4.1 三层输出结构满足不同角色需求输出层级文件名示例核心内容使用场景原始层raw_match_log_20240520.csv每条 GPS 点匹配的road_id、匹配距离、原始坐标数据质量审计、匹配算法调优指标层ttr_daily_summary_20240520.csv(road_id, time_bin, ttr_index, congestion_level, sample_count)交管值班员日报、信号配时调整依据空间层ttr_geo_20240520.geojson每条道路 geometry ttr_index属性支持 QGIS/Mapbox 直接加载领导汇报 PPT、大屏可视化底图生成ttr_geo_20240520.geojson的关键代码# 将指标结果 merge 回路网 GeoDataFrame roads_with_ttr roads.merge( df_merged[[road_id, time_bin, ttr_index, congestion_level]], left_onroad_id, right_onroad_id, howleft ) # 按时间切片例如取早高峰 07:30–08:30 peak_mask roads_with_ttr[time_bin].between(2024-05-20 07:30:00, 2024-05-20 08:30:00) roads_peak roads_with_ttr[peak_mask].copy() # 保存为 GeoJSON确保 CRS 正确 roads_peak.to_file(ttr_geo_20240520.geojson, driverGeoJSON, crsEPSG:4326)注意to_file(..., crsEPSG:4326)必须显式指定 CRS否则 QGIS 加载时会错位。若路网原始 CRS 是EPSG:32650UTM 50N需先roads roads.to_crs(epsg4326)转换。4.2 指数可信度交叉验证三板斧光跑出数字不够得让人信。我们用以下三法交叉验证同比校验取上周同一天同一时段计算ttr_index变化率若全城 20% 路段变化率 ±30%触发人工核查大概率是数据源异常空间一致性校验对相邻路段共享端点计算|ttr_i - ttr_j| / max(ttr_i, ttr_j)若 0.5 且sample_count均 20则标记为“空间跳跃异常”需检查匹配是否跨路物理合理性校验对ttr_index 2.0的路段反查其avg_speed是否 15 km/h 且pct_slow 70%否则视为计算溢出如自由流基准被异常值污染该组结果置为NaN。# 物理校验示例 df_validated df_merged.copy() mask_unphysical ( (df_merged[ttr_index] 2.0) ~((df_merged[avg_speed] 15) (df_merged[pct_slow] 70)) ) df_validated.loc[mask_unphysical, ttr_index] np.nan df_validated.loc[mask_unphysical, congestion_level] 待核查5. 进阶技巧用congestion_index.py做实时预警与归因分析5.1 实时滚动计算从“日报”升级为“分钟级哨兵”把脚本改造成守护进程每 5 分钟读取新入库的 GPS 数据只计算最新一个时间桶如2024-05-20 14:20:00–14:35:00并与前 30 分钟均值比对# 实时模式核心逻辑伪代码 last_bin get_latest_time_bin() # 如 2024-05-20 14:20:00 current_data load_new_gps_since(last_bin - pd.Timedelta(35T)) # 仅聚合 last_bin 这一组 current_agg current_data.groupby(road_id).agg({ speed_kmh: [median, lambda x: (x10).mean()], }).round(2) # 与历史基线比取过去 7 天同 bin 的 ttr_index 均值 baseline load_baseline(last_bin) alert_roads current_agg.merge(baseline, onroad_id, howinner) alert_roads[delta_ttr] alert_roads[ttr_index] - alert_roads[baseline_ttr] # 触发预警delta_ttr 0.3 且 sample_count 10 alert_list alert_roads[alert_roads[delta_ttr] 0.3].sort_values(delta_ttr, ascendingFalse)部署提示用systemd启动 Python 守护进程stdout 重定向到/var/log/congestion_alert.log配合logrotate防止日志爆炸预警结果写入 Redis 的 Sorted Set前端轮询获取。5.2 拥堵归因不只是“哪里堵”而是“为什么堵”单纯指数无法指导治理。我们在ttr_index基础上叠加两个归因维度归因类型计算逻辑输出字段业务价值流量驱动型pct_slow 75%且sample_count同比 40%cause 流量激增提示扩容或限行事故驱动型std_speed 5 km/h且max_speed 20 km/hcause 局部瘫痪调度交警/拖车信号驱动型avg_speed在 20–35 km/h 区间且pct_slow波动剧烈5 分钟内升降 30%cause 信号配时失配推送至信号优化平台# 归因判定逻辑接续 df_merged df_merged[cause] 未知 mask_flow (df_merged[pct_slow] 75) (df_merged[sample_count] df_merged[sample_count].shift(7*96).fillna(0) * 1.4) mask_accident (df_merged[std_speed] 5) (df_merged[max_speed] 20) mask_signal ( (df_merged[avg_speed].between(20, 35)) (df_merged[pct_slow].rolling(window4).std() 30) # 4 个 15 分钟桶1 小时内标准差 ) df_merged.loc[mask_flow, cause] 流量激增 df_merged.loc[mask_accident, cause] 局部瘫痪 df_merged.loc[mask_signal, cause] 信号配时失配5.3 我的血泪经验别急着画热力图先做“拥堵指纹”聚类做过 12 个城市项目后我养成一个习惯不直接渲染全城热力图而是先对road_id做拥堵时序指纹聚类。用sklearn.cluster.KMeans对每条路 24 小时的ttr_index序列96 维向量聚类你会发现第 1 类早高峰单峰型CBD 主干道→ 适配绿波带优化第 2 类早晚双峰午间小峰型高校周边→ 适配错峰信号第 3 类全天高位型施工围挡路段→ 适配绕行诱导# 指纹聚类示意需先 pivot 成 road_id × time_bin 宽表 pivot_df df_merged.pivot(indexroad_id, columnstime_bin, valuesttr_index).fillna(0) from sklearn.cluster import KMeans kmeans KMeans(n_clusters4, random_state42) pivot_df[cluster] kmeans.fit_predict(pivot_df.values) # 输出每类代表路段用于报告 cluster_rep pivot_df.groupby(cluster).apply( lambda x: x.index[0] # 取每类第一条路 )这一步多花 20 分钟却能让后续所有分析——从大屏展示到治理方案——直击要害。技术不是堆参数而是帮人看清问题结构。希望帮到你。本文还有配套的精品资源点击获取