企业官网建设流程全解析

简介：基于高德地图API自动处理、分析与预测车速数据的完整项目，面向交通数据挖掘方向的毕设学生和开发者。项目打通了从实时车速获取、多策略区域爬取、数据清洗与归一化，到时空分布特征分析与车速预测建模的完整链路，覆盖时间序列、机器学习与深度学习等预测思路，可用于理解交通流变化规律及构建实时稳定预测模型。压缩包共12个文件、约305KB，包含Python爬虫与数据处理脚本、实验效果图、HTML可视化页面及说明文档，脚本对高德API调用、数据预处理和模型训练都有清晰实现，便于快速复现与二次开发。已有74人学习下载，适合用来参考真实道路车速数据的采集方案、异常值清洗方式以及预测模型的代码落地。

1. 车速数据不是“拿到就能用”的，高德API只是第一步

做交通方向毕设时，很多人第一步就去调高德地图API，以为拿到路况JSON就能直接训练模型。真实情况是：高德返回的roads数组里，每条道路被切成了不等长的segments，同一路段不同分钟返回的segment边界可能不同，速度字段还会夹杂0值和明显偏离物理上限的脏数据。如果拿这样的数据直接做时序分析，结果基本是乱的。从实际项目里拆出来的三个爬虫脚本（crawl_rectangle.py、crawl_road.py、crawl_square.py）正好覆盖了高德路况API的三种抓取方式，下面把JSON扁平化、数据清洗、时空分析和车速预测整个链路走一遍。适合正准备做交通数据分析、智能导航或拥堵预测的从业者和毕业生。

2. 高德API车速数据抓取：矩形、道路与区域三种爬法怎么选

2.1 三个爬虫脚本的分工与边界

高德路况API没有提供“按行政区批量拉取”的接口，所以项目里才要写三个爬虫脚本覆盖不同查询维度。

crawl_rectangle.py对应的是/v3/traffic/status/rectangle接口，入参是矩形左下角和右上角的经纬度，返回该矩形范围内所有道路的路况。这个接口适合做网格化扫描，比如把城市切成2km×2km的小方格，然后用多线程每隔几分钟轮询一遍。但矩形边长不能无限大，高德对矩形范围有限制，我实测超过一定范围后返回的数据会被截断，所以更安全的做法是把大矩形拆成若干个小矩形再并发请求。

crawl_road.py走的是/v3/traffic/status/road接口，直接传道路名和城市编码，返回这条道路的完整路况。它的优势是数据干净、道路ID稳定，适合对指定路段做长期速度跟踪。缺点是高德的道路名匹配存在歧义，比如“北四环”可能对应不同区段，需要结合adcode或city参数消歧。实际上我建议优先使用道路ID而不是道路名，但路况接口的road版本只接受名称，只能靠拼音或名称扩展来降低歧义。

crawl_square.py的命名容易让人误以为是正方形抓取，实际上它是在拿到行政区边界polygon后，把多边形拆成多个小矩形，再复用crawl_rectangle.py的逻辑逐块抓取。项目里的xingzheng.html保存的是行政区边界数据，所以这个脚本是为了支撑“整个区”的路况统计而写的。如果你只需要单个路段，完全可以不跑它。

这三个脚本的适用场景对比如下：

因为高德API使用GCJ-02坐标系，如果你的矩形边界来自WGS84数据源（比如GPS轨迹），直接传给API会有几百米的偏移。项目里我在crawl_rectangle.py前面加了一层坐标转换，把矩形顶点先转成GCJ-02再请求。坐标转换的代价是额外一次的经纬度偏移计算，但对后续按坐标匹配segment非常重要。

2.2 请求参数与返回结构解析

以rectangle接口为例，核心请求如下：

