基于Python的车辆驾驶行为分析与司机聚类评分实战

简介：以运输车辆驾驶行为分析为案例的Python数据分析实战教程，面向物流与交通行业数据分析人员，也适合正在学习Pandas、NumPy、Matplotlib、Seaborn等库的Python初学者。资料从环境配置讲起，说明如何采集车辆GPS定位、速度、加速度、急加速急减速事件，并结合气象与道路状况数据；随后系统讲解Pandas读取CSV/Excel数据、缺失值与异常值处理、时间戳转换等预处理。特征工程部分重点演示急加速急减速事件识别、每公里急加速急减速次数、速度标准差、加速度标准差等特征构建，提供可直接运行的示例代码。分析环节涵盖描述性统计、速度分布直方图、按车辆ID的驾驶行为箱线图、加速度变化曲线和相关性热力图，并结合结果讨论驾驶员安全培训与路线优化应用。资源为单个PDF文档，体积仅126KB，内容精炼，适合作为课程案例、实训项目或毕业设计参考。目前已有458人学习浏览，读者可从中获得完整数据分析流程、业务分析思路和可迁移到实际项目的代码模板。

1. 货车怎么开最费油，数据说了算

车辆驾驶行为分析在物流和运输行业一直有个老问题：同样的路线、同样的车型，不同司机开出来的油耗和事故率能差出 20% 以上。靠老司机带新人、靠安全员盯监控，既慢又覆盖不全。我的做法是把车载 GPS 终端和 CAN 总线里埋了几年的原始数据翻出来，用 Python 做一轮完整的「采集—清洗—特征工程—建模—可视化」分析，用聚类算法把司机分成激进型、平稳型、经济型，再结合超速、急加速、急刹、急转弯等事件给每个司机出驾驶行为评分。这篇文章就把这条链路里能直接照抄的代码、参数和坑，按实操顺序拆开讲。适合手里已有车辆轨迹 CSV 或 数据库表、但还没跑通完整分析流程的 Python 数据分析工程师。

2. 驾驶行为分析的原始数据长什么样，先做清洗再说

2.1 GPS 轨迹表和 CAN 报文表的常见字段

做车辆驾驶行为分析，数据来源无非两路：OBD 终端回传的 GPS 轨迹，和 CAN 总线解析出的车辆状态报文。GPS 轨迹表里通常有车辆编号、定位时间、经度、纬度、瞬时速度、方向角，CAN 报文表里则有发动机转速、油门踏板开度、制动踏板状态、横向加速度这些高频字段。两张表通过车辆编号和时间戳关联，但时间精度往往不一致——GPS 是秒级，CAN 是毫秒级，第一步就是统一时间基准。

SELECT vehicle_no, DATE_FORMAT(gps_time, '%Y-%m-%d %H:%i:%s') AS ts_sec, lng, lat, speed_kmh, heading FROM gps_log WHERE gps_date = '2024-06-01' AND lng BETWEEN 73 AND 135 AND lat BETWEEN 18 AND 54;

这段 SQL 在做两件事：截断毫秒时间戳，方便后续和 CAN 表做关联；同时把经纬度限制在中国范围内，过滤掉漂移点。如果车辆去了边境或园区外，说明定位模块可能被遮挡或天线故障。还需要说明的是，lng、lat 越界的直接丢弃，因为轨迹漂移点会在后面算里程和转角时放大误差，宁可丢不能留。

2.2 pandas 处理缺失值和重复轨迹

实际拿到的 CSV 用 pandas 读进来之后，第一件要检查的事情是重复 GPS 点——很多终端在信号弱或停车时会原地刷新，产生同一时间戳、同一经纬度的多条记录。重复点不处理，算出来的单日里程会偏大，急加速事件次数也会被误判。

import pandas as pd import numpy as np df = pd.read_csv('vehicle_gps_2024.csv', parse_dates=['gps_time']) df = df.drop_duplicates(subset=['vehicle_no', 'gps_time']) df = df.sort_values(['vehicle_no', 'gps_time']).reset_index(drop=True) # 停车状态标记：速度接近0且超过3分钟视为停车 df['is_stop'] = (df['speed_kmh'] < 0.5).astype(int) df['stop_group'] = (df['is_stop'].diff() != 0).cumsum() stop_mask = df['is_stop'] == 1 df.loc[stop_mask, 'stop_duration_min'] = ( df.loc[stop_mask, 'gps_time'].groupby(df.loc[stop_mask, 'stop_group']).diff().dt.total_seconds() / 60 ) df['stop_duration_min'] = df['stop_duration_min'].fillna(0) df = df[df['stop_duration_min'] <= 30]

