移位numpy数组中的元素

Question

这个问题在底部有自己的答案。使用预先分配的数组。

多年前this question 的后续行动，numpy 中是否有规范的“移位”功能？我没有看到来自 the documentation 的任何内容。

这是我正在寻找的简单版本：

def shift(xs, n):
    if n >= 0:
        return np.r_[np.full(n, np.nan), xs[:-n]]
    else:
        return np.r_[xs[-n:], np.full(-n, np.nan)]

使用这个就像：

In [76]: xs
Out[76]: array([ 0.,  1.,  2.,  3.,  4.,  5.,  6.,  7.,  8.,  9.])

In [77]: shift(xs, 3)
Out[77]: array([ nan,  nan,  nan,   0.,   1.,   2.,   3.,   4.,   5.,   6.])

In [78]: shift(xs, -3)
Out[78]: array([  3.,   4.,   5.,   6.,   7.,   8.,   9.,  nan,  nan,  nan])

_{这个问题来自我昨天尝试write a fast rolling_product。我需要一种方法来“转移”累积产品，而我所能想到的就是复制 np.roll() 中的逻辑。}

---

所以np.concatenate() 比np.r_[] 快得多。这个版本的函数性能要好很多：

def shift(xs, n):
    if n >= 0:
        return np.concatenate((np.full(n, np.nan), xs[:-n]))
    else:
        return np.concatenate((xs[-n:], np.full(-n, np.nan)))

---

更快的版本只是预先分配数组：

def shift(xs, n):
    e = np.empty_like(xs)
    if n >= 0:
        e[:n] = np.nan
        e[n:] = xs[:-n]
    else:
        e[n:] = np.nan
        e[:n] = xs[-n:]
    return e

上面的建议就是答案。使用预先分配的数组。

Answer 1

不是 numpy，但 scipy 提供了您想要的 shift 功能，

import numpy as np
from scipy.ndimage.interpolation import shift

xs = np.array([ 0.,  1.,  2.,  3.,  4.,  5.,  6.,  7.,  8.,  9.])

shift(xs, 3, cval=np.NaN)

默认是从数组外部引入一个常量值cval，这里设置为nan。这给出了所需的输出，

array([ nan, nan, nan, 0., 1., 2., 3., 4., 5., 6.])

负移的作用类似，

shift(xs, -3, cval=np.NaN)

提供输出

array([  3.,   4.,   5.,   6.,   7.,   8.,   9.,  nan,  nan,  nan])

Answer 2

对于那些只想复制和粘贴最快实现 shift 的人，有一个基准和结论（见最后）。另外，我引入了fill_value参数并修复了一些bug。

基准测试

import numpy as np
import timeit

# enhanced from IronManMark20 version
def shift1(arr, num, fill_value=np.nan):
    arr = np.roll(arr,num)
    if num < 0:
        arr[num:] = fill_value
    elif num > 0:
        arr[:num] = fill_value
    return arr

# use np.roll and np.put by IronManMark20
def shift2(arr,num):
    arr=np.roll(arr,num)
    if num<0:
         np.put(arr,range(len(arr)+num,len(arr)),np.nan)
    elif num > 0:
         np.put(arr,range(num),np.nan)
    return arr

# use np.pad and slice by me.
def shift3(arr, num, fill_value=np.nan):
    l = len(arr)
    if num < 0:
        arr = np.pad(arr, (0, abs(num)), mode='constant', constant_values=(fill_value,))[:-num]
    elif num > 0:
        arr = np.pad(arr, (num, 0), mode='constant', constant_values=(fill_value,))[:-num]

    return arr

# use np.concatenate and np.full by chrisaycock
def shift4(arr, num, fill_value=np.nan):
    if num >= 0:
        return np.concatenate((np.full(num, fill_value), arr[:-num]))
    else:
        return np.concatenate((arr[-num:], np.full(-num, fill_value)))

