3. 牵引试验数据异常值识别与标记

异常识别的三层规则与判定顺序

牵引试验数据中的异常值不能只靠一个阈值判断。传感器毛刺、通信丢帧和真实工况突变在数值表现上可能都表现为“偏离邻域”，但它们的物理成因、持续时间和处理方式完全不同。本节采用物理量程 → 统计分布 → 时序连续性三层递进规则，先剔除物理上不可能的值，再用统计方法识别离群点，最后通过时序特征区分三类异常。

规则实现与参数设定

以下 Python 实现以 100 Hz 采样率、单次牵引启动工况（约 30 s）为示例场景。物理量程依据城轨车辆牵引系统典型参数设定：牵引力 ±400 kN、速度 0–120 km/h、牵引电流 ±1500 A。统计层采用滑动窗口中位数与 MAD（中位数绝对偏差），窗口长度取 51 点（约 0.5 s），阈值系数 3.5，兼顾对短时毛刺的敏感性与对工况突变的容忍度。

import numpy as np
import pandas as pd

def detect_anomalies(df, channels, fs=100.0):
    """
    df: 对齐后数据集，含 time_s 及各通道列
    channels: dict，通道名 -> (物理下限, 物理上限)
    返回: 带标记的 df 与异常清单 DataFrame
    """
    dt = 1.0 / fs
    win = 51          # 滑动窗口点数（约0.5s）
    k_mad = 3.5       # MAD阈值系数
    records = []

    for ch, (lo, hi) in channels.items():
        x = df[ch].values.astype(float)
        n = len(x)
        flag = np.zeros(n, dtype=int)   # 0正常 1毛刺 2丢帧 3工况突变 4物理无效

        # ---- 第一层：物理量程 ----
        phys_bad = (x < lo) | (x > hi) | np.isnan(x)
        flag[phys_bad] = 4

        # ---- 第二层：滑动MAD统计 ----
        half = win // 2
        for i in range(n):
            if flag[i] == 4:
                continue
            s = max(0, i - half)
            e = min(n, i + half + 1)
            seg = x[s:e]
            seg = seg[~np.isnan(seg)]
            if len(seg) < 5:
                continue
            med = np.median(seg)
            mad = np.median(np.abs(seg - med))
            if mad == 0:
                mad = 1e-9
            if abs(x[i] - med) > k_mad * 1.4826 * mad:
                flag[i] = -1   # 暂标记为统计离群，待第三层分类

        # ---- 第三层：时序连续性分类 ----
        i = 0
        while i < n:
            if flag[i] == -1:
                j = i
                while j < n and flag[j] == -1:
                    j += 1
                run_len = j - i
                # 检查时间戳间隔是否异常（丢帧）
                if i > 0 and j < n:
                    gap = df['time_s'].iloc[j] - df['time_s'].iloc[i-1]
                    if gap > 2.5 * dt * (run_len + 1):
                        flag[i:j] = 2   # 丢帧
                    elif run_len <= 2:
                        flag[i:j] = 1   # 毛刺
                    else:
                        flag[i:j] = 3   # 工况突变
                elif run_len <= 2:
                    flag[i:j] = 1
                else:
                    flag[i:j] = 3
                i = j
            else:
                i += 1

        # 记录异常
        for idx in np.where(flag != 0)[0]:
            records.append({
                'time_s': round(df['time_s'].iloc[idx], 3),
                'channel': ch,
                'raw_value': x[idx],
                'flag': int(flag[idx]),
                'type': {1:'毛刺', 2:'丢帧', 3:'工况突变', 4:'物理无效'}[flag[idx]]
            })
        df[f'{ch}_flag'] = flag

    anomaly_df = pd.DataFrame(records)
    return df, anomaly_df

三类异常的判读与处理策略

结果验证要点

• 毛刺验证：随机抽取 10 个标记为毛刺的点，确认其前后各 3 个采样点均在正常范围内，且毛刺幅值通常超过局部中位数的 5 倍以上。
• 丢帧验证：检查标记为丢帧的时间段内，原始时间戳间隔是否确实大于正常采样间隔的 2.5 倍；若时间戳均匀但数值恒定，应结合通信协议判断是否为保持寄存器输出。
• 工况突变验证：将标记为工况突变的区段与牵引指令通道叠加对比，确认突变起始时刻与指令变化时刻的偏差在 100 ms 以内，且变化方向一致。
• 误判检查：统计三类异常的数量占比。若毛刺占比超过 5%，说明阈值过严或传感器接触不良；若工况突变占比超过 20%，需检查 MAD 窗口是否过短导致正常动态过程被误判。