去重时保留第一条即可，因为同一秒内的坐标差异对速度计算没有意义；is_stop用 0.5 km/h 的阈值是为了兼容低速蠕行工况，比如在停车场挪车或堵车缓行，这类点如果在后续用于计算转角或加速度，会制造大量伪急转弯事件。stop_duration_min超过 30 分钟的高速公路停车可能是事故或严重故障，这类轨迹段会显著影响作息判断，直接切掉比修复更稳妥。

3. 驾驶行为特征工程：从原始轨迹算出超速、急刹车和急转弯

3.1 加速度计算与急加速急减速识别

驾驶行为分析里的急加速、急减速、急转弯都没有一个全国统一标准，物流公司通常参照 JT/T 883 营运车辆安全运行标准或者自己企业制定的阈值。常见做法是用相邻两个 GPS 点的速度差除以时间差得到纵向加速度，再进行滑窗平滑，消除定位噪声带来的毛刺。

df = df.sort_values(['vehicle_no', 'gps_time']).reset_index(drop=True) df['dt_sec'] = df.groupby('vehicle_no')['gps_time'].diff().dt.total_seconds() df['v_prev'] = df.groupby('vehicle_no')['speed_kmh'].shift(1) df['accel_ms2'] = (df['speed_kmh'] - df['v_prev']) * 1000 / 3600 / df['dt_sec'].clip(lower=1) # 用滚动窗口去噪，窗口内取中位数 df['accel_smooth'] = df['accel_ms2'].rolling(5, min_periods=2, center=True).median() hard_accel = (df['accel_smooth'] > 2.5) & (df['speed_kmh'] > 20) hard_brake = (df['accel_smooth'] < -2.5) & (df['speed_kmh'] > 20) # 聚合到每次行程 trip_id = (df['vehicle_no'].ne(df['vehicle_no'].shift()) | (df['gps_time'].diff().dt.total_seconds() > 1800)).cumsum() df['trip_id'] = trip_id event_cnt = df.groupby('trip_id').agg( 急加速次数=('accel_smooth', lambda s: ((s > 2.5) & (df.loc[s.index, 'speed_kmh'] > 20)).sum()), 急刹车次数=('accel_smooth', lambda s: ((s < -2.5) & (df.loc[s.index, 'speed_kmh'] > 20)).sum()) ).reset_index()

加速度阈值 2.5 m/s² 对应约 9 km/h 每秒的变化率，比一般家用车的日常起步略大，比紧急制动小。低于 20 km/h 时的急加速不统计，因为红绿灯起步频繁、速度本就低，容易把正常启动误判为激进驾驶，这也是在做规则提取时最容易搞错的细节——只设加速度阈值、不设下限速度，会让市区司机的不良驾驶事件数量虚高。dt_sec.clip(lower=1)防止设备重复上报瞬间时间差为 0 导致除零。

3.2 用转向角变化率识别急转弯和变道

横向行为的判定仅靠 GPS 不够，最好叠加陀螺仪 z 轴角速度。如果没有陀螺仪，可以用方向角 heading 的差分近似横向角速度，但 GPS 方向角在低速时抖动很大，需要先平滑。

df['heading_smooth'] = df.groupby('vehicle_no')['heading'].transform( lambda s: s.rolling(5, min_periods=2, center=True).apply( lambda x: np.degrees(np.arctan2(np.sin(np.radians(x)).mean(), np.cos(np.radians(x)).mean())) ) ) df['heading_diff'] = df.groupby('vehicle_no')['heading_smooth'].diff() df['heading_diff'] = ((df['heading_diff'] + 180) % 360) - 180 # 映射到[-180, 180] df['yaw_rate_dps'] = df['heading_diff'] / df['dt_sec'].clip(lower=1) sharp_turn = (df['yaw_rate_dps'].abs() > 30) & (df['speed_kmh'] > 30)

方向角平滑这里用了圆周均值法，直接用普通rolling().mean()会在 0° 和 360° 交叉处产生跳变，比如 350° 和 10° 的平均值会算出 180° 这种完全错误的结果。圆周均值先转成 sin、cos 再求均值，再转回角度，能天然处理角度环绕。角速度阈值给到 30°/s，约等于 0.52 rad/s，对应半径比较小的弯道或紧急变道；速度阈值取 30 km/h 是为了排除停车场里原地打轮的情况——低速时方向角本来就会快速变化，但车身并没有发生真正的横向失稳。

3.3 分箱统计驾驶行为评分

