使用numpy实现列表滑动的向量化实现

Question

从给定的numpy数组[1,2,3,4]和窗口wz=2（每个元素前后两个元素）中，我必须得到一些对（中央元素，窗口中的元素）进行操作。 不存在元素的对可以跳过或替换为零。 所以在这个例子中，我必须得到以下结果：

[[1., 0.]
 [2., 1.]
 [3., 2.]
 [4., 3.]
 [1., 2.]
 [2., 3.]
 [3., 4.]
 [4., 0.]
 [1., 0.]
 [2., 0.]
 [3., 1.]
 [4., 2.]
 [1., 3.]
 [2., 4.]
 [3., 0.]
 [4., 0.]]

我的实现非常低效，看起来像这样：

x = np.array([1,2,3,4])
l = x.shape[0]
for i in range(1, m):
    init = np.empty((x.shape[0]*2,2))
    init[:,0] = np.append(x, x)
    init[:l,1] = np.pad(x, (i,0), mode='constant')[:l]
    init[-l:,1] = np.pad(x, (0,i), mode='constant')[-l:]
    corpus.extend(init)

请问有人能提供更高效的解决方案吗？ 在其他我实现过的简单测试数据和变量上，我得到了以下结果：

285 µs ± 19.3 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)
379 µs ± 7.68 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)

- Ivan Telnov

Answer 1

这里有一个NumPy风格的方法：

In [23]: a = np.array([1,2,3,4])
In [24]: arr = np.hstack((a-1, a+1, a - 2, a+ 2))
In [25]: mask = ~np.in1d(arr, a)
In [26]: arr[mask] = 0
In [27]: np.column_stack((np.tile(a, 4), arr))
Out[27]: 
array([ [1, 0],
        [2, 1],
        [3, 2],
        [4, 3],
        [1, 2],
        [2, 3],
        [3, 4],
        [4, 0],
        [1, 0],
        [2, 0],
        [3, 1],
        [4, 2],
        [1, 3],
        [2, 4],
        [3, 0],
        [4, 0]])

- Mazdak

---

也许这是一个愚蠢的问题，但是这里的 ~ 是做什么用的？ - erocoar

它的作用类似于逻辑非（logical_not），将True转换为False，将False转换为True。 - Mazdak

Answer 2

如果x是一些数据，比如单词或随机值，我们需要重新组合它，我们可以使用numpy中的重新索引机制。 零替换版本

x = np.array([1,2,3,4])
wz = 2
zero = 0

让我们建立索引矩阵。

ri = np.arange(-wz,wz+1)+np.arange(x.shape[0]).reshape(-1,1)
print(ri) 

