Numpy Vektorisierung von Sliding-Window-Betrieb

Question

Ich habe folgende numpy Arrays:

arr_1 = [[1,2],[3,4],[5,6]] # 3 X 2 
arr_2 = [[0.5,0.6],[0.7,0.8],[0.9,1.0],[1.1,1.2],[1.3,1.4]] # 5 X 2 

arr_1 ist eindeutig ein 3 X 2 Array, während ein arr_25 X 2 Array ist.

Jetzt ohne Schleifenbildung, ich möchte elementarweise arr_1 und arr_2 multiplizieren, so dass ich eine Schiebefenstertechnik (Fenstergröße 3) auf arr_2 anwende.

Example: 

Multiplication 1: np.multiply(arr_1,arr_2[:3,:]) 

Multiplication 2: np.multiply(arr_1,arr_2[1:4,:]) 

Multiplication 3: np.multiply(arr_1,arr_2[2:5,:]) 

Ich mag dies eine Matrixmultiplikation Form in irgendeiner Art tun, um es schneller als meine aktuelle Lösung, die von der Form:

for i in (2): 
    np.multiply(arr_1,arr_2[i:i+3,:]) 

Also, wenn die Anzahl der Zeilen in arr_2 ist groß (in der Größenordnung von Zehntausenden), skaliert diese Lösung nicht wirklich sehr gut.

Jede Hilfe würde sehr geschätzt werden.

Answer 1

Wir können NumPy broadcasting verwenden, um diese gleitenden Fensterindizes vektorisiert zu erstellen. Dann können wir einfach in arr_2 mit denen zu einem 3D Array indexieren und Element-weise Multiplikation mit 2D Array arr_1, die wiederum auf broadcasting bringt wieder führen.

So würden wir eine vektorisierte Implementierung wie so haben -

W = arr_1.shape[0] # Window size 
idx = np.arange(arr_2.shape[0]-W+1)[:,None] + np.arange(W) 
out = arr_1*arr_2[idx] 

Runtime testen und verifizieren Ergebnisse -

In [143]: # Input arrays 
    ...: arr_1 = np.random.rand(3,2) 
    ...: arr_2 = np.random.rand(10000,2) 
    ...: 
    ...: def org_app(arr_1,arr_2): 
    ...:  W = arr_1.shape[0] # Window size 
    ...:  L = arr_2.shape[0]-W+1 
    ...:  out = np.empty((L,W,arr_1.shape[1])) 
    ...:  for i in range(L): 
    ...:  out[i] = np.multiply(arr_1,arr_2[i:i+W,:]) 
    ...:  return out 
    ...: 
    ...: def vectorized_app(arr_1,arr_2): 
    ...:  W = arr_1.shape[0] # Window size 
    ...:  idx = np.arange(arr_2.shape[0]-W+1)[:,None] + np.arange(W) 
    ...:  return arr_1*arr_2[idx] 
    ...: 

In [144]: np.allclose(org_app(arr_1,arr_2),vectorized_app(arr_1,arr_2)) 
Out[144]: True 

In [145]: %timeit org_app(arr_1,arr_2) 
10 loops, best of 3: 47.3 ms per loop 

In [146]: %timeit vectorized_app(arr_1,arr_2) 
1000 loops, best of 3: 1.21 ms per loop 

Answer 2

Dies ist ein schöner Fall die Geschwindigkeit der as_strided und Divakar Rundfunk zu testen.

In [281]: %%timeit 
    ...: out=np.empty((L,W,arr1.shape[1])) 
    ...: for i in range(L): 
    ...: out[i]=np.multiply(arr1,arr2[i:i+W,:]) 
    ...: 
10 loops, best of 3: 48.9 ms per loop 
In [282]: %%timeit 
    ...: idx=np.arange(L)[:,None]+np.arange(W) 
    ...: out=arr1*arr2[idx] 
    ...: 
100 loops, best of 3: 2.18 ms per loop 
In [283]: %%timeit 
    ...: arr3=as_strided(arr2, shape=(L,W,2), strides=(16,16,8)) 
    ...: out=arr1*arr3 
    ...: 
1000 loops, best of 3: 805 µs per loop 

Create Numpy array without enumerating array für mehr von einem Vergleich dieser Methoden.