GPS轨迹降噪：Python API与算法流水线

简介：面向需要清洗GPS轨迹数据的开发者与算法学习者，这份资源聚焦噪点剔除与异常点处理，结合高德、百度轨迹服务接口与基于点间距离的降噪思路，可用于智能交通、物流跟踪和户外运动记录等场景。压缩包共3个文件，均为Python脚本，整体约7KB，分别承担高德接口封装、百度鹰眼接口封装以及降噪主流程演示，便于在本地串联请求、解析与算法计算。已有153人学习。读者可据此理解距离阈值判定、异常点筛除与轨迹平滑的基本实现，掌握调用在线地图API获取清洗结果的方法；在线接口适合快速处理批量轨迹，本地距离法则便于理解原理和二次调整，也可参考脚本组织方式完成自己的轨迹预处理流程。

1. 从一团毛线到一条干净路线：GPS 轨迹噪点剔除 API 到底解决什么问题

去年帮一个做车队管理的小团队看后台，地图上把一天的 GPS 轨迹一撒，整条线路像被人拿毛线团砸过：车明明停在园区门口，轨迹却在两个街区之间来回甩了十几趟。司机没绕路，是定位在漂。这类问题在物流、外卖、骑行记录、运动 App 里几乎是标配，谁都躲不掉。

这份 GPS 轨迹噪点剔除的 Python 代码，做的事很具体：把序列里的脏点挑出来扔掉或拉回来，再包一层 HTTP API，让业务后端不用关心算法细节，传一串点进去，拿一串干净点出来。里面既有能直接 import 的函数，也有能独立跑起来的服务。它适合两类人：手里已经有轨迹数据、被漂移点折磨过的后端；以及刚接触时空数据、想找个能改的起点的人。

2. 工程骨架：六个模块撑起一个可跑的降噪服务

拿到一份代码，我习惯先不读算法，先看目录。目录结构基本能告诉你作者是怎么切分职责的，也能看出他有没有把「算法」和「接口」这两件事分开。这个项目切得还算干净，核心逻辑全在core/下，app/只负责协议层，两者之间只通过普通 Python 函数调用，不互相依赖。

2.1 目录结构与模块职责

gps-denoiser/ ├── app/ │ ├── main.py # FastAPI 入口：路由、异常兜底、启动钩子 │ ├── schemas.py # Pydantic 请求/响应模型，字段校验都在这 │ └── settings.py # 阈值默认值集中管理，方便按业务调 ├── core/ │ ├── geo.py # 距离、方位角、米制投影等几何工具 │ ├── filters.py # 速度粗筛、MAD 滑窗、一维卡尔曼 │ ├── simplify.py # Douglas-Peucker 抽稀 │ └── pipeline.py # 把上面几步串成一条流水线 ├── tests/ │ └── test_filters.py # 用构造数据验证每种噪点能不能被干掉 ├── requirements.txt └── README.md

geo.py只做纯计算，不碰任何状态，所以它最好测；filters.py是重灾区，每个 filter 都设计成「输入点列 → 输出点列 + 标记位」，方便你在中间任何一步插桩看效果；pipeline.py是编排层，只负责按顺序调用并统计每一步删了多少点。这种分层最大的好处是：你想换掉卡尔曼、只留速度粗筛，改一个文件就行，API 层一个字都不用动。

常见做法是把所有逻辑塞进一个utils.py，能跑，但等你调试到第三天、想单独验证「到底是哪一步把我这条正常轨迹砍了一半」的时候，就会后悔。这个项目在结构上避开了这个坑。

2.2 依赖与虚拟环境

依赖非常克制，没有一上来就拉 scipy 和 pandas：

python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install -r requirements.txt

fastapi>=0.110 uvicorn[standard]>=0.27 pydantic>=2.6 numpy>=1.26

四个包，fastapi管 HTTP，pydantic管字段校验，numpy管向量化计算，uvicorn是跑服务用的。卡尔曼那部分我一般手写十几行，不引filterpy，因为一维匀速模型的协方差更新就那么几步，自己写反而更容易跟业务参数对齐，也少一个依赖。如果你后续要上无迹卡尔曼或者多传感器融合，再引库不迟。

