1. 项目缘起：从“一堆数据”到“一个场景”

如果你正在做自动驾驶或者高级驾驶辅助系统（ADAS）的研究，尤其是跟车、变道、超车这些决策相关的算法，那你大概率听说过HighD数据集。这个由德国亚琛工业大学汽车工程研究所（ika）发布的自然驾驶数据集，在业内名气不小。它包含了德国高速公路上超过16.5万辆车、总里程超过4.5万公里的轨迹数据，精度高，场景丰富，是研究车辆微观行为的宝贵资源。

但问题来了，当你兴冲冲地下载了那几十个G的CSV文件，打开一看，可能会瞬间懵掉。数据字段密密麻麻，从车辆ID、帧号、到长宽高、速度、加速度、车道ID，再到前车后车ID，一应俱全。数据是有了，可怎么用呢？比如，你想研究“超车”这个行为。在HighD里，并没有一个现成的标签叫“is_overtaking”。你需要自己从海量的轨迹点中，把那些发生了超车行为的片段“挖”出来。

更进一步，研究超车不能只看主角车自己。超车是一个与周围环境，尤其是与被超车辆（邻近车辆）强交互的过程。你需要准确地筛选出在超车事件发生前后，与主车相关的邻近车辆数据。这个“筛选”的过程，就是本次要解决的核心问题：如何用Python，从HighD这一大盘“原料”里，精准地烹饪出“超车变道邻近车辆数据”这道“菜”。这不仅仅是写几行pandas的filter那么简单，它涉及到对数据结构的理解、对驾驶场景的定义、对时空关系的判断，以及如何高效处理大规模数据。接下来，我就结合自己的实操经验，把整个流程掰开揉碎了讲清楚。

2. 庖丁解牛：深入理解HighD数据结构与核心字段

在动手写代码之前，我们必须像熟悉自己的工具一样熟悉HighD的数据结构。盲目处理只会事倍功半。HighD数据集主要包含两个核心文件：*_tracks.csv（轨迹文件）和*_tracksMeta.csv（元数据文件）。此外还有记录场景信息的*_recordingMeta.csv，但对于车辆筛选，前两者是关键。

2.1 轨迹文件 (*_tracks.csv): 每一帧的瞬间状态

这个文件记录了所有车辆在每一帧下的状态。你可以把它想象成一个超大的Excel表，每一行是某个车辆在某个时刻的一张“快照”。它的核心字段包括：

• trackId:车辆的唯一标识符。这是串联所有数据的“主键”，在同一段录制（Recording）内是唯一的。
• frame: 帧编号。从0开始连续递增，代表了时间序列。HighD的帧率通常是25Hz（每秒25帧），所以相邻帧间隔0.04秒。
• x,y: 车辆中心在图像坐标系下的位置（单位：米）。特别注意：HighD使用的是图像坐标系，原点在左上角，x轴向右，y轴向下。这与我们通常理解的地图坐标系（北东天）或车辆坐标系（前左天）不同，在计算相对位置和距离时需要小心。
• xVelocity,yVelocity: 在x和y方向上的速度分量（米/秒）。
• xAcceleration,yAcceleration: 在x和y方向上的加速度分量（米/秒²）。
• frontSightDistance,backSightDistance: 到车道前/后可见距离内最近物体的距离。这个信息有时可用于辅助判断，但并非绝对可靠。
• dhw(Distance Headway),thw(Time Headway): 到前车（Preceding Vehicle）的距离和时间车头时距。这是筛选前车的直接依据。如果车辆前方没有车，这些值会被设为-1。
• precedingId,followingId: 在同一车道内，本车的前车和后车的trackId。这是筛选同车道邻近车辆的关键字段！如果不存在，值为0。
• leftPrecedingId,leftAlongsideId,leftFollowingId: 在左侧相邻车道，本车的前车、并行车、后车的trackId。
• rightPrecedingId,rightAlongsideId,rightFollowingId: 在右侧相邻车道，本车的前车、并行车、后车的trackId。
• laneId: 车辆所在车道的ID。通常从最左侧车道开始编号（如1, 2, 3...）。这是判断变道行为的核心依据之一。

