Оптимизатор Производительности

Анализ производительности process_orders

Допущения

1. Процент прошедших фильтр. Пузырьковая сортировка на всех 50 000 заказах шла бы не 12 секунд, а около 1,5–2 минут: ~1,25·10⁹ сравнений по ~80 нс. Наблюдаемые ~12 с соответствуют примерно 15–20 тыс. заказов, прошедших фильтры (completed и не в блэклисте). Это косвенно подтверждает, что главная горячая точка — сортировка.
2. Блэклист. BLACKLISTED_CUSTOMERS может перезагружаться из БД или переприсваиваться в рантайме. Поэтому множество строится на каждом вызове (~800 элементов, ~20 мкс), а не кешируется навсегда.
3. Типы полей. customer_id и category хешируемые (строки). amount — int или float без NaN.
4. Тесты и processed_at. Тесты не сравнивают processed_at побайтно, потому что значение зависит от времени. Тем не менее семантика «метка времени на каждый заказ» сохранена.

Найденные проблемы

Отсортированы по влиянию, от большего к меньшему.

1. Пузырьковая сортировка ради топ-10 — критично

• Что не так: полная сортировка O(M²) по M обработанным заказам, хотя нужны только 10 максимальных элементов.
• Влияние: при M ≈ 17 000 это ~1,5·10⁸ итераций интерпретатора с двумя обращениями к словарю в каждой. Это ~90–95 % всего времени.
• Причина: алгоритм с квадратичной сложностью, к тому же на чистом Python.
• Решение: heapq.nlargest(10, ..., key=...) — O(M·log 10) ≈ O(M), внутри C-код. Важно для тестов: nlargest документированно эквивалентен sorted(..., reverse=True)[:n], то есть стабилен. Исходная пузырьковая сортировка тоже стабильна: при равенстве сумм обмена нет. Порядок заказов с одинаковой суммой не изменится.

2. Линейный поиск в блэклисте — высокое влияние

• Что не так: x in list из 800 элементов — O(B) на каждый заказ.
• Влияние: до 50 000 × 800 = 4·10⁷ сравнений строк, то есть ~0,3–1 с. Для «чистых» клиентов, которых большинство, всегда проходится весь список.
• Решение: set — O(1) в среднем. Построение O(B) один раз на батч.

3. processed_orders хранит 50 000 словарей, которые не возвращаются — память и аллокации

• Что не так: полный список нужен только для len() и топ-10. Но для каждого заказа создаётся словарь из 5 ключей (~230 Б), ISO-строка (~75 Б) и float. Итого ~16 МБ и 100 000+ аллокаций на батч.
• Решение: хранить лёгкий кортеж (final_amount, order, processed_at). Ссылка на исходный order не копирует данные. Словари собираются только для 10 победителей.

4. datetime.utcnow().isoformat() на каждый заказ — среднее влияние

• Что не так: форматирование в строку — ~0,5–1 мкс на заказ, хотя результат нужен только для 10 записей.
• Решение: на заказ берём только datetime.utcnow() (дешёво, ~0,2 мкс, семантика сохраняется). isoformat() вызывается только для топ-10.
• Отдельно: utcnow() объявлен deprecated в Python 3.12+. На 3.11 предупреждений нет, но при апгрейде появятся DeprecationWarning.

5. Конкатенация строк += в цикле — низкое или среднее влияние

• Что не так: CPython оптимизирует s += t in-place, только когда на строку нет других ссылок. Это деталь реализации, не гарантия. При срабатывании realloc с копированием поведение деградирует до O(M·L) и создаёт фрагментацию памяти.
• Решение: список частей и "".join(...) — гарантированно O(M), одна итоговая аллокация.

6. Полный перебор DISCOUNT_RULES на каждый заказ — низкое влияние

