4、轨迹数据预处理:数据清洗、插值与压缩

大家好,我是老张。做轨迹回放这些年,我踩过最多的坑,其实不在算法本身,而在数据预处理阶段。

你想想看,拿到的原始轨迹数据,那叫一个「脏」。GPS漂移、重复点、丢帧……直接扔进算法里,结果肯定一塌糊涂。我刚开始带项目时,就吃过这个亏——回放出来的轨迹,车辆直接穿墙而过,客户当场黑脸。

所以今天咱们聊聊数据预处理。说白了,就是给原始数据「洗个澡」,让它变得干净、连续、轻量。我习惯把预处理分成三步:清洗、插值、压缩。下面这张图能帮你快速建立整体认知:

轨迹数据预处理核心流程 原始轨迹数据 步骤1:数据清洗 去重 · 去噪 步骤2:数据插值 线性插值 · 样条插值 步骤3:数据压缩 道格拉斯-普克算法

4.1 数据清洗:去重与去噪

数据清洗,我习惯先做去重,再做去噪。顺序别搞反了——先去掉重复点,再去处理异常点,逻辑上更顺畅。

4.1.1 去重

重复点怎么来的?最常见的情况是:设备在同一个位置连续上报了多次。比如车辆等红灯时,GPS没动,但上报了5条几乎一样的数据。

我个人习惯的做法是:按时间戳排序后,判断相邻点的经纬度差。如果差值小于某个阈值(比如0.00001度,约1米),就认为是重复点,只保留第一个。

核心思路: 时间连续 + 空间静止 → 判定为重复
def remove_duplicates(points, threshold=0.00001):
    """
    points: [(lat, lng, timestamp), ...]
    threshold: 经纬度差值阈值
    """
    cleaned = [points[0]]
    for i in range(1, len(points)):
        lat_diff = abs(points[i][0] - points[i-1][0])
        lng_diff = abs(points[i][1] - points[i-1][1])
        if lat_diff > threshold or lng_diff > threshold:
            cleaned.append(points[i])
    return cleaned
小技巧: 阈值别设太死。城市道路和高速场景不一样,我一般会根据采样频率动态调整。比如1秒采一次,阈值设0.00001;10秒采一次,阈值可以放大到0.00005。

4.1.2 去噪

去噪,说白了就是干掉那些「离谱」的点。GPS信号受高楼、隧道、天气影响,偶尔会蹦出一个离群点。比如车辆明明在A点,突然跳到了500米外的B点,下一秒又跳回来。

我曾经在做一个物流项目时,遇到一辆货车轨迹里有个点直接漂到了海里。回放时,货车「游泳」了3秒钟……客户截图发过来,我那个尴尬啊。

去噪我常用两种方法:

  • 速度阈值法: 计算相邻点之间的速度,如果超过合理范围(比如城市道路超过120km/h),就判定为噪声点。
  • 距离突变法: 如果某点与前后两点的距离都远大于正常步长,直接剔除。
def remove_noise_by_speed(points, max_speed=33.3):
    """
    max_speed: 最大合理速度,单位m/s(约120km/h)
    """
    cleaned = [points[0]]
    for i in range(1, len(points)):
        lat1, lng1, t1 = points[i-1]
        lat2, lng2, t2 = points[i]
        # 简化计算:1度经纬度约111km
        dist = ((lat2-lat1)**2 + (lng2-lng1)**2)**0.5 * 111000
        dt = abs(t2 - t1)
        if dt > 0:
            speed = dist / dt
            if speed <= max_speed:
                cleaned.append(points[i])
    return cleaned
注意: 速度阈值法有个坑——如果连续两个点都是噪声,计算出的速度可能正常。我建议结合前后3-5个点做滑动窗口判断,效果更稳。

4.2 数据插值:线性插值与样条插值

清洗完数据,你会发现轨迹里有很多「空洞」。设备丢帧、信号中断,都会导致时间不连续。插值就是把这些洞补上。

嗯,这里要注意:插值不是万能的。如果丢帧超过10秒,我建议直接标记为「数据缺失」,而不是强行补一个假点。

4.2.1 线性插值

线性插值最简单,也最常用。它假设物体在两个已知点之间做匀速直线运动。说白了,就是「两点之间画条直线,中间的点按比例取」。我80%的场景都用它,够用且快。