注意：precedingId/followingId与leftPrecedingId等字段是互斥的。precedingId特指同车道的前车。当车辆变道时，这些关联ID会动态变化，这为我们捕捉交互对象提供了直接线索。

2.2 元数据文件 (*_tracksMeta.csv): 每辆车的全局信息

这个文件记录了每辆车在整个录制过程中的整体属性，一行对应一辆车（一个trackId）。重要字段包括：

• trackId: 与轨迹文件对应。
• initialFrame,finalFrame: 该车辆首次和末次出现的帧号。可以快速计算车辆存在的总时长。
• numFrames: 车辆出现的总帧数。
• width,length: 车辆的宽度和长度（米），用于计算安全边界或绘制车辆轮廓。
• class: 车辆类型（如Car,Truck等）。在研究超车行为时，你可能需要区分小车超卡车和卡车超小车等不同场景。

理解这些字段是第一步。接下来，我们需要基于这些字段，定义出什么是我们关心的“超车变道”事件。

3. 定义事件：如何从数据中识别一次“超车变道”

在代码中，我们不能模糊地说“找超车”，必须给出精确的、可量化的定义。一个典型的超车变道事件，可以分解为以下几个阶段，我们可以用数据字段来刻画每个阶段：

1. 跟随阶段：主车（Ego Vehicle）在原始车道（例如车道2）内，跟随一辆速度较慢的前车（precedingId有效且不为0）。此时，dhw（距离）和thw（时距）可能较小，主车速度可能低于期望速度。
2. 意图与准备阶段：主车产生超车意图。在数据上可能表现为：开始接近前车（dhw减小）、计算与相邻车道车辆的空间关系（观察leftPrecedingId,leftFollowingId等字段是否显示有足够空隙）。
3. 变道执行阶段：这是最关键的阶段。主车的laneId发生变化，例如从车道2变为车道1（向左变道）或车道3（向右变道）。在HighD中，一次laneId的跳变（如2->1）通常就代表了一次变道事件。我们需要找到laneId发生变化的起止帧。
4. 超越阶段：在主车变道至目标车道后，其x坐标（沿道路方向）逐渐超过原车道前车的x坐标。注意：由于HighD使用图像坐标系，x轴代表道路方向（录制方向），所以比较x坐标可以判断纵向位置的先后。
5. 并回阶段（可选）：在超越完成后，主车可能再次变道回原车道。这定义了完整的“超车-并回”行为。

基于以上阶段，我们可以定义一个超车变道事件的数据化条件：

• 必要条件：主车的laneId在连续帧中发生了变化（例如，在帧区间[100, 150]内，laneId从2变成了1）。
• 核心条件：在变道发生前的一段时间内（例如变道前2秒），主车在原车道有一个有效的precedingId（即存在一辆前车）。在变道发生后的一段时间内（例如变道后3秒），主车在目标车道的x坐标最终超过了那辆原前车的x坐标。
• 辅助条件（用于净化数据）：变道过程持续时间合理（例如1-5秒）；变道前后主车速度大于被超车辆速度；变道过程中与目标车道前后车保持安全距离（通过leftPrecedingId等关联车辆的相对距离判断）。

有了清晰的定义，我们就可以开始用Python来寻找这些事件了。首要任务是把数据高效地读进来。

4. 实战第一步：使用Pandas高效加载与预处理数据

HighD的单个轨迹文件就可能包含数十万行，直接全量读入内存可能效率低下。我们需要有策略地进行加载和初步过滤。

4.1 分块读取与初步筛选

如果你的内存充足，直接使用pandas.read_csv是最简单的。但为了稳健，尤其是处理多个文件时，可以考虑分块（chunk）读取。