• Что не так: O(R) на заказ, из них 4 из 5 правил заведомо не той категории. При росте числа правил влияние растёт линейно.
• Решение: индекс category → [(min_amount, discount), ...], строится один раз на батч. Логика выбора максимальной скидки сохранена один в один.

7. Повторные обращения к словарю

• Что не так: order["category"] читается до 4 раз, order["amount"] — дважды.
• Решение: локальные переменные. Это экономит единицы процентов, но делает код чище.

8. Архитектурные моменты вне функции

Их важно проверить до того, как считать задачу решённой.

• Где на самом деле 900 МБ. Сама функция аллоцирует порядка 20–40 МБ. Входной list[dict] на 50 000 заказов после JSON-парсинга занимает ~50–80 МБ. Пиковые 900 МБ почти наверняка набираются в другом месте:

  • сырое тело запроса;
  • Pydantic-модели плюс model_dump(), то есть двойное хранение;
  • jsonable_encoder при сериализации ответа (он рекурсивно копирует структуру).

  Кроме того, glibc malloc в многопоточном процессе (FastAPI threadpool) держит арены и не возвращает память ОС. RSS остаётся высоким после батча. Проверьте переменную окружения MALLOC_ARENA_MAX=2 в task definition: это конфигурация, код не меняется, откат тривиален.

• Блокировка event loop. Если process_orders вызывается напрямую из async def-эндпоинта, весь инстанс не обслуживает другие запросы на время обработки. Используйте def-эндпоинт или await run_in_threadpool(process_orders, orders) из starlette.concurrency, который уже есть в зависимостях. Быстрее от этого не станет (GIL), но соседние запросы перестанут ждать.

Почему не pandas и не multiprocessing

• pandas: pd.DataFrame(list_of_dicts) на 50 000 строк стоит ~100–200 мс, то есть больше, чем вся оптимизированная функция, и в 2–3 раза увеличивает память. Отчёт всё равно требует построчного форматирования. Series.sum() использует попарное суммирование, поэтому выручка будет отличаться в последних знаках, и точные assert'ы в тестах упадут. pandas оправдан, если данные уже колоночные (из БД или parquet) или батчи вырастут до миллионов строк.
• multiprocessing на 2 vCPU: сериализация (pickle) 50 000 словарей в процесс и обратно дороже самой обработки, плюс удвоение памяти.

Оптимизированный код

import heapq
from datetime import datetime
from operator import itemgetter

BLACKLISTED_CUSTOMERS = [
    "cust_1042", "cust_2091", "cust_3355", "cust_4820", "cust_5567",
    # в реальном списке ~800 идентификаторов, подгружается из БД при старте
]

DISCOUNT_RULES = [
    {"category": "electronics", "min_amount": 1000, "discount": 0.10},
    {"category": "electronics", "min_amount": 5000, "discount": 0.15},
    {"category": "clothing", "min_amount": 500, "discount": 0.05},
    {"category": "books", "min_amount": 300, "discount": 0.07},
    {"category": "home", "min_amount": 2000, "discount": 0.08},
]

TOP_N = 10


def _build_discount_index(rules: list[dict]) -> dict[str, list[tuple[float, float]]]:
    """Группирует правила по категории: category -> [(min_amount, discount), ...].

    Строится один раз на батч за O(R), чтобы на каждом заказе
    не перебирать правила чужих категорий.
    """
    index: dict[str, list[tuple[float, float]]] = {}
    for rule in rules:
        index.setdefault(rule["category"], []).append(
            (rule["min_amount"], rule["discount"])
        )
    return index


