3. 轨道几何数据异常值与缺失段识别

异常形态与判别口径

原始轨道几何检测数据在进入超限判别前，必须先做一次“数据体检”。本案例针对三类典型异常：尖刺（单点或极短里程内突变）、零值段（连续采样值恒为 0 或接近 0）、缺失区间（采样点丢失或里程跳变）。核心难点在于：尖刺既可能是传感器瞬时干扰，也可能是真实轨道缺陷（如钢轨擦伤、接头低塌）；零值段既可能是传感器掉线，也可能是数据采集软件写入的占位值。因此判别不能只看幅值，必须结合持续里程长度、相邻点梯度与同里程其他通道的一致性三个维度。

检测实现：滑动窗口 + 中位数滤波

尖刺检测采用“中位数滤波残差”法：对每个采样点，取前后各 2 点（共 5 点窗口）的中位数作为局部基准，残差 = 原始值 − 中位数。中位数对孤立尖刺不敏感，因此残差能稳定暴露单点突变。零值段用游程扫描，缺失区间用里程差分。以下代码可直接运行，输入为按里程升序排列的字典列表。

import statistics

SAMPLE_STEP = 0.25          # 标称采样间隔 (m)
SPIKE_RESIDUAL = 3.0        # 尖刺残差阈值 (mm)
SPIKE_GRAD = 2.0            # 相邻梯度阈值 (mm/0.25m)
ZERO_ABS = 0.05             # 零值判定绝对值 (mm)
ZERO_MIN_RUN = 5            # 零值段最少连续点数
GAP_FACTOR = 1.5            # 缺失里程差倍数

def detect_anomalies(records, channel):
    """records: [{'mileage': float, 'gauge': float, ...}, ...] 按里程升序"""
    n = len(records)
    vals = [r.get(channel) for r in records]
    miles = [r['mileage'] for r in records]
    flags = [None] * n

    # 1) 缺失区间：里程跳变或字段为空
    for i in range(1, n):
        dm = miles[i] - miles[i-1]
        if dm > GAP_FACTOR * SAMPLE_STEP or vals[i] is None:
            flags[i] = '缺失'

    # 2) 尖刺：5 点中位数残差 + 梯度
    for i in range(2, n-2):
        if vals[i] is None:
            continue
        win = [v for v in vals[i-2:i+3] if v is not None]
        if len(win) < 3:
            continue
        med = statistics.median(win)
        resid = abs(vals[i] - med)
        grad = abs(vals[i] - vals[i-1]) if vals[i-1] is not None else 0
        if resid > SPIKE_RESIDUAL and grad > SPIKE_GRAD:
            flags[i] = '尖刺'

    # 3) 零值段：游程扫描
    run_start = None
    for i in range(n):
        is_zero = vals[i] is not None and abs(vals[i]) < ZERO_ABS
        if is_zero and run_start is None:
            run_start = i
        elif not is_zero and run_start is not None:
            if i - run_start >= ZERO_MIN_RUN:
                for k in range(run_start, i):
                    flags[k] = '零值段'
            run_start = None
    if run_start is not None and n - run_start >= ZERO_MIN_RUN:
        for k in range(run_start, n):
            flags[k] = '零值段'

    return flags

def merge_segments(records, flags, channel):
    """将连续同类型异常点合并为区段清单"""
    segs, i, n = [], 0, len(records)
    while i < n:
        if flags[i] is None:
            i += 1
            continue
        j = i
        while j + 1 < n and flags[j+1] == flags[i]:
            j += 1
        segs.append({
            'start_mileage': records[i]['mileage'],
            'end_mileage': records[j]['mileage'],
            'type': flags[i],
            'channel': channel,
            'points': j - i + 1
        })
        i = j + 1
    return segs

示例数据与运行结果

构造一段 40 点、采样间隔 0.25 m 的轨距数据（单位 mm，标称值 1435），人为注入三类异常：第 10 点尖刺 +8 mm、第 20–26 点置零、第 32 点后里程跳变 1.0 m 模拟缺失。

import random
random.seed(7)
records = []
base = 1435.0
for i in range(40):
    m = 1000.0 + i * SAMPLE_STEP
    v = base + random.uniform(-0.4, 0.4)
    if i == 10:
        v = base + 8.0          # 尖刺
    if 20 <= i <= 26:
        v = 0.0                 # 零值段
    if i >= 32:
        m += 1.0                # 里程跳变，模拟缺失
    records.append({'mileage': m, 'gauge': v})

flags = detect_anomalies(records, 'gauge')
segs = merge_segments(records, flags, 'gauge')
for s in segs:
    print(s)

预期输出（示例数据直接推出，非实测）：

{'start_mileage': 1002.5, 'end_mileage': 1002.5, 'type': '尖刺', 'channel': 'gauge', 'points': 1}
{'start_mileage': 1005.0, 'end_mileage': 1006.5, 'type': '零值段', 'channel': 'gauge', 'points': 7}
{'start_mileage': 1008.25, 'end_mileage': 1009.75, 'type': '缺失', 'channel': 'gauge', 'points': 8}

异常区段清单交付

清单生成后，按 verdict 分流：传感器异常点从数据集中剔除或线性插补；真实缺陷点保留原值；缺失区间在后续区段统计中标注为无效，不参与均值与峰值计算。这样既避免异常值污染超限判别，又不会把真实缺陷误删。