Коллекции стандартной библиотеки: deque, Counter, defaultdict и ChainMap
Содержание курса
Скользящее окно в потоке данных через deque(maxlen)
Представьте задачу: датчик непрерывно шлёт измерения температуры, и вам нужно вычислять среднее арифметическое последних трёх показаний. Если накапливать значения в обычном списке list, то для поддержания фиксированного окна придётся либо удалять нулевой элемент через del window[0], либо брать срез window[-3:]. Обе операции неэффективны: удаление первого элемента списка сдвигает все остальные элементы в памяти за , а срез создаёт новый объект списка.
В модуле collections для таких задач есть двусторонняя очередь deque. При её инициализации можно передать параметр maxlen:
from collections import deque
window = deque(maxlen=3)
for temp in [18.5, 19.0, 20.2, 21.1]:
window.append(temp)
print(list(window))
Вывод программы:
[18.5]
[18.5, 19.0]
[18.5, 19.0, 20.2]
[19.0, 20.2, 21.1]
Когда очередь заполнена до размера maxlen, вызов append() автоматически отбрасывает элемент с противоположного конца. При добавлении вправо вытесняется крайний левый элемент, а при вызове appendleft() — крайний правый. Обе операции выполняются за константное время , потому что deque реализован как двусвязный список блоков, а не монолитный динамический массив.
Используем deque(maxlen=K) для расчёта скользящего среднего на непрерывном потоке данных:
from collections import deque
from collections.abc import Iterable
def running_average(stream: Iterable[float | int], window_size: int) -> list[float]:
window = deque(maxlen=window_size)
averages: list[float] = []
for value in stream:
window.append(value)
averages.append(sum(window) / len(window))
return averages
sensor_data = [10, 20, 30, 40, 50]
print(running_average(sensor_data, window_size=3))
Вывод программы:
[10.0, 15.0, 20.0, 30.0, 40.0]
Пока элементов меньше трёх, среднее рассчитывается по фактической длине len(window). Как только окно заполняется, размер очереди фиксируется, и старые значения вытесняются без явных проверок и срезов.
