This computes the "rolling max" of A (similar to rolling average) over a sliding window of length K:
import numpy as np
A = np.random.rand(100000)
K = 10
rollingmax = np.array([max(A[j:j+K]) for j in range(len(A)-K)])
but I think it is far from optimal in terms of performance.
I know that the pandas library has rolling_max, but in my project, I don't want to use this new dependance.
Question: is there a simple way to compute the rolling maximum with numpy only?
I guess this little trick using strides and as_strided will do the job:
def max_rolling1(a, window,axis =1):
shape = a.shape[:-1] + (a.shape[-1] - window + 1, window)
strides = a.strides + (a.strides[-1],)
rolling = np.lib.stride_tricks.as_strided(a, shape=shape, strides=strides)
return np.max(rolling,axis=axis)
and for comaprison purpose I defined another function based on your algorithm :
def max_rolling2(A,K):
rollingmax = np.array([max(A[j:j+K]) for j in range(len(A)-K)])
return rollingmax
and the comparison by timeit on my laptop is :
with :
A = np.random.rand(100000)
K = 10
%timeit X = max_rolling2(A,K)
170 ms ± 19.8 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)
%timeit X = max_rolling1(A,K)
> 3.75 ms ± 479 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
The solution is totally similar to Divakar's answer here (full credit to him) but the final cropping of the array has different indices in this context:
maximum_filter1d(A, size=K)[K//2:-((K+1)//2)]
Example:
import numpy as np
from scipy.ndimage.filters import maximum_filter1d
A = np.random.randint(0, 10, (50))
K = 5
rollingmax = np.array([max(A[j-K:j]) for j in range(K,len(A))])
rollingmax2 = np.array([max(A[j:j+K]) for j in range(len(A)-K)])
rollingmax3 = maximum_filter1d(A,size=K)[K//2:-((K+1)//2)]
print A, rollingmax, rollingmax2, rollingmax3
[6 7 7 9 4 5 4 7 2 0 3 3 5 9 4 6 6 1 5 2 7 5 7 7 5 6 0 9 0 5 9 3 7 1 9 5 3 7 5 1 6 9 6 0 5 1 5 5 4 9]
[9 9 9 9 7 7 7 7 5 9 9 9 9 9 6 6 7 7 7 7 7 7 7 9 9 9 9 9 9 9 9 9 9 9 9 7 7 9 9 9 9 9 6 5 5]
[9 9 9 9 7 7 7 7 5 9 9 9 9 9 6 6 7 7 7 7 7 7 7 9 9 9 9 9 9 9 9 9 9 9 9 7 7 9 9 9 9 9 6 5 5]
[9 9 9 9 7 7 7 7 5 9 9 9 9 9 6 6 7 7 7 7 7 7 7 9 9 9 9 9 9 9 9 9 9 9 9 7 7 9 9 9 9 9 6 5 5]
Just tested some of the code above, and it returns a few non expected results:
input: test_array_input=np.array([1, 3, 4, 5, 4, 9, 7, 2, 4])
output with rolling max (above) yields: [ 1 3 4 5 4 9 7 2] when checking max over 2 fields, i.e. K=1
whereas would expect: [3 4 5 5 9 9 7 4] --> i,.e first field max of 1 & 3
I've implemented:
def udf_rolling_maximum_array(inputarray,numfieldsrolling):
# example: input ( 1 2 3 4 5 6 2 3 4) -- 9 fields
# example: rolling of 2 numfieldsrolling
# example: output ( 2 3 4 5 6 6 3 4) -- 8 fields
# example: rolling of 3 numfieldsrolling
# example: output (3 4 5 6 6 6 4) -- 7 fields
import numpy as np
# setup result array
k=len(inputarray)-numfieldsrolling+1
rollingmax=np.empty(k)
for i in range(k): rollingmax[i] = max(inputarray[i:i+numfieldsrolling])
return rollingmax #return rolling average array as output of UDF
The solutions given above rollingmax,rollingmax1,rollingmax2 in the original post failed my unit test - cf my expectations. See examples in the code text for behaviour I expected. I've not optimised my code for speed - as I dont'have huge data sets.
Was happy to find these solutions - until I tried with large values for K. Having an array of ~6M floats, K = 25000....takes very long
