简介：这份PDF面向具备Python基础、希望切入交通与物流数据分析场景的学习者，以运输车辆驾驶行为分析为主线，串联数据采集、预处理、统计分析与可视化全流程。内容围绕GPS定位、速度、加速度及急加速急减速事件等监控数据展开，讲解缺失值与异常值处理、时间戳转换、特征工程等关键环节，并给出可直接参考的代码示例。资源包共1个PDF文件，约126KB，以图文与代码结合的形式呈现，便于边看边练。目前已有460人学习。读者可借此掌握用Pandas、NumPy、Matplotlib、Seaborn完成驾驶行为特征构建与图表输出的方法，理解速度分布、急加速急减速频次、相关性热力图等分析思路，并迁移到安全培训、路线优化等实际场景，适合作为Python数据分析实战的案例教程与编程练习参考。
1. 运输车辆驾驶行为分析：从一份 GPS 轨迹到可复现的 Python 分析链路
手里拿到一批运输车辆的 GPS 轨迹数据，第一反应往往不是画图，而是先确认它到底能回答什么问题。运输车辆驾驶行为分析，本质是用车载定位、CAN 总线或 OBD 采集到的时序数据，还原出急加速、急减速、急转弯、超速、疲劳驾驶这几类风险事件，再按车辆、司机、线路做聚合打分。它解决的是车队管理里最实际的问题：哪台车在烧胎、哪个司机在深夜超速、哪条线路的急刹密度异常高。适合有 Python 基础、手头有轨迹或 CAN 抓包数据、想把数据分析落到业务报表上的从业者。这一章先把数据形态和整体链路讲清楚，后面几章再逐段拆开做。
2. 数据从哪来、长什么样：GPS 轨迹与 CAN 抓包两条路
做驾驶行为分析，数据源决定了后面所有算法的写法。常见的有两类：一类是 GPS/北斗定位终端回传的轨迹点，字段通常是时间戳、经纬度、瞬时速度、方向角；另一类是 CAN 抓包数据分析拿到的车辆总线数据，包含车速、转速、油门开度、刹车状态、档位。前者容易拿到、覆盖广，但采样频率低（常见 10 到 30 秒一个点），后者精度高但需要硬件和协议解析。选哪条路，取决于你要识别的事件粒度。
2.1 两类数据源的字段结构与适用边界
GPS 轨迹数据的典型结构是宽表，一行一个定位点。字段大致是vehicle_id、ts、lon、lat、speed、direction、status。它的优势是部署成本低，几乎所有营运车辆都装了定位终端；劣势是采样间隔大，急加速这种持续两三秒的事件很容易被漏掉或误判。我一般会先看采样间隔的分布，如果中位数超过 20 秒，就只做超速和疲劳驾驶这类宏观事件，不硬做急加速。
CAN 抓包数据是另一回事。通过 OBD 接口或直接接总线，能拿到 10Hz 甚至更高的车速和踏板信号。字段会多出rpm、throttle、brake、gear。它的边界在于：不同车型的 CAN 报文 ID 和解析规则不一样，需要 DBC 文件或逆向，落地成本高。如果只是做车队级的行为评分，GPS 数据通常够用；如果要做驾驶风格建模或事故还原，CAN 数据才值得投入。
2.2 用 pandas 读入并做第一轮体检
拿到数据先别急着算事件，先做体检。下面这段代码读入一份 CSV 轨迹，检查缺失、时间顺序和采样间隔。这是后面所有分析的地基，跳过这步后面全是玄学。
import pandas as pd import numpy as np # 读入轨迹数据，parse_dates 直接把时间列转成 datetime df = pd.read_csv("gps_track.csv", parse_dates=["ts"]) # 按车辆和时间排序，保证后续差分计算方向正确 df = df.sort_values(["vehicle_id", "ts"]).reset_index(drop=True) # 基础体检：缺失率、时间跨度、采样间隔 print("缺失率:\n", df.isna().mean()) print("车辆数:", df["vehicle_id"].nunique()) print("时间跨度:", df["ts"].min(), "->", df["ts"].max()) # 计算每台车相邻点的时间差（秒） df["dt"] = df.groupby("vehicle_id")["ts"].diff().dt.total_seconds() print("采样间隔分位数:\n", df["dt"].describe(percentiles=[0.5, 0.9, 0.99]))逻辑说明：parse_dates让时间列直接可用，省去后面反复转换。groupby + diff是按车分组算间隔，不能全局算，否则跨车的第一条记录会产生负值或异常大值。参数上，重点看dt的 50 分位和 99 分位：50 分位代表常规采样间隔，99 分位如果远大于它，说明有大量丢点，需要标记或插值。速度字段如果单位是 km/h，后面算加速度要除以 3.6 转成 m/s，这个坑很常见。
2.3 坐标清洗与异常点剔除
GPS 漂移是绕不开的问题。车辆停在地下车库或隧道时，定位点会跳到几百米外，速度也可能瞬间飙到 200 km/h。处理办法是先按速度阈值粗筛，再用相邻点的位移距离交叉验证。
def haversine(lon1, lat1, lon2, lat2): # 半正矢公式计算两点球面距离，单位米 R = 6371000 phi1, phi2 = np.radians(lat1), np.radians(lat2) dphi = np.radians(lat2 - lat1) dlambda = np.radians(lon2 - lon1) a = np.sin(dphi/2)**2 + np.cos(phi1)*np.cos(phi2)*np.sin(dlambda/2)**2 return 2 * R * np.arcsin(np.sqrt(a)) # 计算相邻点位移距离 df["prev_lon"] = df.groupby("vehicle_id")["lon"].shift(1) df["prev_lat"] = df.groupby("vehicle_id")["lat"].shift(1) df["dist"] = haversine(df["lon"], df["lat"], df["prev_lon"], df["prev_lat"]) # 用位移/时间得到推算速度，与上报速度对比 df["calc_speed"] = df["dist"] / df["dt"] * 3.6 # km/h # 标记异常：上报速度超过 150 或推算速度与上报速度差异过大 df["is_outlier"] = (df["speed"] > 150) | ((df["calc_speed"] - df["speed"]).abs() > 50) print("异常点占比:", df["is_outlier"].mean())逻辑说明：haversine是球面距离的标准算法，比直接用经纬度差乘系数精确。shift(1)配合groupby保证不跨车取点。calc_speed是用位移反推的速度，如果它和上报速度差超过 50 km/h，基本可以判定是漂移。参数上，150 km/h 是营运车辆的合理上限，可按车型调整；50 km/h 的差异阈值偏宽松，城市工况可以收紧到 30。异常点不要直接删，先标记，后面统计事件时排除即可，删了会破坏时间连续性。
3. 驾驶事件识别：急加速、急减速、急转弯怎么算才靠谱
事件识别是这类分析的核心。很多人一上来就设阈值，结果同一批数据换个阈值结论全变。问题出在没有先做平滑和分段。原始 GPS 速度有噪声，直接差分得到的加速度抖动极大，必须先用滑动窗口平滑，再在平滑序列上找事件。这一章把三类事件的算法和参数讲透。
3.1 加速度计算与滑动窗口平滑
加速度是速度对时间的导数。GPS 速度本身有 ±2 km/h 的误差，差分后误差被放大。常见做法是先对速度做中值滤波或移动平均，再差分。
# 按车分组做滑动平均，窗口 3 个点，减少噪声 df["speed_smooth"] = df.groupby("vehicle_id")["speed"].transform( lambda s: s.rolling(window=3, min_periods=1, center=True).mean() ) # 速度转 m/s 后差分求加速度 df["speed_ms"] = df["speed_smooth"] / 3.6 df["accel"] = df.groupby("vehicle_id")["speed_ms"].diff() / df["dt"] # 再对加速度做一次平滑，抑制差分带来的尖峰 df["accel_smooth"] = df.groupby("vehicle_id")["accel"].transform( lambda s: s.rolling(window=3, min_periods=1, center=True).mean() ) print(df[["speed", "speed_smooth", "accel", "accel_smooth"]].describe())逻辑说明：两次平滑是关键。第一次平滑速度，第二次平滑加速度。窗口选 3 是因为 GPS 采样间隔大，窗口太大事件会被抹平。center=True让平滑值对齐当前点，避免相位偏移。参数上，如果采样间隔小于 5 秒，窗口可以放大到 5；如果间隔超过 30 秒，平滑意义不大，直接做宏观统计。min_periods=1保证首尾点不丢。
3.2 急加速与急减速的阈值设定
阈值没有绝对标准，行业里常用的分档是：加速度大于 2.5 m/s² 算急加速，小于 -3.0 m/s² 算急减速。这两个值来自车辆动力学和舒适性研究，但实际项目要按车型和业务调。
# 定义阈值 ACCEL_TH = 2.5 # 急加速阈值 m/s^2 DECEL_TH = -3.0 # 急减速阈值 m/s^2 # 标记事件 df["harsh_accel"] = df["accel_smooth"] > ACCEL_TH df["harsh_brake"] = df["accel_smooth"] < DECEL_TH # 统计每台车的事件次数 event_summary = df.groupby("vehicle_id").agg( accel_count=("harsh_accel", "sum"), brake_count=("harsh_brake", "sum"), total_points=("ts", "count") ).reset_index() event_summary["accel_per_100km"] = event_summary["accel_count"] / event_summary["total_points"] * 100 print(event_summary)逻辑说明：阈值直接决定事件数量，所以必须结合业务校准。accel_per_100km用点数做归一化是粗略做法，更准确的是用里程归一化，需要先累加位移距离。参数上，重型货车急加速阈值可以降到 1.5，因为货车动力弱，2.5 几乎触发不了；城市公交的急减速阈值可以放宽到 -2.5，因为频繁启停是常态。这里没有后悔药，阈值定错整份报告都偏。
3.3 急转弯识别：用方向角变化率而不是横向加速度
急转弯没有直接的横向加速度传感器时，用方向角变化率来近似。方向角是 GPS 上报的direction字段，单位度。转弯时方向角快速变化，变化率超过阈值即判定为急转弯。
# 方向角差分，处理 0/360 跨越 df["dir_diff"] = df.groupby("vehicle_id")["direction"].diff() # 把角度差归一化到 -180 到 180 df["dir_diff"] = (df["dir_diff"] + 180) % 360 - 180 # 方向角变化率，单位 度/秒 df["yaw_rate"] = df["dir_diff"] / df["dt"] # 急转弯阈值：方向角变化率超过 15 度/秒，且车速大于 20 km/h df["harsh_turn"] = (df["yaw_rate"].abs() > 15) & (df["speed"] > 20) print("急转弯事件数:", df["harsh_turn"].sum())逻辑说明：方向角跨越 0 度和 360 度时直接差分会出现 ±360 的跳变，(x + 180) % 360 - 180是标准归一化写法。加车速条件是因为低速掉头也会产生大方向角变化，但那不是危险驾驶。参数上，15 度/秒对应中等速度下的急弯，高速工况可以降到 10；车速门槛 20 km/h 是经验值，可按线路调整。这个指标对 GPS 方向角精度敏感，如果方向角字段缺失或噪声大，建议放弃急转弯识别，只做加减速和超速。
4. 从事件到评分：按车辆和司机的聚合与可视化
识别出事件只是中间产物，业务要的是排名和看板。这一章讲怎么把逐点事件聚合成车辆级、司机级指标，再用可视化把结论呈现出来。聚合的关键是归一化，否则跑得多的车事件数天然高，排名没有意义。
4.1 里程计算与事件密度归一化
归一化前必须先算里程。用前面 haversine 算出的相邻点距离累加即可，注意排除异常点。
# 排除异常点后累加里程 df_clean = df[~df["is_outlier"]].copy() df_clean["dist_km"] = df_clean["dist"] / 1000 mileage = df_clean.groupby("vehicle_id")["dist_km"].sum().reset_index() mileage.columns = ["vehicle_id", "total_km"] # 合并事件统计 event_summary = df_clean.groupby("vehicle_id").agg( accel_count=("harsh_accel", "sum"), brake_count=("harsh_brake", "sum"), turn_count=("harsh_turn", "sum") ).reset_index() score = mileage.merge(event_summary, on="vehicle_id") # 每百公里事件数 for col in ["accel_count", "brake_count", "turn_count"]: score[col + "_per100km"] = score[col] / score["total_km"] * 100 print(score.sort_values("brake_count_per100km", ascending=False).head(10))逻辑说明：先过滤异常点再算里程，否则漂移点会虚增里程。per100km是行业通用口径，方便横向对比。参数上，如果某台车总里程不足 10 公里，样本太小，建议单独标记不参与排名。合并用merge而不是concat，因为里程和事件统计是分别聚合的，索引不一定对齐。
4.2 驾驶行为评分模型：加权与分档
单一指标排名不够，业务通常要一个综合分。常见做法是对各事件密度做归一化后加权求和，再映射到 0 到 100 分。
from sklearn.preprocessing import MinMaxScaler # 选取指标列 cols = ["accel_count_per100km", "brake_count_per100km", "turn_count_per100km"] scaler = MinMaxScaler() score[cols + "_norm"] = scaler.fit_transform(score[cols]) # 加权：急减速权重最高，急加速次之，急转弯最低 weights = {"accel_count_per100km_norm": 0.35, "brake_count_per100km_norm": 0.45, "turn_count_per100km_norm": 0.20} score["risk_score"] = sum(score[k] * w for k, w in weights.items()) # 映射到 0-100，分数越高风险越大 score["risk_score"] = score["risk_score"] * 100 print(score[["vehicle_id", "risk_score"]].sort_values("risk_score", ascending=False))逻辑说明：MinMaxScaler把各指标压到 0 到 1，消除量纲差异。权重按业务风险定，急减速最容易导致追尾，权重给到 0.45。参数上，权重不是固定的，可以按车队历史事故数据做回归校准。如果样本里某指标全为 0，MinMaxScaler会产生除零，需要加判断或改用StandardScaler。这个评分是相对分，只适合同一批数据内部排名，跨批次比较要重新拟合。
4.3 用 matplotlib 出图：事件分布与车辆排名
分析结果最终要给人看。两张图最实用：一张是事件类型分布，一张是车辆风险排名。
import matplotlib.pyplot as plt plt.rcParams["font.sans-serif"] = ["SimHei"] # 中文显示 plt.rcParams["axes.unicode_minus"] = False fig, axes = plt.subplots(1, 2, figsize=(14, 5)) # 左图：事件类型总数 event_totals = score[["accel_count", "brake_count", "turn_count"]].sum() axes[0].bar(event_totals.index, event_totals.values, color=["#4C72B0", "#C44E52", "#55A868"]) axes[0].set_title("各类驾驶事件总数") axes[0].set_ylabel("事件次数") # 右图：风险分前 10 车辆 top10 = score.nlargest(10, "risk_score") axes[1].barh(top10["vehicle_id"].astype(str), top10["risk_score"], color="#C44E52") axes[1].set_title("风险评分前 10 车辆") axes[1].set_xlabel("风险评分") axes[1].invert_yaxis() plt.tight_layout() plt.savefig("driving_behavior_report.png", dpi=150)逻辑说明：barh横向条形图适合展示排名，invert_yaxis让最高分排在最上面。中文字体必须设置，否则显示方块。参数上，dpi=150兼顾清晰度和文件大小，用于报告足够。如果车辆数超过 30，建议只画前 20 或改用箱线图看分布。出图不是终点，把score表导出成 Excel 或写入数据库，才能接到看板工具里。
5. 避坑与排查：这类分析最容易翻车的五个地方
驾驶行为分析看起来是标准的数据处理，实际做起来坑很集中。下面五条是我在不同项目里反复遇到的，每条按现象、原因、解决写清楚。
5.1 采样间隔不一致导致加速度全错
现象：同一批数据里，有的车加速度算出来正常，有的车全是几十 m/s² 的离谱值。原因：不同终端的回传频率不一样，有的 10 秒一个点，有的 60 秒一个点，而代码里用固定窗口平滑，间隔大的车差分误差被放大。解决：先按dt分组，对不同采样间隔的数据分别设平滑窗口，或者统一重采样到固定间隔再做差分。重采样用resample配合插值，但插值会引入虚假平滑，事件识别要谨慎。
5.2 经纬度顺序写反导致距离算成零
现象：haversine 算出的距离全是 0 或极小值。原因：很多数据源里经纬度列的顺序是lat, lon，而函数签名是lon, lat，传参时没注意。解决：读入后先打印前几行确认列含义，或者用geopy这类库直接传经纬度字段。这个坑血泪经验，排查半天以为是坐标系问题，其实是参数顺序。
5.3 阈值直接套用导致事件数为零或爆炸
现象：急加速事件数为 0，或者占了总点数的 30%。原因：阈值没有按车型和采样率校准。采样间隔大时，平滑后的加速度峰值被削平，2.5 的阈值触发不了；采样密集时噪声大，阈值太低全是误报。解决：先画加速度分布直方图，看 95 分位和 99 分位在哪，再定阈值。我一般把阈值定在 99 分位附近，保证事件是少数异常。
5.4 跨天数据的时间戳时区不统一
现象：疲劳驾驶统计里，同一台车在凌晨出现大量驾驶时长。原因：部分终端上报 UTC 时间，部分上报本地时间，混在一起后跨天判断全乱。解决：读入时统一转成 UTC 或统一转成本地时间，用tz_localize和tz_convert处理。如果数据里没有时区标记，按数据来源分别处理，不要假设。
5.5 异常点过滤太狠导致里程偏短
现象：车辆总里程明显低于实际，事件密度虚高。原因：把速度异常点连同正常点一起删了，尤其是隧道和城市峡谷路段，连续多个点被标记为异常。解决：异常点只标记不删除，算里程时用相邻正常点插值补回，或者用地图匹配后的里程做基准。过滤比例超过 5% 就要警惕，超过 10% 说明清洗规则有问题。
6. 进阶技巧：把分析接到实时链路与看板
前面几章是离线批处理的完整链路，但车队管理往往要准实时。这一章讲两个进阶方向：一是把事件识别逻辑改造成流式处理，二是把结果接到看板工具。最后说一个我常用的验证习惯。
流式处理的核心是把按车分组的差分改成滑动状态。离线时用groupby + diff一次算完，实时时每台车维护一个最近 N 个点的缓冲区，新点到达时更新缓冲并计算加速度。下面是一个简化的流式事件检测骨架。
from collections import deque class VehicleState: def __init__(self, window=5): # 每台车维护最近 window 个 (ts, speed_ms) 点 self.buf = deque(maxlen=window) def update(self, ts, speed_kmh): self.buf.append((ts, speed_kmh / 3.6)) if len(self.buf) < 2: return None # 用首尾点算平均加速度，等价于滑动窗口差分 (t0, v0), (t1, v1) = self.buf[0], self.buf[-1] dt = (t1 - t0).total_seconds() if dt <= 0: return None accel = (v1 - v0) / dt if accel > 2.5: return "harsh_accel" if accel < -3.0: return "harsh_brake" return None # 模拟逐点输入 states = {} events = [] for row in df.itertuples(): st = states.setdefault(row.vehicle_id, VehicleState()) ev = st.update(row.ts, row.speed) if ev: events.append((row.vehicle_id, row.ts, ev)) print("流式检测事件数:", len(events))逻辑说明：deque的maxlen自动淘汰旧点，省去手动管理。用首尾点算平均加速度，和离线滑动窗口效果接近，但计算量恒定。参数上，窗口大小决定灵敏度，窗口小响应快但噪声大，窗口大平滑好但延迟高。实时场景一般取 3 到 5 个点。这个骨架没有处理异常点和丢点，生产环境要加上超时重置和异常过滤。
接到看板时，我一般把score表按固定周期写入数据库，看板工具直接查表。字段保留vehicle_id、risk_score、各事件密度、统计时间窗。这样看板不用做计算，只做展示，性能可控。验证分析结果是否靠谱，我有一个习惯：随机抽三台车，把它们的原始轨迹在地图上画出来，人工核对事件点位置是否合理。这一步能发现大部分阈值和清洗问题，比看统计报表有效得多。数据分析的误区之一就是只看聚合数字，不回看原始数据。希望帮到你。
本文还有配套的精品资源，点击获取
