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

"""§13 сборщик итогового советующего отчёта Site Finder v2 — ЧИСТОЕ маппирование.

#988 (955-A2, Site Finder v2 / «GG-форсайт» ТЗ §13), EPIC 11 «Отчёт». Это **СБОРЩИК**
(assembler): он НАПОЛНЯЕТ структурный контейнер `SiteFinderReport` (#987) уже-
посчитанными advisory-выводами форсайт-стека (#980…#986 / §9.1-9.3) + dict'ом
`analyze` (вывод эндпоинта analyze_parcel) и считает отчётную уверенность через
`compute_report_confidence` (#990). Своей §9.x-математики НЕ пересобирает и в БД НЕ
ходит — берёт уже-готовые входы (их считает будущий вызывающий: #961 API / #957 чат)
и РАСКЛАДЫВАЕТ их по восьми типизированным секциям §13. Поэтому он ЮНИТ-ТЕСТИРУЕТСЯ
БЕЗ БД. Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM (шаблонные RU-строки, не генерация).

ADVISORY (зеркало всего стека #980…#986 / #987 / #990): отчёт лишь СОБИРАЕТ advisory-
выводы — `advisory` ВСЕГДА True, прокидывается во все под-вызовы (в т.ч. cap-потолок
#990 'medium'). Не основание для инвест-решения.

ВОСЕМЬ СЕКЦИЙ §13 — ИЗ ЧЕГО собирается каждая:
  1. exec_summary  — СИНТЕЗ заголовка/вердикта/ключевых чисел из сильнейших сигналов
     (deficit_index, рекомендованный класс/mix, overall-скор, уверенность). Шаблонные
     RU-строки (НЕ LLM).
  2. market_now    — из `analyze` dict (абсорбция/цены/конкуренты/POI) + `market_metrics`
     (§9.2) + `supply_layers` (§9.3). Defensive `.get()` (analyze loosely-typed).
  3. future_market — `forecasts` (per-горизонт #952), `future_supply` (§9.3), будущие
     конкуренты, сводка `scenarios` (#984).
  4. product_tz    — `recommendation_overlay` (#983: класс §10.2 / mix / commercial §10.4
     / USP §10.5 + §16-причины).
  5. scenarios     — `scenarios` (три ScenarioForecast #984, by_scenario).
  6. scoring       — `product_scores` (#985, 10 скоров) + `special_indices` (#986, 6) +
     overall.
  7. confidence    — `compute_report_confidence(...)` (#990) из извлечённых сигналов
     качества данных (deal_count/analog_count/domrf_coverage/history_months/confounded +
     component-confidences). ReportConfidenceResult.as_dict() ложится в слот.
  8. meta          — cad_num/district/segment/horizons/advisory/schema_version.

GRACEFUL (дух всего стека): любой вход None → его секция частично заполнена / пуста —
отчёт ВАЛИДЕН как частичный (#987: все поля Optional). НИКОГДА не бросает: каждый вход
нормализуется через `_as_dict_or` (dataclass с `.as_dict()` ИЛИ уже-dict ИЛИ None),
чтобы вызывающий мог передать как объекты, так и их `.as_dict()`-словари.
"""

from __future__ import annotations

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

from app.services.forecasting.confidence_engine import (
    Confidence,
    compute_report_confidence,
)
from app.services.forecasting.report import (
    ReportConfidence,
    ReportConfidenceLevel,
    ReportExecSummary,
    ReportFutureMarket,
    ReportMarketNow,
    ReportMeta,
    ReportProductTz,
    ReportScenarios,
    ReportScoring,
    SiteFinderReport,
)

logger = logging.getLogger(__name__)

# ── Named-константы ───────────────────────────────────────────────────────────

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

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

# Ранг уверенности (для выбора «худшего» компонента при синтезе exec_summary).
_CONFIDENCE_RANK: dict[Confidence, int] = {"low": 0, "medium": 1, "high": 2}

# RU-метки уровня уверенности для exec_summary-вердикта.
_LEVEL_RU: dict[ReportConfidenceLevel, str] = {
    "high": "высокая",
    "medium": "средняя",
    "low": "низкая",
}

# Порог deficit_index, выше которого сигнал «строить» считаем сильным (зеркало духа
# special_indices._VOID_THRESHOLD: умеренно-сильный дефицит на лог-шкале #980 [−1,+1]).
_STRONG_DEFICIT_THRESHOLD: float = 0.25

# Нейтральная зона |deficit_index|, внутри которой рынок считаем сбалансированным
# (#1745 «баланс»). Зеркало frontend forecast-helpers.DEFICIT_BALANCE_EPS = 0.05.
_SIGNAL_BALANCE_EPS: float = 0.05

# #1959 honesty-guard: если ВСЕ ячейки сетки прижаты к ±1.0 (дефицит не различает
# сегменты — degenerate-fallback, как было до фикса посегментации), категоричные
# вердикты «строить»/«избегать» вводят в заблуждение (на самом деле сигнал не
# дискриминирует). Тогда вместо вердикта показываем честную плашку. |di| ≥ этого
# порога считаем «прижато к пределу шкалы».
_DEGENERATE_CLAMP_THRESHOLD: float = 0.999
_DEGENERATE_SEGMENT_NOTE: str = "недостаточно гранулярных данных для посегментного вывода"

# Сегментный горизонт по умолчанию для извлечения сигналов из forecasts (мес). Берём
# 12 — типовой средне-срочный продуктовый горизонт (зеркало #982/#983/#986 default).
_PRIMARY_HORIZON_MONTHS: int = 12


def _as_dict_or(x: Any) -> dict[str, Any] | None:
    """Нормализовать вход в JSON-safe dict: dataclass-с-`as_dict()` / dict / None. PURE.

    Вызывающий (#961 API / #957 чат) может передать КАК объект под-сервиса (например
    `MarketMetrics`), ТАК и его уже-посчитанный `.as_dict()`-словарь. Приводим к единой
    форме (плоский JSON-safe dict):
      • None → None (graceful — секция останется пустой).
      • объект с методом `as_dict` → `x.as_dict()` (под-сервис сам отдаёт JSON-safe вид).
      • уже-dict → как есть.
    Любой иной тип (мусор) → None (НЕ бросаем — отчёт валиден как частичный).

    Args:
        x: dataclass-инстанс под-сервиса, его `.as_dict()`-словарь, или None.

    Returns:
        JSON-safe dict или None.
    """
    if x is None:
        return None
    as_dict = getattr(x, "as_dict", None)
    if callable(as_dict):
        result = as_dict()
        return result if isinstance(result, dict) else None
    if isinstance(x, dict):
        return x
    return None


