gendesign/backend/app/services/forecasting/orchestrator.py

"""§9.x → §22 оркестратор: посчитать форсайт-слои и собрать `SiteFinderReport`.

Step 3a проводки форсайта (955-A2). §22-сборщик `assemble_report` (#988) ЧИСТЫЙ — он
лишь РАСКЛАДЫВАЕТ уже-посчитанные advisory-входы по восьми секциям, в БД не ходит и
§9.x-сервисы не зовёт. ВЫЗЫВАЮЩИЙ (этот модуль) обязан посчитать §9.x-слои и скормить
их сборщику. Здесь — этот вызывающий: ОРКЕСТРАЦИОННЫЙ ШОВ, НЕ новая §9.x-математика.

Что делает `build_site_finder_report`:
  1. Дефолт сегмента (audit Q1): если `obj_class` не задан — выводим МОДАЛЬНЫЙ (самый
     частый) класс конкурентов из `analyze["competitors"]`. Нет конкурентов / нет
     классов → obj_class=None (ЕКБ-wide all-class).
  2. Считает §9.x-слои, КАЖДЫЙ через `_safe_call` (зеркало product_scoring._safe_call):
     сбой/тонкие данные одного слоя → этот слой None, отчёт всё равно собирается
     (graceful: тонкий/сбойный слой ДЕГРАДИРУЕТ отчёт, НЕ роняет его целиком). Сами
     §9.x-модули уже graceful внутри — это belt-and-suspenders на оркестрационном шве.
  3. Раскладывает слои в `assemble_report` (каждый в свой слот, с ПРАВИЛЬНОЙ —
     ГЕТЕРОГЕННОЙ! — сигнатурой: forecasts/scenarios/special_indices берут `horizons=`,
     score_card/future_supply берут `horizon_months=`, overlay берёт `target_class=`).

ADVISORY: весь форсайт-стек советующий (cap ≤ 'medium', #990) — отчёт лишь СОБИРАЕТ
advisory-выводы, не основание для инвест-решения. `advisory=True` прокидывается всюду.

ГЕТЕРОГЕННЫЕ сигнатуры §9.x (audit поймал ловушки — НЕ унифицированы):
  • compute_market_metrics(db, *, district, premise_kind)           — §9.2
  • compute_all_layers(db, *, district, premise_kind)               — §9.3 (список строк)
  • compute_future_supply_pressure(db, *, district, horizon_months) — §9.3 давление
  • compute_demand_supply_forecast(db, *, spec, district, cad_num, horizons)   — §9.8
  • compute_scenarios(db, *, spec, district, cad_num, horizons)      — §11
  • compute_score_card(db, *, spec, district, cad_num, horizon_months)  — §14.2
  • compute_special_indices(db, *, spec, district, cad_num, horizons)   — §25
  • build_forecast_overlay(db, *, district, cad_num, horizon_months, target_class) — §10

МАКРО-СВЕЖЕСТЬ (audit MEDIUM, отложено в 3b): domrf-месячный макро ~123 дня устаревает
(key_rate свежий). У §22-схемы v1.0 (frozen `SiteFinderReport`/`ReportMeta`) НЕТ чистого
слота под caveat без смены контракта (новое поле + bump schema_version) — что вне 3a.
Поэтому здесь только ЛОГИРУЕМ `macro_as_of` (последний наблюдаемый месяц макро), чтобы
сигнал не терялся; проводка его в тело отчёта — follow-up 3b (вместе с endpoint/Celery).
"""

from __future__ import annotations

import logging
from collections import Counter
from collections.abc import Sequence
from typing import Any

from sqlalchemy.orm import Session

from app.services.forecast_request_cache import forecast_cache
from app.services.forecasting.demand_supply_forecast import compute_demand_supply_forecast
from app.services.forecasting.macro_series import get_monthly_macro
from app.services.forecasting.product_scoring import compute_score_card
from app.services.forecasting.recommendation import build_forecast_overlay
from app.services.forecasting.report import SiteFinderReport
from app.services.forecasting.report_assembler import assemble_report
from app.services.forecasting.sales_series import SegmentSpec
from app.services.forecasting.scenarios import compute_scenarios
from app.services.forecasting.special_indices import compute_special_indices
from app.services.site_finder.future_supply import compute_future_supply_pressure
from app.services.site_finder.market_metrics import compute_market_metrics
from app.services.site_finder.supply_layers import SupplyLayerRow, compute_all_layers

