Найти макс (и мин) на подвижном интервале с помощью Python

Question

У меня есть массив как

[5.5, 6.0, 6.0, 6.5, 6.0, 5.5, 5.5, 5.0, 4.5]. 

всех чисел этого массива различаются по 0,5, а максимальная разность двух последовательных чисел также 0,5 (они могут быть такими же, как в примере). и есть подвижный интервал, или ящик, который охватывает, например, 3 последовательные номера, например:

[(5.5, 6.0, 6.0), 6.5, 6.0, 5.5, 5.5, 5.0, 4.5] # min: 5.5, max: 6.0 

и коробка движется к правой по одному:

[5.5, (6.0, 6.0, 6.5), 6.0, 5.5, 5.5, 5.0, 4.5] # min: 6.0, max: 6.5 

[5.5, 6.0, (6.0, 6.5, 6.0), 5.5, 5.5, 5.0, 4.5] # min: 6.0, max: 6.5 

вопрос является , как я могу найти min и max чисел внутри поля для каждого окна времени?

Я могу обработать его, когда размер блока и массива мал, как этот пример, но мне нужно применить его к размеру массива размером 100000 и размеру коробки 10000. Используя мой метод (я вычисляю каждый макс и мин, используя for- цикл для каждого окна времени проходит), потребовалось слишком много времени (у меня есть еще 100 массивов, которые нужно выполнить и нужно многократно запускать). Существует некоторое ограничение по времени, поэтому мне нужно запустить его, как один расчет за 0,5 секунды.

Answer 1

l = [5.5, 6.0, 6.0, 6.5, 6.0, 5.5, 5.5, 5.0, 4.5] 

windoSize = 3 

for i in range(0,len(l)-windowSize+1): 

    print max(l[i:i+windoSize]) 

выход:

6.0 
6.5 
6.5 
6.5 
6.0 
5.5 
5.5 

Answer 2

Посмотрите на rolling windows от панд:

>>> import pandas as pd 
>>> L = [5.5, 6.0, 6.0, 6.5, 6.0, 5.5, 5.5, 5.0, 4.5] 
>>> a = pd.DataFrame(L) 
>>> pd.rolling_max(a, 3) 
    0 
0 NaN 
1 NaN 
2 6.0 
3 6.5 
4 6.5 
5 6.5 
6 6.0 
7 5.5 
8 5.5 
>>> pd.rolling_min(a, 3) 
    0 
0 NaN 
1 NaN 
2 5.5 
3 6.0 
4 6.0 
5 5.5 
6 5.5 
7 5.0 
8 4.5 

Answer 3

Это прокатное окно, которое может быть реализовать в пандах как другой ответ показывает.

Если, однако, вы хотите реализовать его самостоятельно, следующий код будет полезен. Этот код может быть оптимизирован далее и может быть более питоническим, но он должен хорошо понимать, что происходит в алгоритме.

Первоначально минимальные и максимальные значения найдены для начального окна. Как только это инициализируется, мы обрабатываем вспомогательный массив как очередь, и только 2 значения становятся важными, добавляется новое значение и удаляется старое значение.

Если старое значение является минимальным или максимальным, мы пересчитали минимум или максимум, в противном случае мы проверяем, является ли новое значение новым максимумом или минимумом.

def updateMinMaxValues(minVal,maxVal,val): 
    if val < minVal: 
     minVal = val 
    if val > maxVal: 
     maxVal= val 
    return minVal,maxVal 

values = [5.5, 6.0, 6.0, 6.5, 6.0, 5.5, 5.5, 5.0, 4.5] 
windowSize = 3 
minVal,maxVal = min(values[:windowSize]),max(values[:windowSize]) 

print(minVal,maxVal) 
for stepIndex in range(windowSize,len(values)): 
    oldVal,newVal = values[stepIndex-windowSize],values[stepIndex] 
    if oldVal == minVal: 
     minVal = min(values[stepIndex-windowSize+1:stepIndex+1]) 
    if oldVal == maxVal: 
     maxVal = max(values[stepIndex-(windowSize)+1:stepIndex+1]) 
    minVal,maxVal = updateMinMaxValues(minVal,maxVal,newVal) 
    print(minVal,maxVal) 

приводит:

5.5 6.0 
6.0 6.5 
6.0 6.5 
5.5 6.5 
5.5 6.0 
5.0 5.5 
4.5 5.5 

Answer 4

Не уверен, что если есть способ эффективно использовать медленную скользящую структуру потока чисел.

Я решил, что лучший общий способ сделать это - с приоритетными очередями. Я оставил свое описание того, как это сделать ниже. Это O (log (window_size)) за новое число в окне.

Однако комментарий Вима на исходное сообщение указывает на то, что есть O (1) алгоритм, описанный в этой статье: Implement a queue in which push_rear(), pop_front() and get_min() are all constant time operations

