ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

高德交通态势爬取实战:从Web API到拥堵画像的完整方案

高德交通态势爬取实战:从Web API到拥堵画像的完整方案

简介:面向交通数据分析与地理信息系统开发者的高德交通态势爬取资源包,提供从高德地图接口获取实时路况信息并转为可分析格式的代码方案。压缩包内共2个文件,核心为Python脚本与打包归档,整体仅2KB,适合初学者快速阅读爬取逻辑。脚本实现交通态势请求与返回数据解析,并包含导入地理信息系统前的数据预处理思路,便于后续空间可视化与拥堵规律分析。资源虽精简,但覆盖高德接口调用、反爬策略要点、数据清洗与存储等关键环节,可帮助用户理解交通态势数据从采集到地理信息应用的完整链路。目前已有1867人学习浏览,适合需要快速上手高德交通数据抓取,或希望扩展地理信息系统与Python集成技能的开发者和研究人员。

1. 高德交通态势爬取是什么:一条 URL 拿全城路况,先别急着写代码

做城市交通分析的人,大概都干过这种笨事:电脑开着满屏路况的高德地图,每隔几分钟截一张图当历史数据。图攒多了才发现只能看不能算——没有速度、没有拥堵等级,想统计早晚高峰根本无从下手。高德交通态势爬取,就是把页面上的彩色路况,变成程序里带速度、带状态、带方向的结构化记录。

高德开放平台有个交通态势 Web API:给一个矩形范围或坐标点,就返回区域内所有道路的实时状态——畅通缓行、平均车速、方向、轨迹坐标。这意味着不用再人肉截图,而是把真实路况当成普通数据源,定时拉取、入库、再交给统计模型。

适合做城市交通仿真、同城配送调度、路况大屏和交通方向论文的人。成本很低:一个 Key、一台能跑 Python 的机器。下面严格按落地顺序讲,从选型到避坑,代码可以直接抄。

2. 高德交通态势爬取选型:Web API 与高德地图瓦片两条路的边界

在写第一行代码之前,先要回答一个问题:这个“交通态势”到底从哪来。相关检索里高频出现的高德地图瓦片,和交通态势爬取经常被放在一起,但这两条路的技术难度、数据形态、合规风险完全不一样。我的决策顺序是:先想清楚最终要的是“能统计的数据”还是“能还原的画面”,再决定走哪条路。选错了,返工成本不是几小时,是几天。

2.1 官方 Web API:结构化 JSON 与配额是硬边界

高德 Web 服务里有两个交通态势接口,一个是按矩形范围查询,一个是按坐标点周边查询。矩形接口是主力,参数只有 key、rectangle、level、extensions 四类,返回的是路段级路况快照。所谓“路段级”,意思是接口返回的是某条路的一段,包含路名、拥堵状态、平均速度、方向、轨迹坐标串,而不是车辆轨迹或车流密度。这一点先摆清楚,避免后面拿着结果找不存在的“车流量”字段。

这套接口最值钱的地方在于数据是结构化的。每条道路记录可以直接对到一张表,字段含义明确,不需要做视觉识别。代价是配额约束非常现实:控制台里每个 Key 的日配额和 QPS 限制是硬指标,网格切细了、轮询加密了,配额消耗会肉眼可见地涨。我在前期调研时习惯先跑一天小范围任务,用实际请求量反推配额够不够支撑全城覆盖,而不是等到上线第二天被限流了才回头改参数。

2.2 高德地图瓦片路况层:像素颜色转状态的两个难点

另一条路是爬高德地图瓦片。瓦片是地图渲染的基本单位,路况作为独立图层叠加在底图上,瓦片里不同颜色的像素代表不同拥堵程度:绿色畅通、黄色缓行、红色拥堵、深红严重拥堵。这套方案看起来漂亮,因为下载下来的瓦片天然就是“画面”,能做视觉回放,也没有显式配额。但真正动手会发现两个硬坑。

第一个坑是颜色到状态的映射不稳定。同一个绿色在不同缩放级别下可能对应不同速度区间,而且瓦片边缘有抗锯齿,像素颜色是混合色,不是纯色。只靠“取一个像素判断颜色”做出来的状态,噪点会多到没法用。常见的做法是取瓦片中心区域的若干像素做多数投票,并且只接受接近纯色的结果,宁缺毋滥。

