如何识别噪声数据较低时期的开始和结束？

Question

我每天大约每隔 1 分钟就有一次嘈杂的数据。

这是一个简单的版本：

如何识别以黄色标记的噪音较小且价值较低的时段的开始和结束索引值？

这是测试数据：

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

arr = np.array([8,9,7,3,6,3,2,1,2,3,1,2,3,2,2,3,2,2,5,7,8,9,15,20,21])

plt.plot(arr)
plt.show()

Answer 1

好吧，如果您只想要那个“区域”，您需要某种方法来查找特定范围内的点。我们怎么能做到这一点？好吧，我们可能应该从找到数组的最小值开始，然后在同一个数组中找到属于指定偏差的其他值：

def lows(arr, dev=0):
    lim = min(arr) + dev
    pts = []
    for i,e in enumerate(arr):
        if e <= lim:
            pts.append((i,e))
    return pts

上述函数返回落在指定范围内的点列表。下限是输入数组的最小值，上限是最小值加上您将提供的偏差。例如，如果您希望所有点都在最小值的 1 以内：

plt.plot(arr)
for pt in lows(arr, 1):
    circle = plt.Circle(pt, 0.2, color='g')
    plt.gca().add_patch(circle)
plt.show()

Answer 2

对于给定的点，我们可以根据某些标准决定保留/屏蔽它：

1. 它的邻居是否在某个三角洲内？
2. 是否在最小值的某个阈值之内？
3. 它是否在一个连续的块中？

注意：由于您标记并导入了 pandas，为了方便起见，我将使用 pandas，但可以使用纯 numpy/matplotlib 实现相同的想法。

---

如果所有较低的时段都在同一水平附近

然后一个简单的方法是使用具有最小阈值的邻居增量（尽管要小心真实数据中的异常值）：

s = pd.Series(np.hstack([arr, arr]))

delta = 2
threshold = s.std()

# check if each point's neighbors are within `delta`
mask_delta = s.diff().abs().le(delta) & s.diff(-1).abs().le(delta)

# check if each point is within `threshold` of the minimum
mask_threshold = s < s.min() + threshold

s.plot(label='raw')
s.where(mask_threshold & mask_delta).plot(marker='*', label='delta & threshold')

---

如果较低的时期处于不同的水平

然后全局最小阈值将不起作用，因为某些时期会太高。在这种情况下，尝试使用相邻块的邻居增量：

# shift the second period by 5
s = pd.Series(np.hstack([arr, arr + 5]))

delta = 2
blocksize = 10

# check if each point's neighbors are within `delta`
mask_delta = s.diff().abs().le(delta) & s.diff(-1).abs().le(delta)

# check if each point is in a contiguous block of at least `blocksize`
masked = s.where(mask_delta)
groups = masked.isnull().cumsum()
blocksizes = masked.groupby(groups).transform('count').mask(masked.isnull())
mask_contiguous = blocksizes >= blocksize

s.plot(label='raw')
s.where(mask_contiguous).plot(marker='*', label='delta & contiguous')

Answer 3

您可以尝试通过测量附近值的方差来检测噪声较小的点。

例如，对于每个点，您可以查看它之前的最后 N 个值并计算它们的标准差，然后在标准差低于某个阈值时标记该点。

以下代码使用 pandas 系列的 rolling 方法应用此过程。

std_thresh = 1
window_len = 5

s = pd.Series([8,9,7,3,6,3,2,1,2,3,1,2,3,2,2,3,2,2,5,7,8,9,15,20,21])

# Create a boolean mask which marks the less noisy points
marked = s.rolling(window=window_len).std() < std_thresh

# Whenever a new point is marked, mark also the other points of the window (see discussion below)
for i in range(window_len + 1, len(marked)):
    if marked[i] and ~marked[i-1]:
        marked[i - (window_len-1) : i] = True
        
plt.plot(s)
plt.scatter(s[marked].index, s[marked], c='orange')

您可以尝试更改window_len（计算标准值的窗口长度）和std_thresh（窗口标准值小于标记的点）的值，并根据您的需要调整它们。

请注意，rolling 考虑在每个点结束的窗口，因此，无论何时遇到噪声较小的点段，它们中的第一个 window_len-1 将不会被标记。这就是为什么我在定义 marked 之后在代码中包含了 for 循环。