提示：pydantic建议锁 2.x。1.x 和 2.x 的校验器写法差别很大，网上抄来的示例混着用会直接报ConfigError。

2.3 启动服务与第一条 curl 请求

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

--reload只在开发时开，它监听了文件变更，生产环境开着会白白吃 CPU。跑起来之后访问/docs能看到自动生成的交互文档，这是 FastAPI 自带的好处，调试时可以直接在页面上改参数试。

来一条最小请求：

curl -X POST http://127.0.0.1:8000/api/v1/denoise \ -H "Content-Type: application/json" \ -d '{ "points": [ {"lat": 31.2304, "lon": 121.4737, "ts": 1710000000000, "acc": 8.0}, {"lat": 31.2305, "lon": 121.4738, "ts": 1710000003000, "acc": 9.0}, {"lat": 31.2451, "lon": 121.4902, "ts": 1710000005000, "acc": 120.0}, {"lat": 31.2307, "lon": 121.4740, "ts": 1710000008000, "acc": 7.5} ], "profile": "vehicle" }'

第三个点就是典型的跳点：两秒内位移接近两公里，精度半径还标了 120 米。请求里profile是预设档位，vehicle、walk、bike三档对应不同的速度上限和窗口大小，业务方不用自己算阈值，选个档就行。响应会返回清洗后的点列、每个点的保留标记，以及一份「各阶段淘汰统计」，方便你在线上定位是速度筛还是窗口筛动的手。

3. 核心算法拆解：一条轨迹从脏到干净的完整链路

GPS 噪点不是一种东西。至少有四类：跳点（瞬时定位跑到几百米外）、抖动（静止时点在一个圈里乱蹦）、漂移（整段缓慢偏移，像被风吹走）、冗余（一秒十个点，全都对但没用）。指望一个算法全吃掉是不现实的，所以这条流水线是分层的，一层治一种病。

3.1 距离计算与预处理：先把单位统一到米

所有阈值的物理意义都应该是米和秒，不是经纬度度数。原因很直白：纬度一度的距离基本恒定，经度一度在赤道是 111 公里，在北纬 60 度只剩 55 公里。你要是在度数上定阈值，同一份代码在广州好用，到了北方就全乱。

import math import numpy as np EARTH_R = 6371008.8 # 地球平均半径，米 def haversine_m(lat1, lon1, lat2, lon2): """两点球面距离（米）。轨迹类计算别用欧氏距离糊弄，近距离误差也很明显。""" p1, p2 = math.radians(lat1), math.radians(lat2) dphi = p2 - p1 dlam = math.radians(lon2 - lon1) a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlam / 2) ** 2 return 2 * EARTH_R * math.asin(math.sqrt(a)) def haversine_vec(lat, lon): """向量化版本：输入同长度的 lat/lon 数组，返回相邻点距离数组，长度 n-1。""" lat_r = np.radians(lat) lon_r = np.radians(lon) dphi = np.diff(lat_r) dlam = np.diff(lon_r) a = np.sin(dphi / 2) ** 2 + np.cos(lat_r[:-1]) * np.cos(lat_r[1:]) * np.sin(dlam / 2) ** 2 return 2 * EARTH_R * np.arcsin(np.sqrt(np.clip(a, 0, 1)))

haversine_vec里的np.clip(a, 0, 1)不是多余的。浮点误差会让a偶尔蹦到 1.0000000002，arcsin直接给你nan，然后整条轨迹的距离数组全废。这种问题在测试集上跑不出来，上线跑真实数据才炸，属于血泪经验。

预处理还要做三件事，缺一不可：

• 按ts排序并去重。设备补传、离线缓存合并都会让时间戳乱序，乱序之后算出来的速度是负数或无穷大。
• 时间戳统一单位。有的设备给毫秒，有的给秒，有的给"2024-03-09T12:00:00Z"字符串。我一般在 schema 层就强制转成毫秒整数，后面全部按毫秒算。
• 精度字段归一。acc、hdop、radius各家叫法不同，单位可能是米也可能是分米，进流水线之前统一成米。

