python - Numpy 中两个相关数组的矢量化操作

Question

我有一个 n x n numpy 数组，其中包含所有成对距离和另一个 1 x n 数组包含一些评分指标。

例子:

import numpy as np
import scipy.spatial.distance

dists = scipy.spatial.distance.squareform(np.array([3.2,4.1,8.8,.6,1.5,9.,5.0,9.9,10.,1.1]))

array([[  0. ,   3.2,   4.1,   8.8,   0.6],
       [  3.2,   0. ,   1.5,   9. ,   5. ],
       [  4.1,   1.5,   0. ,   9.9,  10. ],
       [  8.8,   9. ,   9.9,   0. ,   1.1],
       [  0.6,   5. ,  10. ,   1.1,   0. ]])

score = np.array([19., 1.3, 4.8, 6.2, 5.7])

array([ 19. ,   1.3,   4.8,   6.2,   5.7])

因此，请注意得分数组的第 i 个元素对应于距离数组的第 i 行。

我需要做的是向量化这个过程:

1. 对于分数数组中的第i个值，找到所有其他大于第i个值的值并记下它们的索引
2. 然后，在距离数组的第 i 行中，获取所有具有与上述步骤 1 中所述相同索引的距离，并返回最小距离
3. 如果score数组中第i个值最大，则将距离最小的设置为距离数组中找到的最大距离

这是一个未向量化的版本:

n = score.shape[0]
min_dist = np.full(n, np.max(dists))
for i in range(score.shape[0]):
    inx = numpy.where(score > score[i])
    if len(inx[0]) > 0:
        min_dist[i] = np.min(dists[i, inx])

min_dist

array([ 10. ,   1.5,   4.1,   8.8,   0.6])

这可行，但在速度方面效率很低，而且我的阵列预计会大得多。我希望通过使用更快的矢量化操作来实现相同的结果来提高效率。

更新:根据 Oliver W. 的回答，我提出了自己的想法，不需要复制距离数组

def new_method (dists, score):
    mask = score > score.reshape(-1,1)
    return np.ma.masked_array(dists, mask=~mask).min(axis=1).filled(dists.max()) 

理论上可以把它写成一行，但对于未经训练的人来说，阅读它已经有点挑战了。

Answer 1

下面给出了一种可能的矢量化解决方案。

import numpy as np
import scipy.spatial.distance

dists = scipy.spatial.distance.squareform(np.array([3.2,4.1,8.8,.6,1.5,9.,5.0,9.9,10.,1.1]))
score = np.array([19., 1.3, 4.8, 6.2, 5.7])

def your_method(dists, score):
    dim = score.shape[0]
    min_dist = np.full(dim, np.max(dists))
    for i in range(dim):
        inx = np.where(score > score[i])
        if len(inx[0]) > 0:
            min_dist[i] = np.min(dists[i, inx])
    return min_dist

def vectorized_method_v1(dists, score):
    mask = score > score.reshape(-1,1)
    dists2 = dists.copy()  # get rid of this in case the dists array can be changed
    dists2[np.logical_not(mask)] = dists.max()
    return dists2.min(axis=1)    

这些小阵列的速度增益并不那么令人印象深刻(在我的机器上大约是 3 倍)，所以我将使用更大的阵列进行演示:

dists = scipy.spatial.distance.squareform(np.random.random(50*99))
score = np.random.random(dists.shape[0])
print(dists.shape)
%timeit your_method(dists, score)
%timeit vectorized_method_v1(dists, score)

## -- End pasted text --
(100, 100)
100 loops, best of 3: 2.98 ms per loop
10000 loops, best of 3: 125 µs per loop

接近 24 的因数。