def _as_dict_list(items: Sequence[Any] | None) -> list[dict[str, Any]]:
    """Нормализовать последовательность входов в список JSON-safe dict'ов. PURE.

    Каждый элемент через `_as_dict_or`; None-элементы (нормализовавшиеся в None)
    отбрасываем. None/пустой вход → [] (graceful). Сохраняет порядок.
    """
    if not items:
        return []
    out: list[dict[str, Any]] = []
    for item in items:
        normalized = _as_dict_or(item)
        if normalized is not None:
            out.append(normalized)
    return out


# ──────────────────────────────────────────────────────────────────────────────
# Извлечение сигналов качества данных для §15-уверенности (#990). Каждый pure,
# graceful: вход None/тонкий → None (#990 трактует None как «нет сигнала»).
# ──────────────────────────────────────────────────────────────────────────────


def _deal_count(analyze: dict[str, Any], market_metrics: dict[str, Any] | None) -> int | None:
    """Число сделок (продаж) за окно — для deal_count #990. PURE.

    Предпочитаем `market_metrics.n_sold` (§9.2 — прямой счётчик проданных лотов выборки);
    fallback на `analyze.market_pulse.competitors_total` НЕ берём (это число ЖК, не
    сделок — пошло бы в analog_count). Нет сигнала → None (#990 → тянет в low).
    """
    if market_metrics is not None:
        n_sold = market_metrics.get("n_sold")
        if isinstance(n_sold, int):
            return n_sold
    return None


def _analog_count(analyze: dict[str, Any], market_metrics: dict[str, Any] | None) -> int | None:
    """Число ЖК-аналогов в выборке — для analog_count #990. PURE.

    Предпочитаем `market_metrics.obj_count` (§9.2 — число отдельных ЖК выборки, ТОЧНО
    то, что #990 ждёт). Fallback — число конкурентов из `analyze` (market_pulse
    competitors_total ИЛИ длина списка competitors): это окружение участка, тоже мера
    «сколько объектов-аналогов рядом». Нет сигнала → None.
    """
    if market_metrics is not None:
        obj_count = market_metrics.get("obj_count")
        if isinstance(obj_count, int):
            return obj_count
    pulse = analyze.get("market_pulse")
    if isinstance(pulse, dict):
        total = pulse.get("competitors_total")
        if isinstance(total, int):
            return total
    competitors = analyze.get("competitors")
    if isinstance(competitors, list):
        return len(competitors)
    return None


def _domrf_coverage(analyze: dict[str, Any], supply_layers: dict[str, Any] | None) -> float | None:
    """Покрытие domrf↔objective ∈ [0,1] — для domrf_coverage #990. PURE.

    Главный sparse-риск проекта (~2.5%). Источники по приоритету (единица ЯВНАЯ
    per-branch — НЕ угадываем по величине, иначе настоящий sub-1% процент типа 0.8%
    спутался бы с долей 0.8 = 80% и инфлировал бы confidence в exactly near-zero кейсе,
    который §15 призван флагать):
      • `supply_layers.domrf_coverage` — уже ДОЛЯ ∈ [0,1] (0.025) → берём как есть.
      • `analyze.market_data_coverage_pct` — всегда ПРОЦЕНТ (2.5 == 2.5%) → /100 → доля.
    Нет сигнала → None (#990 → тянет в low: слой §9.3 недооценён).
    """
    if supply_layers is not None:
        coverage = supply_layers.get("domrf_coverage")
        if isinstance(coverage, int | float) and not isinstance(coverage, bool):
            return _clamp_fraction(float(coverage))
    pct = analyze.get("market_data_coverage_pct")
    if isinstance(pct, int | float) and not isinstance(pct, bool):
        return _clamp_fraction(float(pct) / 100.0)
    return None


def _clamp_fraction(value: float) -> float:
    """Зажать долю покрытия в [0,1] (защита от грязных данных). PURE.

    Единица приводится у источника (`_domrf_coverage`: percent-ветка делит на 100,
    fraction-ветка — как есть) — здесь только clamp, БЕЗ догадок про percent-vs-fraction.
    """
    return max(0.0, min(1.0, value))


def _history_months(
    market_metrics: dict[str, Any] | None, forecasts: Sequence[dict[str, Any]]
) -> int | None:
    """Глубина ряда (мес) — для history_months #990. PURE.

    `market_metrics.window_months` (§9.2 окно наблюдения продаж) — единственный явный
    сигнал глубины истории в доступных входах. Нет → None (#990 → короткий ряд тянет
    вниз). forecasts оставлены в сигнатуре для будущего расширения (на сейчас ряд
    глубины в них не выносится).
    """
    if market_metrics is not None:
        window = market_metrics.get("window_months")
        if isinstance(window, int) and window > 0:
            return window
    return None


def _deal_count_months(market_metrics: dict[str, Any] | None) -> int | None:
    """Окно наблюдения для deal_count (мес) — для deal_count_months #990. PURE.

    Читает тот же `market_metrics.window_months` (§9.2), что и `_history_months` —
    именно за это окно считается n_sold. Нет → None (#990 пропускает суффикс «за N мес»).
    """
    if market_metrics is not None:
        window = market_metrics.get("window_months")
        if isinstance(window, int) and window > 0:
            return window
    return None


def _confounded(forecasts: Sequence[dict[str, Any]]) -> bool:
    """Пересекает ли окно прогноза шок-период — для confounded #990. PURE.

    Канонический ключ — `confounded` (см. `DemandSupplyForecast.as_dict()`, #1222:
    прокинут из §9.5 macro_coefficient / §9.6 rate_sensitivity). Исторический алиас
    `is_confounded_window` оставляем ради forward-совместимости со старыми форматами
    forecast-диктов. Всегда через `.get()` (без default → None) — отсутствие ключа
    НЕ KeyError, а «нет сигнала» (None ≠ True → False). Если хоть один forecast
    True → отчётное окно считаем confounded (оценки смещены, #990 → НИКОГДА не
    'high'). Нет флага нигде → False (чистое окно).
    """
    for f in forecasts:
        if f.get("confounded") is True or f.get("is_confounded_window") is True:
            return True
    return False


