根据另一个参考数组从一个数组中选择紧密匹配

Question

我有一个数组A 和一个引用数组B。 A 的大小至少与 B 一样大。例如

A = [2,100,300,793,1300,1500,1810,2400]
B = [4,305,789,1234,1890]

B实际上是信号在指定时间的峰值位置，A包含稍后时间的峰值位置。但A 中的一些元素实际上并不是我想要的峰值（可能是由于噪音等原因），我想根据B 在A 中找到“真实”的峰值。 A 中的“真实”元素应该与B 中的元素接近，在上面给出的示例中，A 中的“真实”元素应该是A'=[2,300,793,1300,1810]。在这个例子中应该很明显100,1500,2400 不是我们想要的，因为它们与 B 中的任何元素都相距甚远。如何在 python/matlab 中以最有效/最准确的方式编码？

Answer 1

方法#1：使用NumPy broadcasting，我们可以在输入数组之间寻找绝对元素减法，并使用适当的阈值从A 中过滤掉不需要的元素。对于给定的样本输入，90 的阈值似乎有效。

因此，我们将有一个实现，就像这样 -

thresh = 90
Aout = A[(np.abs(A[:,None] - B) < thresh).any(1)]

示例运行 -

In [69]: A
Out[69]: array([   2,  100,  300,  793, 1300, 1500, 1810, 2400])

In [70]: B
Out[70]: array([   4,  305,  789, 1234, 1890])

In [71]: A[(np.abs(A[:,None] - B) < 90).any(1)]
Out[71]: array([   2,  300,  793, 1300, 1810])

方法#2：基于this post，这是一种使用np.searchsorted 的内存高效方法，这对于大型数组可能至关重要 -

def searchsorted_filter(a, b, thresh):
    choices = np.sort(b) # if b is already sorted, skip it
    lidx = np.searchsorted(choices, a, 'left').clip(max=choices.size-1)
    ridx = (np.searchsorted(choices, a, 'right')-1).clip(min=0)
    cl = np.take(choices,lidx) # Or choices[lidx]
    cr = np.take(choices,ridx) # Or choices[ridx]
    return a[np.minimum(np.abs(a - cl), np.abs(a - cr)) < thresh]

示例运行 -

In [95]: searchsorted_filter(A,B, thresh = 90)
Out[95]: array([   2,  300,  793, 1300, 1810])

---

运行时测试

In [104]: A = np.sort(np.random.randint(0,100000,(1000)))

In [105]: B = np.sort(np.random.randint(0,100000,(400)))

In [106]: out1 = A[(np.abs(A[:,None] - B) < 10).any(1)]

In [107]: out2 = searchsorted_filter(A,B, thresh = 10)

In [108]: np.allclose(out1, out2)  # Verify results
Out[108]: True

In [109]: %timeit A[(np.abs(A[:,None] - B) < 10).any(1)]
100 loops, best of 3: 2.74 ms per loop

In [110]: %timeit searchsorted_filter(A,B, thresh = 10)
10000 loops, best of 3: 85.3 µs per loop

---

2018 年 1 月更新，性能进一步提升

我们可以通过使用从np.searchsorted(..., 'left') 获得的索引以及absolute 计算来避免np.searchsorted(..., 'right') 的第二次使用，就像这样 -

def searchsorted_filter_v2(a, b, thresh):
    N = len(b)

    choices = np.sort(b) # if b is already sorted, skip it

    l = np.searchsorted(choices, a, 'left')
    l_invalid_mask = l==N
    l[l_invalid_mask] = N-1
    left_offset = choices[l]-a
    left_offset[l_invalid_mask] *= -1    

    r = (l - (left_offset!=0))
    r_invalid_mask = r<0
    r[r_invalid_mask] = 0
    r += l_invalid_mask
    right_offset = a-choices[r]
    right_offset[r_invalid_mask] *= -1

    out = a[(left_offset < thresh) | (right_offset < thresh)]
    return out

更新了测试进一步加速的时间 -

In [388]: np.random.seed(0)
     ...: A = np.random.randint(0,1000000,(100000))
     ...: B = np.unique(np.random.randint(0,1000000,(40000)))
     ...: np.random.shuffle(B)
     ...: thresh = 10
     ...: 
     ...: out1 = searchsorted_filter(A, B, thresh)
     ...: out2 = searchsorted_filter_v2(A, B, thresh)
     ...: print np.allclose(out1, out2)
True

In [389]: %timeit searchsorted_filter(A, B, thresh)
10 loops, best of 3: 24.2 ms per loop

In [390]: %timeit searchsorted_filter_v2(A, B, thresh)
100 loops, best of 3: 13.9 ms per loop

深入挖掘 -

In [396]: a = A; b = B

In [397]: N = len(b)
     ...: 
     ...: choices = np.sort(b) # if b is already sorted, skip it
     ...: 
     ...: l = np.searchsorted(choices, a, 'left')

In [398]: %timeit np.sort(B)
100 loops, best of 3: 2 ms per loop

In [399]: %timeit np.searchsorted(choices, a, 'left')
100 loops, best of 3: 10.3 ms per loop

似乎searchsorted 和sort 占用了几乎所有的运行时间，它们似乎对这种方法至关重要。因此，使用这种基于排序的方法似乎无法进一步改进。

Answer 2

您可以使用bsxfun 找到A 中每个点与B 中每个值的距离，然后使用@ 找到最接近B 中每个值的A 中点的索引987654326@.

[dists, ind] = min(abs(bsxfun(@minus, A, B.')), [], 2)

如果您使用的是 R2016b，由于自动广播，bsxfun 可以被删除

[dists, ind] = min(abs(A - B.'), [], 2);

如果您怀疑B 中的某些值不是真正的峰值，那么您可以设置一个阈值并移除任何大于该值的距离。

threshold = 90;
ind = ind(dists < threshold);

然后我们可以使用ind 索引到A

output = A(ind);

Answer 3

您可以使用 MATLAB interp1 函数来完全满足您的需求。
选项nearest用于查找最近点，无需指定阈值。

out = interp1(A, A, B, 'nearest', 'extrap');

与其他方法比较：

A = sort(randi([0,1000000],1,10000));

B = sort(randi([0,1000000],1,4000));

disp('---interp1----------------')
tic
    out = interp1(A, A, B, 'nearest', 'extrap');
toc
disp('---subtraction with threshold------')
%numpy version is the same
tic
    [dists, ind] = min(abs(bsxfun(@minus, A, B.')), [], 2);
toc

结果：

---interp1----------------
Elapsed time is 0.00778699 seconds.
---subtraction with threshold------
Elapsed time is 0.445485 seconds.

interp1 可用于大于 10000 和 4000 的输入，但在 subtrction 方法中出现内存不足错误。