3.2 速度与精度粗筛：先把最离谱的点砍掉

第一层最便宜也最有效。相邻两点距离除以时间差就是速度，超过物理上限的直接判为跳点。同时把精度半径过大的点标记出来——精度 200 米的点，本身就没有参考价值。

import numpy as np def speed_filter(points, max_speed=42.0, max_acc=50.0, max_gap_s=300.0): """ points: dict 列表，含 lat/lon/ts(ms)/acc(m) max_speed: 速度上限 m/s。42 约等于 150km/h，覆盖绝大多数道路场景 max_acc: 精度上限，超过这个值的点直接淘汰 max_gap_s: 相邻点时间差超过这个秒数就不做速度判断（中间断档了） 返回: (保留的索引列表, 被淘汰的索引列表) """ n = len(points) if n < 2: return list(range(n)), [] lat = np.array([p["lat"] for p in points]) lon = np.array([p["lon"] for p in points]) ts = np.array([p["ts"] for p in points], dtype=np.int64) acc = np.array([p.get("acc", 0.0) or 0.0 for p in points]) dist = np.concatenate([[0.0], haversine_vec(lat, lon)]) # 首点补 0，对齐长度 dt = np.concatenate([[0.0], np.diff(ts) / 1000.0]) # 转秒 with np.errstate(divide="ignore", invalid="ignore"): speed = np.where(dt > 0, dist / dt, 0.0) keep, drop = [], [] for i in range(n): # 精度太差，或者速度超限但时间差在合理范围内（排除断档造成的假超速） if acc[i] > max_acc: drop.append(i) elif 0 < dt[i] <= max_gap_s and speed[i] > max_speed: drop.append(i) else: keep.append(i) return keep, drop

几个参数值得掰开说。max_speed=42是个偏保守的上限，城市配送场景调到 30 也够；做赛道或者越野摩托的得往上放。max_acc=50针对的是普通手机 GPS，在楼宇密集区 50 到 100 米很常见，如果你用的是专业设备，这个值可以压到 10。max_gap_s=300是关键：设备熄屏、隧道、地下车库都会造成几分钟的断档，这时候下一跳距离再远也正常，不能用速度去判它。

注意这里只返回索引，不直接删点。原因在于速度筛会把「真实高速+单点漂移」误伤，后面还有一层回溯逻辑要参考原始序列，先把索引留着，编排层再决定最终删哪些。

3.3 滑动窗口 MAD：收拾那些「不算离谱但一直抖」的点

速度筛对付大跳点很灵，但对付抖动没辙。车停着不动，点在一百米范围里来回蹦，每个点的速度都不超限，可轨迹画出来就是个毛球。这种要用滑窗统计。

我用的是 Hampel 的思路：以每个点为中心取前后各k个点，算这组数的中位数和 MAD（中位数绝对偏差），偏离中位数超过n_sigma倍 MAD 的判为离群。跟 3σ 比，它抗污染能力强得多——窗口里就算混进一两个真跳点，中位数也不会被带跑。

import numpy as np def mad_outlier_flags(values, window=7, n_sigma=3.0): """ values: 一维序列，可以是纬度、经度，也可以是相邻点距离 window: 单侧窗口大小，实际窗口长度 = 2*window+1 n_sigma: 偏离几个 MAD 判为离群 返回: 布尔数组，True 表示离群 """ v = np.asarray(values, dtype=float) n = len(v) flags = np.zeros(n, dtype=bool) if n < 3: return flags scale = 1.4826 # 让 MAD 在正态分布下与标准差等价 for i in range(n): lo, hi = max(0, i - window), min(n, i + window + 1) seg = v[lo:hi] med = np.median(seg) mad = np.median(np.abs(seg - med)) * scale if mad < 1e-9: # 全静止的窗口，MAD 为 0，不做判断 continue if abs(v[i] - med) / mad > n_sigma: flags[i] = True return flags