import pandas as pd import numpy as np from pathlib import Path # 假设我们只处理一个录制文件 tracks_path = Path('./data/01_tracks.csv') meta_path = Path('./data/01_tracksMeta.csv') # 读取元数据，我们可能先根据元数据筛选车辆类型（如只选小汽车） df_meta = pd.read_csv(meta_path) car_ids = df_meta[df_meta['class'] == 'Car']['trackId'].tolist() # 读取轨迹数据，可以只读取我们关心的列，节省内存 usecols = ['trackId', 'frame', 'x', 'y', 'xVelocity', 'yVelocity', 'laneId', 'precedingId', 'followingId', 'leftPrecedingId', 'leftFollowingId', 'leftAlongsideId', 'rightPrecedingId', 'rightFollowingId', 'rightAlongsideId'] df_tracks = pd.read_csv(tracks_path, usecols=usecols) # 可选：如果只研究小汽车，提前过滤 df_tracks = df_tracks[df_tracks['trackId'].isin(car_ids)]

4.2 关键衍生字段计算

原始数据中的速度是分量，我们通常更关心车辆沿道路方向的速度（纵向速度）。由于HighD的x轴大致是道路方向，我们可以用xVelocity来近似。更严谨的做法是计算合速度在道路方向上的投影，但这需要道路方向向量，比较复杂。对于筛选和初步分析，用xVelocity作为纵向速度是常见的简化。

# 计算纵向速度（近似），取绝对值，因为xVelocity可能为负（如果录制方向与x轴相反？需根据recordingMeta确认） # 通常HighD中，车辆前进方向是x轴正方向 df_tracks['lonVelocity'] = df_tracks['xVelocity'] # 计算加速度 df_tracks['lonAcceleration'] = df_tracks['xAcceleration']

4.3 数据排序与分组

为了后续按车辆、按时间查找数据，正确的排序至关重要。

# 确保数据按车辆ID和帧号排序 df_tracks.sort_values(by=['trackId', 'frame'], inplace=True) # 重置索引 df_tracks.reset_index(drop=True, inplace=True)

预处理完成后，我们手头就是一个干净的、按车辆和时间排好序的DataFrame。接下来就是最核心的部分：算法实现。

5. 核心算法实现：逐车扫描，定位变道与筛选邻近车

我们的目标是：遍历每一辆可能作为主车的车辆，找出它所有的变道事件，并为每个事件筛选出相关的邻近车辆数据。这里采用一个清晰的、模块化的思路。

5.1 第一步：为单辆车找出所有变道点

我们首先定义一个函数，输入一辆车的轨迹数据（一个DataFrame），输出一个列表，列表中的每个元素是一个字典，代表一次变道事件，包含start_frame（变道开始帧）、end_frame（变道结束帧）、from_lane（起始车道）、to_lane（目标车道）。

def find_lane_changes_for_vehicle(vehicle_df): """ 找出单辆车的所有变道事件。 参数: vehicle_df: 单辆车的轨迹DataFrame，已按frame排序。 返回: list of dict: 每个dict包含一次变道事件的信息。 """ lane_changes = [] # 获取车道ID序列 lane_ids = vehicle_df['laneId'].values frames = vehicle_df['frame'].values i = 0 n = len(lane_ids) while i < n: current_lane = lane_ids[i] start_frame = frames[i] # 寻找车道ID变化的点 j = i + 1 while j < n and lane_ids[j] == current_lane: j += 1 # 如果j < n，说明在j处车道发生了变化 if j < n: # 变道结束帧可以认为是变化前的最后一帧，或者变化后的第一帧？这里定义为变化前的最后一帧。 # 更精确的做法是寻找车道ID变化的过渡区间，但HighD数据中变道通常发生在1-几帧内。 # 我们简单地将车道ID发生变化的帧区间[start_frame, frames[j])定义为一次变道。 # 实际上，frames[j]是新车道的起始帧。 end_frame = frames[j-1] # 旧车道的最后一帧 to_lane = lane_ids[j] # 新车道的ID lane_changes.append({ 'start_frame': start_frame, # 变道开始（旧车道最后一帧？这里需要明确） 'end_frame': end_frame, # 变道结束（旧车道最后一帧） 'change_frame': frames[j], # 新车道的起始帧，这是一个关键帧 'from_lane': current_lane, 'to_lane': to_lane }) # 从新车道的起始点开始继续搜索 i = j else: break # 上述简单算法可能会把短暂的GPS抖动识别为变道。需要过滤。 # 过滤条件：变道前后车道ID应该是合理的（如相邻车道），且变道过程持续时间不能太短。 filtered_changes = [] for lc in lane_changes: duration = lc['change_frame'] - lc['start_frame'] # 假设帧率25Hz，合理变道时间通常在1-4秒，即25-100帧。过滤掉持续时间过短（如<10帧）的“抖动”。 if duration >= 10 and abs(lc['to_lane'] - lc['from_lane']) == 1: filtered_changes.append(lc) return filtered_changes