# preallocate empty array and assign slice by chrisaycock
def shift5(arr, num, fill_value=np.nan):
    result = np.empty_like(arr)
    if num > 0:
        result[:num] = fill_value
        result[num:] = arr[:-num]
    elif num < 0:
        result[num:] = fill_value
        result[:num] = arr[-num:]
    else:
        result[:] = arr
    return result

arr = np.arange(2000).astype(float)

def benchmark_shift1():
    shift1(arr, 3)

def benchmark_shift2():
    shift2(arr, 3)

def benchmark_shift3():
    shift3(arr, 3)

def benchmark_shift4():
    shift4(arr, 3)

def benchmark_shift5():
    shift5(arr, 3)

benchmark_set = ['benchmark_shift1', 'benchmark_shift2', 'benchmark_shift3', 'benchmark_shift4', 'benchmark_shift5']

for x in benchmark_set:
    number = 10000
    t = timeit.timeit('%s()' % x, 'from __main__ import %s' % x, number=number)
    print '%s time: %f' % (x, t)

基准测试结果：

benchmark_shift1 time: 0.265238
benchmark_shift2 time: 0.285175
benchmark_shift3 time: 0.473890
benchmark_shift4 time: 0.099049
benchmark_shift5 time: 0.052836

结论

shift5 是赢家！这是 OP 的第三个解决方案。

Answer 3

基准测试和引入 Numba

1。总结

• 接受的答案 (scipy.ndimage.interpolation.shift) 是本页列出的最慢解决方案。
• 当数组大小小于 ~25.000 时，Numba (@numba.njit) 提供了一些性能提升
• “任何方法”在数组大小较大 (>250.000) 时同样适用。
• 最快的选择真的取决于
  (1)  数组的长度
  (2)  您需要完成的班次。
• 下面是本页 (2020-07-11) 上列出的所有不同方法的时序图片，使用常量 shift = 10。可以看出，对于小数组大小，某些方法的使用率超过 +2000%时间比最好的方法。

2。具有最佳选择的详细基准

• 如果您想要优秀的多面手，请选择 shift4_numba（定义见下文）

3。代码

3.1shift4_numba

• 良好的多面手；最大 20% wrt。使用任何数组大小的最佳方法
• 中等数组大小的最佳方法：~ 500
• 警告：Numba jit（即时编译器）仅在您多次调用修饰函数时才会提高性能。第一次调用的时间通常比后续调用长 3-4 倍。使用ahead of time compiled numba，您可以进一步提升性能。

import numba

@numba.njit
def shift4_numba(arr, num, fill_value=np.nan):
    if num >= 0:
        return np.concatenate((np.full(num, fill_value), arr[:-num]))
    else:
        return np.concatenate((arr[-num:], np.full(-num, fill_value)))

3.2. shift5_numba