第二个坑是瓦片坐标换算。瓦片的 z/x/y 编号和经纬度之间的换算是 Web Mercator 标准,但高德的瓦片序号规则和其他地图不完全一致,需要先抓包确认当前服务端的地址和参数格式;另外高德坐标本身是 GCJ-02,和 WGS-84 还有一次偏移。这些叠加起来,实现成本至少是 Web API 路线的三倍。还有合规问题:低频、自用、做视觉校核可以接受;大批量下载并长期存储、再分发,就明显越过正常使用边界了。

2.3 选型结论:什么场景走 API、什么场景才需要碰瓦片

对比维度Web API(推荐主线)路况瓦片(备选)
数据形态JSON 路段记录,含 status/speed/direction图片像素,颜色代表状态
结构化程度高,直接入库低,需要像素反推且有抖动
配额限制有日配额与 QPS 限制无显式配额,但有越界风险
实现成本半天跑通至少两天,要处理坐标换算
适合场景统计、建模、大屏、历史回放画面还原、视觉校核

结论很简单:要做趋势统计、拥堵指数、路段画像,一律走 Web API;只有当你需要还原“当时屏幕上看到的画面”,比如做路况回放视频,才值得碰瓦片。我见过一个项目两种都要,用 API 入库打底,瓦片只按需截取重点路段的图做视觉对照,各干各的活,这个组合反而是最省心的。

3. 高德交通态势爬取最小脚本:矩形接口参数、返回字段与一次跑通

这一章的目标是让你在半小时内拿到第一批真实路况数据。先别想全城覆盖,那一步是第四章的事。先把一个矩形请求通、字段看懂、能存下来,后面所有工作都是在这个基础上做放大。

3.1 申请 Web 服务 Key:选错类型会白忙一场

申请位置在高德开放平台控制台:创建一个应用,然后添加 Key,服务平台那一项必须选“Web 服务”。我见过不少人栽在第一步——选了“Web 端(JS API)”类型,拿这个 Key 去调 restapi 接口,返回的都是权限错误,排查半天才发现 Key 类型不对。这两个 Key 的权限域完全不同,不能混用。

Key 拿到手先做两件事。第一,不要写死在代码里,用环境变量或者本地配置文件,更不要提交到 Git 仓库。第二,到控制台的配额页看两眼:日配额总量、当前已用、QPS 限制。这几个数字决定了第四章的调度参数怎么设,越早知道越好。

export AMAP_KEY="你的Web服务Key"

之后在 Python 里读取:

import os AMAP_KEY = os.environ.get("AMAP_KEY", "")

环境变量方式的好处是换账号、换 Key 不用改代码,服务器上做定期轮换也方便。如果团队里多人共用数据,建议每人一个环境变量文件,避免 Key 在聊天记录里传来传去。

3.2 矩形接口最小脚本:rectangle 参数的正确拼法

矩形接口的请求地址是https://restapi.amap.com/v3/traffic/status/rectangle,核心参数的拼写规则是:rectangle=左下角经度,左下角纬度;右上角经度,右上角纬度,中间用分号分隔,坐标顺序是经度在前。

import os import requests AMAP_KEY = os.environ.get("AMAP_KEY", "") URL = "https://restapi.amap.com/v3/traffic/status/rectangle" def fetch_rectangle(lon_min, lat_min, lon_max, lat_max, level=""): params = { "key": AMAP_KEY, "rectangle": f"{lon_min},{lat_min};{lon_max},{lat_max}", "level": level, "extensions": "all", } resp = requests.get(URL, params=params, timeout=10) resp.raise_for_status() return resp.json()

参数说明:rectangle是左下角和右上角两个经纬度点拼成的闭区间;level用于按道路等级过滤数据,数值从 1 到 8 依次对应高速、城市快速路、国道、省道、县道、乡镇道路等,等级越低数值越大,记不清就不传,默认全等级返回;extensions传all时字段最全,首次摸底建议用all,等确认了需要的字段再考虑用base省流量。

先拿一个很小的矩形做冒烟测试,比如北京城区内约 1 公里的范围:

