feat(recommend): horizon-aware forecast-overlay for recommend_mix (#982, 953-A)

Additive opt-in: when horizon_months is passed, attach an ADVISORY forecast
overlay under scope["forecast"] via new forecasting/recommendation.py
(demand_supply when cad_num set → #981 rank_segments; demand_only otherwise →
per-room-bucket demand pace, NO fabricated supply). Default path stays
byte-identical — overlay is try/except-guarded and never crashes the live mix.

- schemas/recommend.py: horizon_months + cad_num inputs; RecommendForecast{Segment,
  Overlay} output models (ride open scope dict; RecommendMixOutput 4 fields intact)
- forecasting/recommendation.py: pure room/class bridges (inverse of _BUCKET_PRETTY,
  drift-guarded) + build_forecast_overlay (graceful, advisory always True)
- analytics_queries.recommend_mix: +horizon_months/cad_num params + guarded tail
- api/v1/analytics.py: endpoint passthrough
- tests: bridges both directions, both modes, graceful, invariance, crash isolation

NOTE: live recommend endpoint now has a DORMANT path into the advisory forecast
engine (frontend sends no horizon_months yet); wiring point when #951 backtest lands.
45 tests; 413 forecasting+velocity green. opus code-review ✅ (7 live-endpoint gates).

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/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"}