logger = logging.getLogger(__name__)

# Горизонты прогноза по умолчанию (мес) — зеркало §9.x default grid (#952/#984/#986).
_DEFAULT_HORIZONS: tuple[int, ...] = (6, 12, 18, 24)

# Основной горизонт для single-горизонтных сервисов (score_card / overlay / future_supply).
# Берём 12 (типовой средне-срочный продуктовый горизонт) если он в сетке, иначе первый.
_PREFERRED_PRIMARY_HORIZON: int = 12

# Тип помещения — единый дефолт всего §9.x-стека (жилой сегмент ЕКБ).
_PREMISE_KIND: str = "квартира"

# Весь отчёт советующий (зеркало advisory-cap стека) — прокидывается в #990 cap.
_ADVISORY: bool = True

# Маппинг layer-номера → ключ агрегата supply_layers-сводки (#950, §9.3).
_LAYER_UNIT_KEY: dict[int, str] = {1: "open_units", 2: "hidden_units", 3: "future_units"}


def _modal_obj_class(competitors: Any) -> str | None:
    """Вывести МОДАЛЬНЫЙ (самый частый) класс конкурентов как дефолт сегмента. PURE.

    audit Q1: user-дефолт сегмента = модальный класс конкуренции вокруг участка. Читаем
    `obj_class` (или legacy-ключ `class`) каждого dict'а-конкурента, считаем Counter,
    берём самый частый. Ничьи разрешаются детерминированно (по первому встреченному —
    стабильный порядок Counter.most_common для равных частот).

    Graceful: competitors не список / пуст / ни у одного нет класса → None
    (ЕКБ-wide all-class сегмент — честно «класс не уточнён»).

    Args:
        competitors: `analyze["competitors"]` (loosely-typed: ожидаем список dict'ов).

    Returns:
        Модальный класс (str) или None.
    """
    if not isinstance(competitors, list):
        return None
    classes: list[str] = []
    for comp in competitors:
        if not isinstance(comp, dict):
            continue
        value = comp.get("obj_class")
        if not (isinstance(value, str) and value):
            value = comp.get("class")
        if isinstance(value, str) and value:
            classes.append(value)
    if not classes:
        return None
    # most_common(1) — детерминированно: при равенстве частот сохраняется порядок вставки.
    return Counter(classes).most_common(1)[0][0]


def _primary_horizon(horizons: Sequence[int]) -> int:
    """Основной горизонт для single-горизонтных сервисов. PURE.

    _PREFERRED_PRIMARY_HORIZON (12 мес) если он в сетке, иначе первый горизонт сетки.
    Пустая сетка не ожидается (вызывающий даёт дефолт), но на всякий случай → 12.
    """
    if _PREFERRED_PRIMARY_HORIZON in horizons:
        return _PREFERRED_PRIMARY_HORIZON
    return horizons[0] if horizons else _PREFERRED_PRIMARY_HORIZON


def _safe_call(label: str, db: Session, fn: Any) -> Any:
    """Вызвать §9.x-сервис graceful: сбой → None + logger.exception (не crash отчёта).

    Зеркало product_scoring._safe_call: любой §9.x-слой может бросить (тонкие данные / нет
    геометрии cad_num / сбой SQL) — логируем и возвращаем None, чтобы соответствующая
    секция отчёта осталась пустой/частичной, а остальной отчёт собрался. Намеренно ловим
    широкий Exception (изоляция одного слоя от отчёта) с ОБЯЗАТЕЛЬНЫМ logger.exception —
    НЕ молчаливое глотание. §9.x уже graceful внутри; это belt-and-suspenders на шве.

    SAVEPOINT (#2464 cluster A finding #1): все §9.x-слои шарят ОДИН `db`-Session на
    отчёт (module docstring `forecast_request_cache.py`). Без обёртки сбойный
    `db.execute` внутри слоя оставляет транзакцию Postgres в состоянии aborted
    («current transaction is aborted, commands ignored until end of transaction
    block») — КАЖДЫЙ последующий слой на той же сессии тоже падает, хотя его данные
    были бы доступны. `with db.begin_nested():` заводит SAVEPOINT НА ВЕСЬ вызов слоя
    (слой может делать несколько `db.execute` внутри себя — например §9.6 внутри
    §9.8/§11); при исключении SAVEPOINT откатывается автоматически (ROLLBACK TO
    SAVEPOINT), внешняя транзакция остаётся рабочей для следующего слоя. НЕ
    `db.rollback()` — тот откатил бы ВСЮ внешнюю транзакцию (см. `backend.md` §
    SAVEPOINT pattern, established anti-pattern).

    Args:
        label: имя слоя для лога (диагностика какой §9.x-вызов деградировал).
        db: общая §22-сессия (для SAVEPOINT вокруг вызова слоя).
        fn: нулевой-аргумент thunk вокруг §9.x-вызова.

    Returns:
        Результат `fn()` или None при исключении.
    """
    try:
        with db.begin_nested():
            return fn()
    except Exception:
        logger.exception("orchestrator: §9.x layer %s failed → section degraded", label)
        return None