对轨迹，光对纬度、经度分别做过滤是不够的，因为一个点可能经纬度各自都在合理范围，但组合起来就跑偏了。我一般会再对「相邻点距离序列」跑一次同样的逻辑，两次的离群标记取并集。这样横向漂移和纵向漂移都能覆盖到。

window=7意味着窗口 15 个点，按 1 秒采样算就是前后 7 秒。采样频率高的话（10Hz）这个窗口只有 1.5 秒，太窄，这时候要么加大window，要么先降采样到 1Hz 再处理。我一般选后者——先把轨迹按时间抽稀到 1Hz，后面所有阈值都好定得多。

3.4 卡尔曼平滑与 Douglas-Peucker 抽稀

筛完之后轨迹点少了，但不代表好看。剩下的点还是带着测量噪声，画出来锯齿感明显。这时候上一维卡尔曼做平滑，模型用最简单的匀速（CV）模型就够了。

def kalman_smooth_1d(measurements, q=1e-4, r=1e-2): """ 一维匀速模型卡尔曼，对单条坐标序列做平滑 q: 过程噪声，越大越信测量，轨迹更贴合原始点 r: 测量噪声，越大越信预测，轨迹更平滑 """ n = len(measurements) if n == 0: return [] x = np.array([measurements[0], 0.0]) # 状态: [位置, 速度] P = np.eye(2) * 1.0 F = np.array([[1.0, 1.0], [0.0, 1.0]]) # dt 已归一化为 1，采样间隔需先统一 Q = np.eye(2) * q H = np.array([[1.0, 0.0]]) out = [measurements[0]] for z in measurements[1:]: x = F @ x P = F @ P @ F.T + Q y = z - (H @ x)[0] S = (H @ P @ H.T)[0, 0] + r K = (P @ H.T / S).flatten() x = x + K * y P = (np.eye(2) - np.outer(K, H)) @ P out.append(x[0]) return out

q和r的比例才是关键，绝对值意义不大。r/q大，平滑力度强，转弯会被抹圆；r/q小，贴合原始点但噪声回潮。我的经验是先固定q=1e-4，把r从1e-2往上试，看转弯处的偏差能不能接受。

平滑完还有个问题：点太多。一分钟几百个点，前端画图卡、存库也占地方。Douglas-Peucker 抽稀就是干这个的，它保留「偏离直线超过阈值」的关键点，把中间那些近似共线的全丢掉。

def douglas_peucker(points, epsilon_m=3.0): """points: [(x, y), ...]，坐标为米制。epsilon_m 是允许的最大偏离距离""" if len(points) < 3: return points x0, y0 = points[0] x1, y1 = points[-1] dx, dy = x1 - x0, y1 - y0 norm = (dx * dx + dy * dy) ** 0.5 max_d, idx = -1.0, 0 for i in range(1, len(points) - 1): px, py = points[i] if norm < 1e-9: # 首尾重合，退化成点距 d = ((px - x0) ** 2 + (py - y0) ** 2) ** 0.5 else: d = abs(dy * (px - x0) - dx * (py - y0)) / norm if d > max_d: max_d, idx = d, i if max_d > epsilon_m: left = douglas_peucker(points[:idx + 1], epsilon_m) right = douglas_peucker(points[idx:], epsilon_m) return left[:-1] + right return [points[0], points[-1]]

epsilon_m=3.0表示允许偏离直线 3 米。这个值不宜大：配送轨迹在小区里拐来拐去，阈值给到 10 米，路网形状就没了，导航回放会明显看出「切角」。抽稀前记得把经纬度投到米制平面上，直接拿度数算epsilon是另一种经典翻车法。

3.5 编排层：把四步串成一条流水线

单个算法都跑通了，接下来要把它们按正确顺序串起来。顺序有讲究：粗筛必须在前，不然跳点会污染滑窗的中位数；平滑必须在滤波之后，不然卡尔曼要把漂移也一起信了。

