gendesign/backend/app/services/forecast_request_cache.py

"""Request-scoped memoization для §22-форсайта (#1129) — БЕЗ смены сигнатур сервисов.

ПРОБЛЕМА (профиль #1129): один холодный §22-отчёт (`build_site_finder_report`,
orchestrator.py) собирает 8 §9.x-слоёв, и эти слои МАССОВО пере-вызывают ОДНИ И ТЕ ЖЕ
горизонт-инвариантные БД-загрузки с ИДЕНТИЧНЫМИ аргументами. Пример амплификации на
один отчёт (сетка 6/12/18/24, default what_to_build-сетка 3×5):
  • `compute_scenarios` гоняет `compute_demand_supply_forecast` ТРИЖДЫ (conservative/
    base/aggressive) × 4 горизонта;
  • `special_indices`/`recommendation` дважды зовут `rank_segments` → 15 ячеек ×
    `compute_demand_supply_forecast`;
  • КАЖДАЯ `compute_demand_supply_forecast`-ячейка внутри зовёт §9.4
    `compute_demand_normalization` (→ §9.6 `compute_rate_regime_sensitivity` →
    `get_monthly_macro` + 2× `build_sales_series`) + §9.3
    `compute_future_supply_pressure` (→ `compute_market_metrics`) + §9.7
    `get_competitors`.
Итог — сотни повторных roundtrip'ов к БД, почти все с СОВПАДАЮЩИМИ аргументами
(один и тот же `district`, один и тот же `spec`, один и тот же `cad_num`,
горизонт-инвариантный макро-ряд). Это case (c) аудита #1129: ОБЩАЯ дорогая загрузка
пере-выполняется N раз — мемоизировать её ОДИН раз на отчёт убирает дублирование БЕЗ
конкурентности и БЕЗ изменения результата.

РЕШЕНИЕ: per-request (per-отчёт) кэш на `ContextVar`. Контекст-менеджер
`forecast_cache()` активирует свежий dict на время сборки ОДНОГО отчёта; декоратор
`cached(...)` оборачивает дорогие §9.x-сервисы и:
  • ВНУТРИ активного контекста → результат считается один раз на уникальный ключ
    аргументов и переиспользуется (тот же объект — все наши результаты frozen-dataclass
    либо read-only-список frozen-dataclass'ов, см. ниже: callers их НЕ мутируют, поэтому
    шаринг инстанса детерминирован и безопасен);
  • ВНЕ контекста (обычные standalone-вызовы: API `/competitors`, юнит-тесты, прочие
    эндпоинты) → функция выполняется КАК ОБЫЧНО, без кэша. Нулевое изменение поведения
    за пределами сборки §22-отчёта.

ПОЧЕМУ ContextVar, а не аргумент-кэш: не надо протаскивать cache-параметр через ~10
гетерогенных сигнатур §9.x. ContextVar автоматически скоупится сборкой отчёта и
изолирован per-thread/per-task (важно, если позже сборка уйдёт в ThreadPool — у
каждого потока свой контекст; здесь же §22 идёт в одном Celery-`def`-воркере).

КОРРЕКТНОСТЬ (жёсткий гейт #1129): кэш возвращает ТОТ ЖЕ инстанс для совпадающих
ключей. Все мемоизируемые продьюсеры возвращают НЕИЗМЕНЯЕМЫЕ значения —
`@dataclass(frozen=True)` (MarketMetrics / FutureSupplyPressure / DemandSupplyForecast /
DemandNormalization / RateSensitivity / MacroCoefficient / SalesSeries / MonthlyMacro) и
Pydantic-модель CompetitorsResponse — а вызывающий код их только ЧИТАЕТ (атрибуты,
list-comprehension, `for`), нигде не мутирует список/поля на месте. Поэтому шаринг
кэш-инстанса даёт ПОБАЙТОВО тот же собранный отчёт, что и без кэша.

Ключи строятся ЯВНЫМ key-builder'ом на каждую функцию (не «магический» по сигнатуре):
нормализуем только те аргументы, что РЕАЛЬНО влияют на результат, в hashable-tuple.
`db`-сессия в ключ НЕ входит (на время отчёта она одна; данные за сборку не меняются).
"""

from __future__ import annotations

import logging
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from functools import wraps
from typing import Any, TypeVar, cast

logger = logging.getLogger(__name__)

# Активный per-отчёт кэш (или None вне сборки отчёта). Хранит {ключ: результат}.
# default=None ⇔ «контекст не активен» → cached(...) выполняет функцию без кэша.
_CACHE: ContextVar[dict[Any, Any] | None] = ContextVar("forecast_request_cache", default=None)

F = TypeVar("F", bound=Callable[..., Any])

# Сентинел для «значение отсутствует» — отличаем закэшированный None от промаха.
_MISS = object()


@contextmanager
def forecast_cache() -> Iterator[dict[Any, Any]]:
    """Активировать свежий per-отчёт кэш на время блока (один §22-отчёт).

    Внутри блока все `cached(...)`-сервисы мемоизируются в общий dict; на выходе он
    отбрасывается (кэш живёт ровно одну сборку отчёта — никакой кросс-отчётной утечки
    устаревших данных). Реентерабельность: вложенный `forecast_cache()` НЕ заводит
    новый кэш — переиспользует уже активный (idempotent), чтобы мемоизация была общей
    на всю сборку, даже если вход обёрнут дважды.
    """
    existing = _CACHE.get()
    if existing is not None:
        # Уже внутри активного контекста — переиспользуем (не сбрасываем накопленное).
        yield existing
        return
    cache: dict[Any, Any] = {}
    token = _CACHE.set(cache)
    try:
        yield cache
    finally:
        _CACHE.reset(token)
        logger.debug("forecast_cache: closed (entries=%d)", len(cache))


def cached(key: Callable[..., Any], *, label: str) -> Callable[[F], F]:
    """Декоратор: мемоизировать результат функции per-отчёт по явному ключу аргументов.

    Args:
        key: builder ключа — принимает те же *args/**kwargs, что обёрнутая функция
            (включая `db` первым позиционным — его игнорируем), и возвращает hashable
            идентификатор того, что РЕАЛЬНО влияет на результат. Должен быть
            детерминирован и полностью покрывать варьирующиеся входы.
        label: короткое имя для namespacing ключа и для debug-лога (имя функции).

    Returns:
        Декоратор, оборачивающий sync-функцию `(db, ...) -> Result`.

    Вне активного `forecast_cache()` обёртка просто зовёт оригинал (без кэша) — обычные
    standalone-вызовы (API/тесты) не затрагиваются.
    """

    def decorate(fn: F) -> F:
        @wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            cache = _CACHE.get()
            if cache is None:
                # Нет активного контекста → поведение по умолчанию, без мемоизации.
                return fn(*args, **kwargs)
            cache_key = (label, key(*args, **kwargs))
            hit = cache.get(cache_key, _MISS)
            if hit is not _MISS:
                return hit
            result = fn(*args, **kwargs)
            cache[cache_key] = result
            return result

        return cast(F, wrapper)

    return decorate


def is_active() -> bool:
    """Активен ли сейчас per-отчёт кэш (для диагностики/тестов). PURE-ish."""
    return _CACHE.get() is not None