如何基于阈值与后续值差异扩展NumPy异常检测掩码？

我有一个包含浮点值的np.array数组，以及一个布尔类型的mask（True/False）。需要实现以下逻辑：

• 计算掩码中标记为True的数组元素与其后续元素的差值绝对值，若该值小于自定义阈值（示例中为600），则将后续元素对应的掩码位置标记为True。
• 若后续元素是NaN，对应掩码位置保持False。
• 掩码中最多允许连续5个True。

示例输入 #

import numpy as np
x = np.array([1.5, 16000, 16100, 2.5, np.nan, 3.1, 3.4, -15000, 4.1, np.nan])
mask = np.array([False, True, False, False, False, False, False, True, False, False])
threshold = 600

计算逻辑 #

• 索引1的元素16000与后续元素16100的差值绝对值为100，小于阈值600 → 索引2的掩码设为True
• 索引7的元素-15000与后续元素4.1的差值绝对值为14995.9，大于阈值 → 索引8的掩码保持False

期望输出 #

mask_new = np.array([False, True, True, False, False, False, False, True, False, False])

我尝试过np.where但无法处理后续元素的差值计算，需要可行的解决方案。

以下是基于NumPy的高效实现，涵盖所有需求：

步骤1：初始化新掩码 #

先复制原掩码作为操作基础，避免修改原数据：

mask_new = mask.copy()

步骤2：定位原掩码的True位置 #

获取所有需要检查后续元素的索引：

true_indices = np.where(mask_new)[0]

步骤3：遍历检查后续元素 #

逐个处理原掩码中的True位置，判断后续元素是否满足标记条件：

for idx in true_indices:
    next_idx = idx + 1
    # 跳过数组越界的情况
    if next_idx >= len(x):
        continue
    # 后续元素为NaN时不标记
    if np.isnan(x[next_idx]):
        continue
    # 计算差值绝对值并判断是否小于阈值
    if abs(x[idx] - x[next_idx]) < threshold:
        mask_new[next_idx] = True

步骤4：限制连续True的最大长度（最多5个） #

通过计算连续True的起止索引，截断过长的连续标记：

# 计算连续True的累积和，用于定位连续段
consecutive = np.concatenate([[0], np.cumsum(mask_new), [0]])
# 获取连续True的起始索引
starts = np.where(consecutive[1:] - consecutive[:-1] == 1)[0]
# 获取连续True的结束索引（不包含）
ends = np.where(consecutive[1:] - consecutive[:-1] == -1)[0]

# 遍历所有连续段，截断超过5个的部分
for s, e in zip(starts, ends):
    segment_length = e - s
    if segment_length > 5:
        mask_new[s+5:e] = False

完整测试代码 #

import numpy as np

x = np.array([1.5, 16000, 16100, 2.5, np.nan, 3.1, 3.4, -15000, 4.1, np.nan])
mask = np.array([False, True, False, False, False, False, False, True, False, False])
threshold = 600

# 初始化新掩码
mask_new = mask.copy()
# 获取原掩码True的索引
true_indices = np.where(mask_new)[0]

# 检查后续元素并更新掩码
for idx in true_indices:
    next_idx = idx + 1
    if next_idx >= len(x):
        continue
    if np.isnan(x[next_idx]):
        continue
    if abs(x[idx] - x[next_idx]) < threshold:
        mask_new[next_idx] = True

# 处理连续True的长度限制
consecutive = np.concatenate([[0], np.cumsum(mask_new), [0]])
starts = np.where(consecutive[1:] - consecutive[:-1] == 1)[0]
ends = np.where(consecutive[1:] - consecutive[:-1] == -1)[0]

for s, e in zip(starts, ends):
    if (e - s) > 5:
        mask_new[s+5:e] = False

print(mask_new)
# 输出：[False  True  True False False False False  True False False]

关键说明 #

• 仅遍历原掩码中的True位置，避免不必要的计算，保证效率
• 严格处理NaN元素，不会误标记NaN对应的掩码位置
• 连续True的限制通过精准定位连续段实现，逻辑清晰且高效

内容的提问来源于stack exchange，提问作者Lisa