def _summarize_supply_layers(rows: Sequence[SupplyLayerRow] | None) -> dict[str, Any] | None:
    """Свернуть строки §9.3-слоёв (#950) в сводку-dict для слота `supply_layers`. PURE.

    `compute_all_layers` отдаёт ПЛОСКИЙ список SupplyLayerRow (L1+L2+L3). Слот
    `supply_layers` сборщика (#988 `_build_market_now`/`_domrf_coverage`) ждёт ОДИН dict,
    не список — поэтому агрегируем: Σ units по слою (open/hidden/future_units) + сырой
    список строк (`rows`) для прозрачности. domrf_coverage здесь НЕ выводим (нет дешёвого
    продьюсера в сервисах) — сборщик возьмёт fallback из `analyze.market_data_coverage_pct`.

    Graceful: None/пустой список → None (слот останется пустым — секция частична).

    Args:
        rows: вывод `compute_all_layers` (или None при сбое слоя).

    Returns:
        Сводка-dict (open_units/hidden_units/future_units/n_rows/rows) или None.
    """
    if not rows:
        return None
    units_by_key: dict[str, int] = {"open_units": 0, "hidden_units": 0, "future_units": 0}
    row_dicts: list[dict[str, Any]] = []
    for row in rows:
        row_dicts.append(row.as_dict())
        key = _LAYER_UNIT_KEY.get(int(row.layer))
        if key is not None and row.units_estimate is not None:
            units_by_key[key] += int(row.units_estimate)
    return {
        "open_units": units_by_key["open_units"],
        "hidden_units": units_by_key["hidden_units"],
        "future_units": units_by_key["future_units"],
        "n_rows": len(row_dicts),
        "rows": row_dicts,
    }


def _macro_as_of(macro: Any) -> str | None:
    """Последний наблюдаемый месяц макро-ряда (ISO) — для caveat-лога свежести. PURE.

    `get_monthly_macro` отдаёт MonthlyMacro по ВОЗРАСТАНИЮ month по непрерывной сетке;
    последний элемент — самый поздний месяц с (carry-forward) данными. Используем только
    для ЛОГА свежести (audit MEDIUM) — проводка в тело отчёта отложена в 3b.

    Graceful: пустой/None ряд → None.
    """
    if not macro:
        return None
    last = macro[-1]
    month = getattr(last, "month", None)
    isoformat = getattr(month, "isoformat", None)
    if callable(isoformat):
        return isoformat()
    return str(month) if month is not None else None


def _normalize_district(district: Any) -> str | None:
    """Привести `district` к str|None — защита от silent-degrade при dict-входе. PURE.

    `analyze["district"]` в этой кодовой базе — `dict` вида
    ``{"district_name": "Верх-Исетский", "dist_to_center": 0.0, "median_price_per_m2": ...}``.
    Штатный caller (`workers/tasks/forecast.py:123`) явно извлекает `district_name`:
    `district = row.district or analyze["district"]["district_name"]`. Но новые callers
    (тесты, расширения чата, ad-hoc эндпоинты) легко передают сырой dict без знания
    этой конвенции — тогда внутри §9.x-слоёв `compute_market_metrics(district=<dict>)`
    падает с `TypeError: unhashable type: 'dict'` в `forecast_request_cache.wrapper`,
    `_safe_call` это проглатывает → секции `future_market.forecasts_by_horizon=[]` и
    `scenarios.by_scenario={}` тихо остаются пустыми (silent degrade, не 500).

    Здесь — defensive normalization на входе оркестратора: dict → `district_name`,
    str → как есть, None → None, всё остальное → None (с warning'ом).
    """
    if district is None or isinstance(district, str):
        return district
    if isinstance(district, dict):
        name = district.get("district_name")
        if isinstance(name, str) and name:
            return name
        logger.warning(
            "build_site_finder_report: district передан как dict без district_name (%r) — None",
            sorted(district.keys()),
        )
        return None
    logger.warning(
        "build_site_finder_report: district неподдерживаемого типа %s — None",
        type(district).__name__,
    )
    return None


