简介面向交通数据分析与地理信息系统开发者的高德交通态势爬取资源包提供从高德地图接口获取实时路况信息并转为可分析格式的代码方案。压缩包内共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, paramsparams, timeout10) 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~30 未知、1 畅通、2 缓行、3 拥堵转 int 存储speedstring平均速度单位 km/h清洗后转 floatdirectionstring道路方向描述参与去重主键polylinestring轨迹点串分号分隔按分号切分后统计点数解析函数写成这样顺便把状态码检查掉def parse_roads(data): if data.get(status) ! 1: raise RuntimeError(fAPI 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_lat0.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_lat0.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 只用于辅助排序。千万别把 speed0 当“静止”去算拥堵那是把脏数据当成了信号。阈值 130 是按国内高速限速习惯定的如果你监测区域含特殊快速路可以按需调。5.5 现象五白天超时凌晨秒回爬虫看天吃饭现象日志里白天大量 timeout凌晨一片绿越重试越严重最终整轮任务挂掉要手动重启。原因白天高德服务本身负载高公共配额下响应变慢timeout 设得太短反而触发重试风暴把 QPS 顶到限制值形成恶性循环。日志里看到成片的 timeout先怀疑自己的重试策略而不是服务端。解决把基础 timeout 放宽到 10 秒重试改成指数退避最多两次def fetch_with_retry(*args, retries2, base_delay1.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的粒度分桶每桶算一个“拥堵路段占比”。占比用 status3 的路段数除以总路段数这个指标对拥堵起讫时间非常敏感。def hourly_profile(conn, day_start, day_end, bucket_size300): 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 交给可视化团队做热力图。我现在每两周跑一次全量画像和上一轮对比一旦发现某条路的拥堵时段大规模漂移就先查采集是不是漏了格子再查城市路网是不是有临时施工。校验脚本永远比分析脚本先写这个习惯帮我省掉了后面一大半排查时间。希望帮到你。本文还有配套的精品资源点击获取