data = fetch_rectangle(116.30, 39.90, 116.31, 39.91) print(data.get("info")) print(len(data.get("trafficinfo", {}).get("roads", [])))

提示:rectangle 参数最容易写反的是经纬度顺序和左下/右上的对应关系。写反了接口不会报错,但数据会落到完全不相干的区域,排查时先从这两点查起。

3.3 返回字段拆解:speed、status、direction 的清洗口径

接口返回的 JSON 结构大致如下:

{ "status": "1", "info": "OK", "trafficinfo": { "description": "区域内部分路段缓行", "roads": [ { "name": "北四环中路", "status": "3", "direction": "东", "speed": "18", "polyline": "116.32,39.98;116.35,39.98" } ] } }

核心字段就五个,做个表直接对应清洗口径:

字段类型含义入库建议
namestring道路名称直接存储
statusstring "0"~"3"0 未知、1 畅通、2 缓行、3 拥堵转 int 存储
speedstring平均速度,单位 km/h清洗后转 float
directionstring道路方向描述参与去重主键
polylinestring轨迹点串,分号分隔按分号切分后统计点数

解析函数写成这样,顺便把状态码检查掉:

def parse_roads(data): if data.get("status") != "1": raise RuntimeError(f"API error: {data.get('info')}") ti = data.get("trafficinfo") or {} return [ { "name": r.get("name", ""), "status": r.get("status", "0"), "speed": r.get("speed", "0"), "direction": r.get("direction", ""), "polyline": r.get("polyline", ""), } for r in ti.get("roads", []) ]

注意 HTTP 返回里 status 是字符串,不是整数。后面入库前我会再转一次 int,避免统计时把"3"和数字3混在一起,这种类型错位在 SQL 聚合时最容易翻车。

另外还有一个周边接口https://restapi.amap.com/v3/traffic/status/around,参数是point=经度,纬度和radius,语义上很像高德地图 App 里“获取当前位置”之后拉取周边路况的做法。如果监测对象是固定热点,比如某个商圈或枢纽,用这个接口比维护一个矩形网格更省请求量。后面全城覆盖的部分,我仍以矩形接口为主线讲,因为它的边界更可控。

4. 把交通态势爬虫做成定时监控:网格切分、入库与调度参数

单矩形跑通只是开始,真正要的是连续采集。这一章解决三个问题:怎么用一个接口覆盖整个城市、数据存到哪里才不会乱、轮询参数怎么定才不会把配额打爆。顺序不能反,先定网格,再定表结构,最后才是调度。

4.1 网格切分生成器:按纬度修正经度步长

矩形接口的单次查询范围有限,覆盖全城必须切网格。网上的常见做法是把城市边界按固定步长切成小矩形,逐个请求。步长的选择有个细节:纬度 0.005 度大约对应 555 米,但经度方向的实际距离要乘纬度的余弦,同一个步长在哈尔滨和在三亚对应的物理宽度完全不同。

import math def gen_grid(lon_min, lat_min, lon_max, lat_max, step_lat=0.005): mid_lat = math.radians((lat_min + lat_max) / 2) step_lon = step_lat * math.cos(mid_lat) lat = lat_min while lat < lat_max: lon = lon_min while lon < lon_max: yield ( lon, lat, min(lon + step_lon, lon_max), min(lat + step_lat, lat_max), ) lon += step_lon lat += step_lat

逻辑说明:生成器每次产出一个四元组,分别是左下角经度、左下角纬度、右上角经度、右上角纬度,正好对应 fetch_rectangle 的四个位置参数。step_lon用中心纬度的余弦修正,保证格子尽量接近正方形,避免高纬度地区格子横向过长、超出矩形接口的参数上限。

参数怎么调:step_lat=0.005是起步值,跑一轮看返回数据的完整度;如果大量格子 roads 为空,先缩小步长而不是怀疑接口。全城网格数约等于城市边长除以步长后的平方,例如 40 公里 × 40 公里的城区范围,0.005 度步长大概会产生 80 × 80 个格子,这一轮数据量心里要有数。

4.2 SQLite 入库与自然去重:主键设计决定数据质量