def build_site_finder_report(
    db: Session,
    *,
    analyze: dict[str, Any],
    cad_num: str,
    district: str | None,
    obj_class: str | None = None,
    horizons: Sequence[int] = _DEFAULT_HORIZONS,
) -> SiteFinderReport:
    """Посчитать §9.x-слои и собрать §22 `SiteFinderReport` для участка (ОРКЕСТРАЦИЯ).

    PERF (#1129): активируем per-отчёт кэш (`forecast_cache`) на ВСЮ сборку. §9.x-слои
    массово пере-вызывают одни и те же горизонт/сегмент-инвариантные БД-загрузки с
    совпадающими аргументами (профиль: get_competitors ×69, market_metrics ×124,
    get_monthly_macro ×290 на один отчёт) — кэш считает каждую уникальную загрузку ОДИН
    раз и переиспользует тот же (frozen, read-only) результат. Вывод ПОБАЙТОВО идентичен
    несжатому пути (детерминизм §16/§26 сохранён); ускорение — чисто за счёт устранения
    дублирующих roundtrip'ов, БЕЗ конкурентности. Кэш живёт ровно одну сборку (на выходе
    из `with` отбрасывается — никакой кросс-отчётной утечки).

    DEFENSIVE (#1130 follow-up): `district` нормализуется на входе — если caller передал
    сырой dict из `analyze["district"]` (а не извлечённое `district_name`), достанем имя.
    Подробности — `_normalize_district`. PURE-нормализация, без изменения остального флоу.
    """
    district = _normalize_district(district)
    with forecast_cache():
        return _build_site_finder_report_impl(
            db,
            analyze=analyze,
            cad_num=cad_num,
            district=district,
            obj_class=obj_class,
            horizons=horizons,
        )