5.2 第二步：判断一次变道是否为“超车”变道

不是所有变道都是为了超车。可能是为了下匝道、避让或单纯换道。我们需要用前面定义的核心条件来验证。

def is_overtaking_lane_change(vehicle_df, lane_change_event, df_tracks_all): """ 判断一次变道事件是否是一次超车变道。 参数: vehicle_df: 主车的轨迹DataFrame。 lane_change_event: find_lane_changes_for_vehicle返回的一个事件字典。 df_tracks_all: 完整的轨迹DataFrame，用于查找其他车辆。 返回: tuple: (bool, dict) 是否为超车变道，以及包含被超车辆ID等信息的详情字典。 """ ego_id = vehicle_df['trackId'].iloc[0] change_frame = lane_change_event['change_frame'] from_lane = lane_change_event['from_lane'] to_lane = lane_change_event['to_lane'] # 1. 寻找变道前，在主车原车道的前车 (precedingId) # 取变道前一段时间（如2秒，50帧）的数据 look_back_start = max(vehicle_df['frame'].min(), change_frame - 50) pre_change_data = vehicle_df[(vehicle_df['frame'] >= look_back_start) & (vehicle_df['frame'] < change_frame)] if pre_change_data.empty: return False, {} # 获取变道前最接近变道时刻的precedingId # 注意：precedingId可能在整个阶段变化，我们取最后时刻的值 last_preceding_id = pre_change_data.iloc[-1]['precedingId'] if pd.isna(last_preceding_id) or last_preceding_id == 0: # 变道前没有前车，可能不是跟随超车，可能是自由流变道 return False, {} # 2. 获取被超车辆（leading_vehicle）的轨迹 leading_df = df_tracks_all[df_tracks_all['trackId'] == last_preceding_id] # 3. 判断在变道后一段时间内，主车是否超越了前车 # 取变道后一段时间（如3秒，75帧）的数据 look_forward_end = min(vehicle_df['frame'].max(), change_frame + 75) post_change_data = vehicle_df[(vehicle_df['frame'] > change_frame) & (vehicle_df['frame'] <= look_forward_end)] if post_change_data.empty or leading_df.empty: return False, {} # 需要确保在变道后时间段内，被超车辆仍然存在 # 计算主车和被超车辆在变道后每一帧的纵向位置（x坐标） # 我们需要一个时间对齐的对比。可以取两个数据集在时间帧上的交集。 common_frames = np.intersect1d(post_change_data['frame'].values, leading_df['frame'].values) if len(common_frames) < 10: # 共同帧太少，无法可靠判断 return False, {} ego_x_post = post_change_data.set_index('frame')['x'].reindex(common_frames) lead_x_post = leading_df.set_index('frame')['x'].reindex(common_frames) # 判断在共同时间序列的后期，主车x是否持续大于被超车辆x（即完成了超越） # 可以看最后若干帧的平均位置差 if len(common_frames) > 20: check_frames = common_frames[-20:] else: check_frames = common_frames[-5:] ego_x_check = ego_x_post.loc[check_frames].mean() lead_x_check = lead_x_post.loc[check_frames].mean() is_overtaken = ego_x_check > lead_x_check if is_overtaken: # 还可以添加更多验证，例如变道前主车速度是否小于前车速度（有超车动机）？ # 变道前后主车速度是否提升？ return True, { 'ego_id': ego_id, 'leading_id': last_preceding_id, 'change_frame': change_frame, 'from_lane': from_lane, 'to_lane': to_lane, 'overtake_completion_frame': common_frames[-1] # 以最后共同帧作为超越完成参考 } else: return False, {}