输出：

  [[-2, -1,  0,  1,  2],
   [-1,  0,  1,  2,  3],
   [ 0,  1,  2,  3,  4],
   [ 1,  2,  3,  4,  5]

现在，如果我们将零添加到 x 中作为最后一个元素，我们可以用它的索引替换错误的索引。

np.place(ri,(ri<0)|(ri>x.shape[0]),x.shape[0]) #replace wrong indexes
np.vstack((
    np.hstack((x,[zero]))[ri].reshape(1,-1),#extending x with zero and reindexing 
    np.tile(x,2*wz+1)) #repeating basic `x` to each window position
    )#.T #uncomment .T to make it vertical   

输出：

 ([[0, 0, 1, 2, 3, 0, 1, 2, 3, 4, 1, 2, 3, 4, 0, 2, 3, 4, 0, 0],
   [1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4]])

---

跳过的版本

同样的想法，但顺序略有不同：生成完整的索引矩阵[window_index,x_index]，然后排除错误的配对，最后重新索引'x'。

x = np.array([1,2,3,4])
wz = 2
ri = np.vstack((
    (np.arange(-wz,wz+1)+np.arange(x.shape[0]).reshape(-1,1)).ravel(),#same index matrix flaten 
    np.tile(np.arange(x.shape[0]),2*wz+1) #repeating `x` indexes to each window position
    )) 
x[ri[:,(ri[0]>=0)&(ri[0]<x.shape[0])]]#.T #uncomment .T to make it vertical   

输出：

 [[1, 2, 3, 1, 2, 3, 4, 1, 2, 3, 4, 2, 3, 4],
  [3, 4, 1, 3, 4, 1, 2, 3, 4, 1, 2, 4, 1, 2]]

---

更新1（错误修复） 从窗口中排除零以避免成对重复。

x = np.array([1,2,3,4])
wz = 2
ri = np.vstack(((
        np.hstack(( np.arange(-wz,0), #remove zero from window
                    np.arange(1,wz+1)))+
        np.arange(x.shape[0]).reshape(-1,1)).ravel(), #same index matrix flaten 
    np.tile(np.arange(x.shape[0]),2*wz) #repeating `x` indexes to each window position
    )) 
x[ri[:,(ri[0]>=0)&(ri[0]<x.shape[0])]]#.T #uncomment .T to make it vertical   

输出：

  [[2, 3, 1, 3, 4, 1, 2, 4, 2, 3],
   [3, 4, 2, 3, 4, 1, 2, 3, 1, 2]]

---

请查看有关所使用函数的文档np.arange, np.reshape, np.place, np.hstack, broadcasting rules和indexing。

- ilia timofeev

---

Ilia，谢谢你，你的解决方案正是我一直在寻找的，可以接受初始数组和窗口大小的随机输入参数。然而输出结果与我的不同。我发现矩阵索引的问题 - 用于构建它的初始数组不应该包含零[-2 -1 2 1]，而应该是[-2 -1 0 1 2]。 - Ivan Telnov

@IvanTelnov你进行了性能测试吗？新的基准测试结果如何？ - ilia timofeev

我的先前解决方案（与我在问题中给出的不同）：每次循环304微秒± 8.25微秒（7次运行，每个1000个循环的平均值±标准差）与您的方法相比： 每次循环84.2微秒± 748纳秒（7次运行，每个10000个循环的平均值±标准差）。 - Ivan Telnov

为什么要优化如此小的时间？也许我们应该优化更大的范围以获得更有价值的改进？ - ilia timofeev

每个输入数组都表示其中单词的索引。因此，循环的大小不能超过非常长句子中的单词数量。 - Ivan Telnov

不幸的是，我不能将所有句子连接成一个大数组，然后在其中使用滑动窗口函数，因为窗口不应该从不同的句子中获取单词，所以在这两种情况下，我都要遍历给定的数组（句子），然后使用滑动窗口函数。这就是为什么在真正大的数组上进行测试对我来说没有意义。然而，性能差异确实随着句子数量的增加而增加，从2倍更好（如我所示）到6倍，但这更多是累积性能。现在我在思考如何改变我的算法，以更好地利用当前解决方案的潜在性能。 - Ivan Telnov

Answer 3

numpy方法更受青睐，但对于那些感兴趣的人，这里是一种函数式方法：

给定

import functools as ft


# Helper function
def curry(f):
    @ft.wraps(f)
    def wrapped(arg):
        try:
            return f(arg)
        except TypeError:
            return curry(ft.wraps(f)(ft.partial(f, arg)))
    return wrapped

代码

lst = [1, 2, 3, 4]
c = curry(lambda x, y: x + y)
funcs = [c(-1), c(1), c(-2), c(2)]
set_ = set(lst)


[[x, 0] if fn(x) not in set_ else [x, fn(x)] for fn in funcs for x in lst]

输出

[[1, 0],
 [2, 1],
 [3, 2],
 [4, 3],
 [1, 2],
 [2, 3],
 [3, 4],
 [4, 0],
 [1, 0],
 [2, 0],
 [3, 1],
 [4, 2],
 [1, 3],
 [2, 4],
 [3, 0],
 [4, 0]]

---

细节

在列表推导的双重for循环中，迭代了一个柯里化函数列表，并且每个函数都被应用于主列表(lst)的每个元素。柯里化允许您通过传递一些参数(例如1、-1、-2、2)来计算新值，然后稍后再传递主列表中的元素。

创建元组，例如(主要元素，计算元素)。列表推导的条件部分将0替换为未在主列表中找到的计算元素。

还可以参见这个curry函数的实现。

- pylang