def _component_confidences(
    market_metrics: dict[str, Any] | None,
    future_supply: dict[str, Any] | None,
    forecasts: Sequence[dict[str, Any]],
    product_scores: dict[str, Any] | None,
    special_indices: dict[str, Any] | None,
) -> list[tuple[str, Confidence]]:
    """Собрать per-service confidence вкладывающих под-сервисов — для #990. PURE.

    #990 свернёт их weakest-link (MIN) вместе с сырыми счётчиками. Берём confidence-
    метку каждого доступного под-вывода (market_metrics §9.2 / future_supply §9.3 /
    каждый per-горизонт forecast #952 / product_scores #985 / special_indices #986).
    Только whitelisted 'high|medium|low' (мусор/None отбрасываем).

    #1737: возвращаем ПАРЫ (имя-сервиса, уровень) — машинное имя под-сервиса
    («market_metrics» / «future_supply» / «product_scores» / «special_indices» /
    «forecasts»), чтобы #990 пронёс ИМЯ сервиса в RU-метку фактора и UI не показывал
    безличный «Компонент N».
    """
    out: list[tuple[str, Confidence]] = []
    for name, source in (
        ("market_metrics", market_metrics),
        ("future_supply", future_supply),
        ("product_scores", product_scores),
        ("special_indices", special_indices),
    ):
        conf = _confidence_of(source)
        if conf is not None:
            out.append((name, conf))
    # #1958: forecasts отдаёт по одному confidence НА КАЖДЫЙ горизонт (#952: 3/6/12/24
    # мес). Раньше каждый горизонт эмитился отдельной парой → в списке факторов #990
    # «Прогноз спрос/предложение» дублировался ×N (обычно ×4) с одинаковой меткой.
    # Сворачиваем в ОДИН фактор weakest-link'ом (MIN ранга по горизонтам) — худший
    # горизонт тянет общий прогноз вниз, как и остальной weakest-link стек.
    forecast_confs = [c for c in (_confidence_of(f) for f in forecasts) if c is not None]
    if forecast_confs:
        weakest = min(forecast_confs, key=lambda c: _CONFIDENCE_RANK[c])
        out.append(("forecasts", weakest))
    return out


def _confidence_of(source: dict[str, Any] | None) -> Confidence | None:
    """Достать whitelisted confidence-метку из под-вывода (или None). PURE."""
    if not isinstance(source, dict):
        return None
    conf = source.get("confidence")
    if conf in _CONFIDENCE_RANK:
        return conf  # type: ignore[return-value]  # сужено проверкой членства
    return None


# ──────────────────────────────────────────────────────────────────────────────
# Извлечение опорных чисел для exec_summary-синтеза. Каждый pure, graceful → None.
# ──────────────────────────────────────────────────────────────────────────────


def _primary_deficit_index(forecasts: Sequence[dict[str, Any]]) -> float | None:
    """deficit_index основного горизонта (или первого доступного). PURE.

    Предпочитаем _PRIMARY_HORIZON_MONTHS (12 мес — продуктовый горизонт); если его нет —
    первый forecast с не-None deficit_index. Нет → None.
    """
    primary = next(
        (f for f in forecasts if f.get("horizon_months") == _PRIMARY_HORIZON_MONTHS),
        None,
    )
    if primary is not None and primary.get("deficit_index") is not None:
        di = primary["deficit_index"]
        return float(di) if isinstance(di, int | float) and not isinstance(di, bool) else None
    for f in forecasts:
        di = f.get("deficit_index")
        if isinstance(di, int | float) and not isinstance(di, bool):
            return float(di)
    return None


def _primary_months_of_inventory(forecasts: Sequence[dict[str, Any]]) -> float | None:
    """months_of_inventory основного горизонта (или первого доступного). PURE.

    Зеркало `_primary_deficit_index`, но для MOI (#952 ДИСКРИМИНИРУЮЩИЙ companion к
    дефициту). Предпочитаем _PRIMARY_HORIZON_MONTHS (12 мес); иначе первый forecast
    с не-None months_of_inventory. Нет → None.
    """
    primary = next(
        (f for f in forecasts if f.get("horizon_months") == _PRIMARY_HORIZON_MONTHS),
        None,
    )
    if primary is not None and primary.get("months_of_inventory") is not None:
        moi = primary["months_of_inventory"]
        return float(moi) if isinstance(moi, int | float) and not isinstance(moi, bool) else None
    for f in forecasts:
        moi = f.get("months_of_inventory")
        if isinstance(moi, int | float) and not isinstance(moi, bool):
            return float(moi)
    return None


def _recommended_class(product_tz: dict[str, Any] | None) -> str | None:
    """Рекомендованный класс из product_tz-вывода (overlay class_reco / obj_class). PURE."""
    if not isinstance(product_tz, dict):
        return None
    obj_class = product_tz.get("obj_class")
    if isinstance(obj_class, str) and obj_class:
        return obj_class
    class_reco = product_tz.get("class_reco")
    if isinstance(class_reco, dict):
        reco_class = class_reco.get("obj_class")
        if isinstance(reco_class, str) and reco_class:
            return reco_class
    return None


def _overall_score(product_scores: dict[str, Any] | None) -> float | None:
    """overall-скор из карты #985 (ProductScoreCard.as_dict()['overall']). PURE."""
    if not isinstance(product_scores, dict):
        return None
    overall = product_scores.get("overall")
    if isinstance(overall, int | float) and not isinstance(overall, bool):
        return float(overall)
    return None


# ──────────────────────────────────────────────────────────────────────────────
# Per-section builders — каждый PURE, graceful (входы None → пустая/частичная секция).
# ──────────────────────────────────────────────────────────────────────────────


