feat(forecasting): §10.2/10.4/10.5 + §16 reasons on forecast-overlay (#983, 953-B)

Additively extend #982 build_forecast_overlay: class recommendation (§10.2,
strongest aggregate deficit per class), USP-from-deficits (§10.5, top-3),
degraded-honest commercial signal (§10.4, premise_kind=нежилое → available:False
+caveat when thin, never fabricates), and a §16 structural reason
(why/drivers/rejected/what_would_change/confidence/advisory) on every ranked
segment. §16 rejected-alternatives derived from the what_to_build ranking
runner-ups. Deterministic, no LLM, advisory, no new SQL. All #982 keys/behavior
preserved; schema extended additively (RecommendReason + optional fields).
31 new tests (86 total recommendation+velocity green).

backend/app/schemas/recommend.py

@@ -62,13 +62,36 @@ class RecommendComparable(BaseModel):
     buildings_n: int | None = None
 
 
+class RecommendReason(BaseModel):
+    """§16 структурная причина одного советующего пункта overlay (#983, 953-B). ADVISORY.
+
+    Объясняет, ПОЧЕМУ пункт попал в совет, ЧТО на него повлияло и ЧТО его изменило бы:
+      • `why` — RU-предложение (сегмент, deficit_index на горизонте, спрос/предложение).
+      • `drivers` — драйверы сигнала ([{factor, value, direction}]).
+      • `rejected` — отвергнутые альтернативы из РАНКИНГА (runner-up'ы / негатив-дефицит).
+      • `what_would_change` — рычаги §9.x, которые сдвинули бы индекс (ставка/предложение/
+        горизонт), полными RU-предложениями.
+    `advisory` ВСЕГДА True (наследует advisory-статус §9.x). Все поля схемы — для
+    explainability, НЕ основание для инвест-решения. Это «контракт-намёк»: overlay
+    пробрасывает причины как открытые dict'ы (см. RecommendForecastSegment.reason).
+    """
+
+    why: str
+    drivers: list[dict[str, Any]] = []
+    rejected: list[dict[str, Any]] = []
+    what_would_change: list[str] = []
+    confidence: Literal["high", "medium", "low"]
+    advisory: bool
+
+
 class RecommendForecastSegment(BaseModel):
     """Одна ранжированная ячейка §9.7 forecast-overlay (#982). ADVISORY.
 
     `deficit_index`: в demand_supply-режиме ∈ [−1,+1] (>0 строить / <0 затоварка,
     supply-based); в demand_only-режиме ∈ (0,1] = ПРОКСИ относительной силы спроса
     (NOT supply-based — геометрии участка нет). `balance_units` (demand−supply)
-    None в demand_only (предложение неизмеримо без cad_num).
+    None в demand_only (предложение неизмеримо без cad_num). `reason` (#983, §16) —
+    структурная причина ячейки (RecommendReason-shaped dict) или None.
     """
 
     bucket: str
@@ -76,16 +99,25 @@ class RecommendForecastSegment(BaseModel):
     deficit_index: float
     balance_units: float | None = None
     confidence: Literal["high", "medium", "low"]
+    reason: dict[str, Any] | None = None
 
 
 class RecommendForecastOverlay(BaseModel):