采集数据最怕重复和错位。SQLite 在单机场景下够用,关键是表结构把主键设计好,让同一条道路在同一时刻只留一条记录。

CREATE TABLE IF NOT EXISTS traffic ( ts INTEGER NOT NULL, grid TEXT NOT NULL, road_name TEXT, status INTEGER, speed REAL, direction TEXT, coord_count INTEGER, PRIMARY KEY (ts, grid, road_name, direction) );

字段说明:ts是 Unix 时间戳,这一轮采集的统一时间;grid是格子左下角的坐标字符串,用来定位数据在哪个网格里;coord_count是 polyline 按分号切分后的点数,间接代表这段路在本格子里的覆盖长度。主键用ts + grid + road_name + direction,同一时刻、同一格子、同一道路且同一方向,只允许一条记录。

写入用INSERT OR REPLACE,好处是任务重跑同一批时间片时不会产生脏数据:

import sqlite3 def save_batch(conn, rows, ts): conn.executemany( "INSERT OR REPLACE INTO traffic VALUES (?,?,?,?,?,?,?)", [(ts, *row) for row in rows], ) conn.commit()

逻辑说明:rows 是列表,每个元素是七元组,直接展开成一条记录。注意 ts 是外部统一传入,不是每行取当前时间,否则同一轮数据的时间戳会漂移,后面的时间聚合就做不准了。

数据量估算:全城 6400 个格子、每格平均 5 条道路,一轮就是 3.2 万条记录,5 分钟一轮、一天 288 轮,日增约 900 万条。SQLite 单表能扛,但两周后查询会明显变慢,我一般按月建表或者在 ts 上建索引,再写个定期清理旧数据的小任务。真要长期大规模采集,就换 PostgreSQL 或时序数据库,表结构可以原样搬过去。

4.3 调度与配额预算:轮询周期、超时退避怎么定

调度不一定要引入重型框架,一个带对齐功能的死循环在单机场景下足够。关键是两个参数:INTERVAL控制轮询周期,REQUEST_GAP控制单格请求间隔。

import time INTERVAL = 300 # 每轮间隔,单位秒 REQUEST_GAP = 0.3 # 格子之间的请求间隔,单位秒 def run_forever(bounds): conn = sqlite3.connect("traffic.db") while True: ts = int(time.time()) rows = [] for cell in gen_grid(*bounds): grid = f"{cell[0]:.4f},{cell[1]:.4f}" try: data = fetch_rectangle(*cell) for r in parse_roads(data): rows.append( ( grid, r["name"], int(r["status"]), float(r["speed"]), r["direction"], len(r["polyline"].split(";")), ) ) except Exception as exc: print(f"{time.time()} cell {grid} failed: {exc}") time.sleep(REQUEST_GAP) save_batch(conn, rows, ts) print(f"{ts}: saved {len(rows)} roads") time.sleep(INTERVAL - time.time() % INTERVAL)

逻辑说明:每轮开始时固定一个 ts,整轮所有记录共用这个时间戳;单个格子失败只打日志,不中断整轮,缺口由下一轮自然补上;最后的 sleep 是对齐到下一个INTERVAL整数倍,避免轮询周期越跑越漂。

配额预算要先算再跑:网格数乘以每日轮数就是一天的请求量。100 个格子、5 分钟一轮,每天约 2.88 万次请求;6400 个格子同样的频率就是 184 万次,很可能直接超配额。超了先拉大INTERVAL到 600 甚至 900 秒,或者只覆盖重点片区。不要用多个 Key 并发硬冲,平台规则不允许,整批 Key 被限流后反而更被动。

5. 高德交通态势爬取避坑清单:配额、坐标系、空数据与限流的真实处理

下面五条是从我自己的定时任务日志里筛出来的高频问题,按出现概率排序。每条按现象、原因、解决三段写,前三条属于“数据坏了但程序不报错”的类型,最坑,因为它们不会让任务中断,只会让结果悄悄变脏。等你发现的时候,脏数据已经攒了好几天。

5.1 现象一:接口返回正常但 roads 一直空

现象:status 是"1"、info 是"OK",但trafficinfo.roads是空数组,日志里看不出任何异常,程序照常入库,只是每轮入库条数接近零。