def _build_market_now(
    analyze: dict[str, Any],
    market_metrics: dict[str, Any] | None,
    supply_layers: dict[str, Any] | None,
) -> ReportMarketNow:
    """§13.2 market_now — текущий рынок из analyze + §9.2/§9.3. PURE, defensive.

    `market_metrics` (§9.2) / `supply_layers` (§9.3) кладём как есть (их as_dict уже
    JSON-safe). `competitors` берём из analyze (defensive: список dict'ов конкурентов).
    `summary` синтезируем из доступных абсорбции/цены/числа конкурентов. analyze loosely-
    typed → всё через `.get()`.
    """
    competitors = analyze.get("competitors")
    competitor_list = (
        [c for c in competitors if isinstance(c, dict)] if isinstance(competitors, list) else []
    )
    summary = _market_now_summary(analyze, market_metrics)
    return ReportMarketNow(
        market_metrics=market_metrics,
        competitors=competitor_list,
        supply_layers=supply_layers,
        summary=summary,
    )


def _market_now_summary(
    analyze: dict[str, Any], market_metrics: dict[str, Any] | None
) -> str | None:
    """Короткий RU-текст про текущий рынок из доступных сигналов. PURE, шаблонный.

    Собирает фразу из абсорбции (§9.2 unit_velocity), средней цены (analyze) и числа
    конкурентов. Ни одного сигнала → None (graceful — секция без summary валидна).
    """
    parts: list[str] = []
    if market_metrics is not None:
        velocity = market_metrics.get("unit_velocity")
        if isinstance(velocity, int | float) and not isinstance(velocity, bool):
            parts.append(f"абсорбция ~{round(float(velocity), 1)} ед./мес")
    avg_price = analyze.get("market_avg_price_per_m2")
    if isinstance(avg_price, int | float) and not isinstance(avg_price, bool):
        parts.append(f"средняя цена ~{round(float(avg_price)):,} ₽/м²".replace(",", " "))
    # #1634: НЕ через _analog_count — он отдаёт market_metrics.obj_count (число ЖК во
    # всей district-wide/микрорайонной выборке §9.2), что НЕ равно «конкурентов рядом».
    # Метка честно следует источнику: obj_count → «в выборке района», локальный fallback
    # из analyze (competitors_total / len(competitors)) → «рядом».
    if market_metrics is not None and isinstance(market_metrics.get("obj_count"), int):
        parts.append(f"{market_metrics['obj_count']} ЖК в выборке района")
    else:
        n_local: int | None = None
        pulse = analyze.get("market_pulse")
        if isinstance(pulse, dict) and isinstance(pulse.get("competitors_total"), int):
            n_local = pulse["competitors_total"]
        elif isinstance(analyze.get("competitors"), list):
            n_local = len(analyze["competitors"])
        if n_local is not None:
            parts.append(f"{n_local} ЖК-конкурентов рядом")
    if not parts:
        return None
    return "Текущий рынок: " + ", ".join(parts) + "."


def _build_future_market(
    forecasts: Sequence[dict[str, Any]],
    future_supply: dict[str, Any] | None,
    future_competitors: list[dict[str, Any]],
    scenarios_summary: dict[str, Any] | None,
) -> ReportFutureMarket:
    """§13.3 future_market — прогноз из forecasts/future_supply/scenarios. PURE.

    `forecasts_by_horizon` — список per-горизонт #952; `future_supply` (§9.3) как есть;
    `future_competitors` — будущие конкуренты (из forecasts или явный аргумент);
    `scenarios_summary` — сводка #984. `summary` синтезируем из дефицита основного
    горизонта. Всё пустое → пустая секция (graceful).
    """
    summary = _future_market_summary(forecasts)
    return ReportFutureMarket(
        forecasts_by_horizon=list(forecasts),
        future_supply=future_supply,
        future_competitors=future_competitors,
        scenarios_summary=scenarios_summary,
        summary=summary,
    )


def _future_market_summary(forecasts: Sequence[dict[str, Any]]) -> str | None:
    """Короткий RU-текст про будущий рынок (дефицит/затоварка осн. горизонта). PURE.

    Индекс дефицита — АБСОЛЮТНЫЙ вердикт; months_of_inventory (если есть) —
    ДИСКРИМИНИРУЮЩИЙ headline «≈N мес конкурирующего предложения» (различает
    затоваренные сегменты, где дефицит у всех липнет к −1).
    """
    di = _primary_deficit_index(forecasts)
    if di is None:
        return None
    if di >= _STRONG_DEFICIT_THRESHOLD:
        verdict = "выраженный дефицит предложения (сигнал «строить»)"
    elif di > 0:
        verdict = "умеренный дефицит предложения"
    elif di == 0:
        verdict = "баланс спроса и предложения"
    else:
        verdict = "затоварка (предложения больше спроса)"
    moi = _primary_months_of_inventory(forecasts)
    moi_part = f", ≈{round(moi, 1)} мес конкурирующего предложения" if moi is not None else ""
    return f"Прогноз: {verdict}, индекс дефицита {_signed(di)}{moi_part} на горизонте."


def _build_product_tz(product_tz: dict[str, Any] | None) -> ReportProductTz:
    """§13.4 product_tz — рекомендация продукта из overlay #983. PURE, defensive.

    Маппит overlay-форму (#983 build_forecast_overlay as-dict): класс §10.2 (class_reco
    / obj_class), квартирография (mix / ranked_segments), commercial §10.4, USP §10.5,
    §16-причины (reasons / class_reco.reason). overlay None → пустая секция (graceful).
    """
    if not isinstance(product_tz, dict):
        return ReportProductTz()
    obj_class = _recommended_class(product_tz)
    mix = _extract_mix(product_tz)
    commercial = (
        product_tz.get("commercial") if isinstance(product_tz.get("commercial"), dict) else None
    )
    usp = _extract_list(product_tz, "usp")
    reasons = _extract_reasons(product_tz)
    summary = _product_tz_summary(obj_class, usp)
    return ReportProductTz(
        obj_class=obj_class,
        mix=mix,
        commercial=commercial,
        usp=usp,
        reasons=reasons,
        summary=summary,
    )


