2D数组中每个对角线的最大值

Question

我有一个数组，需要动态窗口内的滚动差的最大值。

a = np.array([8, 18, 5,15,12])
print (a)
[ 8 18  5 15 12]

所以首先我将差异本身创造出来：

b = a - a[:, None]
print (b)
[[  0  10  -3   7   4]
 [-10   0 -13  -3  -6]
 [  3  13   0  10   7]
 [ -7   3 -10   0  -3]
 [ -4   6  -7   3   0]]

然后将上三角矩阵替换为0：

c = np.tril(b)
print (c)
[[  0   0   0   0   0]
 [-10   0   0   0   0]
 [  3  13   0   0   0]
 [ -7   3 -10   0   0]
 [ -4   6  -7   3   0]]

最后需要每条对角线上的最大值，这意味着：

max([0,0,0,0,0]) = 0  
max([-10,13,-10,3]) = 13
max([3,3,-7]) = 3
max([-7,6]) = 6
max([-4]) = -4

所以预期输出为：

[0, 13, 3, 6, -4]

有什么好的向量化解决方案吗？还是有其他期望输出的方法吗？

- jezrael

Answer 1

使用ndarray.diagonal

v = [max(c.diagonal(-i)) for i in range(b.shape[0])]
print(v) # [0, 13, 3, 6, -4]

- Guy

Answer 2

不确定这种方法在涉及高级索引时的效率如何，但这是一种实现的方式：

import numpy as np

a = np.array([8, 18, 5, 15, 12])
b = a[:, None] - a
# Fill lower triangle with largest negative
b[np.tril_indices(len(a))] = np.iinfo(b.dtype).min  # np.finfo for float
# Put diagonals as rows
s = b.strides[1]
diags = np.ndarray((len(a) - 1, len(a) - 1), b.dtype, b, offset=s, strides=(s, (len(a) + 1) * s))
# Get maximum from each row and add initial zero
c = np.r_[0, diags.max(1)]
print(c)
# [ 0 13  3  6 -4]

---

编辑：

另一个选择是使用Numba，可能不是你想要的，例如像这样使用：

import numpy as np
import numba as nb

def max_window_diffs_jdehesa(a):
    a = np.asarray(a)
    dtinf = np.iinfo(b.dtype) if np.issubdtype(b.dtype, np.integer) else np.finfo(b.dtype)
    out = np.full_like(a, dtinf.min)
    _pwise_diffs(a, out)
    return out

@nb.njit(parallel=True)
def _pwise_diffs(a, out):
    out[0] = 0
    for w in nb.prange(1, len(a)):
        for i in range(len(a) - w):
            out[w] = max(a[i] - a[i + w], out[w])

a = np.array([8, 18, 5, 15, 12])
print(max_window_diffs(a))
# [ 0 13  3  6 -4]

将这些方法与原始方法进行比较：

import numpy as np
import numba as nb

def max_window_diffs_orig(a):
    a = np.asarray(a)
    b = a - a[:, None]
    out = np.zeros(len(a), b.dtype)
    out[-1] = b[-1, 0]
    for i in range(1, len(a) - 1):
        out[i] = np.diag(b, -i).max()
    return out

def max_window_diffs_jdehesa_np(a):
    a = np.asarray(a)
    b = a[:, None] - a
    dtinf = np.iinfo(b.dtype) if np.issubdtype(b.dtype, np.integer) else np.finfo(b.dtype)
    b[np.tril_indices(len(a))] = dtinf.min
    s = b.strides[1]
    diags = np.ndarray((len(a) - 1, len(a) - 1), b.dtype, b, offset=s, strides=(s, (len(a) + 1) * s))
    return np.concatenate([[0], diags.max(1)])

def max_window_diffs_jdehesa_nb(a):
    a = np.asarray(a)
    dtinf = np.iinfo(b.dtype) if np.issubdtype(b.dtype, np.integer) else np.finfo(b.dtype)
    out = np.full_like(a, dtinf.min)
    _pwise_diffs(a, out)
    return out

@nb.njit(parallel=True)
def _pwise_diffs(a, out):
    out[0] = 0
    for w in nb.prange(1, len(a)):
        for i in range(len(a) - w):
            out[w] = max(a[i] - a[i + w], out[w])

np.random.seed(0)
a = np.random.randint(0, 100, size=100)
r = max_window_diffs_orig(a)
print((max_window_diffs_jdehesa_np(a) == r).all())
# True
print((max_window_diffs_jdehesa_nb(a) == r).all())
# True

%timeit max_window_diffs_orig(a)
# 348 µs ± 986 ns per loop (mean ± std. dev. of 7 runs, 1000 loops each)
%timeit max_window_diffs_jdehesa_np(a)
# 91.7 µs ± 1.3 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
%timeit max_window_diffs_jdehesa_nb(a)
# 19.7 µs ± 88.1 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)

np.random.seed(0)
a = np.random.randint(0, 100, size=10000)
%timeit max_window_diffs_orig(a)
# 651 ms ± 26 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
%timeit max_window_diffs_jdehesa_np(a)
# 1.61 s ± 6.19 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
%timeit max_window_diffs_jdehesa_nb(a)
# 22 ms ± 967 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

第一种方法对于较小的数组可能更好，但对于较大的数组效果不佳。而另一方面，Numba在所有情况下都表现得相当好。

- jdehesa

Answer 3

你可以使用 numpy.diagonal ：

a = np.array([8, 18, 5,15,12])
b = a - a[:, None]
c = np.tril(b)
for i in range(b.shape[0]):
    print(max(c.diagonal(-i)))

输出：

0
13
3
6
-4

- Sayandip Dutta

---

我认为矢量化，不需要循环。 - jezrael

Answer 4

这是一个使用 strides 的矢量化解决方案 -

from skimage.util import view_as_windows

n = len(a)
z = np.zeros(n-1,dtype=a.dtype)
p = np.concatenate((a,z))

s = view_as_windows(p,n)
mask = np.tri(n,k=-1,dtype=bool)[:,::-1]
v = s[0]-s
out = np.where(mask,v.min()-1,v).max(1)

使用单循环以提高内存效率 -

n = len(a)
out = [max(a[:-i+n]-a[i:]) for i in range(n)]

为更好地利用数组内存，请使用np.max代替max。

- Divakar

---

@jezrael 取决于数据大小。对于大尺寸，我认为使用切片+最大值的循环方式可能会因为内存效率而获胜。 - Divakar

Answer 5

你可以利用非正方形数组的形状从 (N+1, N) 转换为 (N, N+1)，这样对角线就会出现在列中。

from scipy.linalg import toeplitz
a = toeplitz([1,2,3,4], [1,4,3])
# array([[1, 4, 3],
#        [2, 1, 4],
#        [3, 2, 1],
#        [4, 3, 2]])
a.reshape(3, 4)
# array([[1, 4, 3, 2],
#        [1, 4, 3, 2],
#        [1, 4, 3, 2]])

您可以像这样使用它（请注意，我已经交换了符号并将下三角设置为零）。

smallv = -10000  # replace this with np.nan if you have floats

a = np.array([8, 18, 5,15,12])
b = a[:, None] - a

b[np.tril_indices(len(b), -1)] = smallv
d = np.vstack((b, np.full(len(b), smallv)))

d.reshape(len(d) - 1, -1).max(0)[:-1]
# array([ 0, 13,  3,  6, -4])

- Nils Werner