import requests def fetch_road_speed(rectangle, api_key): url = "https://restapi.amap.com/v3/traffic/status/rectangle" params = { "rectangle": rectangle, # 左下角经度,纬度;右上角经度,纬度 "key": api_key, "extensions": "all", # 返回segments，不传则只有道路级别状态 "level": 6, # 1~6，6表示所有道路等级 "output": "json", } resp = requests.get(url, params=params, timeout=10) resp.raise_for_status() data = resp.json() if data.get("status") == "1": return data["trafficinfo"]["roads"] else: return []

rectangle参数必须是左下角在前、右上角在后，用分号分隔。顺序反了不会报错，但返回的区域会和你预想的不一致。level取值1到6，分别对应高速、国道、省道、县道、乡镇村道和所有道路。做城市拥堵分析时传6，如果只关心快速路可以传1，减少返回数据量。extensions=all是关键，只有这个参数才会让每个road下面挂出segments数组，speed和status都在segment里。

返回的JSON是一个三级嵌套结构。trafficinfo下面有roads数组，每个road有name、status、direction、angle和segments。每个segment又有polyline、speed、status、length等字段。polyline是一串“经度,纬度;经度,纬度”格式的坐标，要拆出来才能算真实长度。segment的切分规则不公开，所以我不建议把segment序号当稳定ID使用，最好用polyline的首尾坐标作为空间匹配键。

2.3 配额与收费：为什么有人吐槽“高德地图API收费坑人”

高德地图API的收费策略常被开发者吐槽，尤其是个人申请后，免费配额和QPS限制都很紧。路况接口按调用次数计费，返回的数据量大小不影响费用，所以真正要优化的是“请求次数”而不是“响应体大小”。如果你的爬虫没有做缓存，按5分钟一次扫全市网格，很容易在早高峰时段把配额打空。

另一个容易被忽略的是错误码处理。配额超限时高德不会抛异常，而是返回一个带status和infocode的JSON，你如果只判断HTTP状态码，程序会陷入盲目重试，反而消耗更多配额。我一般会封装一个带退避逻辑的请求函数：

import time def safe_request(url, params, max_retry=3): for attempt in range(max_retry): try: resp = requests.get(url, params=params, timeout=10) data = resp.json() except Exception: time.sleep(2) continue if data.get("status") == "1": return data time.sleep(2 ** attempt) return None

status == "1"表示请求成功，非1可能是服务异常或配额不足。每次失败后按2的指数次幂退避，比如第一次等2秒、第二次等4秒，而不是立刻重试。这样至少不会在配额超限时把请求数打爆。项目里我还在外层加了SQLite缓存，同一个小矩形在5分钟内只请求一次，具体实现放在第五章。

提示：高德路况接口的status字段取值是0畅通、1缓行、2拥堵、3严重拥堵。但不同接口版本可能返回数值还是字符串，跑通后先打印一条原始数据确认，再写解析逻辑。

3. 数据预处理：把路况JSON变成可训练的时序表

3.1 从JSON到DataFrame的字段抽取

高德返回的嵌套JSON没法直接喂给pandas，需要先展开成“一条segment一行”的扁平表。展开时除了保存速度，还要保留下拉线数据，因为后续做空间关联时需要它。代码：

import pandas as pd def flatten_roads(roads, ts): rows = [] for road in roads: for seg in road.get("segments", []): rows.append({ "timestamp": ts, "road_name": road.get("name"), "roadid": seg.get("roadid"), "speed": seg.get("speed"), "status": seg.get("status"), "length": seg.get("length"), "polyline": seg.get("polyline"), "direction": road.get("direction"), }) return pd.DataFrame(rows)

ts是抓取时刻的时间戳，建议统一成ISO格式字符串，方便后续按时间排序。roadid是segment级ID，同一个物理路段在不同时刻返回的roadid可能不同，因为segment的切分会变化。length单位是米，后面计算加权平均速度时要用它作权重。direction是从道路整体上取下来的，但同一条道路不同segment的方向可能不一致，所以它只能作为参考特征，不能直接当分类标签。

展开后要进行一次“时间桶”对齐。比如抓取任务从9:00:10持续到9:03:50，这些时间戳不能直接当成同一时刻，否则画出的时序曲线会有锯齿。我一般把时间戳向下取整到最近的5分钟格点，比如9:00:10变成9:00:00。这样多个矩形抓取的数据就能对齐到同一时刻。

df["timestamp"] = pd.to_datetime(df["timestamp"]).dt.floor("5min")

floor("5min")是按5分钟下取整，如果你希望以整点对齐，可以用ceil或round，但实测floor最安全，不会引入未来时间。

3.2 清洗噪声与异常值

车速数据有几种典型脏数据：速度直接是0但状态是畅通，速度超过120但路段是城市道路，同一个时间桶内同一条segment重复出现，以及segment长度只有几米导致的瞬时速度异常。清洗顺序我固定为先删重复，再滤异常，最后补缺失。

def clean_speed(df): df = df.sort_values("timestamp") df = df.drop_duplicates(subset=["timestamp", "road_name", "polyline"]) df = df[(df["speed"] >= 5) & (df["speed"] <= 120)] df = df.replace(0, pd.NA) df["speed"] = df.groupby(["road_name", "roadid"])["speed"].transform( lambda x: x.interpolate(method="linear", limit_direction="both") ) return df.dropna(subset=["speed"])

drop_duplicates的subset里包含polyline，因为polyline是segment最细粒度的标识，用它去重能避免重复请求导致的重复行。速度下限设5，上限设120，这是针对城市路段的经验值，如果你的数据里有高速路段，要按实际情况扩展上限。replace(0, pd.NA)把速度为0的非法值转成缺失，再通过interpolate沿时间方向线性插值。分组使用road_name和roadid，避免插值时混入其他道路的数据。

我用过method="time"插值，它会利用时间索引的间隔信息，但要求索引是datetime类型，并且需要先set_index。如果你的时间序列不规则，linear就够用了。要注意的是，interpolate默认从前往后填充，limit_direction="both"可以在序列开头和结尾都回填，避免前几个点因为缺值被整行删除。

3.3 归一化与时间窗口构造

预测模型输入前，速度值要先归一化到0-1区间。归一化参数必须在训练集上拟合，然后同时转换训练集和测试集，否则会把测试集的均值方差泄漏到训练过程中，导致验证结果虚高。

from sklearn.preprocessing import MinMaxScaler scaler = MinMaxScaler(feature_range=(0, 1)) train_speed = train_df[["speed"]].values scaler.fit(train_speed) train_df["speed_scaled"] = scaler.transform(train_speed) test_df["speed_scaled"] = scaler.transform(test_df[["speed"]].values) def make_windows(df, window_size=6): df = df.sort_values("timestamp").reset_index(drop=True) X, y = [], [] for i in range(len(df) - window_size): X.append(df["speed_scaled"].iloc[i:i+window_size].values) y.append(df["speed_scaled"].iloc[i+window_size]) return np.array(X), np.array(y)

MinMaxScaler的好处是保留原始速度的相对关系，坏处是对异常值敏感，所以要在清洗之后做归一化。window_size=6表示用前6个时刻的速度预测第7个时刻，如果抓取间隔是5分钟，就是利用过去30分钟预测未来5分钟。窗口太大反而会把过早的历史状态带进来，增加噪声。

注意：make_windows在实现里假设输入的是单条道路的时序数据。如果你把多条道路混在一起调用，窗口内会出现来自不同道路的速度跳跃，导致预测结果完全失真。正确做法是先按roadid分组，再对每组调用make_windows。

4. 分析车速时空分布与构建预测模型

4.1 时空分布特征计算

数据清洗完成后，我习惯先做探索性分析，再考虑建模。最直接的是按小时和道路分组，统计平均速度和波动情况：

df["hour"] = pd.to_datetime(df["timestamp"]).dt.hour hourly_stats = df.groupby(["road_name", "hour"])["speed"].agg( ["mean", "median", "std", "count"] ).reset_index() pivot_table = hourly_stats.pivot_table( index="hour", columns="road_name", values="mean" )

hourly_stats里std列能提示模型的难度：早晚高峰时段std大，说明同一时刻不同日期的速度差异大，模型需要更多特征来解释这种波动。count列可以检查数据缺失，如果某条道路的count明显比其他道路少，说明爬虫可能漏抓了某个矩形，应该回到爬虫层补抓。

pivot_table变成小时×道路的平均速度矩阵后，可以直接用matplotlib画热力图：

import matplotlib.pyplot as plt plt.figure(figsize=(12, 6)) plt.imshow(pivot_table.values, aspect="auto", cmap="RdYlGn_r") plt.colorbar() plt.xlabel("road_name") plt.ylabel("hour") plt.show()

热力图能直观暴露“潮汐路段”，比如某个方向早高峰拥堵、晚高峰畅通。这种路段在建模时一定要把“当前小时”和“道路方向”作为特征，否则模型会学到平均状态，无法捕捉方向性的拥堵变化。

4.2 时间序列预测：ARIMA还是LSTM

车速预测可以按纯时间序列或回归建模。纯时间序列里，ARIMA适合数据量小、变化平稳的路段，但它是单变量模型，无法引入星期几、节假日等外部特征。LSTM能拟合非线性趋势，但对数据量和训练时间要求高，一周的数据量很难训练出稳定的LSTM。

如果只是想快速拿到一个基准，ARIMA可以这样跑：

from statsmodels.tsa.arima.model import ARIMA series = one_road_df.set_index("timestamp")["speed"] model_arima = ARIMA(series, order=(2, 1, 2)) fit = model_arima.fit() forecast = fit.forecast(steps=3)

这里order=(2,1,2)分别对应AR项、差分阶数和MA项，需要根据ACF/PACF图调整。我试过在几条国道上用ARIMA，效果还行，但放到城市主干道上就不稳定，因为信号在红绿灯周期上下波动，ARIMA总是滞后一拍。

所以最终项目里我选了XGBoost路线，把时间序列预测转化为回归任务。优点是可以自由加入外部特征，训练快，还能输出特征重要性用于排查问题。

4.3 用XGBoost构建车速回归模型

XGBoost建模的关键是特征构造。我用的特征分三组：时间特征（小时、是否周末、是否高峰）、历史速度特征（前三个时刻的速度）、道路特征（道路编码、方向）。注意hour是周期性变量，最好同时保留sin和cos分量，否则模型很难学到0点和23点的相邻关系。

import xgboost as xgb import numpy as np feature_cols = ["hour_sin", "hour_cos", "is_weekend", "is_peak", "speed_t1", "speed_t2", "speed_t3", "road_code"] model = xgb.XGBRegressor( n_estimators=500, max_depth=6, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, early_stopping_rounds=30, random_state=42, ) model.fit(X_train, y_train, eval_set=[(X_val, y_val)], verbose=False)

n_estimators=500配合early_stopping_rounds=30，在验证集损失不再下降时提前停止，避免过拟合。max_depth=6限制树深度，防止模型记住某个特定日期的高峰突变。subsample和colsample_bytree都设为0.8，让每棵树用不同的样本子集和特征子集，增强泛化性。

这里的speed_t1是当前时刻前一个时间点的速度。特征重要性通常显示speed_t1排在第一位，说明车速强自相关，这与交通流理论一致。如果speed_t1的重要性低于hour，要检查是不是窗口构造出错，比如时间排序混乱导致未来数据泄漏。

4.4 模型评估与选择

评估指标我用RMSE和MAPE。RMSE单位是km/h，直观但受拥堵时的大误差影响大；MAPE是百分比误差，更能反映相对偏差。对于城市主干道，MAPE在12%以内算可接受，超过20%就要回头查特征或数据清洗逻辑。

从我的实践看，XGBoost在中等规模数据下综合表现最好。但注意XGBoost本身不感知时间顺序，训练时必须按时间划分训练集和验证集，不能随机切分。我通常在排序后取前70%时间做训练，后30%做验证，这样能评估模型在真实“未来”上的表现。

5. 进阶：把抓取-存储-预测串成定时任务的三个技巧

5.1 用SQLite做请求级缓存

路况数据5分钟内变化不大，高德API的配额又有限，缓存是必须的。我在爬虫层用SQLite做请求级缓存，主键是“矩形坐标+时间桶”，时间桶按5分钟取整：

import sqlite3 conn = sqlite3.connect("traffic_cache.db") conn.execute( "CREATE TABLE IF NOT EXISTS cache " "(rect TEXT, ts_bucket TEXT, data TEXT, PRIMARY KEY(rect, ts_bucket))" )

请求前先查表，命中就直接返回data里的JSON文本，未命中才调API并写回。这样同一个矩形在5分钟内无论被哪个脚本触发，都只会向高德发一次请求。实测能把API调用量降低一半以上，对于毕设配额完全够用。

5.2 增量追加而不是全量覆盖

爬虫每次启动时如果清空数据库，之前的历史数据就全丢了。我建了一张raw_traffic表，主键是timestamp, roadid, segment_hash，其中segment_hash是polyline取MD5后的前16个字符。写入时用INSERT OR REPLACE，这样重复运行的脚本不会产生重复行，又能保留历史记录。断点续抓时，只需查询max(timestamp)，从最后抓取时刻接续，不用重新跑全量。

SELECT MAX(timestamp) FROM raw_traffic;

注意timestamp要使用ISO格式字符串，这样字符串比较和排序与时间顺序一致，不会被数字格式干扰。

5.3 用残差波动判断预测是否失效

预测模型上线后，不能只看训练时的RMSE。我每周会重新计算一批验证集上的残差（真实值减预测值），看残差的滚动标准差。如果标准差突然变大，说明模型已经跟不上当前的交通状态，比如附近道路施工或大型活动改变了通行模式。

更细的做法是画残差热力图，横轴是小时，纵轴是道路，颜色是平均残差。如果发现某条道路在高频时段出现系统性负残差（预测偏快），说明该路段有重复性拥堵原因没有被模型捕捉。我在项目里就遇到过这样的情况，最终发现是附近学校放学，增加一个school_hour特征后MAPE下降了3个百分点。这个排查过程比调参本身更重要。

定时任务不需要上复杂的调度平台，直接写一个run_pipeline.py，在crontab里配置每5分钟执行一次即可：

*/5 * * * * cd /path/to/project && python run_pipeline.py >> pipeline.log 2>&1

crontab的行首*/5表示每5分钟触发一次，日志重定向到pipeline.log方便排查。如果你的机器不会休眠，这套方案足够支撑毕设或小型项目的持续数据采集。

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