有了事件明细后，最终要落到每个司机的可对比评分。我用的是分箱加权法，先按每百公里的急加速/急刹车/急转弯次数分档，再给不同档位赋分，综合得到 0-100 的驾驶评分。这里不直接用原始次数做归一化，是因为不同司机跑的城市、山区、高速占比不一样，绝对值不可比。

trip_stats = event_cnt.merge(trip_km, on='trip_id') trip_stats['rate_accel'] = trip_stats['急加速次数'] / trip_stats['里程_km'] * 100 trip_stats['rate_brake'] = trip_stats['急刹车次数'] / trip_stats['里程_km'] * 100 trip_stats['rate_turn'] = trip_stats['急转弯次数'] / trip_stats['里程_km'] * 100 def score_by_rate(rate, breakpoints, scores): # breakpoints 如 [2, 4, 6, 8]，scores 如 [100, 85, 70, 50, 30] for bp, sc in zip(breakpoints, scores): if rate <= bp: return sc return scores[-1] trip_stats['驾驶评分'] = ( trip_stats['rate_accel'].apply(lambda r: score_by_rate(r, [2,4,6,8], [100,85,70,50,30])) * 0.3 + trip_stats['rate_brake'].apply(lambda r: score_by_rate(r, [1.5,3,5,7], [100,85,70,50,30])) * 0.5 + trip_stats['rate_turn'].apply(lambda r: score_by_rate(r, [0.5,1,2,3], [100,85,70,50,30])) * 0.2 )

权重分配急刹车最高，是因为急刹车不仅费刹车片，还和追尾事故相关性最强；急加速次之，对油耗和传动系统冲击大；急转弯只占 20%，因为它的样本量最小、噪声也最大。分箱阈值要随车队实际水平迭代，我见过油耗极高但事件次数并不突出的车队，说明他们的驾驶行为问题是长时间高速巡航或怠速，而不是激烈的加减速，这时需要再加一个超速时长占比特征。

4. 用聚类算法给司机分型，KMeans 和 DBSCAN 怎么选

4.1 特征归一化与 PCA 降维

事件统计做完之后，每个司机就变成了一行特征向量。我通常会挑 6 个特征进聚类模型：平均车速、超速时长占比、急加速率、急刹率、急转弯率、夜间行驶占比。这六个特征不在一个量纲上——平均车速可能是 60，急转弯率可能只有 0.3，如果不做标准化，聚类结果会被平均车速主导，其他特征等于没喂进去。

from sklearn.preprocessing import StandardScaler from sklearn.decomposition import PCA from sklearn.cluster import KMeans, DBSCAN import matplotlib.pyplot as plt feature_cols = ['平均车速_kmh', '超速时长占比', '急加速率', '急刹率', '急转弯率', '夜间行驶占比'] X = driver_feat[feature_cols].fillna(0) scaler = StandardScaler() X_scaled = scaler.fit_transform(X) pca = PCA(n_components=2, random_state=42) X_pca = pca.fit_transform(X_scaled) print('前两个主成分解释方差比：', pca.explained_variance_ratio_.sum())

PCA 降维主要不是为了聚类本身，而是为了在二维平面上画散点图验证聚类效果。如果前两个主成分累计解释方差不到 70%，说明特征维度间的相关性不强，或者信息被摊得很散，这时可以考虑剔除某些冗余特征再跑一次。有些资料会直接把原始高维特征丢进 KMeans，再用 TSEN 降维画图，但 TSNE 的布局受困惑度参数影响很大，同一份数据两次运行的结果形状可能不一样，用于内部验证可以，用于给管理层汇报很容易被质疑。

4.2 肘部法则和轮廓系数定聚类数量

KMeans 的 K 值不能拍脑袋定 3 或 4，同一个数据集在 3 类和 5 类下的业务解释完全不同。我用的是两个指标交叉验证：肘部法则看簇内误差平方和下降的拐点，轮廓系数看同一类内的紧凑度和不同类间的分离度，两者结合再决定最终 K 值。

from sklearn.metrics import silhouette_score inertia_list, sil_list = [], [] K_range = range(2, 7) for k in K_range: km = KMeans(n_clusters=k, random_state=42, n_init=10) labels = km.fit_predict(X_scaled) inertia_list.append(km.inertia_) sil_list.append(silhouette_score(X_scaled, labels)) for k, inert, sil in zip(K_range, inertia_list, sil_list): print(f'K={k}, 簇内SSE={inert:.2f}, 轮廓系数={sil:.4f}')