Просто сохраняя один из них, который держит мин и макс будет быть лучшим решением на сегодняшний день.

Но для справки здесь моя попытка:

Поддерживать пару очереди приоритетов, один для макс и один для мин, а также добавлять и удалять записи из каждого, каждый раз. Это добавляет довольно много накладных расходов для каждой новой записи [O (log (window_size))], но имеет приятное плавное поведение для каждой записи и хорошую общую эффективность.

Модуль Python heapq является обычным способом реализации очереди приоритетов в Python. Тем не менее, он не поддерживает прямое удаление записей или изменение их приоритета. Это можно сделать, добавив индекс словаря из числа в позицию в очереди, без увеличения вычислительной сложности. Чтобы удалить запись, вы можете обновить ее номер до крайне низкого (или высокого уровня соответственно) и переупаковать, чтобы он переместился в верхнюю часть и может быть удален.

Вот пример, который выглядит нормально, хотя я не проверял:

http://code.activestate.com/recipes/522995-priority-dict-a-priority-queue-with-updatable-prio/

Вам нужно будет неоднозначность записи с тем же значением в словаре, или хранить несколько значений в ключе, так что вы можете найти все экземпляры, когда придет время их удалить.

Answer 5

Сначала мне показалось, что для каждого элемента большого списка требуется минимальное количество операций O (log (window_size)) (см. Мой другой ответ). Но @wim указал мне на самом деле замечательный алгоритм, описанный на @adamax в этом посте:

Implement a queue in which push_rear(), pop_front() and get_min() are all constant time operations

Вот реализация.

Запуск его на предлагаемые 100000 номеров с окном 1000 занимает 0,6 секунды вместо 60 секунд наивного алгоритма.

class MinMaxStack(object): 

    def __init__(self): 
     self.stack = [] 

    def push(self,val): 
     if not self.stack: 
      self.stack = [(val,val,val)] 
     else: 
      _,minimum,maximum = self.stack[-1] 
      if val < minimum: 
       self.stack.append((val,val,maximum)) 
      elif val > maximum: 
       self.stack.append((val,minimum,val)) 
      else: 
       self.stack.append((val,minimum,maximum)) 

    def pop(self): 
     return self.stack.pop() 

    def get_minimax(self): 
     return self.stack[-1][1:] 

    def __len__(self): 
     return len(self.stack) 

class RollingWindow(object): 

    def __init__(self): 
     self.push_stack = MinMaxStack() 
     self.pop_stack = MinMaxStack() 

    def push_only(self,o): 
     self.push_stack.push(o) 

    def push_and_pop(self,o): 
     self.push_stack.push(o) 
     if not self.pop_stack: 
      for i in range(len(self.push_stack.stack)-1): 
       self.pop_stack.push(self.push_stack.pop()[0]) 
      self.push_stack.pop() 
     else: 
      self.pop_stack.pop() 

    def get_minimax(self): 
     if not self.pop_stack: 
      return self.push_stack.get_minimax() 
     elif not self.push_stack: 
      return self.pop_stack.get_minimax() 
     mn1,mx1 = self.pop_stack.get_minimax() 
     mn2,mx2 = self.push_stack.get_minimax() 
     return min(mn1,mn2),max(mx1,mx2) 



import time 
import random 
window = 10000 
test_length = 100000 
data = [random.randint(1,100) for i in range(test_length)] 

s = time.time() 

wr = RollingWindow() 
answer1 = [] 
for i in range(test_length): 
    if i < window: 
     wr.push_only(data[i]) 
    else: 
     wr.push_and_pop(data[i]) 
    answer1.append(wr.get_minimax()) 

print(s-time.time()) 

s = time.time() 
answer2 = [] 
for i in range(test_length): 
    if i+1 < window: 
     current_window = i+1 
    else: 
     current_window = window 
    answer2.append((min(data[i+1-current_window:i+1]),max(data[i+1-current_window:i+1]))) 

print(s-time.time()) 

if answer1 != answer2: 
    print("Test Fail") 

Возможны небольшие улучшения в производительности. Эта версия постоянно растет и сокращает список python, используемый как стек. Это немного быстрее, чтобы никогда не сжимать его и использовать конечный указатель. Но только несколько процентов. Если вы действительно отчаянно нуждались в еще нескольких процентах, вы могли бы объединить два стека в класс окна и уменьшить косвенность в вызовах. Я построил оптимизированную версию, заменив списки с помощью collections.deque и вложив код стека и доведя ее до 0,32 секунды.

Если требуется еще больше скорости, это будет довольно легко кодировать в C или Cython (в частности, для фиксированного размера окна), особенно если вы можете ограничить тип значений в стеках.