Just wondering if there is a way to construct a tumbling window in python. So for example if I have list/ndarray , listA = [3,2,5,9,4,6,3,8,7,9]
. Then how could I find the maximum of the first 3 items (3,2,5) -> 5, and then the next 3 items (9,4,6) -> 9 and so on… Sort of like breaking it up to sections and finding the max. So the final result would be list [5,9,8,9]
Advertisement
Answer
Approach #1: One-liner for windowed-max using np.maximum.reduceat
–
In [118]: np.maximum.reduceat(listA,np.arange(0,len(listA),3)) Out[118]: array([5, 9, 8, 9])
Becomes more compact with np.r_
–
np.maximum.reduceat(listA,np.r_[:len(listA):3])
Approach #2: Generic ufunc way
Here’s a function for generic ufuncs and that window length as a parameter –
def windowed_ufunc(a, ufunc, W): a = np.asarray(a) n = len(a) L = W*(n//W) out = ufunc(a[:L].reshape(-1,W),axis=1) if n>L: out = np.hstack((out, ufunc(a[L:]))) return out
Sample run –
In [81]: a = [3,2,5,9,4,6,3,8,7,9] In [82]: windowed_ufunc(a, ufunc=np.max, W=3) Out[82]: array([5, 9, 8, 9])
On other ufuncs –
In [83]: windowed_ufunc(a, ufunc=np.min, W=3) Out[83]: array([2, 4, 3, 9]) In [84]: windowed_ufunc(a, ufunc=np.sum, W=3) Out[84]: array([10, 19, 18, 9]) In [85]: windowed_ufunc(a, ufunc=np.mean, W=3) Out[85]: array([3.33333333, 6.33333333, 6. , 9. ])
Benchmarking
Timings on NumPy solutions on array data with sample data scaled up by 10000x
–
In [159]: a = [3,2,5,9,4,6,3,8,7,9] In [160]: a = np.tile(a, 10000) # @yatu's soln In [162]: %timeit moving_maxima(a, w=3) 435 µs ± 8.54 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each) # From this post - app#1 In [167]: %timeit np.maximum.reduceat(a,np.arange(0,len(a),3)) 353 µs ± 2.55 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each) # From this post - app#2 In [165]: %timeit windowed_ufunc(a, ufunc=np.max, W=3) 379 µs ± 6.44 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)