轮廓系数的取值范围是 -1 到 1，大于 0.5 说明聚类结构合理，0.3 到 0.5 说明有重叠但仍可用，低于 0.25 就基本等于随机划分，可以考虑换特征或换距离度量。注意 silhouette_score 对样本量敏感，司机总数少于 50 人时轮廓系数的波动会很大，这时我更倾向直接看每个簇的特征均值，手工验证业务上是否分得开——毕竟给管理层解释「轮廓系数 0.42」远不如「激进型司机百公里急刹 3 次，经济型司机 0.4 次」来得直观。

4.3 DBSCAN 处理异常驾驶风格

KMeans 的问题在于它强迫每个司机都进一个簇，哪怕这个司机的驾驶数据完全不可信——比如 GPS 信号丢失后补传的伪轨迹导致车速在 0 到 90 之间反复跳变。这种数据在特征空间里就是个离群点，KMeans 会硬把它塞进最近的某个簇，从而拉偏那个簇的质心。DBSCAN 的好处是不需要预设类别数，并且能把噪声点单独标成 -1。

db = DBSCAN(eps=1.8, min_samples=5, metric='euclidean') db_labels = db.fit_predict(X_scaled) noise_mask = db_labels == -1 print(f'噪声样本数：{noise_mask.sum()}，占比 {noise_mask.mean():.1%}') driver_feat['聚类标签'] = db_labels driver_feat.loc[noise_mask, '聚类标签'] = '异常数据'

DBSCAN 的两个参数中，min_samples 推荐取特征维数的 2 倍再加 1，即 6 个特征就设 13 左右；但如果司机样本总数少，min_samples 设 13 会导致几乎所有点都变成噪声，要结合样本量缩小到 5。eps 我一般从 1.5 开始迭代，每次加 0.1，观察噪声占比曲线的变化——噪声占比突然下降的那段往往对应合适的 eps。聚类完之后，异常数据单独建一张表，不参与车队平均值的计算，同时标记给设备维护人员去查终端状态，这比直接删掉更严谨。

5. 用聚类结果反查事件明细，定位需要干预的具体路段和时间段

聚类模型输出的只是司机层面的结论，真正要落地的安全管理和油耗管理，必须能下钻到具体的时间段和地点。我一般会再把每个簇的司机名单回放到原始 GPS 表里，按路段分组统计事件密集区域，画核密度热力图。这种可视化用 folium 生成交互页面比 matplotlib 静态图更实用，可以直接发给车队长去对应路线上实地核查。

import folium from folium.plugins import HeatMap agg_df = df[df['急加速标志'] == 1][['lat', 'lng']].dropna() m = folium.Map(location=[agg_df['lat'].mean(), agg_df['lng'].mean()], zoom_start=12) HeatMap(agg_df[['lat', 'lng']].values.tolist(), radius=12, blur=18, max_zoom=10).add_to(m) m.save('hard_accel_hotmap.html')

HeatMap 的 radius 参数控制单个点的影响半径，值太大会让热点连成一片，看不出具体路口，我习惯先设 12，缩放级别到 12 左右时基本能精确到路口尺度。这里还需要注意一个常见误区：直接用所有点画热力图会被高频通行的路段淹没，正确做法是只取事件标志为 1 的点做核密度，而不是用全部轨迹做底图。另外 folium 依赖网络加载 OpenStreetMap 底图，在内网部署的服务器上出不来图，解决方案是本地下载底图瓦片目录，或者改用 pyecharts 的地图组件。

夜间驾驶行为的识别也是这里值得单独做的一块。我判断夜间时段不是用固定 22 点到 6 点，而是用日出日落时间动态计算，因为新疆和内地的时差导致同一标准时间下的光照条件完全不同。可以用astral或suntime这个库传入经纬度和日期计算日出日落，再判断每条定位记录是否为夜行。

from astral import LocationInfo from astral.sun import sun import datetime city = LocationInfo('乌鲁木齐', 'China', 'Asia/Shanghai', 43.82, 87.62) s = sun(city.observer, date=datetime.date(2024, 6, 1)) sunset_ms = s['sunset'].hour * 60 + s['sunset'].minute df['夜间'] = (df['gps_time'].dt.hour * 60 + df['gps_time'].dt.minute >= sunset_ms).astype(int)

用动态日落时间算夜行占比，比用固定 18:00 判定的好处在于：夏季新疆 21:30 天还是亮的，用固定阈值会错判大量正常驾驶为夜行驾驶，导致夜行疲劳评分失真。得出的夜间驾驶特征喂回第 4 章的特征矩阵里，就能把「白天正常、夜晚激进」这类混合型司机从纯激进群里分出来，干预方式是完全不同的——前者要调整排班，后者才需要培训。这套「聚类分群—回查事件—时间地点定位—差异干预」的流程，是整个驾驶行为分析项目最有交付价值的部分，比单出一份司机排名表有用得多。

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