def process_orders(orders: list[dict]) -> dict:
    """Обрабатывает батч заказов: считает выручку по категориям,
    применяет скидки и формирует текстовый отчёт с топ-10 товаров."""

    # [#2] set вместо list: проверка вхождения O(1) вместо O(800).
    # Строим на каждый вызов (~20 мкс), чтобы корректно подхватывать
    # перезагрузку BLACKLISTED_CUSTOMERS из БД.
    blacklist = set(BLACKLISTED_CUSTOMERS)

    # [#6] Индекс скидок по категориям.
    discount_index = _build_discount_index(DISCOUNT_RULES)

    revenue_by_category: dict = {}

    # [#5] Копим части отчёта в списке, склеиваем один раз в конце.
    report_parts: list[str] = []

    # [#3] Вместо 50k словарей храним лёгкие кортежи со ссылкой на исходный
    # заказ. Полные записи строим только для топ-10.
    candidates: list[tuple[float, dict, datetime]] = []

    for order in orders:
        # Порядок проверок сохранён как в оригинале (важно для KeyError
        # на «битых» заказах без поля status).
        if order["customer_id"] in blacklist:
            continue
        if order["status"] != "completed":
            continue

        # [#7] Читаем поля один раз.
        category = order["category"]
        amount = order["amount"]

        # Логика выбора скидки идентична исходной: максимум среди подходящих
        # правил, но не меньше 0.0.
        applicable_discount = 0.0
        for min_amount, discount in discount_index.get(category, ()):
            if amount >= min_amount and discount > applicable_discount:
                applicable_discount = discount

        final_amount = amount * (1 - applicable_discount)

        # Тот же порядок суммирования, что и в оригинале, поэтому float-результаты
        # совпадают бит в бит. Ключи идут в порядке первого появления.
        revenue_by_category[category] = revenue_by_category.get(category, 0) + final_amount

        # [#4] Время фиксируем на каждый заказ (семантика сохранена),
        # но дорогой isoformat() делаем только для топ-10.
        candidates.append((final_amount, order, datetime.utcnow()))

        report_parts.append(f"Order {order['order_id']}: {final_amount:.2f} ({category})\n")

    # [#1] O(M·log 10) вместо O(M²). nlargest стабилен (эквивалентен
    # sorted(reverse=True)[:n]), поэтому порядок при равных суммах
    # такой же, как у исходной пузырьковой сортировки.
    top = heapq.nlargest(TOP_N, candidates, key=itemgetter(0))

    top_10_orders = [
        {
            "order_id": order["order_id"],
            "customer_id": order["customer_id"],
            "category": order["category"],
            "final_amount": final_amount,
            "processed_at": processed_at.isoformat(),
        }
        for final_amount, order, processed_at in top
    ]

    return {
        "revenue_by_category": revenue_by_category,
        "processed_count": len(candidates),
        "top_10_orders": top_10_orders,
        "report": "".join(report_parts),
    }

Опциональный шаг (требует согласования, потому что формально меняет поведение). Если бизнесу достаточно одной метки времени на батч, вынесите batch_ts = datetime.utcnow().isoformat() до цикла и уберите datetime из кортежа. Это даст ещё ~10–15 мс и ~2,5 МБ экономии. Все 10 записей получат одинаковый processed_at.

Ожидаемый эффект и проверка

До и после

Обозначения: N — все заказы (50 000), M — прошедшие фильтр, B — размер блэклиста (800), R — число правил, r_c — правил на категорию (≤ 2).

Цифры «после» — оценки по типичной стоимости операций CPython 3.11, а не замер. Их нужно подтвердить скриптом ниже на реалистичных данных.

Сценарий замера

Используется только стандартная библиотека. Сохраните старую версию как orders_old.py, новую — как orders_new.py.

# bench_process_orders.py
import cProfile
import pstats
import random
import time
import tracemalloc

import orders_new
import orders_old

CATEGORIES = ["electronics", "clothing", "books", "home", "toys"]