def _extract_mix(product_tz: dict[str, Any]) -> list[dict[str, Any]]:
    """Квартирография — явный `mix` ИЛИ ranked_segments overlay (формат+дефицит). PURE.

    Overlay #983 не несёт готового `mix`, но его `ranked_segments` (DESC по дефициту) —
    продуктовый ответ «какой формат строить». Берём `mix` если задан, иначе проецируем
    ranked_segments в компактные {bucket, obj_class, deficit_index, projected_demand_units,
    signal} (#1745 — таблица прогноза по форматам). Нет ни того, ни другого → [].

    #1745 ADDITIVE: проброс `projected_demand_units` (прогноз спроса, квартир) и
    деривированного `signal` («строить»/«баланс»/«избегать») из deficit_index. Старые
    потребители mix читают bucket/obj_class/deficit_index — новые поля их не ломают.
    """
    explicit = product_tz.get("mix")
    if isinstance(explicit, list):
        entries = [m for m in explicit if isinstance(m, dict)]
        degenerate = _all_cells_clamped([e.get("deficit_index") for e in entries])
        return [_enrich_mix_entry(m, degenerate=degenerate) for m in entries]
    ranked = product_tz.get("ranked_segments")
    if isinstance(ranked, list):
        segs = [seg for seg in ranked if isinstance(seg, dict)]
        # #1959 honesty-guard: вся сетка прижата к ±1.0 → дефицит не дискриминирует
        # сегменты (degenerate-fallback). Не эмитим категоричные «строить»/«избегать»
        # — вместо них честная плашка `_DEGENERATE_SEGMENT_NOTE`.
        degenerate = _all_cells_clamped([seg.get("deficit_index") for seg in segs])
        return [
            {
                "bucket": seg.get("bucket"),
                "obj_class": seg.get("obj_class"),
                "deficit_index": seg.get("deficit_index"),
                "projected_demand_units": seg.get("projected_demand_units"),
                "signal": (
                    _DEGENERATE_SEGMENT_NOTE
                    if degenerate
                    else _build_signal(seg.get("deficit_index"))
                ),
            }
            for seg in segs
        ]
    return []


def _all_cells_clamped(deficit_indices: Sequence[Any]) -> bool:
    """True если ВСЕ измеримые ячейки прижаты к ±1.0 (degenerate-fallback). PURE.

    #1959 honesty-guard. «Измеримые» = числовые deficit_index (None-ячейки —
    тонкие данные, в знаменатель не идут). Пусто / нет измеримых → False (нечего
    защищать — обычный graceful-путь). Если хотя бы одна ячейка различается
    (|di| < порога), сетка ДИСКРИМИНИРУЕТ → вердикты честны → False. PURE.

    Args:
        deficit_indices: deficit_index ячеек (числа / None / мусор).

    Returns:
        True только когда есть ≥1 измеримая ячейка и ВСЕ они |di| ≥ порога.
    """
    measured = [
        float(d) for d in deficit_indices if isinstance(d, int | float) and not isinstance(d, bool)
    ]
    if not measured:
        return False
    return all(abs(d) >= _DEGENERATE_CLAMP_THRESHOLD for d in measured)


def _enrich_mix_entry(entry: dict[str, Any], *, degenerate: bool = False) -> dict[str, Any]:
    """Дополнить явную mix-ячейку деривированным `signal` (#1745). PURE.

    Не перетирает уже заданные поля (явный `mix` мог нести свой `signal`); добавляет
    `signal` из `deficit_index` только если его нет. Возвращает НОВЫЙ dict (не мутирует
    вход — assembler-чистота).

    #1959 honesty-guard: `degenerate=True` (вся сетка прижата к ±1.0) → вместо
    категоричного сигнала ставим честную плашку `_DEGENERATE_SEGMENT_NOTE` (только
    когда у ячейки не было своего явного `signal`).
    """
    enriched = dict(entry)
    if enriched.get("signal") is None:
        enriched["signal"] = (
            _DEGENERATE_SEGMENT_NOTE if degenerate else _build_signal(enriched.get("deficit_index"))
        )
    return enriched


def _build_signal(deficit_index: Any) -> str | None:
    """Сигнал «строить»/«баланс»/«избегать» из deficit_index ∈ [−1,1] (#1745). PURE.

    Зеркало deficit-семантики §22 (frontend forecast-helpers.deficitWord): >+0.05
    недонасыщенность → «строить», <−0.05 затоварка → «избегать», иначе «баланс».
    deficit_index None (тонкие данные) → None (фронт рисует «тонкие данные», НЕ 0/сигнал).
    """
    if not isinstance(deficit_index, int | float) or isinstance(deficit_index, bool):
        return None
    if deficit_index > _SIGNAL_BALANCE_EPS:
        return "строить"
    if deficit_index < -_SIGNAL_BALANCE_EPS:
        return "избегать"
    return "баланс"


def _extract_list(source: dict[str, Any], key: str) -> list[dict[str, Any]]:
    """Достать список dict'ов по ключу (мусор/None → []). PURE."""
    value = source.get(key)
    if isinstance(value, list):
        return [item for item in value if isinstance(item, dict)]
    return []


def _extract_reasons(product_tz: dict[str, Any]) -> list[dict[str, Any]]:
    """§16-причины — явные `reasons` ИЛИ §16-reason из class_reco/usp overlay. PURE.

    overlay #983 кладёт §16-reason ВНУТРЬ class_reco / каждого usp / ranked_segment.
    Если явного `reasons` нет — поднимаем reason из class_reco (рекомендации класса) на
    уровень секции (главная §16-причина продукта). Нет → [].
    """
    explicit = product_tz.get("reasons")
    if isinstance(explicit, list):
        return [r for r in explicit if isinstance(r, dict)]
    class_reco = product_tz.get("class_reco")
    if isinstance(class_reco, dict):
        reason = class_reco.get("reason")
        if isinstance(reason, dict):
            return [reason]
    return []


def _product_tz_summary(obj_class: str | None, usp: list[dict[str, Any]]) -> str | None:
    """Короткий RU-текст про рекомендованный продукт. PURE, шаблонный."""
    if obj_class is None and not usp:
        return None
    parts: list[str] = []
    if obj_class is not None:
        parts.append(f"рекомендован класс «{obj_class}»")
    if usp:
        parts.append(f"{len(usp)} USP-ниш по дефициту форматов")
    return "Продукт: " + ", ".join(parts) + "."


