简介一套基于 Python 的高德地图交通数据分析项目适合交通物流与数据分析方向的初学者、开发者参考。项目聚焦早高峰通勤场景通过采集指定路段实时路况数据按时间段统计拥堵变化并以地图可视化呈现帮助用户合理选择出门时间缓解“早出门仍迟到、晚出门却畅通”的困扰。压缩包共47个文件大小约1.62MB以15个py模块为逻辑核心配合地图数据文件、前端页面及项目说明文档具备数据获取、地理编码、拥堵绘图与展示等完整结构。包内附用户密钥配置说明、城市adcode对照、依赖清单和README便于快速复现读者可参照说明修改路段与密钥生成自己的通勤时段拥堵分析。目前已有409人学习下载适合想系统了解高德地图接口调用和交通数据可视化流程的读者直接学习或二次开发。1. 高德地图的交通数据分析在做一件什么事早上八点导航屏幕上显示“前方畅通”等你开过去却一脚刹车踩到终点——这不是导航数据不准而是你把“数据”和“分析”当成了同一件事。基于Python实现高德地图的交通数据分析就是要把高德开放接口里的实时路况、路径规划结果变成一份你能按小时、按路段、按方向去拆解的干净数据集再用几张图和几个指标回答“这条路到底什么时候堵、为什么堵、堵多久”。这类被标记为“高分项目”的源码包真正值钱的不是那几百行请求代码而是把数据从接口搬到本地库、再从本地库画成图表的这条完整链路。适合正在做课程设计或毕设、想用真实交通数据做验证的开发者也适合想把路况数据接进大屏或算法实验的数据方向从业者。下面这条链路我会从接口选型、定时拉取、清洗入库、指标计算一路写到避坑全部按可复现的标准来。2. 先用对接口高德交通数据的四个取数通道做交通数据分析代码写得再漂亮接口选错了也是白费。高德开放平台上挂着几十个Web服务API但真正和“交通数据分析”强相关的只有四个。这一章先把选型讲清楚再给一个最小可跑的交通态势请求让你对返回的数据长什么样有个直观印象。2.1 交通态势、路径规划、地理编码、逆地理编码怎么选高德地图的交通数据分析数据源主要来自四个服务API交通态势查询、路径规划、地理编码、逆地理编码。很多人一上来就盯着交通态势但实际项目里路径规划和地理编码往往才是把数据盘活的关键。API服务入参核心返回核心在交通分析中的作用交通态势查询矩形区域或道路名道路名称、拥堵等级、速度片段采集实时路况的原始数据路径规划起点、终点坐标路线距离、预计耗时、导航路径坐标串推算两点间旅行时间补路段空白地理编码结构化地址文本经纬度坐标把以地址描述的起点转成查询坐标逆地理编码经纬度坐标省市区、街道名称、商圈把路段坐标还原成可读的行政区信息选型逻辑是这样的交通态势是主体一次请求传一个矩形接口返回矩形框内各条道路的拥堵等级和分段速度路径规划负责“起点到终点”的旅行时间对比特别是郊区道路没有路况覆盖时只能用路径规划结果去补地理编码和逆地理编码是辅助处理两类场景——把地址文本转成坐标框用于拉数据以及把拉回来的轨迹点还原成道路所属的行政区方便按区域聚合统计。我一般会先把“要回答什么问题”写下来再决定要不要接路径规划。如果你只是做“城市早晚高峰拥堵趋势”交通态势一个接口就够了如果要做“某几条通勤路线的耗时对比”必须加路径规划如果要做“哪个城区的路况最差”逆地理编码那一环省不掉。接口不在多够用、能闭环就行。2.2 申请Key和第一个请求交通态势API的最小调用不管用什么接口第一步都是去高德开放平台创建一个应用勾选Web服务类目拿到一个Key。这个Key就是一个字符串请求时放在参数里带过去。申请过程不需要写代码但要注意一点个人开发者需要实名认证后才会有比较稳定的调用额度课程设计阶段早几天申请别拖到验收前一天才去认证。拿到Key之后先跑通一个最小请求。交通态势查询有两种入口按矩形查和按道路名查。道路名查询适合单条路的监控矩形查询适合做区域分析我们后面整套采集逻辑都基于矩形查询。下面是按矩形查询的最小实现import requests def fetch_rectangle_traffic(rectangle, key): 按矩形区域拉取实时路况 rectangle 格式: 左下经度,左下纬度;右上经度,右上纬度 例如: 116.30,39.80;116.42,39.95 url https://restapi.amap.com/v3/traffic/status/rectangle params { key: key, rectangle: rectangle, level: 6, # 6高速高架1高速7快速路 extensions: all, # all返回分段速度base只返回概要 } resp requests.get(url, paramsparams, timeout10) resp.raise_for_status() return resp.json() # 示例拉取一个小矩形区域替换成你自己的Key data fetch_rectangle_traffic(116.30,39.80;116.42,39.95, your_key_here) # 看一下返回结构的关键层级 roads data[trafficinfo][roads] for road in roads[:3]: print(road.get(name), road.get(status)) for seg in road.get(segments, [])[:2]: print( 分段状态:, seg.get(status), 速度:, seg.get(speed))这段代码里有一个易错点rectangle参数的分隔符是英文分号前后各是一个“经度,纬度”对左下角和右上角。level6是我的常用值它会把高速和高架一起返回适合城市快速路网分析如果只要某一条具体道路用道路名入口更直接后面避坑章节会讲两者在返回值上的差异。extensionsall这个参数很多人会漏掉。默认的base只返回道路级的概要状态一个道路一个状态值拿去做分析会发现维度不够用设成all之后才会返回segments分段数组每个段有独立的状态、速度和坐标串分析时的样本量就从“道路数”变成了“路段数”数据粒度完全不同。我第一次跑这个接口时用的默认值入库后按天聚合发现“畅通”占比高得离谱就是因为一条3公里的路只有一个状态拥堵段被畅通段平均掉了。跑通这一步之后你已经能看到真实的交通数据了。接下来要考虑的是怎么把一个城市的所有道路都拉回来并且让这份数据能支撑后续的时间维度分析。3. 把路况拉下来存起来从JSON到SQLite的完整链路拿到了第一个矩形区域的数据下一步就是把采集范围从一个小矩形扩大到整个目标区域并且按一定时间频率持续拉取。这一章解决两个问题怎么拉才能不超限、不遗漏拉回来的JSON怎么清洗成一行行干净的结构化记录。3.1 分网格定时拉取为什么不能一次请求覆盖全城交通态势接口虽然支持矩形查询但矩形不是画得越大越好。矩形太大时接口返回的道路片段会被截断或稀疏化有些支路直接不返回而且单次请求的数据量过大解析和入库都会变慢。常见做法是把这个大矩形切成几百个小网格循环请求然后把结果合并。网格大小怎么定我一般用0.05度约等于5公里见方。这个尺寸在城市里能覆盖4到10条主干道返回数据量适中。如果网格太小比如0.01度请求次数会膨胀几十倍Key的日配额撑不过半天网格太大比如0.2度支路数据丢得厉害。下面是网格拆分和批量拉取的实现import time import sqlite3 import requests def split_rectangle(min_lng, min_lat, max_lng, max_lat, step0.05): 把一个大矩形切成 step 度见方的小网格 这里用示例坐标 0,0 到 0.1,0.1实际换成目标城市边界 cells [] lng min_lng while lng max_lng: lat min_lat while lat max_lat: right min(lng step, max_lng) top min(lat step, max_lat) cells.append(f{lng:.4f},{lat:.4f};{right:.4f},{top:.4f}) lat step lng step return cells def collect_all(cells, key, sleep_seconds0.2): 按网格循环拉取返回所有记录 all_roads [] for idx, cell in enumerate(cells): try: payload fetch_rectangle_traffic(cell, key) roads payload.get(trafficinfo, {}).get(roads, []) for road in roads: road[_cell] cell road[_batch] idx all_roads.extend(roads) print(f[{idx 1}/{len(cells)}] {cell} - {len(roads)} roads) except Exception as e: print(f[{idx 1}/{len(cells)}] {cell} 失败: {e}) time.sleep(sleep_seconds) # 每个网格之间留间隔防止触发频率限制 return all_roads这段代码里sleep_seconds0.2是我习惯的请求间隔。高德的Web服务API对单Key有QPS限制连续快速请求容易触发频率限制加一个0.2到0.5秒的间隔一整轮几百个网格拉完也就多花几分钟却能避开大部分限流问题。try...except在这里不是可选项——网格拉取是批量任务一个网格失败不应该中断整轮采集记录下来下一轮补拉就行。关于“什么时候拉”的问题交通数据分析至少要覆盖早高峰和晚高峰两个时段。我常用的策略是每5分钟拉一轮从早上7点到晚上9点连续采集。5分钟这个频率能捕捉到拥堵的起落过程又不至于让数据量失控。一天下来按100个网格算每轮100次请求一天168轮大约是16800次请求。这个量对个人开发者Key来说偏大所以实际项目里我会把范围缩小到几条重点通勤走廊或者把频率降到10分钟一轮。配额不是用来被一次跑完的要留着给整个项目周期。3.2 清洗、标准化、入库坐标系和字段口径先定死拉回来的JSON直接存文件不是不行但后面做时间聚合时你会发现查询和分析都很难受。正确做法是尽快摊平成结构化表存进SQLite。SQLite单机足够用不需要额外部署服务交作业或者做Demo都很方便。清洗时有三件事必须处理一是只保留分析要用的字段原始JSON另外存档二是把坐标和时间字段格式统一三是处理缺失的字段值。下面是我常用的解析和建表逻辑import sqlite3 from datetime import datetime def parse_roads_to_rows(roads, collect_ts): 把接口返回的 roads 列表解析成一条条路段记录 rows [] for road in roads: road_name road.get(name, 未知道路) direction road.get(direction, ) for seg in road.get(segments, []): status seg.get(status) speed seg.get(speed) polyline seg.get(polyline, ) # 速度字段可能为空或非法值统一转成 float try: speed_val float(speed) if speed not in (None, ) else None except ValueError: speed_val None rows.append(( road_name, direction, status, speed_val, polyline, collect_ts, )) return rows def init_db(db_pathtraffic.db): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS traffic_raw ( id INTEGER PRIMARY KEY AUTOINCREMENT, road_name TEXT, direction TEXT, status INTEGER, speed REAL, polyline TEXT, collect_ts TEXT, hour INTEGER ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_ts ON traffic_raw(collect_ts) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_road ON traffic_raw(road_name) ) conn.commit() return conn def insert_rows(conn, rows, collect_ts): 批量插入hour 字段单独算好再入库 dt datetime.fromisoformat(collect_ts) hour dt.hour data [r (hour,) for r in rows] conn.executemany( INSERT INTO traffic_raw (road_name, direction, status, speed, polyline, collect_ts, hour) VALUES (?, ?, ?, ?, ?, ?, ?) , data ) conn.commit()这里我多加了一个hour字段直接在入库时把小时数字算好。这样后续做“按小时聚合”时不需要每次都对字符串时间做解析查询会快很多。status字段的类型我用整数存高德交通态势的状态码一般是1、2、3、4对应畅通、缓行、拥堵、严重拥堵用整数存比存中文更省空间后期要用中文再做映射。polyline 字段是这个数据集里最值钱也最容易被忽略的东西。它是一长串经纬度点用来描述路段的实际走向。存这个字段有两个目的一是后续可以用它还原路段的实际空间位置画在地图上二是遇到“同一道路名出现在两个不相连的网格里”这种情况靠名称去重会误伤必须靠坐标串判断是否同一路段。我踩过这个坑某条东西向主干道被切成四段名称完全一样去重时直接把我整懵了后来才发现要用坐标串的首尾点去做连接判断。入库之后每次采集的数据都叠加在traffic_raw表里collect_ts索引能支持按时间段切片。这一章做完你已经拥有了一个“持续增长的交通路况明细表”。下一章要解决的问题是从这一堆明细记录里算出有说服力的交通指标。4. 算对指标拥堵等级占比和平均车速的实操口径数据落库只是第一步验收时真正被追问的是“你的结论从哪来”。这一章给两个最常用的分析口径拥堵等级占比和平均车速并解释为什么很多代码里算出来的“平均速度”会骗人。4.1 平均速度、拥堵指数、等级占比哪种口径更抗噪“这条路平均速度30码”和“这条路30%时间在拥堵”是两个完全不同的问题。交通态势接口返回的数据里speed是路段级的速度status是路段级的拥堵等级。直接用速度求平均有一个问题速度缺失值很多。特别是支路和匝道要么没返回速度要么返回00一进来平均值直接被拉垮。我常用的口径是“拥堵等级占比”。先把每条路段的 status 映射成等级名称再按小时统计各等级出现的频率。这个口径对缺失值不敏感——status 缺失了可以丢弃但不会通过“0值”污染平均值。下面这段代码把明细表转成每个小时各拥堵等级的占比import pandas as pd def load_data(db_pathtraffic.db): conn sqlite3.connect(db_path) df pd.read_sql_query( SELECT road_name, status, speed, collect_ts, hour FROM traffic_raw, conn ) conn.close() return df STATUS_MAP {1: 畅通, 2: 缓行, 3: 拥堵, 4: 严重拥堵} def level_ratio_by_hour(df): 按小时统计每个拥堵等级的占比返回宽表 df df.copy() df df[df[status].notna()] df[level_name] df[status].map(STATUS_MAP) # 按小时和等级计数 cnt df.groupby([hour, level_name]).size().reset_index(namecount) # 转成 hour 为行、等级为列的宽表 pivot cnt.pivot(indexhour, columnslevel_name, valuescount).fillna(0) # 计算占比每一小时的行总和是100% ratio pivot.div(pivot.sum(axis1), axis0) return ratio df load_data() hourly_ratio level_ratio_by_hour(df) print(hourly_ratio.head())这段代码的核心是pivot和div两步。pivot把“长表”变成“宽表”每个小时一行四个等级各一列div用每一行的总和去归一化得到的就是“这个小时里畅通路段占了多少比例、拥堵占了多少比例”。注意pivot之后会丢列顺序如果某一天某个等级一次都没出现fillna(0)会补上但列顺序需要你自己维护。更保险的做法是提前声明一个固定的列顺序[畅通, 缓行, 拥堵, 严重拥堵]然后用reindex对齐。如果确实要用平均速度我建议只对status为2和3的路段算平均并且把速度为零的记录排除掉。这相当于只看“在动但不太快”的路段能反映缓行带的通行效率比全量平均稳定得多。速度口径在“同一道路名跨网格”时会偏小因为同一个路名可能被切成了多段每段速度不同平均之后掩盖了局部的拥堵点。4.2 画一张24小时拥堵趋势图数据计算完之后可视化是让结论“可验收”的关键一步。用 matplotlib 画一张24小时趋势图横轴是小时纵轴是各等级占比一眼就能看出早晚高峰的形态。下面是上一节数据的直接可视化import matplotlib.pyplot as plt def plot_ratio_trend(hourly_ratio, outputtraffic_trend.png): 画小时级拥堵等级占比趋势图 fig, ax plt.subplots(figsize(10, 5)) # 固定颜色顺序避免每次运行颜色漂移 colors {畅通: #2ecc71, 缓行: #f39c12, 拥堵: #e74c3c, 严重拥堵: #8e44ad} for level in [畅通, 缓行, 拥堵, 严重拥堵]: if level in hourly_ratio.columns: ax.plot(hourly_ratio.index, hourly_ratio[level], labellevel, colorcolors[level], markero, markersize3) ax.set_xlabel(hour) ax.set_ylabel(ratio of road segments) ax.set_title(Traffic Level Ratio by Hour) ax.legend() ax.grid(alpha0.3) fig.tight_layout() fig.savefig(output, dpi160) print(saved to, output) plot_ratio_trend(hourly_ratio)画图参数的设置有一点值得说markersize3是为了让每个小时的点不喧宾夺主线条是主视觉dpi160是导出论文和报告可用的清晰度默认的100dpi在放大后会有锯齿。颜色我固定用蓝黄红紫四色和导航软件对拥堵的配色习惯保持一致看图的人不需要额外解释就能理解。如果你拿到的时间粒度是5分钟一条想把它聚合成小时趋势可以先把collect_ts转成datetime再用dt.floor(H)把时间向下取整到小时再按这个新的时间列做groupby。5分钟粒度的小时占比曲线会更毛糙能看到更多波动细节但作为“趋势图”反而干扰判断。我实际分析时5分钟数据用于查具体事件小时数据用于看整体趋势两套口径并存不互相替代。这一章做完你已经能从数据集里产出一张趋势图和一张占比表。第5章要讲的是我在这条链路上反复掉进去的五个坑每一个都曾经让我的数据结果看起来“合理但经不起追问”。5. 避坑高德交通数据拉取与分析中常见的5个坑这一章专门写坑全部是真实踩过的按“现象、原因、解决”三条写你照着排查能少走很多弯路。5.1 坑一配额超限报错比想象中来得更早现象采集脚本跑不到半天请求返回里出现类似“USER_DAILY_QUOTA_EXCEEDED”的错误码之前还能正常返回的数据突然全部失败。原因高德的Web服务API对个人开发者的Key有日配额和每秒并发限制。交通态势接口一次请求消耗配额是按次计算的网格一多、频率一快配额消耗速度超出预期。很多课程设计项目喜欢用循环一次拉全城几百个网格一轮下去配额就见底了。解决第一次跑通后先确认三件事当前Key的配额数值你在高德控制台能查到采集范围是不是真的需要那么大请求间隔是不是太短。我后来的标准做法是先缩小网格范围只保留目标区域内道路密度高的部分用10到15分钟一轮代替5分钟一轮把日请求量控制在配额的一半以内剩下的一半留给补拉和临时手工查询。5.2 坑二坐标偏移导致网格错位路段落不到网格里现象按自己画的矩形去拉数据返回的路段坐标串画出来会整体偏离矩形边界有些路段的坐标甚至和相邻网格的边界交叉看起来像是“拉错了位置”。原因高德地图用的是GCJ-02坐标体系而很多开源地图工具和GPS原始数据默认是WGS-84体系两者相差几百米。如果你拿着WGS-84边界去请求接口返回的GCJ-02坐标自然对不上你用于划分网格的边界框。解决坐标系问题不要想着事后纠正应该在第一步就统一。我习惯直接以高德的GCJ-02坐标系来设定网格边界也就是先在另一个地图工具上选好目标城市的大致四个角点再用一次逆地理编码验证边界上的坐标点归属于预期行政区确认边界没偏出去再切网格。这里不是几行代码能解决的问题而是要建立一套认知在走高德数据链路时所有坐标一律默认为GCJ-02不混用。5.3 坑三拥堵等级在“畅通”和“缓行”之间反复横跳现象同一路段同一时间窗口前后两轮拉取返回的status在1和2之间跳动导致5分钟粒度的等级占比曲线像心电图看不出稳定趋势。原因交通态势接口的状态是融合了多源数据的结果在早晚高峰的切换时段系统对路况状态的判断本来就存在迟滞和抖动。这不是你的代码问题是数据源本身有这个特性。解决不要用单次状态做结论给status做时间维度的滚动平滑。最简单的方法是按“最近N轮拉取”做众数投票比如连续6轮数据里4轮以上是2才判定为缓行。如果不想写复杂逻辑也可以直接在小时聚合那一步做窗口平均把5分钟抖动滤掉。我在前面的level_ratio_by_hour里小时聚合本身就有平滑效果但如果做5分钟粒度分析平滑逻辑必须单独加。5.4 坑四extensionsbase时返回的detail字段内容不足现象拉回来的JSON里roads下面有detail但detail里只有一句文字描述没有分段速度画不了路段级的速度分布。原因这是请求参数里extensions设置的锅。base模式只给道路级概要all模式才有segments分段数组。部分网上流传的示例代码里用的是默认值或base跟着抄的人就拿到了残缺的数据结构。解决在请求参数里显式设置extensionsall并且在解析代码里做防御性判断如果segments为空就跳过这条道路不写进库里。判断逻辑在我的parse_roads_to_rows里已经有了但如果你的代码是从老项目拷的记得检查请求参数不要只看解析函数。5.5 坑五collect_ts时区不统一按小时聚合出现“半夜高峰”现象把采集时间字段collect_ts直接存成字符串后来发现凌晨2点到4点的拥堵占比奇高明显不合理。原因采集脚本可能部署在不同时区的机器上或者中途改过服务器时间。如果有的行是UTC时间写入有的行是本地时间写入按hour字段聚合时同一个小时里混进了两个时区的记录凭空造出了假高峰。解决入库前统一将时间格式化为带时区信息的字符串或者统一按本地时间算好hour字段再入库。我推荐在insert_rows里面直接算hour但前提是collect_ts的时区在脚本里写死。更稳妥的做法是在数据加载到pandas后用pd.to_datetime(..., utcTrue).dt.tz_convert(Asia/Shanghai)强制统一时区再做小时聚合。这里用Asia/Shanghai是因为我的分析对象是国内城市换成你目标城市的IANA时区名即可不要依赖服务器默认时区。6. 进阶把一次性脚本变成可复用的交通数据采集框架前面几章的代码能跑通但离“可复用”还差一步。我习惯把采集逻辑封装成一个类把Key、网格边界、请求间隔、入库路径都作为配置项这样换一个城市或者换一批重点路段时不需要改主流程代码。下面是一个极简框架的骨架import requests import sqlite3 import time import json class TrafficCollector: def __init__(self, key, db_path, rectangle, step0.05, sleep_seconds0.2): self.key key self.db_path db_path self.cells self._split_rectangle(*rectangle, step) self.sleep_seconds sleep_seconds def _split_rectangle(self, min_lng, min_lat, max_lng, max_lat, step): cells [] lng min_lng while lng max_lng: lat min_lat while lat max_lat: cells.append( f{lng:.4f},{lat:.4f}; f{min(lng step, max_lng):.4f},{min(lat step, max_lat):.4f} ) lat step lng step return cells def run_once(self): rows [] for cell in self.cells: payload fetch_rectangle_traffic(cell, self.key) rows.extend(parse_roads_to_rows( payload.get(trafficinfo, {}).get(roads, []), time.strftime(%Y-%m-%dT%H:%M:%S) )) time.sleep(self.sleep_seconds) conn sqlite3.connect(self.db_path) insert_rows(conn, rows, time.strftime(%Y-%m-%dT%H:%M:%S)) conn.close() # 用原始JSON存档可以留一条后悔药 # with open(raw_backup.json, w) as f: # json.dump(payload, f, ensure_asciiFalse)框架化的好处不只是复用更重要的是补拉逻辑好写。采集过程中某个网格失败了你可以在run_once里收集失败的cell下一轮只重试这些失败项而不是全量重拉。我自己的项目里会用collect_ts字段加上一个batch_id每一轮采集写同一个batch_id出了问题能按批次回溯。另一个值得养成的习惯是把“项目说明”当成代码的一部分来写。这个标题里带了“项目说明”四个字说明验收方很看重文档。我的README组织方式是开发环境、Key配置、运行顺序、数据字典、输出物说明其中数据字典一定和代码里的字段名完全一致不能写一套代码一套。数据字典就是把第3章traffic_raw表的字段列出来逐个说明含义、类型和取值范围新接手的人照着字典就能读懂接口返回的JSON结构。最后说一个我自己的习惯每次拿到一批新的交通数据先不做任何可视化先做三个最基础的抽样检查——随机抽10条路段的原始JSON和库里的记录比对检查当天采集的网格覆盖数是不是完整验证凌晨时段的数据占比是否趋近于一个稳定低值。这三个检查通过后我才信得过这份数据。交通数据分析的难点从来不在画图而在“你凭什么相信这张图说的就是现实”。希望这套流程和避坑清单能帮你在自己的项目里少走几段弯路把精力留给真正值得分析的问题。本文还有配套的精品资源点击获取