• 小 (N
• 在任何数组大小上都有良好的性能；与最快的解决方案相比，最大 + 50%。
• 警告：Numba jit（即时编译器）仅在您多次调用修饰函数时才会提高性能。第一次调用的时间通常比后续调用长 3-4 倍。使用ahead of time compiled numba，您可以进一步提升性能。

import numba

@numba.njit
def shift5_numba(arr, num, fill_value=np.nan):
    result = np.empty_like(arr)
    if num > 0:
        result[:num] = fill_value
        result[num:] = arr[:-num]
    elif num < 0:
        result[num:] = fill_value
        result[:num] = arr[-num:]
    else:
        result[:] = arr
    return result

3.3。 shift5

• 数组大小约为 20.000 的最佳方法
• 与shift5_numba 相同，只是删除@numba.njit 装饰器。

4 附录

4.1 使用方法详情

• shift_scipy: scipy.ndimage.interpolation.shift (scipy 1.4.1) - 接受答案的选项，显然是最慢的选择。
• shift1: np.roll 和 out[:num] xnp.nan IronManMark20 & gzc
• shift2: np.roll 和 np.put IronManMark20
• shift3: np.pad 和 slice gzc
• shift4: np.concatenate 和 np.full chrisaycock
• shift5：使用两次result[slice] = x by chrisaycock
• shift#_numba: @numba.njit 修饰过的版本。

shift2 和 shift3 包含当前 numba (0.50.1) 不支持的函数。

4.2 其他测试结果

4.2.1 相对时序，所有方法

• Relative timings, 10% shift, all methods
• Relative timings, constant shift (10), all methods

4.2.2 原始时序，所有方法

• Raw timings, constant shift (10), all methods
• Raw timings, 10% shift, all methods

4.2.3 原始时序，少数最佳方法

• Raw timings with small arrays, constant shift (10), few best methods
• Raw timings with small arrays, 10% shift, few best methods
• Raw timings with large arrays, constant shift (10), few best methods
• Raw timings with large arrays, 10% shift, few best methods

Answer 4

没有一个功能可以满足您的需求。你对转变的定义与大多数人所做的略有不同。移动数组的方法更常见的是循环：

>>>xs=np.array([1,2,3,4,5])
>>>shift(xs,3)
array([3,4,5,1,2])

但是，您可以使用两个函数来做您想做的事情。
考虑a=np.array([ 0., 1., 2., 3., 4., 5., 6., 7., 8., 9.])：

def shift2(arr,num):
    arr=np.roll(arr,num)
    if num<0:
         np.put(arr,range(len(arr)+num,len(arr)),np.nan)
    elif num > 0:
         np.put(arr,range(num),np.nan)
    return arr
>>>shift2(a,3)
[ nan  nan  nan   0.   1.   2.   3.   4.   5.   6.]
>>>shift2(a,-3)
[  3.   4.   5.   6.   7.   8.   9.  nan  nan  nan]

在您给定的函数和您提供的上述代码上运行 cProfile 后，我发现您提供的代码进行了 42 次函数调用，而 shift2 在 arr 为正时进行了 14 次调用，在为负时进行了 16 次调用。 我将在时间上进行试验，看看每种方法在真实数据下的表现如何。

Answer 5

您可以先将ndarray 转换为Series 或DataFrame 与pandas，然后您可以根据需要使用shift 方法。

例子：

In [1]: from pandas import Series

In [2]: data = np.arange(10)

In [3]: data
Out[3]: array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])

In [4]: data = Series(data)

In [5]: data
Out[5]: 
0    0
1    1
2    2
3    3
4    4
5    5
6    6
7    7
8    8
9    9
dtype: int64

In [6]: data = data.shift(3)

In [7]: data
Out[7]: 
0    NaN
1    NaN
2    NaN
3    0.0
4    1.0
5    2.0
6    3.0
7    4.0
8    5.0
9    6.0
dtype: float64

In [8]: data = data.values

In [9]: data
Out[9]: array([ nan,  nan,  nan,   0.,   1.,   2.,   3.,   4.,   5.,   6.])

Answer 6

你也可以用 Pandas 做到这一点：

使用 2356 长的数组：

import numpy as np

xs = np.array([...])

使用 scipy：

from scipy.ndimage.interpolation import shift

%timeit shift(xs, 1, cval=np.nan)
# 956 µs ± 77.9 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)

使用熊猫：

import pandas as pd

%timeit pd.Series(xs).shift(1).values
# 377 µs ± 9.42 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)

在这个例子中，使用 Pandas 比 Scipy 快大约 8 倍

Answer 7

如果您想要 numpy 的单线并且不太关心性能，请尝试：

np.sum(np.diag(the_array,1),0)[:-1]

解释：np.diag(the_array,1) 创建一个矩阵，其中您的数组在对角线上，np.sum(...,0) 按列对矩阵求和，...[:-1] 采用与原始数组大小相对应的元素。使用1 和:-1 作为参数可以让您朝不同的方向转变。

Answer 8

也许np.roll 是你所需要的

arr = np.arange(10)
shift = 2  # shift length
arr_1 = np.roll(arr, shift=shift)
arr_1[:shift] = np.nan

Answer 9

我的解决方案涉及np.roll 和屏蔽数组：