def make_orders(n: int, seed: int = 42) -> list[dict]:
    rnd = random.Random(seed)
    blacklist = orders_new.BLACKLISTED_CUSTOMERS
    orders = []
    for i in range(n):
        orders.append({
            "order_id": f"ord_{i}",
            "customer_id": rnd.choice(blacklist) if rnd.random() < 0.02
                           else f"cust_{rnd.randint(10_000, 99_999)}",
            "status": rnd.choices(["completed", "pending", "cancelled"], [35, 45, 20])[0],
            "category": rnd.choice(CATEGORIES),
            # 10 % заказов с суммой ровно 1000: проверка стабильности при равенстве
            # и граничного условия min_amount
            "amount": 1000 if rnd.random() < 0.1 else round(rnd.uniform(10, 8000), 2),
        })
    return orders


def strip_ts(result: dict) -> dict:
    return {
        **result,
        "top_10_orders": [
            {k: v for k, v in o.items() if k != "processed_at"}
            for o in result["top_10_orders"]
        ],
    }


def check_equivalence() -> None:
    # Для старой версии небольшой батч: пузырьковая сортировка медленная
    for seed in range(5):
        orders = make_orders(3_000, seed)
        old, new = orders_old.process_orders(orders), orders_new.process_orders(orders)
        assert strip_ts(old) == strip_ts(new), f"расхождение при seed={seed}"
    print("✔ Результаты идентичны (кроме processed_at)")


def bench_time(func, orders, repeat: int = 5) -> float:
    times = []
    for _ in range(repeat):
        t0 = time.perf_counter()
        func(orders)
        times.append(time.perf_counter() - t0)
    return min(times)


def bench_mem(func, orders) -> float:
    # Отдельно от замера времени: tracemalloc замедляет выполнение
    tracemalloc.start()
    func(orders)
    _, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return peak / 2**20


if __name__ == "__main__":
    check_equivalence()

    orders = make_orders(50_000)
    print(f"new: {bench_time(orders_new.process_orders, orders):.3f} s, "
          f"peak {bench_mem(orders_new.process_orders, orders):.1f} MiB")
    # Старая версия на 50k может работать десятки секунд, запускайте осознанно:
    print(f"old: {bench_time(orders_old.process_orders, orders, repeat=1):.3f} s, "
          f"peak {bench_mem(orders_old.process_orders, orders):.1f} MiB")

    # Где теперь тратится время
    cProfile.run("orders_new.process_orders(orders)", "new.prof")
    pstats.Stats("new.prof").sort_stats("cumulative").print_stats(10)

Как запускать

1. Локально в образе с лимитами Fargate:

   docker run --rm --cpus=2 --memory=4g -v "$PWD":/app -w /app python:3.11-slim python bench_process_orders.py
   

2. Unit-тесты. Прогоните существующий набор: сигнатура и формат ответа не менялись. Стоит добавить тест на стабильность при равных суммах и на границу amount == min_amount.
3. Поиск реального источника 900 МБ. Включите tracemalloc в эндпоинте на staging и снимите take_snapshot().statistics("lineno") до и после парсинга запроса, обработки и сериализации ответа. Параллельно сравните RSS из CloudWatch Container Insights. Если tracemalloc показывает мало, а RSS высокий, дело в фрагментации аллокатора, и стоит попробовать MALLOC_ARENA_MAX=2.
4. Выкатка. Блю-грин без миграций. Откат — переключение на предыдущий task definition. В метриках следите за p95 латентности эндпоинта и MemoryUtilization.

Уточняющие вопросы

1. Какая доля заказов обычно имеет статус completed и не попадает в блэклист? Это нужно для точной оценки «до».
2. BLACKLISTED_CUSTOMERS загружается один раз при старте или периодически обновляется? Если один раз, множество можно построить на уровне модуля, но выигрыш составит всего ~20 мкс на вызов.
3. Проверяют ли тесты processed_at (формат или различие между записями)? От этого зависит, можно ли перейти на одну метку на батч.
4. Из какого эндпоинта вызывается функция: async def или def? Используется ли response_model или Pydantic-модель для входа?
5. Как измерялись 900 МБ: RSS контейнера или tracemalloc? Включает ли цифра парсинг запроса и сериализацию ответа?