backend/app/api/v1/analytics.py

@@ -254,4 +254,6 @@ def recommend_mix(
         months_window=payload.months_window,
         price_factor=payload.price_factor,
         target_months=payload.target_months,
+        horizon_months=payload.horizon_months,
+        cad_num=payload.cad_num,
     )

backend/app/schemas/recommend.py

@@ -21,6 +21,12 @@ class RecommendMixInput(BaseModel):
     # деления на 0 / pf^elasticity = ∞ при «бесплатной» цене.
     price_factor: float = Field(default=1.0, ge=0.01, le=3.0)
     target_months: int | None = Field(default=None, ge=3, le=120)
+    # #982 (953-A) forecast-overlay opt-in. Оба None по умолчанию → живой микс
+    # БАЙТ-в-БАЙТ как раньше (overlay не добавляется). horizon_months задан →
+    # advisory overlay под scope["forecast"]; cad_num задан → demand_supply-режим
+    # (с supply/конкурентами участка), иначе demand_only (только темп спроса).
+    horizon_months: int | None = Field(default=None, ge=3, le=24)
+    cad_num: str | None = Field(default=None, max_length=40)
 
 
 class RecommendBucket(BaseModel):
@@ -56,6 +62,39 @@ class RecommendComparable(BaseModel):
     buildings_n: int | None = None
 
 