def denoise_pipeline(points, profile="vehicle"): cfg = { "vehicle": dict(max_speed=42.0, max_acc=50.0, window=7, n_sigma=3.0, eps=3.0), "bike": dict(max_speed=12.0, max_acc=40.0, window=5, n_sigma=2.8, eps=2.0), "walk": dict(max_speed=4.0, max_acc=30.0, window=5, n_sigma=2.5, eps=1.5), }[profile] pts = sort_and_dedup(points) # 排序、去重、时间戳归一 keep, _ = speed_filter(pts, cfg["max_speed"], cfg["max_acc"]) pts = [pts[i] for i in keep] lat = [p["lat"] for p in pts] lon = [p["lon"] for p in pts] flag_lat = mad_outlier_flags(lat, cfg["window"], cfg["n_sigma"]) flag_lon = mad_outlier_flags(lon, cfg["window"], cfg["n_sigma"]) pts = [p for p, a, b in zip(pts, flag_lat, flag_lon) if not (a or b)] pts = project_and_smooth(pts) # 转米制 → 卡尔曼 → 转回经纬度 pts = simplify_by_epsilon(pts, cfg["eps"]) # 抽稀 return pts

sort_and_dedup、project_and_smooth、simplify_by_epsilon是三个薄封装，内部就是把3.1到3.4的函数按坐标系拼起来。抽稀放在最后，是因为它只关心几何形状，前面任何一步改变了点位，都需要重新算。

注意：每步都会改变点的数量，所以别在中间环节用原始索引去对业务 ID。每个点带上自己的唯一 ID 走完全程，最后按 ID 回溯，才不会对错行。

4. API 实现：请求契约、参数语义与大轨迹分块

算法在本地跑通和对外提供服务是两件事。前者只关心结果对不对，后者还要管字段缺失、超大请求、并发、超时和错误可读性。

4.1 接口契约：字段、类型与默认值

overrides这个字段是我后来加的，起因是有个业务方的车辆经常跑高速，用vehicle档的 42m/s 上限会被砍掉一部分正常点。与其不停加 profile，不如开放覆盖，让业务方自己调一两个关键参数。

4.2 FastAPI 路由与 Pydantic 模型

from typing import List, Optional, Literal from pydantic import BaseModel, Field, field_validator from fastapi import FastAPI, HTTPException class TrackPoint(BaseModel): lat: float = Field(..., ge=-90, le=90) lon: float = Field(..., ge=-180, le=180) ts: int = Field(..., gt=0, description="毫秒时间戳") acc: Optional[float] = Field(None, ge=0) class DenoiseRequest(BaseModel): points: List[TrackPoint] = Field(..., min_length=1, max_length=20000) profile: Literal["vehicle", "bike", "walk"] = "vehicle" overrides: Optional[dict] = None return_flags: bool = False @field_validator("points") @classmethod def check_span(cls, v): if len(v) < 2: raise ValueError("至少需要两个点") span = max(p.ts for p in v) - min(p.ts for p in v) if span <= 0: raise ValueError("时间戳跨度必须为正，检查单位是否统一为毫秒") return v app = FastAPI(title="GPS Track Denoiser", version="1.0") @app.post("/api/v1/denoise") def denoise(req: DenoiseRequest): try: pts = [p.model_dump() for p in req.points] cleaned, stats = run_pipeline(pts, req.profile, req.overrides, req.return_flags) return {"code": 0, "points": cleaned, "stats": stats} except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) except Exception: # 生产环境记得换成结构化日志，别把堆栈直接吐给调用方 raise HTTPException(status_code=500, detail="denoise failed")

max_length=20000是硬性护栏。不设上限的话，有人传一条十万点的越野轨迹进来，同步接口要跑好几秒，worker 直接被拖住。check_span那个校验看着不起眼，但它把「时间戳是秒不是毫秒」这类问题挡在了入口——不挡的话后面算出来的速度会小一千倍，所有点都被判成静止,结果全错，还特别难查。

4.3 参数语义：阈值到底怎么定

三个档位的默认值不是拍脑袋来的，是按各自场景的物理上限留了余量：

walk档的max_speed给到 4m/s 听着挺快（14km/h），但真有人跑步骑车混进来，给紧了就误删。我宁可放过几个可疑点，也不想把真实轨迹砍断——砍断的轨迹在回放时的观感比噪点还糟。

