Python deque для реального времени: как использовать скользящие окна с помощью collections.deque

Первый звоночек: ваш list убивает latency

Представьте: вы пишете систему мониторинга для AI-пайплайна. Каждую секунду прилетает латентность инференса. Надо держать последние 1000 значений, считать скользящее среднее, выбрасывать старые. Берёте list, делаете my_list.append(new_val) и my_list.pop(0). Работает. На тестовых 10 точках. Потом в проде выясняется, что pop(0) — это O(n). С каждой новой записью Python перекладывает весь массив на один элемент влево. При 10k элементов — тормоза. При 100k — вы теряете real-time.

В этой статье я покажу, почему collections.deque — ваш единственный друг для FIFO-окон в реалтайме. И главное — как не наступить на грабли, которые ждут даже опытных инженеров.

Почему deque, а не просто кольцевой буфер руками?

Теоретически, можно написать свой кольцевой буфер на массиве фиксированной длины с двумя указателями. Но зачем, когда в стандартной библиотеке есть готовая, отлаженная, C-оптимизированная структура? deque (double-ended queue) — это двусторонняя очередь, реализованная как двусвязный список блоков фиксированного размера. Добавление и удаление с обоих концов — O(1) амортизированно. Никакого memmove, никаких сдвигов.

Более того, у deque есть параметр maxlen, который автоматически вытесняет старые элементы при добавлении новых. Это буквально скользящее окно «из коробки». Никаких if-ов и pop-ов вручную.

from collections import deque

# Окно на 1000 последних значений
window = deque(maxlen=1000)

for new_value in stream:
    window.append(new_value)
    # Старые значения удаляются автоматически
    mean = sum(window) / len(window)

Звучит идеально? Почти. Но есть тонкий нюанс: sum(window) на каждом шаге — это O(n). Для окна в 1000 элементов — норм, для 100 000 — уже проблема. Мы к этому вернёмся в разделе про оптимизацию.

Как НЕ надо делать: классическая ошибка с concurrent.futures

Допустим, вы пишете распределённый сборщик метрик. Потоки пушат данные в общий deque. В документации написано, что deque потокобезопасен для отдельных операций append и popleft. Но! Если вы делаете что-то вроде:

# Поток 1
if len(q) > 0:
    val = q.popleft()  # race condition!

Между проверкой длины и pop другой поток мог уже всё съесть. Получаете IndexError или, ещё хуже, потерянные данные. Эту граблю я наступил в 2022, когда писал очередь для логов AI-агентов. С тех пор — только блокировки или однопоточные очереди.

Если вам нужна потокобезопасная очередь с блокировками — используйте queue.Queue. Но если вы строите скользящее окно в одном потоке (а это 90% сценариев мониторинга), deque — идеал.

Практический кейс: скользящая метрика для AI-пайплайна

Представьте: у вас есть сервис инференса на базе архитектуры Spaceduck (5 функций, реально работающих). Вы хотите мониторить latency последних 500 запросов, чтобы при превышении порога — включить дополнительную реплику. Пишем класс скользящего окна с поддержкой быстрой агрегации.

from collections import deque

class SlidingWindow:
    def __init__(self, size: int):
        self._deque = deque(maxlen=size)
        self._sum = 0.0
    
    def add(self, value: float):
        if len(self._deque) == self._deque.maxlen:
            # Вытесняем старый элемент из суммы
            self._sum -= self._deque[0]
        self._sum += value
        self._deque.append(value)
    
    @property
    def mean(self) -> float:
        if not self._deque:
            return 0.0
        return self._sum / len(self._deque)
    
    @property
    def max(self) -> float:
        return max(self._deque) if self._deque else 0.0
    
    @property
    def last_n(self, n: int) -> list:
        return list(self._deque)[-n:]

Теперь вы можете использовать это в Gradio-дашборде, который обновляется каждую секунду. Без тормозов, без лишних аллокаций.

Когда deque не подходит: честно о границах

• Случайный доступ по индексу — deque поддерживает window[500], но это O(n) в худшем случае (если индекс далеко от краёв). Если вам нужно часто обращаться к произвольным элементам — используйте list или numpy.array.
• Сортировка — deque не сортируемая in-place структура. Для поддержания упорядоченности по какому-то ключу — смотрите в сторону bisect на list или sortedcontainers.
• Огромные окна (миллионы элементов) — deque хранит указатели на блоки, каждый блок ~64 элемента. При окне в 10 млн элементов память и накладные расходы на обход могут быть высоки. Здесь лучше подойдёт кольцевой буфер на numpy с фиксированным размером и view.

Но для 95% real-time задач (метрики, логи, скользящие статистики) deque — оптимальный выбор. Особенно если вы мигрируете с list и видите прирост производительности в десятки раз.

Потоковая обработка и распределённые пайплайны

Если ваш пайплайн обрастает распределённостью — один deque на сервер уже не спасает. Тут в игру вступают очереди вроде Ray (распределённые объекты и акторы) или LLMeQueue для балансировки запросов к LLM. Но внутри каждого узла, где нужно скользящее окно с минимальным оверхедом — deque остаётся королём.

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

Типичные ошибки (проверь свой код)

FAQ (или то, о чём молчат документации)

Теперь, когда вы знаете все подводные камни, идите и перепишите свои скользящие окна. Ваш CPU скажет спасибо. А если понадобится распределить нагрузку — посмотрите на Q-обучение для маршрутизации — там deque тоже пригодится для буфера опыта.

Подписаться на канал