5.3 第三步：为一次超车事件筛选所有相关邻近车辆

一旦确认了一次超车变道事件，我们就需要提取事件前后一个时间窗口内，所有与主车有交互的车辆数据。这包括：

1. 被超车辆（Leading Vehicle）：上面已经识别。
2. 原车道后车（Original Lane Following Vehicle）：变道前，主车在原车道的后车（followingId）。
3. 目标车道前车（Target Lane Preceding Vehicle）：变道后，主车在目标车道的前车（根据变道方向，取leftPrecedingId或rightPrecedingId在变道时刻的值）。
4. 目标车道后车（Target Lane Following Vehicle）：变道后，主车在目标车道的后车（leftFollowingId或rightFollowingId）。
5. 其他潜在相关车辆：如目标车道的并排车辆（alongsideId），或者更宽时间窗口内进入交互区域的车辆。

def get_surrounding_vehicles_for_event(ego_id, event_info, df_tracks_all, time_window_seconds=5.0): """ 获取一次超车事件前后时间窗口内，所有相关车辆的轨迹数据。 参数: ego_id: 主车ID。 event_info: is_overtaking_lane_change返回的详情字典。 df_tracks_all: 完整轨迹数据。 time_window_seconds: 时间窗口半长（秒），默认取事件前后5秒。 返回: dict: 键为角色（如'ego', 'leading', 'orig_following', ...），值为对应车辆的轨迹DataFrame（截取时间窗口内）。 """ frame_rate = 25.0 # HighD典型帧率 window_frames = int(time_window_seconds * frame_rate) event_frame = event_info['change_frame'] start_frame = max(0, event_frame - window_frames) end_frame = event_frame + window_frames # 1. 获取主车数据 ego_df = df_tracks_all[(df_tracks_all['trackId'] == ego_id) & (df_tracks_all['frame'] >= start_frame) & (df_tracks_all['frame'] <= end_frame)].copy() if ego_df.empty: return {} result = {'ego': ego_df} leading_id = event_info.get('leading_id') # 2. 获取被超车辆数据 if leading_id: leading_df = df_tracks_all[(df_tracks_all['trackId'] == leading_id) & (df_tracks_all['frame'] >= start_frame) & (df_tracks_all['frame'] <= end_frame)].copy() if not leading_df.empty: result['leading'] = leading_df # 3. 获取原车道后车（变道前一刻） # 找到变道前最近一帧主车的数据 ego_pre_change = ego_df[ego_df['frame'] < event_frame] if not ego_pre_change.empty: last_frame_pre = ego_pre_change['frame'].max() last_row_pre = ego_df[ego_df['frame'] == last_frame_pre].iloc[0] orig_following_id = last_row_pre['followingId'] if pd.notna(orig_following_id) and orig_following_id != 0: orig_following_df = df_tracks_all[(df_tracks_all['trackId'] == orig_following_id) & (df_tracks_all['frame'] >= start_frame) & (df_tracks_all['frame'] <= end_frame)].copy() if not orig_following_df.empty: result['original_following'] = orig_following_df # 4. 获取目标车道前车和后车（变道后一刻） # 找到变道后最近一帧主车的数据 ego_post_change = ego_df[ego_df['frame'] > event_frame] if not ego_post_change.empty: first_frame_post = ego_post_change['frame'].min() first_row_post = ego_df[ego_df['frame'] == first_frame_post].iloc[0] # 判断变道方向 from_lane = event_info['from_lane'] to_lane = event_info['to_lane'] if to_lane < from_lane: # 向左变道 target_preceding_id = first_row_post['leftPrecedingId'] target_following_id = first_row_post['leftFollowingId'] else: # 向右变道 target_preceding_id = first_row_post['rightPrecedingId'] target_following_id = first_row_post['rightFollowingId'] for role, tid in [('target_preceding', target_preceding_id), ('target_following', target_following_id)]: if pd.notna(tid) and tid != 0: vehicle_df = df_tracks_all[(df_tracks_all['trackId'] == tid) & (df_tracks_all['frame'] >= start_frame) & (df_tracks_all['frame'] <= end_frame)].copy() if not vehicle_df.empty: result[role] = vehicle_df return result