4.4 大轨迹的分块与并发

单条两三万点的轨迹，纯 Python 循环跑 MAD 会明显卡。有两条路：

• 向量化 + 分块：按固定长度切段，段间重叠2*window+1个点，保证边界处的判断不受影响，算完再把重叠部分裁掉。
• 降采样优先：先按时间抽稀到 1Hz，再做全部处理。绝大多数原始轨迹是 5Hz 甚至 10Hz，抽稀后点数降到十分之一，后面的开销跟着降。

我一般两条都用：先降采样，再分块。分块的坑在于段与段之间的接缝——如果直接按 5000 点硬切，接缝处周围的点因为缺了一半窗口，会被误判。重叠区必须留够，而且裁剪时只能保留不属于本段主体的部分。

def chunked(items, size=5000, overlap=31): """按 size 切块，相邻块重叠 overlap 个元素，保证窗口统计不缺上下文""" i = 0 while i < len(items): end = min(i + size + overlap, len(items)) yield i, items[i:end] if end >= len(items): break i += size

overlap=31对应window=15，刚好够最宽窗口用。这个函数返回的是start_index + slice，后面合并时按start_index去重就行。

5. 避坑与排查：真跑轨迹时翻车的五个现场

算法在构造数据上都是对的，一上真实数据就开始演。下面这几条是我在实际数据里撞到的，现象、原因、解决按顺序写。

5.1 五个典型故障的排查路径

一、所有点都被判成静止，轨迹一条直线。

现象是降噪后点全挤在一起，速度序列接近 0。原因几乎都是时间戳单位错了：设备给的是秒，代码按毫秒除，dt小了 1000 倍，速度就小了 1000 倍，一路全被 MAD 判成离群后平滑成直线。解决方式是在 schema 层加一个合理性检查：ts小于 10^11 的按秒解释，否则按毫秒，并在响应里带上推断结果，让调用方自己确认。

二、A 城市的轨迹好用，B 城市的一塌糊涂。

现象是同一套参数，南方数据正常，北方数据大量误删。原因可能是「在经纬度上定了距离阈值」，也可能是「米制投影用的固定系数」。解决：所有几何阈值一律走 haversine 或按纬度动态计算的投影，别硬编码111000。

三、停车的十几分钟被吃掉了。

现象是车辆怠速或停靠的那段，降噪后只剩一两个点。原因是静止时定位在几十米范围内抖动，MAD 把这一整段都标成了离群，而抽稀又把剩下的点合并掉了。解决办法是先做静止段识别：连续若干点两两距离都小于acc的中位数，就判为静止簇，整簇用质心代替，而不是逐点剔除。

四、跨天的轨迹被连成了一条横穿地图的直线。

现象是两段毫不相关的行程中间多了一条笔直的线。原因是轨迹按时间排好序后，第一段末尾和第二段开头直接成了相邻点，dt有好几个小时，速度筛因为max_gap_s跳过了判断，抽稀却把它当真了。解决办法是分段：dt超过阈值就切断，每段独立跑流水线，最后按原顺序拼回去。

五、降噪后点少了 60%，业务方说轨迹「断了」。

现象是统计里淘汰率异常高。八成是max_acc定得太紧——很多安卓机在室内或地铁口会连续几分钟给出 100 米以上的精度值，卡 30 米会把这些点全扔掉，路网就断了。解决办法是把「精度差」和「位置错」区分开：精度差但速度正常的点，只降权不删除，交给卡尔曼用大r处理。

提示：这五条里有四条跟参数无关，跟数据规范有关。上线前先把「时间戳单位、坐标系、精度字段单位」这三件事跟数据源方当面对清楚，能省下大半排查时间。

5.2 自查工具：把降噪前后画出来

线上出问题，靠日志里的淘汰率猜原因是很低效的。我在项目里加了个debug=true开关，打开后在响应里带上每一步的中间结果，再配一个小脚本画对比图：

