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

"""§9.7 forecast-overlay для живого квартирография-рекомендатора (#982, 953-A; #983, 953-B).

Это ТОНКИЙ СОВЕТУЮЩИЙ мост между ЖИВЫМ `analytics_queries.recommend_mix` и
форсайт-движком §9.x (#980/#981). Своей §9.x-математики НЕ пересобирает — собирает
ответ из уже-смерженных сервисов и форматирует под schemas.recommend overlay-контракт.
Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM; своего SQL НЕТ.

#983 (953-B, EPIC 9) ADDITIVE-расширяет overlay поверх #982 (старые поля/поведение
НЕ тронуты): §10.2 рекомендация класса (`class_reco`), §10.5 USP-из-дефицитов (`usp`),
§10.4 советующий коммерческий сигнал (`commercial`, degraded-honest) и §16 структурная
причина (`reason`) на КАЖДОМ ранжированном сегменте + на class_reco/usp. §16-причины
ДЕРИВИРУЮТСЯ из ранкинга what_to_build (отвергнутые альтернативы = runner-up'ы /
негатив-дефицит ячейки) — поэтому маршрут через rank_segments корректен и для §16.

ADVISORY (КРИТИЧНО — честность, зеркало what_to_build.py): overlay наследует
advisory-статус §9.x (движок не провалидирован до бэктеста #951). Поэтому
`advisory` ВСЕГДА True, confidence ≤ 'medium' в demand_supply-режиме и 'low' в
demand_only. Цифры — для explainability/прототипа, НЕ основание для инвест-решения.
LIVE-ENDPOINT-SAFE: НИКОГДА не бросает — на любом тонком/сбойном входе возвращает
пустой ranked_segments + warning (вызывающий в recommend_mix ещё и оборачивает всё
в try/except, но мы и сами graceful).

ДВА РЕЖИМА (выбор по наличию геометрии участка `cad_num`):
  • mode="demand_supply" (cad_num задан): полноценный §9.7-ранкер
    `what_to_build.rank_segments` — спрос (§9.4ק9.5) vs предложение (§9.3) +
    конкуренты участка → `deficit_index` ∈ [−1,+1] (>0 строить / <0 затоварка),
    `balance_units`, confidence ≤ 'medium'. DESC-порядок наследуется от ранкера.
  • mode="demand_only" (cad_num=None): БЕЗ геометрии участка supply/конкуренты
    НЕизмеримы → считаем ТОЛЬКО темп спроса (§9.2 × §9.4 × §9.5) per default
    room-bucket, ранжируем DESC по темпу. `deficit_index` здесь = pace/max_pace ∈
    (0,1]; `balance_units=None`, confidence='low', обязательный warning что supply
    НЕ учтён. НИКОГДА не фабрикуем предложение/конкурентов из воздуха.
    ВАЖНО (#1593): сегодня §9.2 unit_velocity (base_pace) и §9.4 norm.coefficient
    ОДИНАКОВЫ для всех 5 room-bucket'ов (market_metrics.compute_market_metrics не
    принимает room_bucket, а §9.4-β фитится district×obj_class без room_bucket),
    поэтому оба постоянных множителя СОКРАЩАЮТСЯ в pace/max_pace и порядок форматов
    определяется ИСКЛЮЧИТЕЛЬНО §9.5 macro_coef = segment_steepness(room_bucket).
    Т.е. фактический deficit_index в demand_only = ПРОКСИ rate-крутизны нарезки, а
    НЕ относительной силы спроса. Честный per-bucket §9.2/§9.4-сигнал требует
    room_bucket-параметра в market_metrics + §9.4/§9.6 (другие файлы + доменное
    решение, как мерить per-bucket спрос) — см. #1593.

ИМПОРТЫ §9.x — ЛОКАЛЬНЫЕ (внутри функций), чтобы избежать import-cycle: пакет
`forecasting/__init__` тянет `affordability`, который импортит
`analytics_queries._current_mortgage_rate`, а тот — часть живого стека. Top-level
здесь — только `sales_series.SegmentSpec` (лёгкий, без обратных зависимостей) +
stdlib/sqlalchemy.
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy.orm import Session

from app.services.forecasting.sales_series import SegmentSpec

logger = logging.getLogger(__name__)

# ── PURE bridge-таблицы (named, unit-тестируемы в обе стороны) ─────────────────

# Forecast room-bucket (Source-B room_area-вокабуляр what_to_build) → live
# bucket-id (ключ analytics_queries._BUCKET_PRETTY). ТОЧНЫЙ инверс _BUCKET_PRETTY;
# 5 литералов ДУБЛИРУЕМ намеренно (избегаем import-цикла analytics_queries →
# recommendation). Drift-guard в тестах сверяет это с inverse(_BUCKET_PRETTY).
_FORECAST_TO_LIVE_BUCKET: dict[str, str] = {
    "Студии 15-30": "1-Студия",
    "1-к 30-45": "2-1-к",
    "2-к 45-60": "3-2-к",
    "3-к 60-80": "4-3-к",
    "80+ м²": "5-80+ м²",
}

# Инверс — live bucket-id → forecast room-bucket (для demand_only прохода по
# дефолтной сетке room-bucket'ов §9.7 и для map_room_bucket_inverse).
_LIVE_TO_FORECAST_BUCKET: dict[str, str] = {v: k for k, v in _FORECAST_TO_LIVE_BUCKET.items()}

# Live target_class (вокабуляр schemas.recommend ClassLiteral, «человеческий»
# регистр) → forecast obj_class (3 массовых класса §9.7 _DEFAULT_CLASSES). None и
# незнакомое → None (вызывающий тогда отдаёт движку дефолтную сетку классов).
_CLASS_TO_FORECAST: dict[str, str] = {
    "Comfort": "комфорт",
    "Comfort+": "комфорт",
    "Business": "бизнес",
    "Elite": "бизнес",
    "Economy": "эконом",
}

# ── #983 named-константы (§10/§16) ─────────────────────────────────────────────

# §10.5 USP-из-дефицитов: сколько самых недообеспеченных сегментов превращаем в
# USP-пункты «дефицит формата X — стройте его». 3 = верхушка сигнала, не «простыня».
_USP_TOP_K: int = 3

# §10.4 коммерция: premise_kind, которым пробуем измерить нежилой сток (вокабуляр
# objective_lots — «нежилое»). objective покрывает в основном жильё, поэтому обычно
# выборка тонкая → degraded-honest путь (см. _commercial_signal). НЕ фабрикуем число.
_COMMERCIAL_PREMISE_KIND: str = "нежилое"

# §10.4: минимум нежилых лотов в выборке, ниже которого сигнал считаем недостаточным
# (degraded-honest). Зеркало духа market_metrics порогов: мало лотов → ненадёжно.
_COMMERCIAL_MIN_LOTS: int = 30

# §16 «что изменило бы индекс» — рычаги §9.x полными RU-предложениями (микрокопия:
# п.п., ≥). Templated-константы (НЕ LLM): ставка §9.4/§9.6, скрытый запас §9.3 Layer2,
# горизонт. Шаблон с {horizon} подставляет текущий горизонт вызова.
_LEVER_RATE: str = (
    "Рост ключевой ставки на ≥1 п.п. снижает нормализованный спрос (§9.4) → индекс падает."
)
_LEVER_SUPPLY: str = (
    "Выход скрытого запаса (Layer2, §9.3) в радиусе участка → предложение растёт → индекс падает."
)
_LEVER_HORIZON_TMPL: str = (
    "Сужение горизонта с {horizon} до 6 мес → меньше поглощённого спроса → индекс может вырасти."
)


def map_room_bucket(forecast_bucket: str | None) -> str | None:
    """Forecast room-bucket → live bucket-id. PURE. Неизвестное/None → None."""
    if forecast_bucket is None:
        return None
    return _FORECAST_TO_LIVE_BUCKET.get(forecast_bucket)


def map_room_bucket_inverse(live_bucket: str | None) -> str | None:
    """Live bucket-id → forecast room-bucket (инверс map_room_bucket). PURE."""
    if live_bucket is None:
        return None
    return _LIVE_TO_FORECAST_BUCKET.get(live_bucket)


def map_class(target_class: str | None) -> str | None:
    """Live target_class → forecast obj_class. PURE. None/неизвестное → None.

    Comfort/Comfort+ сворачиваются в «комфорт», Business/Elite — в «бизнес»,
    Economy — в «эконом» (3 массовых класса §9.7). None → None = «без фильтра
    класса», вызывающий отдаёт движку дефолтную сетку классов.
    """
    if target_class is None:
        return None
    return _CLASS_TO_FORECAST.get(target_class)


# ── #983 PURE-билдеры §10/§16 — без БД, полностью юнит-тестируемы ──────────────
# Работают на live segment-dict'ах (shape _demand_*_overlay: bucket / obj_class /
# deficit_index / balance_units / confidence). `all_ranked` — DESC-проекция ранкинга
# what_to_build; из неё §16 деривирует отвергнутые альтернативы (runner-up'ы).


def _segment_label(segment: dict[str, Any]) -> str:
    """Человекочитаемая метка ячейки для RU-микрокопии §16/§10. PURE.

    «<bucket> (<класс>)» если класс задан, иначе «<bucket>». bucket — уже живой
    pretty-id (например «3-2-к»); None-bucket → «формат» (не падаем на мусоре).
    """
    bucket = segment.get("bucket") or "формат"
    obj_class = segment.get("obj_class")
    return f"{bucket} ({obj_class})" if obj_class else str(bucket)


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


def _direction(value: float) -> str:
    """Направление драйвера для §16 ('+' рост сигнала / '−' давление вниз). PURE."""
    return "+" if value >= 0 else "−"


def _build_reason(
    segment: dict[str, Any],
    deficit_index: float,
    all_ranked: list[dict[str, Any]],
    *,
    horizon_months: int,
    demand_only: bool = False,
    demand: float | None = None,
    supply: float | None = None,
) -> dict[str, Any]:
    """§16 структурная причина одной ячейки (RecommendReason-shaped dict). PURE.

    Собирает why/drivers/rejected/what_would_change/confidence/advisory:
      • why — RU-предложение с deficit_index на горизонте (+ спрос/предложение если
        известны; в demand_only оговаривает, что предложение НЕ учтено).
      • drivers — deficit_index (всегда) + balance_units (если не None) с direction.
      • rejected — ДЕРИВИРУЕТСЯ из `all_ranked`: прочие ячейки (runner-up'ы +
        негатив-дефицит) — вот ПОЧЕМУ маршрут через rank_segments корректен для §16.
        Негативный дефицит помечаем «затоварка», прочее — «слабее сигнал».
      • what_would_change — templated рычаги §9.x (ставка/предложение/горизонт).
    `confidence` НАСЛЕДУЕТСЯ из самой ячейки (передаётся в segment). advisory=True.

    Args:
        segment: live segment-dict рассматриваемой ячейки (с confidence/bucket/класс).
        deficit_index: индекс ячейки (для why/drivers — отдельным аргументом, т.к.
            может быть уже округлён вызывающим).
        all_ranked: DESC-проекция всего ранкинга (для деривации rejected).
        horizon_months: горизонт прогноза (мес) — для why и horizon-рычага.
        demand_only: True → why оговаривает, что supply не учтён (demand_only-режим).
        demand: оценка спроса (ед.) если известна — иначе опускаем из why.
        supply: оценка предложения (ед.) если известна — иначе опускаем из why.

    Returns:
        RecommendReason-shaped dict.
    """
    label = _segment_label(segment)
    why = (
        f"Сегмент {label}: deficit_index {_signed(deficit_index)} на горизонте {horizon_months} мес"
    )
    if demand is not None and supply is not None:
        why += f" (спрос ~{demand:.0f}, предложение ~{supply:.0f})"
    elif demand_only:
        why += " (прокси силы спроса; предложение участка НЕ учтено — нет геометрии)"
    why += "."

    drivers: list[dict[str, Any]] = [
        {
            "factor": "deficit_index",
            "value": round(deficit_index, 2),
            "direction": _direction(deficit_index),
        }
    ]
    balance_units = segment.get("balance_units")
    if balance_units is not None:
        drivers.append(
            {
                "factor": "balance_units",
                "value": round(float(balance_units), 1),
                "direction": _direction(float(balance_units)),
            }
        )

    rejected = _rejected_alternatives(segment, all_ranked)

    what_would_change = [
        _LEVER_RATE,
        _LEVER_SUPPLY,
        _LEVER_HORIZON_TMPL.format(horizon=horizon_months),
    ]

    return {
        "why": why,
        "drivers": drivers,
        "rejected": rejected,
        "what_would_change": what_would_change,
        "confidence": segment.get("confidence", "low"),
        "advisory": True,
    }


def _rejected_alternatives(
    chosen: dict[str, Any], all_ranked: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    """§16 отвергнутые альтернативы из РАНКИНГА (всё, кроме выбранной ячейки). PURE.

    Деривируем из `all_ranked` (DESC): каждая ДРУГАЯ ячейка — отвергнутая альтернатива.
    Идентичность ячейки — по (bucket, obj_class) (live-ключ overlay). reason-ярлык:
    deficit_index < 0 → «затоварка», иначе «слабее сигнал». Сохраняем DESC-порядок
    ранкинга (детерминированно). Самоисключение по ключу — НЕ по object identity.
    """
    chosen_key = (chosen.get("bucket"), chosen.get("obj_class"))
    rejected: list[dict[str, Any]] = []
    for seg in all_ranked:
        if (seg.get("bucket"), seg.get("obj_class")) == chosen_key:
            continue
        di = seg.get("deficit_index")
        di_f = float(di) if di is not None else 0.0
        rejected.append(
            {
                "alternative": _segment_label(seg),
                "deficit_index": round(di_f, 2),
                "reason": "затоварка" if di_f < 0 else "слабее сигнал",
            }
        )
    return rejected


def _recommend_class(
    all_ranked: list[dict[str, Any]], *, horizon_months: int, demand_only: bool = False
) -> dict[str, Any] | None:
    """§10.2 рекомендация класса — сильнейший агрегатный дефицит по классам. PURE.

    Агрегируем `deficit_index` per `obj_class` (среднее по его room-bucket'ам),
    выбираем класс с сильнейшим средним дефицитом. К рекомендации прикрепляем §16
    `_build_reason` (как для синтетической class-level ячейки; rejected — прочие
    классы тем же механизмом). None если нет ранжированных данных или ни у одной
    ячейки нет obj_class (нечего агрегировать — НЕ фабрикуем класс).

    Детерминированно: tie-break по obj_class ASC (стабильно при равных средних).

    Args:
        all_ranked: DESC-проекция ранкинга (live segment-dict'ы).
        horizon_months: горизонт (для §16 reason).
        demand_only: проброс в reason (оговорка supply-excluded).

    Returns:
        {obj_class, mean_deficit_index, n_segments, reason} либо None.
    """
    if not all_ranked:
        return None

    by_class: dict[str, list[float]] = {}
    for seg in all_ranked:
        obj_class = seg.get("obj_class")
        di = seg.get("deficit_index")
        if obj_class is None or di is None:
            continue
        by_class.setdefault(obj_class, []).append(float(di))
    if not by_class:
        return None

    means = {cls: sum(vals) / len(vals) for cls, vals in by_class.items()}
    # Сильнейший средний дефицит; tie-break — obj_class ASC (детерминизм).
    best_class = max(means, key=lambda c: (means[c], _neg_str(c)))
    best_mean = means[best_class]

    # Синтетическая class-level ячейка для §16: confidence наследуем как лучший
    # (max) среди ячеек класса — class-сигнал не слабее своей сильнейшей ячейки.
    class_segments = [s for s in all_ranked if s.get("obj_class") == best_class]
    confidence = _best_confidence(s.get("confidence") for s in class_segments)
    synthetic = {"bucket": "класс", "obj_class": best_class, "confidence": confidence}
    # rejected — прочие КЛАССЫ (агрегатные ячейки), не room-bucket'ы.
    class_cells = [
        {"bucket": "класс", "obj_class": cls, "deficit_index": m, "confidence": confidence}
        for cls, m in means.items()
    ]
    reason = _build_reason(
        synthetic,
        best_mean,
        class_cells,
        horizon_months=horizon_months,
        demand_only=demand_only,
    )

    return {
        "obj_class": best_class,
        "mean_deficit_index": round(best_mean, 3),
        "n_segments": len(class_segments),
        "reason": reason,
    }


def _usp_from_deficits(
    all_ranked: list[dict[str, Any]],
    *,
    horizon_months: int,
    top_k: int = _USP_TOP_K,
    demand_only: bool = False,
) -> list[dict[str, Any]]:
    """§10.5 USP-из-дефицитов — top-K самых недообеспеченных сегментов. PURE.

    `all_ranked` уже DESC по deficit_index → первые top_k = сильнейшая недообеспеченность.
    Каждый → USP-пункт «Дефицит формата X — стройте его» + §16 reason. Пустой вход →
    пустой список (НЕ фабрикуем USP). Берём ТОЛЬКО ячейки с genuine-дефицитом
    (deficit_index > 0): «стройте его» оправдан лишь для реально недонасыщенного формата.
    Затоварка (di ≤ 0) — не белое пятно, в USP не идёт; если в top_k все di ≤ 0 → [].
    Зеркало product_scoring._count_positive_usp (тот же gate di > 0).

    Args:
        all_ranked: DESC-проекция ранкинга (live segment-dict'ы).
        horizon_months: горизонт (для §16 reason).
        top_k: сколько верхних сегментов превратить в USP (по умолчанию _USP_TOP_K).
        demand_only: проброс в reason (оговорка supply-excluded).

    Returns:
        Список USP-пунктов {segment, obj_class, deficit_index, usp_text, reason}.
    """
    usp: list[dict[str, Any]] = []
    for seg in all_ranked[:top_k]:
        di = seg.get("deficit_index")
        if di is None or di <= 0:
            continue
        usp.append(
            {
                "segment": seg.get("bucket"),
                "obj_class": seg.get("obj_class"),
                "deficit_index": round(float(di), 3),
                "usp_text": f"Дефицит формата «{_segment_label(seg)}» — стройте его.",
                "reason": _build_reason(
                    seg,
                    float(di),
                    all_ranked,
                    horizon_months=horizon_months,
                    demand_only=demand_only,
                ),
            }
        )
    return usp


def _neg_str(value: str) -> tuple[int, ...]:
    """Ключ для ASC-tie-break внутри max() (max берёт «больший» → инвертируем). PURE.

    max(..., key) выбирает максимум; чтобы при равном primary взять лексикографически
    МЕНЬШИЙ obj_class, отдаём покодовый кортеж с отрицанием (меньшая строка → больший
    ключ). Детерминированно, без локали.
    """
    return tuple(-ord(ch) for ch in value)


def _best_confidence(values: Any) -> str:
    """MAX confidence среди ячеек (class-сигнал не слабее сильнейшей ячейки). PURE.

    Зеркало _CONFIDENCE-vocab (low<medium<high). Пустой/мусорный вход → 'low'.
    """
    rank = {"low": 0, "medium": 1, "high": 2}
    inv = {0: "low", 1: "medium", 2: "high"}
    ranks = [rank[v] for v in values if v in rank]
    return inv[max(ranks)] if ranks else "low"


# ── DB-оркестратор overlay — тонкий, graceful, live-safe ──────────────────────


def build_forecast_overlay(
    db: Session,
    *,
    district: str | None,
    cad_num: str | None,
    horizon_months: int,
    target_class: str | None,
    market_buckets: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Собрать СОВЕТУЮЩИЙ §9.7 forecast-overlay для recommend_mix. LIVE-SAFE.

    ADVISORY (наследует advisory-статус §9.x, см. module docstring) — overlay
    исключительно explainability, НЕ основание для инвест-решения. НИКОГДА не
    бросает: на тонком/сбойном входе → пустой ranked_segments + warning.

    Режим выбирается по `cad_num`:
      • cad_num задан → mode="demand_supply": §9.7 rank_segments (спрос vs
        предложение + конкуренты участка), маппинг RankedSegment → segment-dict,
        DESC-порядок сохранён.
      • cad_num=None → mode="demand_only": ТОЛЬКО темп спроса (§9.2ק9.4ק9.5)
        per default room-bucket, DESC по темпу, deficit_index = pace/max_pace
        (ПРОКСИ, не supply-based), balance_units=None, confidence='low' +
        обязательный warning что supply/конкуренты не учтены.

    Args:
        db: SQLAlchemy sync Session.
        district: район участка (None → ЕКБ-wide).
        cad_num: кадастровый номер участка; None → demand_only режим.
        horizon_months: горизонт прогноза (мес) — продуктовый «на горизонте».
        target_class: живой класс недвижимости (ClassLiteral) или None.
        market_buckets: живые бакеты recommend_mix (пока для совместимости
            сигнатуры; demand_only ходит по дефолтной §9.7 room-сетке).

    Returns:
        RecommendForecastOverlay-shaped dict (horizon_months/mode/advisory/
        ranked_segments/warnings). advisory ВСЕГДА True.
    """
    if cad_num is not None:
        return _demand_supply_overlay(
            db,
            district=district,
            cad_num=cad_num,
            horizon_months=horizon_months,
            target_class=target_class,
        )
    return _demand_only_overlay(
        db,
        district=district,
        horizon_months=horizon_months,
        target_class=target_class,
    )


def _demand_supply_overlay(
    db: Session,
    *,
    district: str | None,
    cad_num: str,
    horizon_months: int,
    target_class: str | None,
) -> dict[str, Any]:
    """mode="demand_supply": §9.7 rank_segments → overlay-dict. Graceful → []."""
    # Локальный импорт — избегаем import-cycle (forecasting/__init__ → affordability
    # → analytics_queries). См. module docstring.
    from app.services.forecasting.what_to_build import rank_segments

    warnings: list[str] = []
    mapped_class = map_class(target_class)
    # None класс → ОПУСКАЕМ kwarg classes, чтобы применилась движковая дефолтная
    # сетка §9.7 (rank_segments default = _DEFAULT_CLASSES); room_buckets всегда
    # опускаем → дефолтная §9.7 сетка форматов. Передать None нельзя — ранкер
    # итерирует по этим осям (None → TypeError).
    rank_kwargs: dict[str, Any] = {}
    if mapped_class is not None:
        rank_kwargs["classes"] = [mapped_class]

    try:
        ranking = rank_segments(
            db,
            district=district,
            cad_num=cad_num,
            horizon_months=horizon_months,
            **rank_kwargs,
        )
    except ValueError as e:
        # Нет геометрии участка / невалидный вход — НЕ валим живой ответ.
        logger.info(
            "forecast-overlay demand_supply: rank_segments ValueError "
            "(district=%s cad_num=%s horizon=%d) → пустой ранкинг: %s",
            district,
            cad_num,
            horizon_months,
            e,
        )
        warnings.append(
            "demand_supply: ранкинг недоступен (нет геометрии участка или тонкие "
            "данные) — сегменты не ранжированы."
        )
        return _enrich_overlay(
            db,
            horizon_months=horizon_months,
            mode="demand_supply",
            ranked_segments=[],
            warnings=warnings,
            district=district,
            demand_only=False,
        )

    ranked_segments: list[dict[str, Any]] = []
    for seg in ranking.ranked:  # уже DESC по deficit_index
        live_bucket = map_room_bucket(seg.segment.get("room_bucket"))
        if live_bucket is None:
            # Сегмент в неизвестном room-вокабуляре — отбрасываем (не отображаем мусор).
            continue
        ranked_segments.append(
            {
                "bucket": live_bucket,
                "obj_class": seg.segment.get("obj_class"),
                "deficit_index": seg.deficit_index,
                "balance_units": seg.balance_units,
                "confidence": seg.confidence,
            }
        )

    if not ranked_segments:
        warnings.append(
            "demand_supply: все ячейки сетки тонкие (предложение неизмеримо) — "
            "сегменты не ранжированы."
        )

    logger.info(
        "forecast-overlay demand_supply: district=%s cad_num=%s horizon=%d ranked=%d (ADVISORY)",
        district,
        cad_num,
        horizon_months,
        len(ranked_segments),
    )
    return _enrich_overlay(
        db,
        horizon_months=horizon_months,
        mode="demand_supply",
        ranked_segments=ranked_segments,
        warnings=warnings,
        district=district,
        demand_only=False,
    )


def _demand_only_overlay(
    db: Session,
    *,
    district: str | None,
    horizon_months: int,
    target_class: str | None,
) -> dict[str, Any]:
    """mode="demand_only" (cad_num=None): ТОЛЬКО темп спроса per room-bucket.

    БЕЗ геометрии участка supply/конкуренты НЕизмеримы. Считаем темп спроса
    pace = §9.2 unit_velocity × §9.4 norm.coefficient × §9.5 macro.coefficient
    per дефолтный room-bucket, ранжируем DESC. deficit_index = pace/max_pace ∈
    (0,1]. balance_units=None, confidence='low', обязательный warning. НИКОГДА не
    фабрикуем supply. Graceful → [].

    ВАЖНО (#1593, честность ранкинга): §9.2 base_pace и §9.4 norm.coefficient
    сейчас ОДИНАКОВЫ для всех bucket'ов (нет room_bucket-фильтра в
    market_metrics.compute_market_metrics; §9.4-β фитится district×obj_class без
    room_bucket) и СОКРАЩАЮТСЯ в pace/max_pace. Поэтому фактический порядок
    форматов и deficit_index определяются ИСКЛЮЧИТЕЛЬНО §9.5 macro_coef =
    segment_steepness(room_bucket) — это ПРОКСИ rate-крутизны нарезки, НЕ силы
    спроса. Полноценный per-bucket §9.2/§9.4-сигнал = другие файлы + доменное
    решение (#1593); до тех пор не интерпретировать как «самый дефицитный спрос».
    """
    # Локальные импорты — избегаем import-cycle (см. module docstring).
    from app.services.forecasting.demand_normalization import compute_demand_normalization
    from app.services.forecasting.demand_supply_forecast import hold_last_rate
    from app.services.forecasting.macro_coefficient import compute_macro_coefficient
    from app.services.forecasting.macro_series import get_monthly_macro
    from app.services.site_finder.market_metrics import compute_market_metrics

    warnings: list[str] = [
        "demand-only: без геометрии участка (cad_num) supply/конкуренты НЕ учтены"
    ]
    mapped_class = map_class(target_class)

    # Один раз на вызов: §9.2 наблюдаемый темп (base_pace) + hold-last-rate для §9.4.
    metrics = compute_market_metrics(db, district=district)
    base_pace = metrics.unit_velocity
    macro = get_monthly_macro(db)
    rate_future = hold_last_rate(macro, [horizon_months]).get(horizon_months)

    if base_pace is None:
        # Нет наблюдаемого темпа — ранжировать нечего (НЕ фабрикуем 0-сигнал).
        warnings.append(
            "demand-only: нет наблюдаемого темпа продаж (§9.2) — сегменты не ранжированы."
        )
        return _enrich_overlay(
            db,
            horizon_months=horizon_months,
            mode="demand_only",
            ranked_segments=[],
            warnings=warnings,
            district=district,
            demand_only=True,
        )

    paces: list[tuple[str, str | None, float]] = []  # (live_bucket, obj_class, pace)
    for live_bucket, forecast_bucket in _LIVE_TO_FORECAST_BUCKET.items():
        spec = SegmentSpec(
            obj_class=mapped_class,
            room_bucket=forecast_bucket,
            district=district,
        )
        # §9.4 нормализация под будущий режим ставки (β внутри; rate_future None →
        # деградирует к нейтрали внутри себя, передаём 0.0 как placeholder).
        norm = compute_demand_normalization(
            db, spec=spec, rate_future=rate_future if rate_future is not None else 0.0
        )
        # §9.5 макро-коэффициент (ортогонален β); профиль — класс + room_bucket.
        profile: dict[str, Any] = {"room_bucket": forecast_bucket}
        if mapped_class is not None:
            profile["obj_class"] = mapped_class
        macro_coef = compute_macro_coefficient(db, segment_profile=profile)
        pace = base_pace * norm.coefficient * macro_coef.coefficient
        paces.append((live_bucket, mapped_class, pace))

    max_pace = max((p for _, _, p in paces), default=0.0)
    if max_pace <= 0.0:
        warnings.append("demand-only: темп спроса по всем форматам ≤0 — сегменты не ранжированы.")
        return _enrich_overlay(
            db,
            horizon_months=horizon_months,
            mode="demand_only",
            ranked_segments=[],
            warnings=warnings,
            district=district,
            demand_only=True,
        )

    # DESC по темпу; deficit_index = pace/max_pace ∈ (0,1]. NB #1593: base_pace и
    # §9.4 norm одинаковы для всех bucket'ов и сокращаются → порядок = §9.5
    # segment_steepness (rate-крутизна), НЕ сила спроса. Per-bucket §9.2/§9.4 = TODO.
    paces.sort(key=lambda t: t[2], reverse=True)
    ranked_segments: list[dict[str, Any]] = [
        {
            "bucket": live_bucket,
            "obj_class": obj_class,
            "deficit_index": pace / max_pace,
            "balance_units": None,  # supply неизмеримо без cad_num — НЕ фабрикуем
            "confidence": "low",
        }
        for live_bucket, obj_class, pace in paces
    ]

    logger.info(
        "forecast-overlay demand_only: district=%s horizon=%d ranked=%d base_pace=%.3f "
        "(ADVISORY, supply НЕ учтён)",
        district,
        horizon_months,
        len(ranked_segments),
        base_pace,
    )
    return _enrich_overlay(
        db,
        horizon_months=horizon_months,
        mode="demand_only",
        ranked_segments=ranked_segments,
        warnings=warnings,
        district=district,
        demand_only=True,
    )


def _overlay(
    horizon_months: int,
    mode: str,
    ranked_segments: list[dict[str, Any]],
    warnings: list[str],
    *,
    class_reco: dict[str, Any] | None = None,
    usp: list[dict[str, Any]] | None = None,
    commercial: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Собрать RecommendForecastOverlay-shaped dict. advisory ВСЕГДА True. PURE.

    #982-поля (horizon_months/mode/advisory/ranked_segments/warnings) — без изменений.
    #983 ADDITIVE-поля (class_reco §10.2 / usp §10.5 / commercial §10.4) — опциональны,
    дефолты сохраняют #982-байт-в-байт-форму при пустом расширении (None/[]).
    """
    return {
        "horizon_months": horizon_months,
        "mode": mode,
        "advisory": True,
        "ranked_segments": ranked_segments,
        "warnings": warnings,
        "class_reco": class_reco,
        "usp": usp if usp is not None else [],
        "commercial": commercial,
    }


def _commercial_signal(
    db: Session, district: str | None, horizon_months: int
) -> dict[str, Any] | None:
    """§10.4 советующий коммерческий сигнал (темп распродажи нежилого) — degraded-honest.

    Пробует измерить нежилой сток через `compute_market_metrics(premise_kind=
    "нежилое")`. objective покрывает в основном жильё → выборка обычно тонкая. Тогда
    возвращаем degraded-honest {available: False, caveat, advisory} — НЕ фабрикуем число.
    Если данных достаточно (≥ _COMMERCIAL_MIN_LOTS лотов) → советующая оценка ТЕМПА
    РАСПРОДАЖИ нежилого: sell_through_pct = проданные ÷ (проданные+доступные)·100
    (market_metrics §-определение) — прокси ликвидности/спроса на нежилое, НЕ доля
    нежилого в объёме застройки + §16-подобный reason.
    НИКОГДА не бросает: любой сбой движка/импорта → degraded-honest None-сигнал.

    Args:
        db: SQLAlchemy sync Session.
        district: район (None → ЕКБ-wide).
        horizon_months: горизонт (для контекста caveat/reason).

    Returns:
        Советующий dict {available, ...} либо None при полном отсутствии движка.
    """
    # Локальный импорт — зеркало остального модуля (избегаем import-cycle).
    from app.services.site_finder.market_metrics import compute_market_metrics

    caveat = (
        f"коммерция: нет достаточных данных (premise_kind={_COMMERCIAL_PREMISE_KIND}, "
        f"горизонт {horizon_months} мес)"
    )
    try:
        metrics = compute_market_metrics(
            db, district=district, premise_kind=_COMMERCIAL_PREMISE_KIND
        )
        n_lots = _as_int(getattr(metrics, "n_lots", None))
        sell_through = _as_float(getattr(metrics, "sell_through_pct", None))
        confidence = getattr(metrics, "confidence", "low")
    except Exception:
        # Движок не поддерживает коммерческий premise_kind / сбой — честный degrade.
        logger.exception(
            "forecast-overlay commercial: market_metrics failed (district=%s) → degraded",
            district,
        )
        return {"available": False, "caveat": caveat, "advisory": True}

    # Тонкая выборка (objective ~ жильё) / неизмеримая доля → degraded-honest, без
    # фабрикации числа. n_lots None (нечисловой/сбой) тоже трактуем как недостаток.
    if n_lots is None or n_lots < _COMMERCIAL_MIN_LOTS or sell_through is None:
        logger.info(
            "forecast-overlay commercial: thin (district=%s n_lots=%s) → degraded (ADVISORY)",
            district,
            n_lots,
        )
        return {"available": False, "caveat": caveat, "advisory": True}

    # Достаточно данных: советующая оценка ТЕМПА РАСПРОДАЖИ нежилого (прокси
    # ликвидности/спроса), НЕ доли нежилого в объёме застройки (#1635).
    sell_through_pct = round(sell_through, 1)
    confidence = confidence if confidence in ("high", "medium", "low") else "low"
    return {
        "available": True,
        "premise_kind": _COMMERCIAL_PREMISE_KIND,
        # NB #1635: ключ `commercial_share_pct` — МИСЛЕЙБЛ (величина = темп
        # распродажи sell_through, а не доля нежилого в объёме). Это контракт-ключ,
        # читаемый product_scoring._score_commercial (product_scoring.py:474, чей
        # reason на :480 повторяет тот же мислейбл) — переименование в
        # commercial_sell_through_pct cross-file, координировать с product_scoring.py.
        "commercial_share_pct": sell_through_pct,
        "n_lots": n_lots,
        "confidence": confidence,
        "reason": {
            "why": (
                f"Коммерция (нежилое): темп распродажи ~{sell_through_pct}% по {n_lots} "
                f"лотам на горизонте {horizon_months} мес (прокси ликвидности/спроса на "
                f"нежилые помещения, НЕ доля нежилого в объёме застройки)."
            ),
            "drivers": [
                {"factor": "sell_through_pct", "value": sell_through_pct, "direction": "+"},
                {"factor": "n_lots", "value": n_lots, "direction": "+"},
            ],
            "rejected": [],
            "what_would_change": [_LEVER_SUPPLY],
            "confidence": confidence,
            "advisory": True,
        },
        "advisory": True,
    }


def _as_int(value: Any) -> int | None:
    """Безопасно привести значение к int (None/нечисловое → None). Graceful. PURE.

    Защита §10.4 от мок/мусор-атрибутов (MagicMock < int бросил бы TypeError): любой
    нечисловой/bool/сбойный вход → None → degraded-honest путь, без падения.
    """
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    return int(value)


def _as_float(value: Any) -> float | None:
    """Безопасно привести значение к float (None/нечисловое → None). Graceful. PURE."""
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    return float(value)


def _enrich_overlay(
    db: Session,
    *,
    horizon_months: int,
    mode: str,
    ranked_segments: list[dict[str, Any]],
    warnings: list[str],
    district: str | None,
    demand_only: bool,
) -> dict[str, Any]:
    """#983 ADDITIVE-сборка: §16 reason на каждом сегменте + §10.2/§10.5/§10.4. Graceful.

    Мутирует `ranked_segments` IN-PLACE, добавляя `reason` (§16) к каждой ячейке
    (rejected деривируется из всего списка). Затем считает class_reco (§10.2),
    usp (§10.5) и commercial (§10.4) и собирает overlay через `_overlay`. Расширения
    §10.2/§10.5 только при НАЛИЧИИ ранжированных данных (иначе None/[]); commercial
    (§10.4) пробуем всегда (он независим от ранкинга, сам degraded-honest). Все
    #982-поля сохраняются _overlay'ем без изменений.

    Args:
        db: SQLAlchemy sync Session (для §10.4 коммерции).
        horizon_months: горизонт прогноза (мес).
        mode: "demand_supply" | "demand_only".
        ranked_segments: live segment-dict'ы (мутируются — добавляется reason).
        warnings: #982-варнинги (пробрасываются как есть).
        district: район (для §10.4).
        demand_only: True → reason'ы оговаривают, что supply не учтён.

    Returns:
        RecommendForecastOverlay-shaped dict с #983-расширениями.
    """
    for seg in ranked_segments:
        di = seg.get("deficit_index")
        if di is None:
            continue
        seg["reason"] = _build_reason(
            seg,
            float(di),
            ranked_segments,
            horizon_months=horizon_months,
            demand_only=demand_only,
        )

    class_reco = (
        _recommend_class(ranked_segments, horizon_months=horizon_months, demand_only=demand_only)
        if ranked_segments
        else None
    )
    usp = (
        _usp_from_deficits(ranked_segments, horizon_months=horizon_months, demand_only=demand_only)
        if ranked_segments
        else []
    )
    commercial = _commercial_signal(db, district, horizon_months)

    return _overlay(
        horizon_months,
        mode,
        ranked_segments,
        warnings,
        class_reco=class_reco,
        usp=usp,
        commercial=commercial,
    )