def _build_scenarios(
    scenarios: Sequence[dict[str, Any]] | None,
    *,
    collapsed: bool = False,
    collapse_reason: str | None = None,
) -> ReportScenarios:
    """§13.5 scenarios — карта {scenario: сводка} из трёх ScenarioForecast #984. PURE.

    `scenarios` — список из трёх по `as_dict()` (conservative/base/aggressive). Свора-
    чиваем в `by_scenario` (карта по имени сценария) — стабильно для экспортёров/чата.
    None/пусто → пустая секция (graceful).

    §22 audit-issue #1871 P1: `collapsed`/`collapse_reason` — вердикт
    `compute_scenarios` (когда §9.6 β-gate fail → §9.4 coefficient=1.0 → три сценария
    идентичны base). Прокидываем как есть в `ReportScenarios`; экспортёры/фронт сами
    решат рисовать плашку-объяснение вместо таблицы. Защитная нормализация: пустые
    `scenarios` → флаги сбрасываем в дефолт (нет данных — нет вердикта «схлопнулось»,
    `compute_scenarios` сам уже это гарантирует, но belt-and-suspenders).
    """
    if not scenarios:
        return ReportScenarios()
    by_scenario: dict[str, Any] = {}
    for sc in scenarios:
        if not isinstance(sc, dict):
            continue
        name = sc.get("scenario")
        if isinstance(name, str) and name:
            by_scenario[name] = sc
    if not by_scenario:
        return ReportScenarios()
    summary = _scenarios_summary(by_scenario)
    return ReportScenarios(
        by_scenario=by_scenario,
        summary=summary,
        scenarios_collapsed=collapsed,
        scenarios_collapse_reason=collapse_reason,
    )


def _scenarios_summary(by_scenario: dict[str, Any]) -> str | None:
    """Короткий RU-текст про разброс сценариев. PURE, шаблонный."""
    names = [n for n in ("conservative", "base", "aggressive") if n in by_scenario]
    if not names:
        return None
    return f"Сценарии ({len(by_scenario)}): {', '.join(names)} — разброс по конверту ставки."


def _build_scoring(
    product_scores: dict[str, Any] | None, special_indices: dict[str, Any] | None
) -> ReportScoring:
    """§13.6 scoring — 10 продуктовых скоров #985 + 6 спец-индексов #986 + overall. PURE.

    `product_scores` (#985) / `special_indices` (#986) кладём как есть; `overall`
    дублируем из карты #985 (удобство экспортёров). Всё None → пустая секция (graceful).
    """
    return ReportScoring(
        product_scores=product_scores,
        special_indices=special_indices,
        overall=_overall_score(product_scores),
    )


def _build_confidence(
    *,
    analyze: dict[str, Any],
    market_metrics: dict[str, Any] | None,
    supply_layers: dict[str, Any] | None,
    future_supply: dict[str, Any] | None,
    forecasts: Sequence[dict[str, Any]],
    product_scores: dict[str, Any] | None,
    special_indices: dict[str, Any] | None,
    advisory: bool,
) -> ReportConfidence:
    """§13.7 confidence — отчётная уверенность через #990 из извлечённых сигналов. PURE.

    Кормит `compute_report_confidence` (#990) сырыми счётчиками качества данных
    (deal_count/analog_count/domrf_coverage/history_months/confounded) + per-service
    confidence вкладывающих под-сервисов. Результат (level/rationale/factors) ложится в
    слот ReportConfidence. Все сигналы тонкие → #990 честно вернёт 'low' + «недостаточно
    данных» (НИКОГДА не бросает).
    """
    result = compute_report_confidence(
        component_confidences=_component_confidences(
            market_metrics, future_supply, forecasts, product_scores, special_indices
        ),
        deal_count=_deal_count(analyze, market_metrics),
        deal_count_months=_deal_count_months(market_metrics),
        analog_count=_analog_count(analyze, market_metrics),
        domrf_coverage=_domrf_coverage(analyze, supply_layers),
        history_months=_history_months(market_metrics, forecasts),
        confounded=_confounded(forecasts),
        advisory=advisory,
    )
    payload = result.as_dict()
    return ReportConfidence(
        level=payload["level"],
        rationale=payload["rationale"],
        factors=payload["factors"],
    )


def _build_exec_summary(
    *,
    forecasts: Sequence[dict[str, Any]],
    product_tz: dict[str, Any] | None,
    product_scores: dict[str, Any] | None,
    confidence: ReportConfidence,
) -> ReportExecSummary:
    """§13.1 exec_summary — СИНТЕЗ заголовка/вердикта/чисел из сильнейших сигналов. PURE.

    ДЕТЕРМИНИРОВАННЫЙ шаблонный синтез (НЕ LLM):
      • headline — одна фраза «что строить здесь» из рекомендованного класса +
        силы дефицита (или честный «данных недостаточно»).
      • verdict — абзац: дефицит осн. горизонта (АБСОЛЮТ) + months_of_inventory
        (ДИСКРИМИНИРУЮЩИЙ headline «≈N мес конкурирующего предложения») + overall-скор
        + уровень уверенности.
      • key_numbers — плоские опорные числа (deficit_index / months_of_inventory /
        overall_score / confidence).
      • overall_confidence — уровень из §15-секции (#990).

    Любого сигнала нет → соответствующая часть опускается; всё пусто → честный headline
    «недостаточно данных» (секция всё равно валидна).
    """
    di = _primary_deficit_index(forecasts)
    moi = _primary_months_of_inventory(forecasts)
    obj_class = _recommended_class(product_tz)
    overall = _overall_score(product_scores)
    level: ReportConfidenceLevel | None = confidence.level

    headline = _exec_headline(obj_class, di)
    verdict = _exec_verdict(di, moi, overall, level)
    key_numbers = _exec_key_numbers(di, moi, overall, level)

    return ReportExecSummary(
        headline=headline,
        verdict=verdict,
        key_numbers=key_numbers,
        overall_confidence=level,
    )


def _exec_headline(obj_class: str | None, deficit_index: float | None) -> str:
    """Одна фраза-вердикт «что строить здесь». PURE, шаблонный.

    Класс задан + сильный дефицит → «Строить <класс>: рынок недонасыщен». Класс задан,
    дефицита нет/слаб → «Рассмотреть <класс>». Нет класса, но дефицит → сигнал по
    дефициту. Ничего → честный «данных недостаточно».
    """
    strong = deficit_index is not None and deficit_index >= _STRONG_DEFICIT_THRESHOLD
    if obj_class is not None:
        if strong:
            return f"Строить «{obj_class}»: рынок недонасыщен на целевом горизонте."
        return f"Рассмотреть «{obj_class}» как целевой класс продукта."
    if deficit_index is not None:
        if strong:
            return "Локация недонасыщена — есть сигнал «строить» (класс не уточнён)."
        if deficit_index < 0:
            return "Локация затоварена — выход рискован (класс не уточнён)."
        return "Рынок локации сбалансирован (класс не уточнён)."
    return "Данных недостаточно для однозначного вердикта по продукту."