5.4 第四步：主循环与结果整合

最后，我们将上述模块组合起来，遍历所有车辆，收集所有超车事件及其邻近车辆数据。

def extract_all_overtaking_events(df_tracks, df_meta): """ 从整个数据集中提取所有超车事件及邻近车辆数据。 返回一个列表，每个元素是一个字典，代表一个完整的事件数据包。 """ all_events_data = [] unique_vehicle_ids = df_tracks['trackId'].unique() for ego_id in unique_vehicle_ids[:100]: # 示例：先处理前100辆车，避免耗时过长 ego_df = df_tracks[df_tracks['trackId'] == ego_id].copy() if len(ego_df) < 50: # 轨迹太短，忽略 continue # 找出该车的所有变道事件 lane_changes = find_lane_changes_for_vehicle(ego_df) for lc in lane_changes: is_overtake, event_info = is_overtaking_lane_change(ego_df, lc, df_tracks) if is_overtake: # 获取该事件相关的所有车辆数据 surrounding_data = get_surrounding_vehicles_for_event(ego_id, event_info, df_tracks, time_window_seconds=4.0) if surrounding_data: # 将事件信息和车辆数据打包 event_package = { 'event_info': event_info, 'surrounding_data': surrounding_data } all_events_data.append(event_package) print(f"发现超车事件: 主车 {ego_id} 在帧 {event_info['change_frame']} 超越 {event_info['leading_id']}") print(f"总共发现 {len(all_events_data)} 个超车事件。") return all_events_data

运行extract_all_overtaking_events函数，你就能得到一个列表，里面包含了每一个识别出的超车事件的所有关键数据。每个事件包都包含了主车、被超车以及其他邻近车辆在事件前后时间窗口内的完整轨迹。这些数据可以直接用于后续的可视化、特征提取和模型训练。

6. 避坑指南与性能优化实战经验

理论很美好，但实际处理数据时总会遇到各种坑。下面分享几个我踩过坑后总结的经验。

6.1 坐标系的坑：图像坐标系与真实距离

HighD使用的是图像坐标系（x向右，y向下）。这意味着：

• 相对位置计算：车辆A在车辆B的“左边”，并不意味着在真实道路上A在B的左侧车道。你需要根据laneId来判断车道关系，leftPrecedingId等字段已经帮你做好了这种基于车道的关联，比自己用x,y坐标算更可靠。
• 纵向距离：两车在道路方向上的距离，近似等于它们x坐标的差值（假设道路是水平的且与x轴平行）。但严格来说，应该使用dhw（车头时距）字段，它提供了沿车道中心线的距离。在筛选“近距离”车辆时，优先使用dhw，而不是欧几里得距离。
• 速度方向：xVelocity在车辆前进方向与x轴正方向一致时为正。在计算相对速度时要注意符号。

6.2 数据噪声与异常值的处理

• 车道ID抖动：车辆在车道内行驶时，laneId偶尔会有1-2帧的跳变，这通常不是真实的变道。这就是为什么在find_lane_changes_for_vehicle函数中，我们需要过滤持续时间过短的“变道”。阈值可以设为10-15帧（0.4-0.6秒）。
• 关联ID的瞬时消失：precedingId等字段可能在连续帧中短暂变为0或NaN，然后又恢复。这可能是传感器误检或遮挡。在判断“前车”时，最好看一个时间窗口内的稳定状态，而不是单帧数据。例如，在is_overtaking_lane_change函数中，我们查看变道前2秒的数据来确认前车。
• 缺失帧：虽然HighD数据很连续，但极端情况下可能有帧丢失。在按帧对齐不同车辆的数据时（如判断超越时刻），使用np.intersect1d找共同帧是稳健的做法。