原因:最常见是矩形落在没有路网数据的区域,比如大片农田、水域、新建但未采集路网的地带。另一个更隐蔽的原因是level参数把低级道路过滤掉了:当矩形内只有低于该等级的道路时,roads 就是空数组,接口本身没有报错。

解决:排查顺序固定成三步。第一步换一个明显有城市路网的较大矩形,确认接口本身没写错;第二步把 level 参数去掉,改用全等级返回再测一次;第三步用高德地图网页版目测该区域是否真的有条路。如果网页上能看到路但接口返回空,基本就是等级过滤问题,调 level 就能解决。我自己踩过一次:连续三天某个格子 roads 为空,最后发现是矩形边界写进了河道里,坐标肉眼看上去没错,但对到地图上就是一条河。

5.2 现象二:跑三天后频繁报配额超限

现象:定时任务前两三天一切正常,之后每天固定时段返回的 info 变成配额超限相关错误,第二天又自动恢复,循环出现。

原因:配额按自然日计算,还有并发 QPS 口径。网格轮询如果在同一时刻集中发请求,很容易撞上 QPS 闸口;再加上白天可能有其他业务共享同一个 Key,日配额提前打满很正常。固定时段超限,往往证明这个时段有别的任务在抢量。

解决:先去控制台看配额剩余曲线,确认是日配额还是 QPS 限制。日配额超了就拉大INTERVAL,把轮询从 300 秒改成 600 秒,请求量直接砍半;QPS 超了就调大REQUEST_GAP,从 0.3 秒改成 1 秒。再加一层“超限就跳过本轮”的逻辑,宁可这一轮不采,也不要卡死整个任务。真正要长期做商用规模,走商务合作提额,这是正路,其他绕法都不长久。

5.3 现象三:爬到的坐标和 GPS 对不上

现象:拿自己设备的 GPS 点去匹配爬回来的 polyline,位置始终偏了几百米,看起来像高德的数据错了。

原因:高德所有坐标基于 GCJ-02 坐标系,消费级 GPS 设备输出的是 WGS-84,两者之间隔着一次坐标偏移转换。不做转换直接匹配,偏几百米是正常现象,不是数据质量问题。

解决:最简单的做法是让全链路统一在 GCJ-02 下工作,展示时直接用高德地图组件去画,这样前后端都省去转换。如果必须和自有 WGS-84 数据融合,先把 GPS 点转成 GCJ-02 再做匹配,而不是反过来把高德的坐标当 WGS-84 用。转换算法是公开的,但精度只保证到“能对上路”这个级别,别指望做厘米级定位。我现在的项目里统一约定:所有入库坐标都以 GCJ-02 为准,谁传 WGS-84 谁负责转换,这条规则写进了团队接口文档。

5.4 现象四:speed 字段出现 0 和离谱大值

现象:一条断头路 status 是畅通但 speed 是 0;一条快速路 status 是拥堵但 speed 显示 95。用 speed 直接算平均车速,结果忽高忽低没法看。

原因:speed 是路段级别的平均估计值,数据缺失时接口会带回退值;status 和 speed 也不是同一时刻刷新的,高峰期两者错位一两分钟很常见。把这两个字段当成完全同步的数据用,必然出问题。

解决:入库前做一层清洗,把明显越界的值抹掉:

UPDATE traffic SET speed = NULL WHERE speed <= 0 OR speed >= 130;

统计时排除空值;status 和 speed 矛盾时以 status 为主分类,speed 只用于辅助排序。千万别把 speed=0 当“静止”去算拥堵,那是把脏数据当成了信号。阈值 130 是按国内高速限速习惯定的,如果你监测区域含特殊快速路,可以按需调。

5.5 现象五:白天超时凌晨秒回,爬虫看天吃饭

现象:日志里白天大量 timeout,凌晨一片绿;越重试越严重,最终整轮任务挂掉,要手动重启。

原因:白天高德服务本身负载高,公共配额下响应变慢;timeout 设得太短,反而触发重试风暴,把 QPS 顶到限制值,形成恶性循环。日志里看到成片的 timeout,先怀疑自己的重试策略,而不是服务端。

解决:把基础 timeout 放宽到 10 秒,重试改成指数退避,最多两次:

def fetch_with_retry(*args, retries=2, base_delay=1.0): for attempt in range(retries + 1): try: return fetch_rectangle(*args) except Exception: if attempt == retries: raise time.sleep(base_delay * (3 ** attempt))

逻辑说明:第一次失败等 1 秒,第二次失败等 3 秒,最多重试两次,第三次失败直接抛出并跳过当前格子。这个节奏白天不会顶爆 QPS,凌晨也不会因为响应太快而空转。日志里要能看到哪一格失败、失败了几次,否则第二天排查只能靠猜。

这五条并不互斥,白天超时和配额超限经常一起出现。我现在的习惯是每周看一次日志里的失败率、空 roads 比例和速度清洗量,三个指标超过阈值就优先查 Key 配额,再查网格参数。爬虫的维护重点不是写出来的那几天,而是跑起来之后的观察节奏。

6. 用历史交通态势做拥堵画像:交叉验证方法与 5 分钟聚合技巧

数据连续采了两周之后,就该回答最初的问题了:这堆数据能算出什么?最直接的产品是“拥堵画像”——某个片区一天 24 小时里,拥堵是什么时候开始、什么时候缓解、峰值在几点。下面这个方法不用任何第三方库,SQLite 加 Python 标准库就能跑。

6.1 5 分钟聚合函数:一天 288 个点的拥堵占比曲线

把时间戳按照INTERVAL的粒度分桶,每桶算一个“拥堵路段占比”。占比用 status=3 的路段数除以总路段数,这个指标对拥堵起讫时间非常敏感。

def hourly_profile(conn, day_start, day_end, bucket_size=300): buckets = {} query = ( "SELECT ts, status, COUNT(*) FROM traffic " "WHERE ts >= ? AND ts < ? GROUP BY ts, status" ) for ts, status, cnt in conn.execute(query, (day_start, day_end)): bucket = (ts // bucket_size) * bucket_size b = buckets.setdefault(bucket, {"roads": 0, "jam": 0}) b["roads"] += cnt if status == 3: b["jam"] += cnt return { bucket: (b["jam"] / b["roads"] if b["roads"] else 0) for bucket, b in sorted(buckets.items()) }

逻辑说明:bucket = (ts // 300) * 300把任意时间戳归到最近的 5 分钟整数边界,和第四章调度的对齐逻辑保持同一规格,聚合结果才能和采集周期对上。返回值是“时间桶 → 拥堵占比”的字典,画出来就是一条从凌晨平滑到早高峰再回落的曲线,高峰起讫时间一眼可见。

6.2 交叉验证:用 speed 中位数校验 status 质量

聚合之前先做一次质量校验,聚出来的指标才敢用。方法是用 speed 的中位数反推 status 是否合理:畅通路段的中位速度应该明显高于拥堵路段,如果错位,说明采集环节出了问题。

def speed_median_by_status(conn, start, end): rows = conn.execute( "SELECT status, speed FROM traffic " "WHERE ts >= ? AND ts < ? AND speed > 0 AND speed < 130", (start, end), ).fetchall() groups = {} for status, speed in rows: groups.setdefault(status, []).append(speed) result = {} for status in sorted(groups): vals = sorted(groups[status]) result[status] = vals[len(vals) // 2] return result

用法:以天为单位跑一次,输出形如{1: 42, 2: 28, 3: 15}的结果。正常数据里,畅通的中位速度应该比缓行高 10 以上,缓行比拥堵高 8 以上;如果两个等级的中位速度几乎重叠,先回来检查入库时的类型转换和清洗规则,而不是急着下结论说数据不准。

更进阶一步,可以用聚合结果为每条主要道路算“早高峰开始时间”——拥堵占比首次连续三个桶超过阈值的时间点。这个指标可以直接输出成表,给交通管理或者配送调度当排班依据,也可以导出 CSV 交给可视化团队做热力图。我现在每两周跑一次全量画像,和上一轮对比,一旦发现某条路的拥堵时段大规模漂移,就先查采集是不是漏了格子,再查城市路网是不是有临时施工。校验脚本永远比分析脚本先写,这个习惯帮我省掉了后面一大半排查时间。希望帮到你。

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

返回列表