def _exec_verdict(
    deficit_index: float | None,
    months_of_inventory: float | None,
    overall: float | None,
    level: ReportConfidenceLevel | None,
) -> str | None:
    """Развёрнутый абзац-резюме из дефицита, MOI, overall-скора и уверенности. PURE.

    Шаблонная RU-фраза; ни одного сигнала → None (graceful — вердикт опционален).
    months_of_inventory — ДИСКРИМИНИРУЮЩИЙ headline (интерпретируемо «≈N мес
    конкурирующего предложения»), рядом с АБСОЛЮТНЫМ индексом дефицита.
    """
    parts: list[str] = []
    if deficit_index is not None:
        parts.append(f"индекс дефицита {_signed(deficit_index)} на целевом горизонте")
    if months_of_inventory is not None:
        parts.append(f"≈{round(months_of_inventory, 1)} мес конкурирующего предложения")
    if overall is not None:
        parts.append(f"итоговый продуктовый скор {round(overall, 2)}")
    if level is not None:
        parts.append(f"уверенность {_LEVEL_RU.get(level, level)}")
    if not parts:
        return None
    return (
        "Сводный советующий вердикт: "
        + "; ".join(parts)
        + ". Оценка advisory — не основание для инвест-решения."
    )


def _exec_key_numbers(
    deficit_index: float | None,
    months_of_inventory: float | None,
    overall: float | None,
    level: ReportConfidenceLevel | None,
) -> dict[str, Any]:
    """Плоские опорные числа exec_summary (JSON-safe). PURE.

    Кладём только заданные сигналы (None опускаем) — частичный набор валиден.
    months_of_inventory — ДИСКРИМИНИРУЮЩИЙ headline рядом с АБСОЛЮТНЫМ дефицитом.
    """
    numbers: dict[str, Any] = {}
    if deficit_index is not None:
        numbers["deficit_index"] = round(deficit_index, 3)
    if months_of_inventory is not None:
        numbers["months_of_inventory"] = round(months_of_inventory, 1)
    if overall is not None:
        numbers["overall_score"] = round(overall, 3)
    if level is not None:
        numbers["confidence"] = level
    return numbers


def _build_meta(
    *,
    cad_num: str | None,
    district: str | None,
    segment: dict[str, Any] | None,
    horizons: Sequence[int],
    advisory: bool,
) -> ReportMeta:
    """§13.8 meta — контекст отчёта. PURE.

    `generated_advisory`/`schema_version` берут дефолты ReportMeta (#987: ВСЕГДА True /
    _SCHEMA_VERSION). `generated_at` НЕ ставим (PURE-маппер не дёргает текущее время —
    детерминизм тестов; вызывающий/экспортёр при желании проставит).
    """
    return ReportMeta(
        cad_num=cad_num,
        district=district,
        segment=dict(segment) if segment else {},
        horizons=list(horizons),
        generated_advisory=advisory,
    )


# ──────────────────────────────────────────────────────────────────────────────
# Микро-хелперы форматирования (RU-микрокопия). PURE.
# ──────────────────────────────────────────────────────────────────────────────


def _signed(value: float) -> str:
    """'+0.34' / '−0.18' — знак для RU-микрокопии индекса (минус типографский). PURE."""
    return f"+{value:.2f}" if value >= 0 else f"−{abs(value):.2f}"


# ──────────────────────────────────────────────────────────────────────────────
# Публичный API — PURE-сборка отчёта (все входы уже посчитаны вызывающим, без БД).
# ──────────────────────────────────────────────────────────────────────────────


