在 Numpy 中对两个相关数组进行矢量化操作答案

【问题标题】：Vectorize Operations in Numpy for Two Dependent Arrays在 Numpy 中对两个相关数组进行矢量化操作
【发布时间】：2015-01-19 21:54:31
【问题描述】：

我有一个 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 行。

我需要做的是矢量化这个过程：

对于分数数组中的第 i 个值，找到所有其他大于第 i 个值的值并记下它们的索引
然后，在距离数组的第 i 行中，获取具有与上述步骤 1 中所述相同索引的所有距离并返回最小距离
如果分数数组中的第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())

理论上可以使它成为单行字，但对于未经训练的人来说阅读起来已经有点挑战性了。

【问题讨论】：

您是否根据我的方案描述了您的解决方案？我测试了你的，但它慢了大约两倍。另外，如果您要进行微管理，不妨将mask 的条件从> 更改为<=，并去掉not 操作。节省更多 µs。
@OliverW.：完成！我最诚挚的歉意。我还在习惯 SO 的投票系统，但没有意识到投票与接受答案无关。

标签： python numpy vectorization

【解决方案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 倍。

【讨论】：