def linear_interpolate(p1, p2, t):
    """
    p1, p2: (lat, lng, timestamp)
    t: 目标时间戳
    """
    lat1, lng1, t1 = p1
    lat2, lng2, t2 = p2
    ratio = (t - t1) / (t2 - t1)
    lat = lat1 + (lat2 - lat1) * ratio
    lng = lng1 + (lng2 - lng1) * ratio
    return (lat, lng, t)
适用场景: 采样间隔均匀、运动轨迹平滑的场景。比如高速公路上的车辆轨迹。

4.2.2 样条插值

线性插值有个硬伤——拐弯处不平滑。你想想看,车辆在路口转弯,你用直线去连,轨迹就变成了「折线」,看着很生硬。

样条插值就不一样了。它用多项式曲线去拟合数据点,保证曲线在连接处一阶、二阶导数连续。说白了,就是让轨迹「弯」得更自然。

我记得有一次做赛车轨迹回放,客户要求画面必须丝滑。线性插值做出来,赛车过弯像「漂移」——其实是折线导致的视觉跳跃。换成三次样条插值后,效果立马好了。

from scipy.interpolate import CubicSpline
import numpy as np

def spline_interpolate(points, target_times):
    """
    points: [(lat, lng, timestamp), ...]
    target_times: 需要插值的时间点列表
    """
    times = np.array([p[2] for p in points])
    lats = np.array([p[0] for p in points])
    lngs = np.array([p[1] for p in points])
    
    cs_lat = CubicSpline(times, lats)
    cs_lng = CubicSpline(times, lngs)
    
    result = []
    for t in target_times:
        result.append((cs_lat(t), cs_lng(t), t))
    return result
注意: 样条插值在数据点稀疏时容易「过拟合」,产生不合理的震荡。我建议至少每5个点做一次分段样条,别全局拟合。

4.3 数据压缩:道格拉斯-普克算法

数据压缩,说白了就是「删掉不重要的点,保留轨迹形状」。你想想看,一条直线上的100个点,其实只需要首尾2个点就够了。

道格拉斯-普克算法(简称DP算法)是轨迹压缩领域的经典方法。它的核心思想很简单:

  1. 连接轨迹首尾两点,形成一条基线。
  2. 找到离基线最远的点,计算距离。
  3. 如果距离大于阈值,保留该点,并递归处理左右两段。
  4. 如果距离小于阈值,删除中间所有点。

我刚开始用这个算法时,阈值设得太小,压缩率只有10%,等于没压。后来调大了,又压过头了,轨迹变成了「折线」。嗯,阈值这东西,得根据场景调。

def douglas_peucker(points, epsilon):
    """
    points: [(lat, lng), ...]
    epsilon: 距离阈值,单位度
    """
    if len(points) <= 2:
        return points
    
    # 找到离基线最远的点
    dmax = 0
    index = 0
    for i in range(1, len(points) - 1):
        d = perpendicular_distance(points[i], points[0], points[-1])
        if d > dmax:
            dmax = d
            index = i
    
    # 递归处理
    if dmax > epsilon:
        left = douglas_peucker(points[:index+1], epsilon)
        right = douglas_peucker(points[index:], epsilon)
        return left[:-1] + right
    else:
        return [points[0], points[-1]]

def perpendicular_distance(point, line_start, line_end):
    """计算点到直线的垂直距离"""
    # 简化实现,实际项目中用经纬度球面距离
    x0, y0 = point
    x1, y1 = line_start
    x2, y2 = line_end
    return abs((y2-y1)*x0 - (x2-x1)*y0 + x2*y1 - y2*x1) / ((y2-y1)**2 + (x2-x1)**2)**0.5
经验值参考:
场景 推荐阈值(度) 压缩率
城市道路 0.0001 ~ 0.0005 60% ~ 80%
高速公路 0.0005 ~ 0.001 80% ~ 95%
步行轨迹 0.00005 ~ 0.0001 50% ~ 70%
避坑指南: 我曾经在压缩出租车轨迹时,阈值设了0.0005,结果把掉头弯给压没了。后来我加了一个「关键点保护」逻辑——如果某点是转弯点(角度变化大于30度),强制保留。效果好了很多。

好了,数据预处理这块就聊到这儿。清洗、插值、压缩,三步走完,你的轨迹数据基本就「能打」了。下一节咱们开始讲轨迹匹配,那才是真正考验算法功底的地方。


可可精品资料库