I have implemented this runningMax class:
class RunningMax():
def __init__(self,window):
self.window = window
self.values = deque()
self.maxima = deque()
return
def append(self,x):
self.values.append(x)
i = 0
while (len(self.maxima) > 0) and (x > self.maxima[-1]) and (i < min(len(self.values),self.window)):
# pop all values at back that are smaller than new value
self.maxima.pop()
i += 1
self.maxima.append(x)
return
def pop(self):
full = len(self.values) == self.window
if self.maxima[0]==self.values[0] and full:
# if a value is about to leave the sliding window, kick it from maxima
max = self.maxima.popleft()
else:
max = self.maxima[0]
if full:
self.values.popleft()
return max
def compute_max(a,window):
rm = RunningMax(window)
maxs = []
for i,ai in enumerate(a):
rm.append(ai)
maxs.append(rm.pop())
return maxs
The output of compute_max(a,window) compares exactly to:
pd.Series(a).rolling(window,min_periods=1).max()
So, its implemented correctly. Just would like to ask if there is any potential for improvement here? Anything that could be written more optimally basically.
Thanks!