Проблема: list + bisect плохо растёт

В прошлом уроке мы видели: bisect находит элемент в отсортированном list за O(log n). Прекрасно. Но что если нужно вставить элемент, сохранив порядок?

import bisect

xs = [1, 3, 5, 7, 9, 11]

# найти позицию — O(log n), быстро
pos = bisect.bisect_left(xs, 6)

# но вставить — O(n)!
xs.insert(pos, 6)

list.insert(i, x) — это O(n), потому что под капотом нужно сдвинуть все элементы от позиции i до конца на одну ячейку вправо. Из-за этого list + bisect плохо подходит для задач, где данные постоянно меняются.

Если массив используется только для чтения (поиска) — list + bisect отлично. Если же мы хотим вставлять, удалять и поддерживать порядок — нужна другая структура.

SortedList — O(log n) для всех операций

Библиотека sortedcontainers — стандарт де-факто в Python для отсортированных коллекций. Установка: pip install sortedcontainers. Она написана на чистом Python (!) и обгоняет многие реализации на C, потому что использует умную внутреннюю структуру.

from sortedcontainers import SortedList

sl = SortedList()

# вставка — O(log n)
sl.add(5)
sl.add(2)
sl.add(8)
sl.add(1)
# теперь sl == [1, 2, 5, 8]

# поиск — O(log n)
print(2 in sl)             # True
print(sl.index(5))         # 2 — позиция элемента

# удаление — O(log n)
sl.remove(2)
# теперь sl == [1, 5, 8]

# random access по индексу — O(log n)
print(sl[0])               # 1
print(sl[-1])              # 8 (последний)

# range query — O(log n + k), где k — число элементов в диапазоне
left = sl.bisect_left(3)
right = sl.bisect_right(7)
print(sl[left:right])      # [5]

Все операции — add, remove, __contains__, __getitem__, bisect_left/right — работают за O(log n). Это не «log n с большой константой» — это реально быстро.

Что внутри SortedList

Если открыть исходники sortedcontainers, можно удивиться: там нет красно-чёрного дерева, нет AVL, нет skip list. Это список списков.

Структура хранит элементы в виде нескольких отсортированных подсписков. По умолчанию длина каждого подсписка около 1000. Когда подсписок становится слишком большим, он расщепляется на два. Когда слишком маленьким — сливается с соседним.

Когда мы ищем элемент, делаем bisect сначала на массиве максимумов бакетов (быстро, бакетов мало), потом bisect внутри найденного бакета. Сложность — O(log n).

Когда вставляем — то же самое: находим бакет, делаем bisect внутри, вставляем (это O(k), где k — длина бакета, около 1000). Поскольку k константа, операция считается O(log n).

Почему это побеждает self-balancing BST (AVL, red-black)? Потому что в BST каждый узел — отдельный объект Python, с указателями и аллокациями. Pointer chasing убивает cache locality. SortedList же хранит элементы массивами — отлично работает с кэшем CPU.

Сравнение с self-balancing BST

В академических курсах обычно учат AVL-дерево или красно-чёрное дерево — структуры, гарантирующие O(log n) для всех операций через балансировку.

На практике SortedList на 5-10x быстрее питоновской реализации AVL. Поэтому в Python для отсортированной коллекции это де-факто стандарт.

Применение в Data Engineering

Где SortedList реально нужен:

Online median / percentile. Поток событий, нужно поддерживать медиану time-to-process. Простое решение — две SortedList (max-half и min-half) или одна с индексным доступом к середине.

Range queries в скользящем окне. «Сколько событий с латентностью между X и Y в последние 10 минут?». bisect_left(window, X) и bisect_right(window, Y) — разность даёт ответ за O(log n).

Top-K с динамическим обновлением. Когда в потоке нужно поддерживать «топ-100 пользователей по активности» и активность постоянно меняется. SortedList + словарь дают вставку, обновление и срез верхушки за O(log n).

Joining sorted streams. Слияние нескольких отсортированных потоков по времени — merge K sorted lists через heap (об этом будет в модуле 18), но иногда удобнее использовать SortedList в качестве буфера.

from sortedcontainers import SortedList
import time

# имитация: поток событий с латентностью
latencies = SortedList()
window_size = 10000

events = [(time.time(), 42 + i * 0.1) for i in range(100_000)]

for ts, latency in events:
    latencies.add(latency)
    if len(latencies) > window_size:
        # удалить самый старый (тут упрощенно: удалить наименьший — для демо)
        latencies.pop(0)

    # быстрая медиана: индекс len//2
    if len(latencies) % 100 == 0:
        median = latencies[len(latencies) // 2]
        p99 = latencies[int(len(latencies) * 0.99)]

Если бы у нас был обычный list, каждое add и pop(0) было бы O(n). Со SortedList — O(log n). Разница на 10 000 элементах: примерно 1000x ускорение.

SortedDict и SortedSet

В библиотеке sortedcontainers есть ещё две полезные структуры:

from sortedcontainers import SortedDict, SortedSet

# словарь, упорядоченный по ключу
sd = SortedDict()
sd[10] = "ten"
sd[5] = "five"
sd[20] = "twenty"

# итерация по упорядоченным ключам
for k in sd:
    print(k, sd[k])
# 5 five
# 10 ten
# 20 twenty

# range query
print(list(sd.irange(5, 15)))   # [5, 10]

# отсортированное множество
ss = SortedSet([5, 10, 5, 20, 10])
print(ss)   # SortedSet([5, 10, 20])
print(ss.bisect_left(10))   # 1

SortedDict особенно полезен для time-series: ключ — timestamp, значение — данные. Можно быстро получать срезы по времени.

Замеры

Сравним вставку 100 000 случайных элементов:

import random
import timeit
import bisect
from sortedcontainers import SortedList

n = 100_000
data = [random.randint(0, n * 10) for _ in range(n)]

# вариант 1: list + bisect.insort
def insert_list():
    xs = []
    for x in data:
        bisect.insort(xs, x)
    return xs

# вариант 2: SortedList
def insert_sortedlist():
    sl = SortedList()
    for x in data:
        sl.add(x)
    return sl

t1 = timeit.timeit(insert_list, number=1)
t2 = timeit.timeit(insert_sortedlist, number=1)

print(f"list + insort: {t1:.2f} s")
print(f"SortedList:    {t2:.2f} s")
print(f"speedup:       {t1 / t2:.1f}x")

Типичный результат: list + insort около 4-5 секунд (O(n²) из-за сдвигов), SortedList около 0.1 секунды (O(n log n)). Разница в 50 раз на 100 000 элементов, и она растёт с размером данных.

Попробуй сам

Реализуйте online median (медиана в потоке) с помощью SortedList:

from sortedcontainers import SortedList

class OnlineMedian:
    def __init__(self):
        self.values = SortedList()

    def add(self, x):
        self.values.add(x)

    def median(self):
        n = len(self.values)
        if n == 0:
            return None
        if n % 2 == 1:
            return self.values[n // 2]
        return (self.values[n // 2 - 1] + self.values[n // 2]) / 2

om = OnlineMedian()
for x in [3, 1, 4, 1, 5, 9, 2, 6, 5, 3]:
    om.add(x)
    print(f"after {x}: median = {om.median()}")

Каждое добавление O(log n), каждый запрос медианы O(log n) (благодаря индексному доступу). На потоке миллиона событий это работает за миллисекунды.