简介这份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(dropTrue) # 基础体检缺失率、时间跨度、采样间隔 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(window3, min_periods1, centerTrue).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(window3, min_periods1, centerTrue).mean() ) print(df[[speed, speed_smooth, accel, accel_smooth]].describe())逻辑说明两次平滑是关键。第一次平滑速度第二次平滑加速度。窗口选 3 是因为 GPS 采样间隔大窗口太大事件会被抹平。centerTrue让平滑值对齐当前点避免相位偏移。参数上如果采样间隔小于 5 秒窗口可以放大到 5如果间隔超过 30 秒平滑意义不大直接做宏观统计。min_periods1保证首尾点不丢。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, onvehicle_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, ascendingFalse).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, ascendingFalse))逻辑说明MinMaxScaler把各指标压到 0 到 1消除量纲差异。权重按业务风险定急减速最容易导致追尾权重给到 0.45。参数上权重不是固定的可以按车队历史事故数据做回归校准。如果样本里某指标全为 0MinMaxScaler会产生除零需要加判断或改用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, dpi150)逻辑说明barh横向条形图适合展示排名invert_yaxis让最高分排在最上面。中文字体必须设置否则显示方块。参数上dpi150兼顾清晰度和文件大小用于报告足够。如果车辆数超过 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, window5): # 每台车维护最近 window 个 (ts, speed_ms) 点 self.buf deque(maxlenwindow) 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、各事件密度、统计时间窗。这样看板不用做计算只做展示性能可控。验证分析结果是否靠谱我有一个习惯随机抽三台车把它们的原始轨迹在地图上画出来人工核对事件点位置是否合理。这一步能发现大部分阈值和清洗问题比看统计报表有效得多。数据分析的误区之一就是只看聚合数字不回看原始数据。希望帮到你。本文还有配套的精品资源点击获取