-    """§9.7 СОВЕТУЮЩИЙ forecast-overlay поверх живого микса (#982, 953-A).
+    """§9.7 СОВЕТУЮЩИЙ forecast-overlay поверх живого микса (#982, 953-A; #983, 953-B).
 
     Кладётся ТОЛЬКО под scope["forecast"] при заданном horizon_months; живой микс
     (4 поля RecommendMixOutput) НЕ затрагивается. `advisory` ВСЕГДА True (наследует
     advisory-статус §9.x — не основание для инвест-решения). `mode`: demand_supply
     (cad_num задан — supply/конкуренты учтены) или demand_only (cad_num=None — только
     темп спроса; см. warnings).
+
+    #983 (953-B) — ADDITIVE-расширения поверх #982 (старые поля/поведение НЕ тронуты):
+      • `class_reco` — §10.2 рекомендация класса (сильнейший агрегатный дефицит) + §16
+        reason, либо None (нет ранжированных данных).
+      • `usp` — §10.5 USP-из-дефицитов: top-K самых недообеспеченных сегментов как
+        «дефицит формата X — стройте его» (пустой при отсутствии данных).
+      • `commercial` — §10.4 советующий коммерческий сигнал (доля коммерции) или
+        degraded-honest {available: False, caveat, advisory} при тонких данных, либо None.
     """
 
     horizon_months: int
@@ -93,6 +125,9 @@ class RecommendForecastOverlay(BaseModel):
     advisory: bool
     ranked_segments: list[RecommendForecastSegment]
     warnings: list[str]
+    class_reco: dict[str, Any] | None = None
+    usp: list[dict[str, Any]] = []
+    commercial: dict[str, Any] | None = None
 
 
 class RecommendMixOutput(BaseModel):

backend/app/services/forecasting/recommendation.py

@@ -1,10 +1,17 @@
-"""§9.7 forecast-overlay для живого квартирография-рекомендатора (#982, 953-A).
+"""§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' в
@@ -72,6 +79,34 @@ _CLASS_TO_FORECAST: dict[str, str] = {
     "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."""
@@ -99,6 +134,271 @@ def map_class(target_class: str | None) -> str | 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). Берём только ячейки с deficit_index не None.
+
+    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:
+            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 ──────────────────────
 
 
@@ -200,7 +500,15 @@ def _demand_supply_overlay(
             "demand_supply: ранкинг недоступен (нет геометрии участка или тонкие "
             "данные) — сегменты не ранжированы."
         )
-        return _overlay(horizon_months, "demand_supply", [], warnings)
+        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
@@ -231,7 +539,15 @@ def _demand_supply_overlay(
         horizon_months,
         len(ranked_segments),
     )
-    return _overlay(horizon_months, "demand_supply", ranked_segments, warnings)
+    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(
@@ -273,7 +589,15 @@ def _demand_only_overlay(
         warnings.append(
             "demand-only: нет наблюдаемого темпа продаж (§9.2) — сегменты не ранжированы."
         )
-        return _overlay(horizon_months, "demand_only", [], warnings)
+        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():
@@ -298,7 +622,15 @@ def _demand_only_overlay(
     max_pace = max((p for _, _, p in paces), default=0.0)
     if max_pace <= 0.0:
         warnings.append("demand-only: темп спроса по всем форматам ≤0 — сегменты не ранжированы.")
-        return _overlay(horizon_months, "demand_only", [], warnings)
+        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] (ПРОКСИ, НЕ supply-based).
     paces.sort(key=lambda t: t[2], reverse=True)
@@ -321,7 +653,15 @@ def _demand_only_overlay(
         len(ranked_segments),
         base_pace,
     )
-    return _overlay(horizon_months, "demand_only", ranked_segments, warnings)
+    return _enrich_overlay(
+        db,
+        horizon_months=horizon_months,
+        mode="demand_only",
+        ranked_segments=ranked_segments,
+        warnings=warnings,
+        district=district,
+        demand_only=True,
+    )
 
 
 def _overlay(
@@ -329,12 +669,187 @@ def _overlay(
     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."""
+    """Собрать 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. Graceful.
+
+    Пробует измерить нежилой сток через `compute_market_metrics(premise_kind=
+    "нежилое")`. objective покрывает в основном жильё → выборка обычно тонкая. Тогда
+    возвращаем degraded-honest {available: False, caveat, advisory} — НЕ фабрикуем число.
+    Если данных достаточно (≥ _COMMERCIAL_MIN_LOTS лотов) → советующая оценка доли
+    коммерции (sell_through_pct как прокси реализованной доли) + §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}
+
+    # Достаточно данных: советующая оценка реализованной доли коммерции (прокси).
+    share_pct = round(sell_through, 1)
+    confidence = confidence if confidence in ("high", "medium", "low") else "low"
+    return {
+        "available": True,
+        "premise_kind": _COMMERCIAL_PREMISE_KIND,
+        "commercial_share_pct": share_pct,
+        "n_lots": n_lots,
+        "confidence": confidence,
+        "reason": {
+            "why": (
+                f"Коммерция (нежилое): реализованная доля ~{share_pct}% по {n_lots} "
+                f"лотам на горизонте {horizon_months} мес (прокси спроса на нежилые помещения)."
+            ),
+            "drivers": [
+                {"factor": "sell_through_pct", "value": share_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,
+    )

backend/tests/services/forecasting/test_recommendation.py

@@ -1,4 +1,4 @@
-"""Tests для §9.7 forecast-overlay моста (#982, 953-A) recommendation.py.
+"""Tests для §9.7 forecast-overlay моста (#982, 953-A; #983, 953-B) recommendation.py.
 
 Покрывает:
   • PURE bridge-таблицы map_room_bucket / map_room_bucket_inverse — обе стороны +
@@ -12,6 +12,14 @@
         flips, обязательный warning, balance_units None, supply НЕ фабрикуется.
       - graceful: rank_segments бросает ValueError → пустой ranked_segments + warning,
         НЕ исключение.
+  • #983 (953-B) PURE-билдеры §10/§16:
+      - _build_reason — why/drivers/rejected(из runner-up'ов)/what_would_change shape,
+        confidence inherited, advisory True.
+      - _recommend_class — сильнейший агрегатный дефицит; None на пустом.
+      - _usp_from_deficits — top-K по deficit, RU usp_text.
+      - _commercial_signal — degraded → available False + caveat, НИКОГДА не crash.
+      - build_forecast_overlay теперь несёт class_reco/usp/commercial + per-segment reason,
+        а #982-поведение (ключи/порядок/режимы) НЕ сломано.
 
 Mock-based — живой БД не требуют (db = MagicMock; §9.x вызовы патчатся по месту
 ЛОКАЛЬНОГО импорта внутри функций recommendation.py).
@@ -26,6 +34,11 @@ import pytest
 
 from app.services.forecasting.recommendation import (
     _FORECAST_TO_LIVE_BUCKET,
+    _USP_TOP_K,
+    _build_reason,
+    _commercial_signal,
+    _recommend_class,
+    _usp_from_deficits,
     build_forecast_overlay,
     map_class,
     map_room_bucket,
@@ -479,3 +492,381 @@ class TestDemandOnlyOverlay:
         model = RecommendForecastOverlay.model_validate(out)
         assert model.mode == "demand_only"
         assert model.advisory is True
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# #983 (953-B) PURE-билдеры §10/§16
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _seg(
+    bucket: str,
+    deficit_index: float,
+    *,
+    obj_class: str | None = "комфорт",
+    balance_units: float | None = 12.0,
+    confidence: str = "medium",
+) -> dict[str, Any]:
+    """Live segment-dict (shape _demand_*_overlay) для PURE-билдеров #983."""
+    return {
+        "bucket": bucket,
+        "obj_class": obj_class,
+        "deficit_index": deficit_index,
+        "balance_units": balance_units,
+        "confidence": confidence,
+    }
+
+
+# ── §16: _build_reason ─────────────────────────────────────────────────────────
+
+
+class TestBuildReason:
+    def _ranked(self) -> list[dict[str, Any]]:
+        # Фиксированный ранкинг-фикстура (DESC): топ + runner-up + негатив-дефицит.
+        return [
+            _seg("1-Студия", 0.62, balance_units=62.0, confidence="medium"),
+            _seg("3-2-к", 0.20, obj_class="бизнес", confidence="low"),
+            _seg("4-3-к", -0.18, confidence="medium"),
+        ]
+
+    def test_why_mentions_segment_index_and_horizon(self) -> None:
+        ranked = self._ranked()
+        r = _build_reason(ranked[0], 0.62, ranked, horizon_months=12)
+        assert "1-Студия" in r["why"]
+        assert "+0.62" in r["why"]
+        assert "12 мес" in r["why"]
+
+    def test_drivers_deficit_and_balance_with_direction(self) -> None:
+        ranked = self._ranked()
+        r = _build_reason(ranked[0], 0.62, ranked, horizon_months=12)
+        factors = {d["factor"]: d for d in r["drivers"]}
+        assert factors["deficit_index"]["value"] == 0.62
+        assert factors["deficit_index"]["direction"] == "+"
+        assert factors["balance_units"]["value"] == 62.0
+        assert factors["balance_units"]["direction"] == "+"
+
+    def test_balance_units_omitted_when_none(self) -> None:
+        # demand_only-ячейка (balance_units None) → драйвера balance_units нет.
+        seg = _seg("1-Студия", 0.8, balance_units=None, confidence="low")
+        r = _build_reason(seg, 0.8, [seg], horizon_months=12, demand_only=True)
+        factors = {d["factor"] for d in r["drivers"]}
+        assert "balance_units" not in factors
+        assert "deficit_index" in factors
+
+    def test_rejected_derived_from_runner_ups(self) -> None:
+        # rejected = ВСЕ прочие ячейки ранкинга (runner-up + негатив-дефицит).
+        ranked = self._ranked()
+        r = _build_reason(ranked[0], 0.62, ranked, horizon_months=12)
+        alts = {x["alternative"] for x in r["rejected"]}
+        assert "3-2-к (бизнес)" in alts
+        assert "4-3-к (комфорт)" in alts
+        # Сама выбранная ячейка не попадает в отвергнутые.
+        assert "1-Студия (комфорт)" not in alts
+
+    def test_rejected_labels_overstock_vs_weaker(self) -> None:
+        ranked = self._ranked()
+        r = _build_reason(ranked[0], 0.62, ranked, horizon_months=12)
+        by_alt = {x["alternative"]: x["reason"] for x in r["rejected"]}
+        assert by_alt["3-2-к (бизнес)"] == "слабее сигнал"  # положительный, но ниже
+        assert by_alt["4-3-к (комфорт)"] == "затоварка"  # негатив-дефицит
+
+    def test_what_would_change_has_three_levers_with_horizon(self) -> None:
+        ranked = self._ranked()
+        r = _build_reason(ranked[0], 0.62, ranked, horizon_months=9)
+        wwc = r["what_would_change"]
+        assert len(wwc) == 3
+        assert any("ключевой ставки" in s and "п.п." in s for s in wwc)
+        assert any("Layer2" in s for s in wwc)
+        assert any("9 до 6 мес" in s for s in wwc)  # horizon подставлен
+
+    def test_confidence_inherited_from_segment(self) -> None:
+        seg = _seg("1-Студия", 0.5, confidence="low")
+        r = _build_reason(seg, 0.5, [seg], horizon_months=12)
+        assert r["confidence"] == "low"
+
+    def test_advisory_always_true(self) -> None:
+        seg = _seg("1-Студия", 0.5)
+        r = _build_reason(seg, 0.5, [seg], horizon_months=12)
+        assert r["advisory"] is True
+
+    def test_demand_only_why_notes_supply_excluded(self) -> None:
+        seg = _seg("1-Студия", 0.9, balance_units=None, confidence="low")
+        r = _build_reason(seg, 0.9, [seg], horizon_months=12, demand_only=True)
+        assert "предложение участка НЕ учтено" in r["why"]
+
+
+# ── §10.2: _recommend_class ────────────────────────────────────────────────────
+
+
+class TestRecommendClass:
+    def test_picks_strongest_aggregate(self) -> None:
+        # комфорт: mean(0.6, -0.2)=0.2 ; бизнес: mean(0.5, 0.5)=0.5 → бизнес сильнее.
+        ranked = [
+            _seg("1-Студия", 0.6, obj_class="комфорт"),
+            _seg("3-2-к", -0.2, obj_class="комфорт"),
+            _seg("4-3-к", 0.5, obj_class="бизнес"),
+            _seg("5-80+ м²", 0.5, obj_class="бизнес"),
+        ]
+        out = _recommend_class(ranked, horizon_months=12)
+        assert out is not None
+        assert out["obj_class"] == "бизнес"
+        assert out["mean_deficit_index"] == pytest.approx(0.5)
+        assert out["n_segments"] == 2
+        assert out["reason"]["advisory"] is True
+
+    def test_none_on_empty(self) -> None:
+        assert _recommend_class([], horizon_months=12) is None
+
+    def test_none_when_no_obj_class(self) -> None:
+        # Ни у одной ячейки нет класса — агрегировать нечего → None (НЕ фабрикуем).
+        ranked = [_seg("1-Студия", 0.6, obj_class=None)]
+        assert _recommend_class(ranked, horizon_months=12) is None
+
+    def test_reason_rejected_lists_other_classes(self) -> None:
+        # комфорт (0.6) > бизнес (0.5) → выбран комфорт, отвергнут бизнес.
+        ranked = [
+            _seg("1-Студия", 0.6, obj_class="комфорт"),
+            _seg("4-3-к", 0.5, obj_class="бизнес"),
+        ]
+        out = _recommend_class(ranked, horizon_months=12)
+        assert out is not None
+        assert out["obj_class"] == "комфорт"
+        alts = {x["alternative"] for x in out["reason"]["rejected"]}
+        # Отвергнут другой КЛАСС (агрегатная class-level ячейка), не выбранный.
+        assert any("бизнес" in a for a in alts)
+        assert not any("комфорт" in a for a in alts)
+
+
+# ── §10.5: _usp_from_deficits ──────────────────────────────────────────────────
+
+
+class TestUspFromDeficits:
+    def test_top_k_by_deficit(self) -> None:
+        ranked = [
+            _seg("1-Студия", 0.9),
+            _seg("2-1-к", 0.7),
+            _seg("3-2-к", 0.5),
+            _seg("4-3-к", 0.3),
+            _seg("5-80+ м²", 0.1),
+        ]
+        usp = _usp_from_deficits(ranked, horizon_months=12, top_k=3)
+        assert len(usp) == 3
+        # Сохранён DESC-порядок верхушки.
+        assert [u["segment"] for u in usp] == ["1-Студия", "2-1-к", "3-2-к"]
+
+    def test_default_top_k_is_module_const(self) -> None:
+        ranked = [_seg(f"b{i}", 1.0 - i * 0.1) for i in range(10)]
+        usp = _usp_from_deficits(ranked, horizon_months=12)
+        assert len(usp) == _USP_TOP_K
+
+    def test_usp_text_is_russian_deficit_phrasing(self) -> None:
+        usp = _usp_from_deficits([_seg("1-Студия", 0.9)], horizon_months=12)
+        assert usp[0]["usp_text"] == "Дефицит формата «1-Студия (комфорт)» — стройте его."
+
+    def test_each_usp_carries_reason(self) -> None:
+        usp = _usp_from_deficits([_seg("1-Студия", 0.9)], horizon_months=12)
+        assert usp[0]["reason"]["advisory"] is True
+        assert "deficit_index" in {d["factor"] for d in usp[0]["reason"]["drivers"]}
+
+    def test_empty_input_empty_list(self) -> None:
+        assert _usp_from_deficits([], horizon_months=12) == []
+
+
+# ── §10.4: _commercial_signal (degraded-honest, never crash) ──────────────────
+
+
+def _mk_commercial_metrics(
+    n_lots: int, sell_through_pct: float | None, confidence: str = "medium"
+) -> MagicMock:
+    m = MagicMock()
+    m.n_lots = n_lots
+    m.sell_through_pct = sell_through_pct
+    m.confidence = confidence
+    return m
+
+
+class TestCommercialSignal:
+    def test_thin_data_degraded_with_caveat(self) -> None:
+        # Мало нежилых лотов → degraded-honest, БЕЗ фабрикации доли.
+        with patch(_METRICS, return_value=_mk_commercial_metrics(5, None)):
+            out = _commercial_signal(MagicMock(), "Ленинский", 12)
+        assert out is not None
+        assert out["available"] is False
+        assert "нет достаточных данных" in out["caveat"]
+        assert "нежилое" in out["caveat"]
+        assert out["advisory"] is True
+
+    def test_engine_exception_degraded_never_crash(self) -> None:
+        # Движок бросает (не поддерживает commercial premise_kind) → честный degrade.
+        with patch(_METRICS, side_effect=RuntimeError("no premise")):
+            out = _commercial_signal(MagicMock(), "Ленинский", 12)
+        assert out is not None
+        assert out["available"] is False
+        assert out["advisory"] is True
+
+    def test_magicmock_metrics_does_not_crash(self) -> None:
+        # Нечисловой n_lots (MagicMock-атрибут) НЕ должен бросать TypeError.
+        with patch(_METRICS, return_value=MagicMock()):
+            out = _commercial_signal(MagicMock(), "Ленинский", 12)
+        assert out is not None
+        assert out["available"] is False  # нечисловой → недостаток данных
+
+    def test_sufficient_data_advisory_share(self) -> None:
+        # Достаточно лотов + измеримая доля → советующая оценка + reason (НЕ crash).
+        with patch(_METRICS, return_value=_mk_commercial_metrics(120, 42.5, "medium")):
+            out = _commercial_signal(MagicMock(), "Ленинский", 12)
+        assert out is not None
+        assert out["available"] is True
+        assert out["commercial_share_pct"] == 42.5
+        assert out["n_lots"] == 120
+        assert out["confidence"] == "medium"
+        assert out["reason"]["advisory"] is True
+        assert out["advisory"] is True
+
+
+# ── build_forecast_overlay: #983 ADDITIVE-расширения присутствуют ─────────────
+
+
+class TestOverlayForecast983Additions:
+    def _ranking(self) -> WhatToBuildRanking:
+        return _ranking(
+            [
+                _ranked(room_bucket="Студии 15-30", deficit_index=0.6, obj_class="комфорт"),
+                _ranked(room_bucket="2-к 45-60", deficit_index=0.2, obj_class="бизнес"),
+                _ranked(room_bucket="3-к 60-80", deficit_index=-0.3, obj_class="комфорт"),
+            ]
+        )
+
+    def test_overlay_has_new_keys(self) -> None:
+        with patch(_RANK, return_value=self._ranking()):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        assert "class_reco" in out
+        assert "usp" in out
+        assert "commercial" in out
+
+    def test_each_ranked_segment_has_reason(self) -> None:
+        with patch(_RANK, return_value=self._ranking()):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        assert out["ranked_segments"]
+        for seg in out["ranked_segments"]:
+            assert seg["reason"]["advisory"] is True
+            assert "why" in seg["reason"]
+            assert "rejected" in seg["reason"]
+
+    def test_class_reco_present_with_ranked_data(self) -> None:
+        with patch(_RANK, return_value=self._ranking()):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        assert out["class_reco"] is not None
+        assert out["class_reco"]["obj_class"] in ("комфорт", "бизнес")
+
+    def test_usp_top_k_with_ranked_data(self) -> None:
+        with patch(_RANK, return_value=self._ranking()):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        assert 1 <= len(out["usp"]) <= _USP_TOP_K
+        assert out["usp"][0]["segment"] == "1-Студия"  # сильнейший дефицит
+
+    def test_commercial_degraded_on_mock_db(self) -> None:
+        # MagicMock-db → нет реальных нежилых лотов → degraded-honest commercial.
+        with patch(_RANK, return_value=self._ranking()):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        assert out["commercial"]["available"] is False
+        assert out["commercial"]["advisory"] is True
+
+    def test_empty_ranking_class_reco_none_usp_empty(self) -> None:
+        # Пустой ранкинг → class_reco None, usp [], но overlay не падает + commercial есть.
+        with patch(_RANK, return_value=_ranking([])):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        assert out["class_reco"] is None
+        assert out["usp"] == []
+        assert out["commercial"] is not None
+
+    def test_982_keys_still_present_and_unchanged(self) -> None:
+        # #982-инвариант: исходные поля overlay на месте + advisory True + mode корректен.
+        with patch(_RANK, return_value=self._ranking()):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        for key in ("horizon_months", "mode", "advisory", "ranked_segments", "warnings"):
+            assert key in out
+        assert out["mode"] == "demand_supply"
+        assert out["advisory"] is True
+        assert out["horizon_months"] == 12
+
+    def test_demand_only_reasons_note_supply_excluded(self) -> None:
+        # demand_only: per-segment reason оговаривает, что предложение не учтено.
+        with (
+            patch(_METRICS, return_value=_mk_metrics(4.0)),
+            patch(_GET_MACRO, return_value=[]),
+            patch(_HOLD, return_value={12: 18.0}),
+            patch(_NORM, return_value=_mk_coef(1.0)),
+            patch(_MACRO, return_value=_mk_coef(1.0)),
+        ):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num=None,
+                horizon_months=12,
+                target_class=None,
+            )
+        assert out["ranked_segments"]
+        assert all(
+            "предложение участка НЕ учтено" in s["reason"]["why"]
+            for s in out["ranked_segments"]
+        )
+
+    def test_overlay_with_983_fields_validates_against_schema(self) -> None:
+        from app.schemas.recommend import RecommendForecastOverlay
+
+        with patch(_RANK, return_value=self._ranking()):
+            out: dict[str, Any] = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        model = RecommendForecastOverlay.model_validate(out)
+        assert model.class_reco is not None
+        assert model.usp
+        assert model.commercial is not None
+        assert model.ranked_segments[0].reason is not None