如何过滤Numpy数组使每个X值仅对应单个Y值（vactrol电阻数据处理）

问题核心 #

采集得到的原始数据是按时间顺序排列的(x, y)点对序列，理想Vactrol特性是x轴（LED驱动电流/占空比）单调升高、y轴（光敏电阻值）单调降低的连续单值曲线。采集时遇到的“向左回溯”异常，是指序列走到x方向的极值拐点后，后续采样点的x值不再持续增大，反而向x值减小的方向偏移——这部分时序靠后的回溯段就是需要整体剔除的异常区段，最终输出需要满足每个x值仅对应唯一y值的要求。

实现原则 #

• 全程基于原始数据的采集时序做判定，绝对不能先打乱顺序按x值排序再过滤。排序操作会直接丢失“右侧异常段”的时序判定依据，把回溯段的异常点混入正常曲线，完全达不到剔除目标。
• 遍历原始数据过程中，实时记录已走过的最大x值，一旦发现当前点的x值比历史最大x值小，且偏差超过设定的测量噪声容差，就判定为进入回溯段，直接截断数组，保留序列开头到回溯触发点之前的所有有效数据。
• 对截断后的有效段做单值规整：消除同一x值对应的多y重复点，剔除超量程的无效极值，最终形成严格的单x单y映射数组。

可直接运行的NumPy实现代码 #

import numpy as np

def clean_vactrol_dataset(raw_data: np.ndarray, noise_tolerance: float = 0.01) -> np.ndarray:
    """
    清洗带回溯异常的Vactrol测量数据
    raw_data: 原始采集二维数组，形状为(N, 2)，第0列是按采集顺序排列的x值（LED驱动参数），第1列是测得的电阻值
    noise_tolerance: 回溯判定的噪声容差，避免把微小测量抖动误判为回溯
    """
    max_x_history = -np.inf
    # 计算正常采样的x步长，作为容差计算基准
    normal_x_step = np.median(np.abs(np.diff(raw_data[:, 0])))
    cutoff_position = len(raw_data)
    # 遍历找到第一个回溯点的位置
    for sample_idx, (current_x, current_y) in enumerate(raw_data):
        # 当x低于历史最大值超过容差阈值，判定为向左回溯
        if current_x < max_x_history - noise_tolerance * normal_x_step:
            cutoff_position = sample_idx
            break
        max_x_history = max(max_x_history, current_x)
    # 初步截断得到有效数据段
    prelim_valid_seg = raw_data[:cutoff_position]
    # 按x值去重，同一x对应的多个测量y值取平均
    unique_x, group_index = np.unique(prelim_valid_seg[:, 0], return_inverse=True)
    avg_y_for_each_x = np.zeros_like(unique_x)
    for x_group_idx in range(len(unique_x)):
        avg_y_for_each_x[x_group_idx] = np.mean(prelim_valid_seg[group_index == x_group_idx, 1])
    # 组合成最终满足单值要求的曲线数组
    return np.column_stack((unique_x, avg_y_for_each_x))

实操注意点 #

• 回溯判定的容差需要根据实际硬件的测量精度调整，不要设为0，否则容易把正常的测量抖动误判成回溯，提前截断有效曲线。
• 截断后的有效段如果存在随机y值波动，可以叠加滑动窗口滤波，把偏离相邻点均值超过2倍标准差的孤立异常值剔除。
• 过滤过程不要过度追求曲线绝对平滑，核心目标是剔除回溯段、保证单值映射，过度平滑会丢失Vactrol本身的非线性响应特性。

内容的提问来源于stack exchange，提问作者John Janigan-Mills