6.3 大规模数据处理的性能优化

当处理多个录制文件（几十GB）时，纯Python循环会非常慢。以下是一些优化策略：

1. 向量化操作：尽量使用pandas和numpy的向量化函数，避免在DataFrame上使用apply或逐行循环。例如，计算所有车辆的纵向速度，直接用df['xVelocity']赋值。
2. 使用数据索引：对df_tracks按trackId和frame建立索引可以大幅加速查询。

   df_tracks_indexed = df_tracks.set_index(['trackId', 'frame']).sort_index() # 查询特定车辆特定帧的数据会很快 # ego_data = df_tracks_indexed.loc[(ego_id, slice(start_frame, end_frame)), :]

   但注意，多级索引在某些操作上可能不如布尔索引直观。
3. 分治与并行：最耗时的部分是遍历每辆车找变道事件。这个过程是独立的，可以并行。
   • 按车辆分组并行：使用joblib或multiprocessing库，将车辆ID列表分成若干份，分配给多个进程同时处理。

   from joblib import Parallel, delayed def process_one_vehicle(vid, df_tracks, df_meta): # 提取该车辆数据并运行上述查找逻辑 # 返回该车辆发现的所有事件列表 pass all_vehicle_ids = df_tracks['trackId'].unique() results = Parallel(n_jobs=4)(delayed(process_one_vehicle)(vid, df_tracks, df_meta) for vid in all_vehicle_ids[:200]) # 然后合并results

   • 使用Dask：如果数据量极大，可以考虑使用Dask库进行分布式计算，它提供了类似pandas的接口但能处理超出内存的数据。
4. 选择性加载：一开始就通过usecols参数只加载必需的列，能极大减少内存占用和I/O时间。
5. 结果缓存：识别超车事件的计算成本很高。一旦计算完成，应将结果（all_events_data）保存为文件（如pickle或parquet格式），后续分析直接加载结果，避免重复计算。

6.4 定义超车事件的阈值调参

在is_overtaking_lane_change函数中，我们使用了几个关键阈值：

• look_back_start = change_frame - 50：查看变道前2秒的数据来判断前车。
• look_forward_end = change_frame + 75：查看变道后3秒的数据来判断是否完成超越。
• duration >= 10：过滤短于10帧（0.4秒）的虚假变道。

这些阈值需要根据你的具体研究问题进行调整。例如，研究激进超车可能用更短的时间窗口，研究保守超车可能用更长的。建议在少量数据上可视化几个案例，来验证你的阈值是否合理。

7. 从数据到洞察：可视化验证与特征提取

算法跑完了，怎么知道抓取得对不对？可视化是必不可少的验证步骤。同时，提取出来的数据需要转化为可用于机器学习模型的特征。

7.1 使用Matplotlib进行事件可视化

我们可以绘制单个超车事件中，所有相关车辆在时间-空间（x坐标）上的轨迹。