import numpy as np
import numpy.ma as ma # this is for masked array

def shift(arr, shift):
    r_arr = np.roll(arr, shift=shift)
    m_arr = ma.masked_array(r_arr,dtype=float)
    if shift > 0: m_arr[:shift] = ma.masked
    else: m_arr[shift:] = ma.masked
    return m_arr.filled(np.nan)

基本上，我只是使用np.roll 来移动数组，然后使用ma.masked_array 将不需要的元素标记为无效，并用np.nan 填充那些无效位置。我将dtype 设置为float，这样填充np.nan 就不会造成任何问题。

In [11]: shift(arr, 3)
Out[11]: array([nan, nan, nan,  0.,  1.,  2.,  3.,  4.,  5.,  6.])

In [12]: shift(arr, -3)
Out[12]: array([ 3.,  4.,  5.,  6.,  7.,  8.,  9., nan, nan, nan])

Answer 10

这里是支持任意多维数组的快速答案 (shift5) 的概括：

def shift(array, offset, constant_values=0):
  """Returns copy of array shifted by offset, with fill using constant."""
  array = np.asarray(array)
  offset = np.atleast_1d(offset)
  assert len(offset) == array.ndim
  new_array = np.empty_like(array)

  def slice1(o):
    return slice(o, None) if o >= 0 else slice(0, o)

  new_array[tuple(slice1(o) for o in offset)] = (
      array[tuple(slice1(-o) for o in offset)])

  for axis, o in enumerate(offset):
    new_array[(slice(None),) * axis +
              (slice(0, o) if o >= 0 else slice(o, None),)] = constant_values

  return new_array

Answer 11

我想我有一个更快的解决方案：为什么不直接使用 deque ？ 我在@gzc 的基准解决方案中添加了 2 个基准：

def shift6(arr, num, fill_value=np.nan):
    for _ in range(num):
        darr.appendleft(fill_value)

def shift7(arr, num, fill_value=np.nan):
    darr = deque(arr)
    for _ in range(num):
        darr.appendleft(fill_value)

darr = deque(arr)

def benchmark_shift6():
    shift6(arr, 3)
    
def benchmark_shift7():
    shift6(arr, 3)

benchmark_set = ['benchmark_shift1', 'benchmark_shift2', 'benchmark_shift3', 'benchmark_shift4', 'benchmark_shift5', 'benchmark_shift6', 'benchmark_shift7']

在我的笔记本电脑上，输出比提出的任何其他解决方案都要好很多：

%s time: ('benchmark_shift1', 0.08232757700170623) 
%s time: ('benchmark_shift2', 0.0934765400015749) 
%s time: ('benchmark_shift3', 0.14349375600431813) 
%s time: ('benchmark_shift4', 0.03575193700089585) 
%s time: ('benchmark_shift5', 0.01389261399890529) 
%s time: ('benchmark_shift6', 0.0025887360025080852) 
%s time: ('benchmark_shift7', 0.0024806019937386736)

Answer 12

一种无需将代码泄露到案例中的方法

带数组：

def shift(arr, dx, default_value):
    result = np.empty_like(arr)
    get_neg_or_none = lambda s: s if s < 0 else None
    get_pos_or_none = lambda s: s if s > 0 else None
    result[get_neg_or_none(dx): get_pos_or_none(dx)] = default_value
    result[get_pos_or_none(dx): get_neg_or_none(dx)] = arr[get_pos_or_none(-dx): get_neg_or_none(-dx)]     
    return result

使用矩阵可以这样做：

def shift(image, dx, dy, default_value):
    res = np.full_like(image, default_value)

    get_neg_or_none = lambda s: s if s < 0 else None
    get_pos_or_none = lambda s : s if s > 0 else None

    res[get_pos_or_none(-dy): get_neg_or_none(-dy), get_pos_or_none(-dx): get_neg_or_none(-dx)] = \
        image[get_pos_or_none(dy): get_neg_or_none(dy), get_pos_or_none(dx): get_neg_or_none(dx)]
    return res