def assemble_report(
    analyze: dict[str, Any],
    *,
    market_metrics: Any = None,
    supply_layers: Any = None,
    forecasts: Sequence[Any] | None = None,
    future_supply: Any = None,
    scenarios: Sequence[Any] | None = None,
    scenarios_collapsed: bool = False,
    scenarios_collapse_reason: str | None = None,
    recommendation_overlay: Any = None,
    product_scores: Any = None,
    special_indices: Any = None,
    segment: dict[str, Any] | None = None,
    cad_num: str | None = None,
    district: str | None = None,
    horizons: Sequence[int] = _DEFAULT_HORIZONS,
    advisory: bool = _ADVISORY,
) -> SiteFinderReport:
    """§13 собрать `SiteFinderReport` (#987) из analyze + advisory-выводов форсайт-стека.

    ЧИСТОЕ МАППИРОВАНИЕ (assembler #988): НЕ ходит в БД, НЕ зовёт §9.x-сервисы — берёт
    уже-посчитанные входы (их считает вызывающий: #961 API / #957 чат) и РАСКЛАДЫВАЕТ
    их по восьми типизированным секциям §13, попутно считая отчётную уверенность через
    `compute_report_confidence` (#990). ДЕТЕРМИНИРОВАННО, БЕЗ LLM (шаблонные RU-строки).
    ЮНИТ-ТЕСТИРУЕТСЯ БЕЗ БД.

    Входы принимаются КАК dataclass-инстансы под-сервисов, ТАК и их `.as_dict()`-словари
    (нормализуются через `_as_dict_or`) — вызывающий волен передать любое.

    Маппинг секций:
      • market_now    ← analyze (конкуренты/цены/абсорбция) + market_metrics + supply_layers.
      • future_market ← forecasts (per-горизонт) + future_supply + будущие конкуренты +
        сводка scenarios.
      • product_tz    ← recommendation_overlay (класс/mix/commercial/USP + §16-причины).
      • scenarios     ← scenarios (три by_scenario).
      • scoring       ← product_scores (10) + special_indices (6) + overall.
      • confidence    ← compute_report_confidence(...) (#990) из извлечённых сигналов
        качества данных + per-service confidence.
      • exec_summary  ← СИНТЕЗ из сильнейших сигналов (дефицит/класс/overall/уверенность).
      • meta          ← cad_num/district/segment/horizons/advisory/schema_version.

    GRACEFUL: любой вход None → его секция частична/пуста — отчёт валиден как частичный
    (#987). НИКОГДА не бросает. `advisory` прокидывается всюду (в т.ч. cap-потолок #990).

    Args:
        analyze: dict вывода analyze_parcel (loosely-typed; читаем defensive `.get()`).
        market_metrics: §9.2 MarketMetrics (или as_dict / None).
        supply_layers: §9.3 supply-слои (или as_dict / None).
        forecasts: per-горизонт DemandSupplyForecast #952 (или as_dict-список / None).
        future_supply: §9.3 FutureSupplyPressure (или as_dict / None).
        scenarios: три ScenarioForecast #984 (или as_dict-список / None).
        scenarios_collapsed: §22 audit-issue #1871 P1 — `compute_scenarios` поймал
            идентичность conservative/aggressive (failed β-gate в §9.6); фронт/
            экспортёры рисуют плашку-объяснение вместо таблицы. Дефолт False.
        scenarios_collapse_reason: RU-объяснение collapse (если есть) — кладётся в
            ReportScenarios.scenarios_collapse_reason. Дефолт None.
        recommendation_overlay: §10 overlay #983 (или as_dict / None).
        product_scores: §14.2 ProductScoreCard #985 (или as_dict / None).
        special_indices: §25 SpecialIndices #986 (или as_dict / None).
        segment: целевой сегмент (оси §9.x) для meta.
        cad_num: кадастровый номер участка для meta.
        district: район для meta.
        horizons: горизонты прогноза (мес) для meta (по умолчанию _DEFAULT_HORIZONS).
        advisory: советующий-флаг (по умолчанию True; прокидывается в #990 cap).

    Returns:
        SiteFinderReport — все восемь секций заполнены настолько, насколько хватило
        входов (частичный отчёт валиден); `advisory` True; `as_dict()` JSON-сериализуем.
    """
    # ── Нормализация входов (dataclass | as_dict | None → JSON-safe dict | None) ──
    market_metrics_d = _as_dict_or(market_metrics)
    supply_layers_d = _as_dict_or(supply_layers)
    future_supply_d = _as_dict_or(future_supply)
    overlay_d = _as_dict_or(recommendation_overlay)
    product_scores_d = _as_dict_or(product_scores)
    special_indices_d = _as_dict_or(special_indices)
    forecasts_l = _as_dict_list(forecasts)
    scenarios_l = _as_dict_list(scenarios)

    # Будущие конкуренты — из forecasts (§9.7 top-N на горизонте, если они там есть).
    future_competitors = _future_competitors(forecasts_l)
    scenarios_summary = _scenarios_overall_summary(scenarios_l)

    # ── Секции (каждая graceful: None-входы → частичная/пустая) ─────────────────
    market_now = _build_market_now(analyze, market_metrics_d, supply_layers_d)
    future_market = _build_future_market(
        forecasts_l, future_supply_d, future_competitors, scenarios_summary
    )
    product_tz = _build_product_tz(overlay_d)
    scenarios_section = _build_scenarios(
        scenarios_l,
        collapsed=scenarios_collapsed,
        collapse_reason=scenarios_collapse_reason,
    )
    scoring = _build_scoring(product_scores_d, special_indices_d)
    confidence = _build_confidence(
        analyze=analyze,
        market_metrics=market_metrics_d,
        supply_layers=supply_layers_d,
        future_supply=future_supply_d,
        forecasts=forecasts_l,
        product_scores=product_scores_d,
        special_indices=special_indices_d,
        advisory=advisory,
    )
    exec_summary = _build_exec_summary(
        forecasts=forecasts_l,
        product_tz=overlay_d,
        product_scores=product_scores_d,
        confidence=confidence,
    )
    meta = _build_meta(
        cad_num=cad_num,
        district=district,
        segment=segment,
        horizons=horizons,
        advisory=advisory,
    )

    logger.info(
        "assemble_report: cad_num=%s district=%s horizons=%s confidence=%s "
        "forecasts=%d scenarios=%d advisory=%s",
        cad_num,
        district,
        list(horizons),
        confidence.level,
        len(forecasts_l),
        len(scenarios_l),
        advisory,
    )

    return SiteFinderReport(
        exec_summary=exec_summary,
        market_now=market_now,
        future_market=future_market,
        product_tz=product_tz,
        scenarios=scenarios_section,
        scoring=scoring,
        confidence=confidence,
        meta=meta,
        advisory=advisory,
    )


def _future_competitors(forecasts: Sequence[dict[str, Any]]) -> list[dict[str, Any]]:
    """Будущие конкуренты основного горизонта из forecasts (§9.7 top-N). PURE.

    Берём `future_competitors` основного горизонта (_PRIMARY_HORIZON_MONTHS) или первого
    forecast, где список не пуст. Нет → [] (graceful).
    """
    primary = next(
        (f for f in forecasts if f.get("horizon_months") == _PRIMARY_HORIZON_MONTHS),
        None,
    )
    if primary is not None:
        comps = primary.get("future_competitors")
        if isinstance(comps, list) and comps:
            return [c for c in comps if isinstance(c, dict)]
    for f in forecasts:
        comps = f.get("future_competitors")
        if isinstance(comps, list) and comps:
            return [c for c in comps if isinstance(c, dict)]
    return []


def _scenarios_overall_summary(scenarios: Sequence[dict[str, Any]]) -> dict[str, Any] | None:
    """Компактная сводка сценариев для future_market (дефицит осн. горизонта). PURE.

    Из каждого ScenarioForecast (#984) достаём deficit_index основного горизонта →
    {scenario: deficit_index}. Так future_market.scenarios_summary несёт разброс
    дефицита по конверту ставки, не дублируя полную секцию scenarios. Нет → None.
    """
    summary: dict[str, Any] = {}
    for sc in scenarios:
        name = sc.get("scenario")
        sc_forecasts = sc.get("forecasts")
        if not isinstance(name, str) or not isinstance(sc_forecasts, list):
            continue
        di = _primary_deficit_index([f for f in sc_forecasts if isinstance(f, dict)])
        if di is not None:
            summary[name] = round(di, 3)
    return summary or None