import matplotlib.pyplot as plt def plot_overtaking_event(event_package, df_tracks): event_info = event_package['event_info'] surrounding = event_package['surrounding_data'] fig, ax = plt.subplots(figsize=(12, 6)) ego_id = event_info['ego_id'] change_frame = event_info['change_frame'] # 为每种角色定义颜色和标签 roles_config = { 'ego': {'color': 'red', 'label': 'Ego', 'linewidth': 3}, 'leading': {'color': 'blue', 'label': 'Leading', 'linewidth': 2}, 'original_following': {'color': 'green', 'label': 'Orig Follow', 'linestyle': '--'}, 'target_preceding': {'color': 'orange', 'label': 'Target Precede', 'linestyle': ':'}, 'target_following': {'color': 'purple', 'label': 'Target Follow', 'linestyle': ':'}, } for role, df in surrounding.items(): if role in roles_config: config = roles_config[role] # 绘制纵向位置 (x) 随时间 (frame) 的变化 ax.plot(df['frame'], df['x'], color=config['color'], label=config['label'], linewidth=config.get('linewidth', 1), linestyle=config.get('linestyle', '-')) # 标记变道时刻 ax.axvline(x=change_frame, color='black', linestyle='--', alpha=0.5, label='Lane Change') ax.set_xlabel('Frame') ax.set_ylabel('Longitudinal Position (x) [m]') ax.set_title(f'Overtaking Event: Ego {ego_id} -> Leading {event_info.get("leading_id", "N/A")}') ax.legend() ax.grid(True, alpha=0.3) plt.tight_layout() plt.show() # 可视化第一个事件 if all_events_data: plot_overtaking_event(all_events_data[0], df_tracks)

通过看图，你可以直观地检查：主车（红色）是否在变道时刻（黑色虚线）后，其轨迹（x坐标）超过了被超车辆（蓝色）。其他车辆的轨迹也能帮助你理解整个交互场景。

7.2 提取机器学习特征

对于每个超车事件，我们可以从surrounding_data中提取一系列特征，用于后续的驾驶行为分类、风险评估等模型。特征可以分为几类：

• 主车特征：变道前的平均速度、加速度、与前车的dhw、thw。
• 相对运动特征：主车与被超车的相对速度、相对距离（在变道前、变道时刻、超越完成时刻）。
• 间隙特征：变道时，目标车道前后车的距离（dhw与thw的变体），即切入间隙和跟随间隙。
• 时间特征：变道持续时间、从产生意图到开始变道的时间、超越全过程时间。
• 交互特征：是否存在并排车辆、后车是否加速逼近等。

例如，提取变道前主车状态的特征：

def extract_features_from_event(event_package): event_info = event_package['event_info'] surrounding = event_package['surrounding_data'] features = {} ego_df = surrounding.get('ego') leading_df = surrounding.get('leading') change_frame = event_info['change_frame'] if ego_df is not None and leading_df is not None: # 1. 变道前主车状态（取变道前1秒的数据平均） pre_change = ego_df[ego_df['frame'] < change_frame] last_second = pre_change[pre_change['frame'] >= (change_frame - 25)] # 前1秒 if not last_second.empty: features['ego_pre_speed_mean'] = last_second['lonVelocity'].mean() features['ego_pre_acc_mean'] = last_second['lonAcceleration'].mean() # 2. 变道时刻的间隙（需要从原始数据中获取，这里简化） # 假设我们能从ego_df中拿到变道时刻的leftPrecedingDistance等（实际需计算） # features['gap_to_target_lead'] = ... # 3. 超越完成时的相对速度 # 找到超越完成帧附近的数据 completion_frame = event_info.get('overtake_completion_frame', change_frame + 50) ego_post = ego_df[ego_df['frame'] >= completion_frame].iloc[0] if not ego_df[ego_df['frame'] >= completion_frame].empty else None lead_post = leading_df[leading_df['frame'] >= completion_frame].iloc[0] if not leading_df[leading_df['frame'] >= completion_frame].empty else None if ego_post is not None and lead_post is not None: features['relative_speed_at_completion'] = ego_post['lonVelocity'] - lead_post['lonVelocity'] # ... 可以提取更多特征 return features

将这些特征提取出来，组织成一个新的DataFrame，就可以作为机器学习模型的输入了。

整个流程从数据加载、事件定义、算法实现、到优化避坑和结果应用，形成了一个完整的闭环。处理HighD这样的数据集，核心在于对数据字段的深刻理解和对驾驶场景的准确定义。代码本身并不复杂，但其中的逻辑和细节决定了你提取的数据质量。希望这篇近万字的详细拆解，能让你在处理HighD或其他类似轨迹数据时，思路更清晰，操作更顺畅。