+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).
+    """
+
+    bucket: str
+    obj_class: str | None = None
+    deficit_index: float
+    balance_units: float | None = None
+    confidence: Literal["high", "medium", "low"]
+
+
+class RecommendForecastOverlay(BaseModel):
+    """§9.7 СОВЕТУЮЩИЙ forecast-overlay поверх живого микса (#982, 953-A).
+
+    Кладётся ТОЛЬКО под scope["forecast"] при заданном horizon_months; живой микс
+    (4 поля RecommendMixOutput) НЕ затрагивается. `advisory` ВСЕГДА True (наследует
+    advisory-статус §9.x — не основание для инвест-решения). `mode`: demand_supply
+    (cad_num задан — supply/конкуренты учтены) или demand_only (cad_num=None — только
+    темп спроса; см. warnings).
+    """
+
+    horizon_months: int
+    mode: Literal["demand_supply", "demand_only"]
+    advisory: bool
+    ranked_segments: list[RecommendForecastSegment]
+    warnings: list[str]
+
+
 class RecommendMixOutput(BaseModel):
     scope: dict[str, Any]
     buckets: list[RecommendBucket]

backend/app/services/analytics_queries.py

@@ -2171,6 +2171,8 @@ def recommend_mix(
     region_code: int = 66,
     price_factor: float = 1.0,
     target_months: int | None = None,
+    horizon_months: int | None = None,
+    cad_num: str | None = None,
 ) -> dict[str, Any]:
     """Rule-based квартирография recommender v3.1-v3.4.
 
@@ -2804,7 +2806,7 @@ def recommend_mix(
         headline_parts.append(f"ликвидность {liquidity_24mo:.0f}/100")
     headline = " · ".join(headline_parts) if headline_parts else None
 
-    return {
+    result: dict[str, Any] = {
         "scope": {
             "district": district_row["district_name"],
             "district_zk_count": district_row["zk_count"],
@@ -2911,6 +2913,37 @@ def recommend_mix(
         ],
     }
 
+    # #982 (953-A) ADDITIVE OPT-IN: horizon-aware forecast-overlay. Живой микс выше
+    # БАЙТ-в-БАЙТ не тронут — overlay добавляется ТОЛЬКО при заданном horizon_months
+    # и ТОЛЬКО под scope["forecast"] (открытый dict). КРИТИЧНО (live-endpoint safety):
+    # overlay СОВЕТУЮЩИЙ и НИКОГДА не должен уронить живой ответ — вся ветка в
+    # try/except, на сбое кладём {error, advisory} и всё равно возвращаем микс.
+    # Локальный импорт — избегаем потенциального import-cycle (analytics_queries ↔
+    # forecasting) и не утяжеляем top-level живого analytics-стека advisory-движком.
+    if horizon_months is not None:
+        try:
+            from app.services.forecasting.recommendation import build_forecast_overlay
+
+            result["scope"]["forecast"] = build_forecast_overlay(
+                db,
+                district=district_row["district_name"],
+                cad_num=cad_num,
+                horizon_months=horizon_months,
+                target_class=target_class,
+                market_buckets=buckets,
+            )
+        except Exception as e:
+            logger.exception(
+                "recommend_mix: forecast-overlay failed (district=%s cad_num=%s horizon=%s) "
+                "— живой микс возвращён без overlay",
+                district_row["district_name"],
+                cad_num,
+                horizon_months,
+            )
+            result["scope"]["forecast"] = {"error": str(e), "advisory": True}
+
+    return result
+
 
 # ── Cadastral buildings per complex ──────────────────────────────────────────
 

backend/app/services/forecasting/recommendation.py

@@ -0,0 +1,340 @@
+"""§9.7 forecast-overlay для живого квартирография-рекомендатора (#982, 953-A).
+
+Это ТОНКИЙ СОВЕТУЮЩИЙ мост между ЖИВЫМ `analytics_queries.recommend_mix` и
+форсайт-движком §9.x (#980/#981). Своей §9.x-математики НЕ пересобирает — собирает
+ответ из уже-смерженных сервисов и форматирует под schemas.recommend overlay-контракт.
+Всё ДЕТЕРМИНИРОВАННО, БЕЗ LLM; своего SQL НЕТ.
+
+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] — это ПРОКСИ относительной силы спроса, НЕ supply-based дефицит;
+    `balance_units=None`, confidence='low', обязательный warning что supply НЕ учтён.
+    НИКОГДА не фабрикуем предложение/конкурентов из воздуха.
+
+ИМПОРТЫ §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": "эконом",
+}
+
+
+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)
+
+
+# ── 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 _overlay(horizon_months, "demand_supply", [], warnings)
+
+    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 _overlay(horizon_months, "demand_supply", ranked_segments, warnings)
+
+
+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] — ПРОКСИ относительной силы спроса (НЕ supply-based дефицит).
+    balance_units=None, confidence='low', обязательный warning. НИКОГДА не
+    фабрикуем supply. Graceful → [].
+    """
+    # Локальные импорты — избегаем 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 _overlay(horizon_months, "demand_only", [], warnings)
+
+    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 _overlay(horizon_months, "demand_only", [], warnings)
+
+    # DESC по темпу; deficit_index = pace/max_pace ∈ (0,1] (ПРОКСИ, НЕ supply-based).
+    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 _overlay(horizon_months, "demand_only", ranked_segments, warnings)
+
+
+def _overlay(
+    horizon_months: int,
+    mode: str,
+    ranked_segments: list[dict[str, Any]],
+    warnings: list[str],
+) -> dict[str, Any]:
+    """Собрать RecommendForecastOverlay-shaped dict. advisory ВСЕГДА True. PURE."""
+    return {
+        "horizon_months": horizon_months,
+        "mode": mode,
+        "advisory": True,
+        "ranked_segments": ranked_segments,
+        "warnings": warnings,
+    }

backend/tests/services/forecasting/test_recommendation.py

@@ -0,0 +1,481 @@
+"""Tests для §9.7 forecast-overlay моста (#982, 953-A) recommendation.py.
+
+Покрывает:
+  • PURE bridge-таблицы map_room_bucket / map_room_bucket_inverse — обе стороны +
+    unknown→None + DRIFT-GUARD (сверка _FORECAST_TO_LIVE_BUCKET с инверсом
+    analytics_queries._BUCKET_PRETTY — таблицы НЕ должны разъехаться).
+  • map_class — сворачивание классов + None/unknown→None.
+  • build_forecast_overlay через @patch:
+      - demand_supply (cad_num задан): rank_segments замокан → маппинг RankedSegment →
+        segment-dict, advisory True, DESC-порядок сохранён, unknown room-bucket отброшен.
+      - demand_only (cad_num=None): market_metrics/demand_norm/macro замоканы → mode
+        flips, обязательный warning, balance_units None, supply НЕ фабрикуется.
+      - graceful: rank_segments бросает ValueError → пустой ranked_segments + warning,
+        НЕ исключение.
+
+Mock-based — живой БД не требуют (db = MagicMock; §9.x вызовы патчатся по месту
+ЛОКАЛЬНОГО импорта внутри функций recommendation.py).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.services.forecasting.recommendation import (
+    _FORECAST_TO_LIVE_BUCKET,
+    build_forecast_overlay,
+    map_class,
+    map_room_bucket,
+    map_room_bucket_inverse,
+)
+from app.services.forecasting.what_to_build import RankedSegment, WhatToBuildRanking
+
+# Точки ЛОКАЛЬНОГО импорта внутри функций recommendation.py — патчим источник,
+# чтобы перехватить вызов независимо от того, где импорт исполнится.
+_RANK = "app.services.forecasting.what_to_build.rank_segments"
+_METRICS = "app.services.site_finder.market_metrics.compute_market_metrics"
+_NORM = "app.services.forecasting.demand_normalization.compute_demand_normalization"
+_MACRO = "app.services.forecasting.macro_coefficient.compute_macro_coefficient"
+_GET_MACRO = "app.services.forecasting.macro_series.get_monthly_macro"
+_HOLD = "app.services.forecasting.demand_supply_forecast.hold_last_rate"
+
+
+# ── PURE: map_room_bucket / map_room_bucket_inverse ───────────────────────────
+
+
+class TestMapRoomBucket:
+    def test_forward_all_five(self) -> None:
+        assert map_room_bucket("Студии 15-30") == "1-Студия"
+        assert map_room_bucket("1-к 30-45") == "2-1-к"
+        assert map_room_bucket("2-к 45-60") == "3-2-к"
+        assert map_room_bucket("3-к 60-80") == "4-3-к"
+        assert map_room_bucket("80+ м²") == "5-80+ м²"
+
+    def test_inverse_all_five(self) -> None:
+        assert map_room_bucket_inverse("1-Студия") == "Студии 15-30"
+        assert map_room_bucket_inverse("2-1-к") == "1-к 30-45"
+        assert map_room_bucket_inverse("3-2-к") == "2-к 45-60"
+        assert map_room_bucket_inverse("4-3-к") == "3-к 60-80"
+        assert map_room_bucket_inverse("5-80+ м²") == "80+ м²"
+
+    def test_round_trip_forward_then_inverse(self) -> None:
+        for forecast in _FORECAST_TO_LIVE_BUCKET:
+            assert map_room_bucket_inverse(map_room_bucket(forecast)) == forecast
+
+    def test_unknown_forward_none(self) -> None:
+        assert map_room_bucket("чердак") is None
+
+    def test_unknown_inverse_none(self) -> None:
+        assert map_room_bucket_inverse("9-чердак") is None
+
+    def test_none_passthrough(self) -> None:
+        assert map_room_bucket(None) is None
+        assert map_room_bucket_inverse(None) is None
+
+
+class TestBucketTableDriftGuard:
+    """_FORECAST_TO_LIVE_BUCKET ДОЛЖНА быть точным инверсом analytics_queries._BUCKET_PRETTY.
+
+    Если кто-то поменяет один словарь и забудет другой (дублируем 5 литералов ради
+    разрыва import-цикла) — этот тест падает первым.
+    """
+
+    def test_is_exact_inverse_of_bucket_pretty(self) -> None:
+        from app.services.analytics_queries import _BUCKET_PRETTY
+
+        expected_inverse = {pretty: bid for bid, pretty in _BUCKET_PRETTY.items()}
+        assert _FORECAST_TO_LIVE_BUCKET == expected_inverse
+
+    def test_same_cardinality(self) -> None:
+        from app.services.analytics_queries import _BUCKET_PRETTY
+
+        assert len(_FORECAST_TO_LIVE_BUCKET) == len(_BUCKET_PRETTY) == 5
+
+
+# ── PURE: map_class ───────────────────────────────────────────────────────────
+
+
+class TestMapClass:
+    @pytest.mark.parametrize(
+        ("live", "expected"),
+        [
+            ("Comfort", "комфорт"),
+            ("Comfort+", "комфорт"),
+            ("Business", "бизнес"),
+            ("Elite", "бизнес"),
+            ("Economy", "эконом"),
+        ],
+    )
+    def test_folding(self, live: str, expected: str) -> None:
+        assert map_class(live) == expected
+
+    def test_none_to_none(self) -> None:
+        assert map_class(None) is None
+
+    def test_unknown_to_none(self) -> None:
+        assert map_class("Luxury++") is None
+
+
+# ── Helpers для мок-сегментов demand_supply ───────────────────────────────────
+
+
+def _ranked(
+    *,
+    room_bucket: str | None,
+    deficit_index: float,
+    obj_class: str = "комфорт",
+    balance_units: float | None = 12.0,
+    confidence: str = "medium",
+) -> RankedSegment:
+    return RankedSegment(
+        segment={
+            "obj_class": obj_class,
+            "room_bucket": room_bucket,
+            "district": "Ленинский",
+            "price_bucket": None,
+        },
+        deficit_index=deficit_index,
+        balance_units=balance_units,
+        confidence=confidence,  # type: ignore[arg-type]
+    )
+
+
+def _ranking(ranked: list[RankedSegment]) -> WhatToBuildRanking:
+    return WhatToBuildRanking(
+        district="Ленинский",
+        cad_num="66:41:0000000:1",
+        horizon_months=12,
+        ranked=ranked,
+        n_cells_scanned=len(ranked),
+        n_cells_ranked=len(ranked),
+        generated_advisory=True,
+    )
+
+
+# ── build_forecast_overlay: demand_supply (cad_num задан) ──────────────────────
+
+
+class TestDemandSupplyOverlay:
+    def test_mode_and_advisory(self) -> None:
+        ranking = _ranking([_ranked(room_bucket="2-к 45-60", deficit_index=0.5)])
+        with patch(_RANK, return_value=ranking):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class="Comfort",
+            )
+        assert out["mode"] == "demand_supply"
+        assert out["advisory"] is True
+        assert out["horizon_months"] == 12
+
+    def test_maps_segment_buckets_to_live(self) -> None:
+        ranking = _ranking(
+            [
+                _ranked(room_bucket="Студии 15-30", deficit_index=0.9),
+                _ranked(room_bucket="2-к 45-60", deficit_index=0.3),
+            ]
+        )
+        with patch(_RANK, return_value=ranking):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        buckets = [s["bucket"] for s in out["ranked_segments"]]
+        assert buckets == ["1-Студия", "3-2-к"]
+
+    def test_desc_order_preserved(self) -> None:
+        # rank_segments уже отдаёт DESC; overlay не переупорядочивает.
+        ranking = _ranking(
+            [
+                _ranked(room_bucket="80+ м²", deficit_index=0.8),
+                _ranked(room_bucket="1-к 30-45", deficit_index=0.2),
+                _ranked(room_bucket="3-к 60-80", deficit_index=-0.4),
+            ]
+        )
+        with patch(_RANK, return_value=ranking):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        indices = [s["deficit_index"] for s in out["ranked_segments"]]
+        assert indices == [0.8, 0.2, -0.4]
+
+    def test_fields_passed_through(self) -> None:
+        ranking = _ranking(
+            [
+                _ranked(
+                    room_bucket="2-к 45-60",
+                    deficit_index=0.55,
+                    obj_class="бизнес",
+                    balance_units=33.0,
+                    confidence="medium",
+                )
+            ]
+        )
+        with patch(_RANK, return_value=ranking):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=9,
+                target_class="Business",
+            )
+        seg = out["ranked_segments"][0]
+        assert seg["obj_class"] == "бизнес"
+        assert seg["deficit_index"] == 0.55
+        assert seg["balance_units"] == 33.0
+        assert seg["confidence"] == "medium"
+
+    def test_unknown_room_bucket_dropped(self) -> None:
+        ranking = _ranking(
+            [
+                _ranked(room_bucket="2-к 45-60", deficit_index=0.5),
+                _ranked(room_bucket="мансарда", deficit_index=0.9),  # неизвестный → drop
+            ]
+        )
+        with patch(_RANK, return_value=ranking):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        buckets = [s["bucket"] for s in out["ranked_segments"]]
+        assert buckets == ["3-2-к"]
+
+    def test_class_mapped_into_rank_kwargs(self) -> None:
+        ranking = _ranking([_ranked(room_bucket="2-к 45-60", deficit_index=0.5)])
+        with patch(_RANK, return_value=ranking) as mock_rank:
+            build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class="Elite",
+            )
+        # Elite → бизнес, передаётся ранкеру как classes=["бизнес"].
+        assert mock_rank.call_args.kwargs["classes"] == ["бизнес"]
+
+    def test_no_class_omits_classes_kwarg(self) -> None:
+        # target_class=None → classes НЕ передаётся (движковый дефолт _DEFAULT_CLASSES).
+        ranking = _ranking([_ranked(room_bucket="2-к 45-60", deficit_index=0.5)])
+        with patch(_RANK, return_value=ranking) as mock_rank:
+            build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        assert "classes" not in mock_rank.call_args.kwargs
+
+    def test_empty_ranking_yields_warning(self) -> None:
+        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["ranked_segments"] == []
+        assert out["warnings"]
+
+
+# ── build_forecast_overlay: graceful (rank_segments бросает) ───────────────────
+
+
+class TestDemandSupplyGraceful:
+    def test_value_error_yields_empty_plus_warning_no_raise(self) -> None:
+        with patch(_RANK, side_effect=ValueError("нет геометрии участка")):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num="66:41:0000000:1",
+                horizon_months=12,
+                target_class=None,
+            )
+        assert out["mode"] == "demand_supply"
+        assert out["advisory"] is True
+        assert out["ranked_segments"] == []
+        assert out["warnings"]
+
+
+# ── build_forecast_overlay: demand_only (cad_num=None) ─────────────────────────
+
+
+def _mk_metrics(unit_velocity: float | None) -> MagicMock:
+    m = MagicMock()
+    m.unit_velocity = unit_velocity
+    return m
+
+
+def _mk_coef(coefficient: float) -> MagicMock:
+    c = MagicMock()
+    c.coefficient = coefficient
+    return c
+
+
+class TestDemandOnlyOverlay:
+    def test_mode_flips_when_no_cad_num(self) -> None:
+        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["mode"] == "demand_only"
+        assert out["advisory"] is True
+
+    def test_mandatory_supply_warning_present(self) -> None:
+        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 any("supply/конкуренты НЕ учтены" in w for w in out["warnings"])
+
+    def test_balance_units_always_none_no_fabricated_supply(self) -> None:
+        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"], "ожидали 5 ранжированных форматов"
+        assert all(s["balance_units"] is None for s in out["ranked_segments"])
+
+    def test_confidence_low_for_all_segments(self) -> None:
+        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 all(s["confidence"] == "low" for s in out["ranked_segments"])
+
+    def test_deficit_index_is_pace_proxy_zero_to_one(self) -> None:
+        # Разные §9.5-коэффициенты per вызов → разные pace → deficit_index = pace/max ∈ (0,1].
+        macro_coeffs = iter([_mk_coef(c) for c in (0.5, 1.0, 1.5, 1.2, 0.8)])
+        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, side_effect=lambda *a, **k: next(macro_coeffs)),
+        ):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num=None,
+                horizon_months=12,
+                target_class=None,
+            )
+        indices = [s["deficit_index"] for s in out["ranked_segments"]]
+        assert max(indices) == pytest.approx(1.0)  # топ нормирован к 1.0
+        assert all(0.0 < i <= 1.0 for i in indices)
+        # DESC по pace.
+        assert indices == sorted(indices, reverse=True)
+
+    def test_no_velocity_yields_empty_plus_warning(self) -> None:
+        # base_pace None (нет §9.2 темпа) → НЕ фабрикуем сигнал, пустой ранкинг + warning.
+        with (
+            patch(_METRICS, return_value=_mk_metrics(None)),
+            patch(_GET_MACRO, return_value=[]),
+            patch(_HOLD, return_value={12: 18.0}),
+        ):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num=None,
+                horizon_months=12,
+                target_class=None,
+            )
+        assert out["mode"] == "demand_only"
+        assert out["ranked_segments"] == []
+        assert out["warnings"]
+
+    def test_five_default_room_buckets_ranked(self) -> None:
+        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,
+            )
+        buckets = {s["bucket"] for s in out["ranked_segments"]}
+        assert buckets == {"1-Студия", "2-1-к", "3-2-к", "4-3-к", "5-80+ м²"}
+
+    def test_overlay_validates_against_schema(self) -> None:
+        # Sanity: demand_only-выход проходит RecommendForecastOverlay-валидацию.
+        from app.schemas.recommend import RecommendForecastOverlay
+
+        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: dict[str, Any] = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num=None,
+                horizon_months=12,
+                target_class="Comfort",
+            )
+        model = RecommendForecastOverlay.model_validate(out)
+        assert model.mode == "demand_only"
+        assert model.advisory is True

backend/tests/services/test_recommend_mix_velocity.py

@@ -274,11 +274,16 @@ def _run_recommend_mix_full(
     n_competitors: float = 10.0,
     sale_graph_vel_pm: float | None = None,
     area_total_m2: float = 12_000.0,
+    horizon_months: int | None = None,
+    cad_num: str | None = None,
 ) -> dict[str, Any]:
     """Запускает recommend_mix с правильным набором моков.
 
     n_competitors — district+class competitors_weighted, который возвращает
     _competitors_two_dim и используется как знаменатель в rosreestr fallback.
+
+    horizon_months/cad_num — #982 forecast-overlay opt-in (по умолчанию None →
+    живой микс БАЙТ-в-БАЙТ как раньше; существующие вызовы не затронуты).
     """
     from app.services.analytics_queries import recommend_mix
 
@@ -342,6 +347,8 @@ def _run_recommend_mix_full(
             target_class=None,
             months_window=24,
             region_code=66,
+            horizon_months=horizon_months,
+            cad_num=cad_num,
         )
 
 
@@ -529,3 +536,86 @@ class TestObjectivePerBucketPath:
         velocities = [b["velocity_per_month"] for b in result["buckets"]]
         unique = set(round(v, 4) for v in velocities)
         assert len(unique) > 1, "Все objective velocities одинаковые — ошибка маппинга"
+
+
+_FORECAST_PER_BUCKET = {
+    "1-Студия": 2.0,
+    "2-1-к": 6.0,
+    "3-2-к": 4.5,
+    "4-3-к": 3.0,
+    "5-80+ м²": 1.5,
+}
+
+
+class TestForecastOverlayOptIn:
+    """#982 (953-A): forecast-overlay — ADDITIVE OPT-IN + live-endpoint safety."""
+
+    def test_invariance_no_horizon_no_forecast_key(self) -> None:
+        """horizon_months=None → НЕТ scope['forecast'] (живой микс не тронут)."""
+        result = _run_recommend_mix_full(
+            objective_per_bucket=_FORECAST_PER_BUCKET,
+            n_competitors=10,
+            sale_graph_vel_pm=5.0,
+        )
+        assert "forecast" not in result["scope"]
+
+    def test_invariance_top_level_keys_unchanged(self) -> None:
+        """Топ-уровневые ключи ответа неизменны (4 поля контракта) без horizon."""
+        result = _run_recommend_mix_full(
+            objective_per_bucket=_FORECAST_PER_BUCKET,
+            n_competitors=10,
+            sale_graph_vel_pm=5.0,
+        )
+        assert set(result.keys()) == {"scope", "buckets", "summary", "comparables"}
+
+    def test_invariance_byte_identical_scope_when_off(self) -> None:
+        """scope БАЙТ-в-БАЙТ совпадает с/без передачи horizon_months=None."""
+        base = _run_recommend_mix_full(
+            objective_per_bucket=_FORECAST_PER_BUCKET, n_competitors=10, sale_graph_vel_pm=5.0
+        )
+        again = _run_recommend_mix_full(
+            objective_per_bucket=_FORECAST_PER_BUCKET,
+            n_competitors=10,
+            sale_graph_vel_pm=5.0,
+            horizon_months=None,
+        )
+        assert base["scope"] == again["scope"]
+        assert base["buckets"] == again["buckets"]
+        assert base["summary"] == again["summary"]
+
+    def test_opt_in_attaches_overlay_under_scope(self) -> None:
+        """horizon_months задан → scope['forecast'] = результат build_forecast_overlay."""
+        sentinel = {"horizon_months": 12, "mode": "demand_only", "advisory": True}
+        with patch(
+            "app.services.forecasting.recommendation.build_forecast_overlay",
+            return_value=sentinel,
+        ) as mock_overlay:
+            result = _run_recommend_mix_full(
+                objective_per_bucket=_FORECAST_PER_BUCKET,
+                n_competitors=10,
+                sale_graph_vel_pm=5.0,
+                horizon_months=12,
+            )
+        assert result["scope"]["forecast"] is sentinel
+        assert mock_overlay.call_count == 1
+        # district проброшен (resolved district_name), horizon/cad — из payload.
+        assert mock_overlay.call_args.kwargs["horizon_months"] == 12
+        assert mock_overlay.call_args.kwargs["cad_num"] is None
+
+    def test_crash_isolation_overlay_failure_does_not_break_live_mix(self) -> None:
+        """build_forecast_overlay бросает → живой микс ВСЁ РАВНО возвращается."""
+        with patch(
+            "app.services.forecasting.recommendation.build_forecast_overlay",
+            side_effect=RuntimeError("boom"),
+        ):
+            result = _run_recommend_mix_full(
+                objective_per_bucket=_FORECAST_PER_BUCKET,
+                n_competitors=10,
+                sale_graph_vel_pm=5.0,
+                horizon_months=12,
+            )
+        # Микс жив; overlay = {error, advisory}, остальные поля на месте.
+        assert result["buckets"]
+        assert result["scope"]["forecast"]["advisory"] is True
+        assert "boom" in result["scope"]["forecast"]["error"]
+        assert set(result.keys()) == {"scope", "buckets", "summary", "comparables"}