def _build_site_finder_report_impl(
    db: Session,
    *,
    analyze: dict[str, Any],
    cad_num: str,
    district: str | None,
    obj_class: str | None = None,
    horizons: Sequence[int] = _DEFAULT_HORIZONS,
) -> SiteFinderReport:
    """Реальная сборка §22-отчёта (тело #1056). Вызывается ВНУТРИ `forecast_cache()`.

    ШОВ между §9.x-сервисами (которые ходят в БД) и ЧИСТЫМ §22-сборщиком (#988
    `assemble_report`, который только раскладывает уже-посчитанные входы). НЕ новая
    §9.x-математика — только проводка: считаем слои → раскладываем в сборщик.

    Поток:
      1. Дефолт сегмента: obj_class=None → модальный класс конкурентов из `analyze`
         (audit Q1). Строим `spec = SegmentSpec(obj_class=…, district=…)`.
      2. Считаем восемь §9.x-слоёв, КАЖДЫЙ через `_safe_call` (graceful: сбой слоя →
         None → его секция деградирует, отчёт собирается). Сигнатуры ГЕТЕРОГЕННЫ
         (см. module docstring): forecasts/scenarios/special_indices ← `horizons=`;
         score_card/future_supply ← `horizon_months=`; overlay ← `target_class=`.
      3. Раскладываем в `assemble_report` (каждый слой в свой слот; supply-строки
         сворачиваем в сводку-dict; overlay идёт в `recommendation_overlay=`).

    МАКРО-СВЕЖЕСТЬ (audit MEDIUM): логируем `macro_as_of`; проводка в отчёт — 3b
    (нет чистого слота в схеме v1.0 без смены контракта). См. module docstring.

    Args:
        db: SQLAlchemy sync Session (read-only по факту — §9.x только читают).
        analyze: dict вывода analyze_parcel (loosely-typed; читаем defensive `.get()`).
        cad_num: кадастровый номер участка (вход §9.7 конкурентов + центроид POI).
        district: район участка (None → ЕКБ-wide; §9.x резолвят админ-имя → микро).
        obj_class: целевой класс; None → выводим модальный класс конкурентов (Q1).
        horizons: горизонты прогноза (мес; по умолчанию _DEFAULT_HORIZONS).

    Returns:
        SiteFinderReport — восемь секций §22, заполнены настолько, насколько хватило
        §9.x-слоёв (частичный отчёт валиден); advisory=True; `as_dict()` JSON-safe.
    """
    horizon_list = list(horizons)
    primary = _primary_horizon(horizon_list)

    # ── 1. Дефолт сегмента: модальный класс конкурентов (audit Q1) ──────────────
    resolved_class = (
        obj_class if obj_class is not None else _modal_obj_class(analyze.get("competitors"))
    )
    spec = SegmentSpec(obj_class=resolved_class, district=district)

    logger.info(
        "build_site_finder_report: cad_num=%s district=%s obj_class=%s (given=%s) "
        "horizons=%s primary=%d",
        cad_num,
        district,
        resolved_class,
        obj_class,
        horizon_list,
        primary,
    )

    # ── 2. §9.x-слои, каждый graceful через _safe_call (ГЕТЕРОГЕННЫЕ сигнатуры) ──
    market_metrics = _safe_call(
        "market_metrics",
        db,
        lambda: compute_market_metrics(db, district=district, premise_kind=_PREMISE_KIND),
    )
    supply_rows = _safe_call(
        "supply_layers",
        db,
        lambda: compute_all_layers(db, district=district, premise_kind=_PREMISE_KIND),
    )
    supply_layers = _summarize_supply_layers(supply_rows)
    future_supply = _safe_call(
        "future_supply",
        db,
        lambda: compute_future_supply_pressure(
            db, district=district, horizon_months=primary, premise_kind=_PREMISE_KIND
        ),
    )
    forecasts = _safe_call(
        "demand_supply_forecast",
        db,
        lambda: compute_demand_supply_forecast(
            db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
        ),
    )
    # §22 audit-issue #1871 P1: compute_scenarios теперь возвращает кортеж
    # `(scenarios, collapsed, collapse_reason)` — collapse-флаг идёт в payload отчёта,
    # чтобы фронт показал ПОЧЕМУ три сценария идентичны (failed β-gate в §9.6).
    # _safe_call оборачивает любой сбой → None → штатно деградируем (collapse=False).
    scenarios_result = _safe_call(
        "scenarios",
        db,
        lambda: compute_scenarios(
            db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
        ),
    )
    if scenarios_result is None:
        scenarios = None
        scenarios_collapsed = False
        scenarios_collapse_reason: str | None = None
    else:
        scenarios, scenarios_collapsed, scenarios_collapse_reason = scenarios_result
    product_scores = _safe_call(
        "score_card",
        db,
        lambda: compute_score_card(
            db, spec=spec, district=district, cad_num=cad_num, horizon_months=primary
        ),
    )
    special_indices = _safe_call(
        "special_indices",
        db,
        lambda: compute_special_indices(
            db, spec=spec, district=district, cad_num=cad_num, horizons=horizon_list
        ),
    )
    recommendation_overlay = _safe_call(
        "forecast_overlay",
        db,
        lambda: build_forecast_overlay(
            db,
            district=district,
            cad_num=cad_num,
            horizon_months=primary,
            target_class=resolved_class,
        ),
    )

    # ── Макро-свежесть (audit MEDIUM): только лог; проводка в отчёт — 3b ─────────
    macro = _safe_call("monthly_macro", db, lambda: get_monthly_macro(db))
    macro_as_of = _macro_as_of(macro)
    if macro_as_of is not None:
        logger.info(
            "build_site_finder_report: macro_as_of=%s (domrf monthly may be stale; "
            "surfacing into report deferred to 3b)",
            macro_as_of,
        )

    # ── 3. Раскладка в ЧИСТЫЙ §22-сборщик (#988) — каждый слой в свой слот ───────
    return assemble_report(
        analyze,
        market_metrics=market_metrics,
        supply_layers=supply_layers,
        forecasts=forecasts,
        future_supply=future_supply,
        scenarios=scenarios,
        scenarios_collapsed=scenarios_collapsed,
        scenarios_collapse_reason=scenarios_collapse_reason,
        recommendation_overlay=recommendation_overlay,
        product_scores=product_scores,
        special_indices=special_indices,
        segment=spec.as_dict(),
        cad_num=cad_num,
        district=district,
        horizons=horizon_list,
        advisory=_ADVISORY,
    )