import matplotlib.pyplot as plt def plot_compare(raw, steps: dict): """raw: 原始点 (lon, lat)；steps: {"after_speed": [...], "after_mad": [...], "final": [...]}""" fig, axes = plt.subplots(1, len(steps) + 1, figsize=(4 * (len(steps) + 1), 4)) axes[0].plot([p[1] for p in raw], [p[0] for p in raw], ".", ms=2, color="#999") axes[0].set_title(f"raw ({len(raw)})") for ax, (name, pts) in zip(axes[1:], steps.items()): ax.plot([p[1] for p in pts], [p[0] for p in pts], ".-", ms=3, lw=0.6) ax.set_title(f"{name} ({len(pts)})") plt.tight_layout() plt.savefig("compare.png", dpi=150)

plot_compare的入参是「原始点 + 各阶段输出」，图会横着摆一排，一眼能看出是哪一步把轨迹砍断的。注意matplotlib的坐标是(x, y)，经度在前纬度在后，写反了图会转 90 度——这个错我犯过不止一次。这个脚本不进生产依赖，放在tools/下按需跑就行。

6. 二次开发：把降噪能力接进自己的业务链路

这套东西要落地，多半不是当独立服务跑，而是嵌进已有的数据处理链路。我踩过的顺序是：先把它当成纯函数库用，等参数调稳了再抽成服务。

6.1 换算法与换打分

pipeline.py里每一步都是可替换的。想加自己的规则，最简单的方式是写一个签名一致的函数，然后插到合适的位置：

def reject_by_geofence(points, fence_polygon): """把落在已知不可达区域（水面、封闭园区）的点直接标为离群""" from matplotlib.path import Path path = Path(fence_polygon) return [not path.contains_point((p["lon"], p["lat"])) for p in points]

这个reject_by_geofence跟mad_outlier_flags返回的都是布尔数组，语义一致（True表示要删），所以能直接插进编排层。接入时只要在 MAD 之后、平滑之前调用一次就行。matplotlib.path.Path做点在多边形内判断够用了，别为这个引 shapely。

如果场景对精度要求更高，可以把 3.4 的一维卡尔曼换成二维联合滤波，把经纬度一起建模，顺带估计速度向量。代价是参数变多、调起来更费劲，我一般只在「轨迹要用来算里程」的场景才上。

6.2 离线批量跑与效果验证

线上服务解决的是「实时回传的轨迹」，离线批处理解决的是「历史数据清洗」。后者用同一套core/就行，外面套个读库写库的循环。

改造后首要的问题不是跑得快不快，是「怎么证明改得对」。我一般看四个指标：

保留率低于 0.7 说明误删严重，高于 0.95 说明基本没干活。平均偏移和最大偏移用来防「平滑过度」——卡尔曼参数调猛了，轨迹会被拉成一条弧线，保留率看着正常，但偏移会明显上去。长度变化率能抓出「抽稀过狠」和「噪点导致里程虚高」两种反向问题。

验证数据我一般取三份：一份干净的（出租车在开阔路段的轨迹）、一份极脏的（地铁口、高架桥下）、一份带长时间静止的。三份都过一遍，指标都在区间内才算改对了。只看一份干净数据就上线，翻车是迟早的事。

def evaluate(raw, cleaned, eps_pairs): """eps_pairs: 预计算的 (原始点, 原始点下标) 用于最近邻查找，别在循环里重算""" keep_rate = len(cleaned) / len(raw) offsets = [nearest_distance(p, eps_pairs) for p in cleaned] return { "keep_rate": round(keep_rate, 3), "offset_mean": round(sum(offsets) / len(offsets), 2), "offset_max": round(max(offsets), 2), "length_ratio": round(polyline_length(cleaned) / polyline_length(raw), 3), }

nearest_distance就是逐个算 haversine 取最小，点数上万时是 O(n²)，慢但够用；真要快就上 KD 树，把经纬度投到米制平面再建索引，别直接对度数建树，纬度越高误差越大。

从那以后我每次改完滤波参数，都会强制走一遍这三份数据的对比，把四个指标打出来贴在 PR 描述里。有次偷懒只跑了干净样本，结果上线后停车段全被削掉，回滚花了一个下午——这类账，算一次就够了。希望这份拆解帮到你。

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