backend/app/services/forecasting/recommendation.py

@@ -30,15 +30,14 @@ LIVE-ENDPOINT-SAFE: НИКОГДА не бросает — на любом то
     room-bucket, ранжируем DESC по темпу. `deficit_index` здесь = pace/max_pace ∈
     (0,1]; `balance_units=None`, confidence='low', обязательный warning что supply
     НЕ учтён. НИКОГДА не фабрикуем предложение/конкурентов из воздуха.
-    ВАЖНО (#1593): сегодня §9.2 unit_velocity (base_pace) и §9.4 norm.coefficient
-    ОДИНАКОВЫ для всех 5 room-bucket'ов (market_metrics.compute_market_metrics не
-    принимает room_bucket, а §9.4-β фитится district×obj_class без room_bucket),
-    поэтому оба постоянных множителя СОКРАЩАЮТСЯ в pace/max_pace и порядок форматов
-    определяется ИСКЛЮЧИТЕЛЬНО §9.5 macro_coef = segment_steepness(room_bucket).
-    Т.е. фактический deficit_index в demand_only = ПРОКСИ rate-крутизны нарезки, а
-    НЕ относительной силы спроса. Честный per-bucket §9.2/§9.4-сигнал требует
-    room_bucket-параметра в market_metrics + §9.4/§9.6 (другие файлы + доменное
-    решение, как мерить per-bucket спрос) — см. #1593.
+    #1593 (ИСПРАВЛЕНО): §9.2 base_pace теперь per-bucket через
+    MarketMetrics.velocity_by_room (ед./мес из objective_lots per комнатность).
+    Маппинг forecast_bucket → metric-ключи: _FORECAST_TO_METRIC_BUCKETS.
+    "80+ м²" = сумма bucket'ов "4" + "5+" (оба крупные форматы). Если данных
+    по конкретному bucket'у нет — fallback на агрегатный unit_velocity, что
+    лучше чем фабрикация 0. §9.4 norm.coefficient по-прежнему без per-bucket
+    фита (β фитится district×obj_class), поэтому он одинаков между bucket'ами
+    и СОКРАЩАЕТСЯ в pace/max_pace — это честно задокументировано.
 
 ИМПОРТЫ §9.x — ЛОКАЛЬНЫЕ (внутри функций), чтобы избежать import-cycle: пакет
 `forecasting/__init__` тянет `affordability`, который импортит
@@ -76,6 +75,18 @@ _FORECAST_TO_LIVE_BUCKET: dict[str, str] = {
 # дефолтной сетке 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()}
 
+# #1593: forecast room-bucket → список ключей MarketMetrics.velocity_by_room.
+# Ключи velocity_by_room — вокабуляр _room_bucket() из market_metrics
+# ("студия","1","2","3","4","5+"). "80+ м²" накрывает 4-комнатные и 5+
+# (большие квартиры), поэтому суммируем оба бакета.
+_FORECAST_TO_METRIC_BUCKETS: dict[str, list[str]] = {
+    "Студии 15-30": ["студия"],
+    "1-к 30-45": ["1"],
+    "2-к 45-60": ["2"],
+    "3-к 60-80": ["3"],
+    "80+ м²": ["4", "5+"],
+}
+
 # Live target_class (вокабуляр schemas.recommend ClassLiteral, «человеческий»
 # регистр) → forecast obj_class (3 массовых класса §9.7 _DEFAULT_CLASSES). None и
 # незнакомое → None (вызывающий тогда отдаёт движку дефолтную сетку классов).
@@ -576,14 +587,11 @@ def _demand_only_overlay(
     (0,1]. balance_units=None, confidence='low', обязательный warning. НИКОГДА не
     фабрикуем supply. Graceful → [].
 
-    ВАЖНО (#1593, честность ранкинга): §9.2 base_pace и §9.4 norm.coefficient
-    сейчас ОДИНАКОВЫ для всех bucket'ов (нет room_bucket-фильтра в
-    market_metrics.compute_market_metrics; §9.4-β фитится district×obj_class без
-    room_bucket) и СОКРАЩАЮТСЯ в pace/max_pace. Поэтому фактический порядок
-    форматов и deficit_index определяются ИСКЛЮЧИТЕЛЬНО §9.5 macro_coef =
-    segment_steepness(room_bucket) — это ПРОКСИ rate-крутизны нарезки, НЕ силы
-    спроса. Полноценный per-bucket §9.2/§9.4-сигнал = другие файлы + доменное
-    решение (#1593); до тех пор не интерпретировать как «самый дефицитный спрос».
+    #1593 (ИСПРАВЛЕНО): §9.2 base_pace теперь per-bucket через
+    MarketMetrics.velocity_by_room. §9.4 norm.coefficient одинаков между
+    bucket'ами (β без room_bucket фита) и СОКРАЩАЕТСЯ в pace/max_pace — это
+    задокументировано, не скрыто. Ранкинг теперь отражает реальные per-bucket
+    темпы спроса из objective_lots, а не только §9.5 macro_coef.
     """
     # Локальные импорты — избегаем import-cycle (см. module docstring).
     from app.services.forecasting.demand_normalization import compute_demand_normalization
@@ -597,9 +605,10 @@ def _demand_only_overlay(
     ]
     mapped_class = map_class(target_class)
 
-    # Один раз на вызов: §9.2 наблюдаемый темп (base_pace) + hold-last-rate для §9.4.
+    # Один раз на вызов: §9.2 наблюдаемый темп (aggregate + per-bucket) + hold-last-rate.
     metrics = compute_market_metrics(db, district=district)
-    base_pace = metrics.unit_velocity
+    base_pace = metrics.unit_velocity  # агрегатный fallback (используется если нет per-bucket)
+    vel_by_room = metrics.velocity_by_room  # #1593: per-bucket из objective_lots
     macro = get_monthly_macro(db)
     rate_future = hold_last_rate(macro, [horizon_months]).get(horizon_months)
 
@@ -625,6 +634,22 @@ def _demand_only_overlay(
             room_bucket=forecast_bucket,
             district=district,
         )
+        # §9.2 #1593: per-bucket velocity из velocity_by_room (objective_lots
+        # per комнатность). _FORECAST_TO_METRIC_BUCKETS даёт список metric-ключей
+        # для данного forecast_bucket ("80+ м²" = "4"+"5+"). Суммируем ед./мес по
+        # бакетам. Честный 0 при известных бакетах — НЕ fallback. Fallback на
+        # агрегатный base_pace только когда velocity_by_room=None (пустая
+        # выборка) или bucket неизвестен в _FORECAST_TO_METRIC_BUCKETS.
+        metric_keys = _FORECAST_TO_METRIC_BUCKETS.get(forecast_bucket, [])
+        if vel_by_room is not None and metric_keys:
+            bucket_velocity: float = sum(vel_by_room.get(k, 0.0) for k in metric_keys)
+            # Если нет продаж ни в одном из ключей — честный 0 (НЕ fallback к
+            # aggregate: лучше показать реальный 0 чем маскировать его aggregate).
+            bucket_base_pace: float = bucket_velocity
+        else:
+            # vel_by_room None (пустая выборка) или неизвестный forecast_bucket →
+            # aggregate fallback сохраняет graceful поведение.
+            bucket_base_pace = float(base_pace)
         # §9.4 нормализация под будущий режим ставки (β внутри). rate_future None
         # (hold_last_rate не дал ставку) → НЕ применяем §9.4: 0.0-placeholder дал бы
         # delta=−rate_window_avg → exp(β·delta) и клэмп к _NORM_MAX (макс. аплифт),
@@ -640,7 +665,7 @@ def _demand_only_overlay(
         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
+        pace = bucket_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)
@@ -656,9 +681,10 @@ def _demand_only_overlay(
             demand_only=True,
         )
 
-    # DESC по темпу; deficit_index = pace/max_pace ∈ (0,1]. NB #1593: base_pace и
-    # §9.4 norm одинаковы для всех bucket'ов и сокращаются → порядок = §9.5
-    # segment_steepness (rate-крутизна), НЕ сила спроса. Per-bucket §9.2/§9.4 = TODO.
+    # DESC по темпу; deficit_index = pace/max_pace ∈ (0,1].
+    # #1593: base_pace теперь per-bucket из velocity_by_room → ранкинг отражает
+    # реальный спрос по комнатности. §9.4 norm одинаков между bucket'ами и
+    # сокращается в pace/max_pace (β без per-bucket фита — честно задокументировано).
     paces.sort(key=lambda t: t[2], reverse=True)
     ranked_segments: list[dict[str, Any]] = [
         {
@@ -672,12 +698,13 @@ def _demand_only_overlay(
     ]
 
     logger.info(
-        "forecast-overlay demand_only: district=%s horizon=%d ranked=%d base_pace=%.3f "
-        "(ADVISORY, supply НЕ учтён)",
+        "forecast-overlay demand_only: district=%s horizon=%d ranked=%d "
+        "base_pace_agg=%.3f vel_by_room=%s (ADVISORY, supply НЕ учтён, #1593 per-bucket)",
         district,
         horizon_months,
         len(ranked_segments),
         base_pace,
+        vel_by_room,
     )
     return _enrich_overlay(
         db,

backend/app/services/site_finder/market_metrics.py

@@ -65,6 +65,7 @@ def _market_metrics_key(
         premise_kind,
     )
 
+
 Confidence = Literal["high", "medium", "low"]
 
 # Лот считается «зависшим» (overstock), если он в продаже дольше этого числа
@@ -111,6 +112,9 @@ class MarketMetrics:
     demand_concentration: float | None  # Херфиндаль продаж по комнатности (0..1)
     price_sensitivity: float | None  # эластичность цена↔спрос (slope, обычно < 0)
     price_sensitivity_source: str | None  # 'regression' | 'fallback' | None
+    # #1593: ед./мес per room-bucket (ключи — вокабуляр _room_bucket():
+    # "студия","1","2","3","4","5+"). None если нет данных в окне.
+    velocity_by_room: dict[str, float] | None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -136,6 +140,8 @@ class MarketMetrics:
             "demand_concentration": _round_or_none(self.demand_concentration, 3),
             "price_sensitivity": _round_or_none(self.price_sensitivity, 4),
             "price_sensitivity_source": self.price_sensitivity_source,
+            # velocity_by_room намеренно не сериализуется: это внутренний
+            # pipeline-атрибут (потребляется recommendation.py напрямую).
         }
 
 
@@ -409,6 +415,15 @@ def compute_market_metrics(
     liquidity = _liquidity_index(sold_by_room)
     overstock = _overstock_index(n_long_unsold, n_available)
     demand_conc = _demand_concentration(sold_by_room)
+    # #1593: per-bucket velocity — ед./мес по каждой комнатности. Ключи зеркалят
+    # _room_bucket() ("студия","1","2","3","4","5+"). При has_sample=False нет
+    # смысла делить 0 лотов → None (graceful, зеркало unit_velocity поведения).
+    # window_months > 0 защищает от ZeroDivisionError (зеркало _monthly_rate()).
+    vel_by_room: dict[str, float] | None = (
+        {bkt: float(cnt) / float(window_months) for bkt, cnt in sold_by_room.items()}
+        if has_sample and sold_by_room and window_months > 0
+        else None
+    )
 
     # ── price_sensitivity — reuse analytics_queries._elasticity_coef ───────────
     price_sensitivity, price_sensitivity_source = _price_sensitivity(
@@ -450,6 +465,7 @@ def compute_market_metrics(
         demand_concentration=demand_conc,
         price_sensitivity=price_sensitivity,
         price_sensitivity_source=price_sensitivity_source,
+        velocity_by_room=vel_by_room,
     )
 
 

backend/tests/services/forecasting/test_recommendation.py

@@ -329,9 +329,13 @@ class TestDemandSupplyGraceful:
 # ── build_forecast_overlay: demand_only (cad_num=None) ─────────────────────────
 
 
-def _mk_metrics(unit_velocity: float | None) -> MagicMock:
+def _mk_metrics(
+    unit_velocity: float | None,
+    velocity_by_room: dict[str, float] | None = None,
+) -> MagicMock:
     m = MagicMock()
     m.unit_velocity = unit_velocity
+    m.velocity_by_room = velocity_by_room
     return m
 
 
@@ -453,6 +457,71 @@ class TestDemandOnlyOverlay:
         assert out["ranked_segments"] == []
         assert out["warnings"]
 
+    def test_per_bucket_velocity_drives_ranking(self) -> None:
+        """#1593: velocity_by_room даёт реальные per-bucket §9.2-темпы.
+
+        При одинаковом §9.5 macro_coef=1.0 ранкинг должен определяться
+        исключительно per-bucket velocity_by_room, а не aggregate base_pace.
+        Студии ("студия") → самый высокий темп → первые в ранкинге.
+        """
+
+        # velocity_by_room: студии продаются в 5× быстрее чем 1-к, остальные ≈ 0.
+        vel_by_room = {
+            "студия": 10.0,  # "Студии 15-30" → "студия"
+            "1": 2.0,  # "1-к 30-45" → "1"
+            "2": 1.0,  # "2-к 45-60" → "2"
+            "3": 0.5,  # "3-к 60-80" → "3"
+            "4": 0.1,  # "80+ м²" → "4" + "5+"
+            "5+": 0.1,
+        }
+        with (
+            patch(_METRICS, return_value=_mk_metrics(4.0, velocity_by_room=vel_by_room)),
+            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)),  # нейтральный §9.5 (изолируем §9.2)
+        ):
+            out = build_forecast_overlay(
+                MagicMock(),
+                district="Ленинский",
+                cad_num=None,
+                horizon_months=12,
+                target_class=None,
+            )
+        buckets = [s["bucket"] for s in out["ranked_segments"]]
+        # При одинаковом macro_coef=1.0 ранкинг = velocity_by_room:
+        # студии (10.0) > 1-к (2.0) > 2-к (1.0) > 3-к (0.5) > 80+ (0.1+0.1=0.2)
+        assert buckets[0] == "1-Студия", f"ожидали студии первыми, получили {buckets}"
+        assert buckets[1] == "2-1-к", f"ожидали 1-к вторыми, получили {buckets}"
+        assert buckets[-1] == "5-80+ м²", f"ожидали 80+ м² последними, получили {buckets}"
+        # deficit_index нормирован к 1.0 для топа
+        assert out["ranked_segments"][0]["deficit_index"] == pytest.approx(1.0)
+
+    def test_fallback_to_aggregate_when_no_velocity_by_room(self) -> None:
+        """#1593: при velocity_by_room=None возвращаемся к aggregate base_pace.
+
+        Ranking в этом случае определяется §9.5 macro_coef (старое поведение),
+        но не крашит и не фабрикует данные.
+        """
+        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, velocity_by_room=None)),
+            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,
+            )
+        # Graceful: 5 сегментов, deficit_index ∈ (0, 1].
+        assert len(out["ranked_segments"]) == 5
+        assert max(s["deficit_index"] for s in out["ranked_segments"]) == pytest.approx(1.0)
+
     def test_five_default_room_buckets_ranked(self) -> None:
         with (
             patch(_METRICS, return_value=_mk_metrics(4.0)),

backend/tests/services/site_finder/test_market_metrics.py

@@ -275,6 +275,7 @@ class TestAsDict:
             demand_concentration=0.3333333,
             price_sensitivity=-1.234567,
             price_sensitivity_source="regression",
+            velocity_by_room={"1": 1.5, "2": 0.5},
         )
         d = m.as_dict()
         assert d["absorption_rate"] == 0.0476
@@ -304,6 +305,7 @@ class TestAsDict:
             demand_concentration=None,
             price_sensitivity=None,
             price_sensitivity_source=None,
+            velocity_by_room=None,
         )
         d = m.as_dict()
         assert d["absorption_rate"] is None
@@ -421,6 +423,14 @@ class TestComputeMarketMetricsShape:
         assert m.price_sensitivity == -1.4
         assert m.price_sensitivity_source == "regression"
         assert m.confidence == "high"
+        # #1593: velocity_by_room — per-bucket ед./мес из sold_by_room.
+        # _FULL_SALES: rooms_int=1 (30 ед.), rooms_int=2 (20 ед.), rooms_int=0 (10 ед.)
+        # window_months=6 → "1": 30/6=5.0, "2": 20/6≈3.333, "студия": 10/6≈1.667
+        assert m.velocity_by_room is not None
+        assert set(m.velocity_by_room.keys()) == {"1", "2", "студия"}
+        assert m.velocity_by_room["1"] == pytest.approx(30.0 / 6)
+        assert m.velocity_by_room["2"] == pytest.approx(20.0 / 6)
+        assert m.velocity_by_room["студия"] == pytest.approx(10.0 / 6)
 
     def test_elasticity_reuse_invoked_with_district(self) -> None:
         db = _mock_db(_FULL_STOCK, _FULL_SALES)
@@ -516,9 +526,7 @@ class TestPriceSensitivityDistrictResolution:
     случайно совпадал в обоих вокабулярах).
     """
 
-    def test_admin_resolves_to_micros_in_elasticity_call(
-        self, _patch_resolver: MagicMock
-    ) -> None:
+    def test_admin_resolves_to_micros_in_elasticity_call(self, _patch_resolver: MagicMock) -> None:
         # Резолвер: 'Кировский' → ['Втузгородок','ЖБИ'] (как на проде).
         _patch_resolver.side_effect = lambda _db, d: (
             ["Втузгородок", "ЖБИ"] if d == "Кировский" else None
@@ -542,9 +550,7 @@ class TestPriceSensitivityDistrictResolution:
         assert value == -1.2
         assert source == "regression"
 
-    def test_resolver_none_passes_empty_list_ekb_wide(
-        self, _patch_resolver: MagicMock
-    ) -> None:
+    def test_resolver_none_passes_empty_list_ekb_wide(self, _patch_resolver: MagicMock) -> None:
         # Резолвер None ('не определён' / нет чистых алиасов) → передаём [] →
         # _elasticity_coef уходит в EKB-wide ветку (без district-фильтра).
         _patch_resolver.side_effect = lambda _db, _d: None
@@ -555,9 +561,7 @@ class TestPriceSensitivityDistrictResolution:
         _args, kwargs = elast.call_args
         assert kwargs["districts"] == []
 
-    def test_none_district_skips_resolver_and_elasticity(
-        self, _patch_resolver: MagicMock
-    ) -> None:
+    def test_none_district_skips_resolver_and_elasticity(self, _patch_resolver: MagicMock) -> None:
         # district=None → _price_sensitivity сразу возвращает (None, None), не зовёт ни
         # резолвер, ни _elasticity_coef.
         db = MagicMock()
@@ -600,6 +604,45 @@ class TestPriceSensitivityDistrictResolution:
         assert source is None
 
 
+class TestVelocityByRoom:
+    """#1593: MarketMetrics.velocity_by_room — per-bucket ед./мес."""
+
+    def test_velocity_by_room_computed_from_sold_by_room(self) -> None:
+        # sold_by_room: "1" → 30, "2" → 20, "студия" → 10 ед. за 6 мес.
+        db = _mock_db(_FULL_STOCK, _FULL_SALES)
+        with patch(_ELAST, return_value={"elasticity": -1.4, "source": "regression"}):
+            m = compute_market_metrics(db, district="Автовокзал", window_months=6)
+        assert m.velocity_by_room is not None
+        assert m.velocity_by_room["1"] == pytest.approx(30.0 / 6)
+        assert m.velocity_by_room["2"] == pytest.approx(20.0 / 6)
+        assert m.velocity_by_room["студия"] == pytest.approx(10.0 / 6)
+
+    def test_velocity_by_room_none_when_no_sample(self) -> None:
+        # n_lots=0 → выборка пуста → velocity_by_room None (зеркало unit_velocity).
+        empty_stock = {
+            "n_lots": 0,
+            "n_sold": 0,
+            "n_available": 0,
+            "obj_count": 0,
+            "n_long_unsold": 0,
+        }
+        db = _mock_db(empty_stock, [])
+        with patch(_ELAST, return_value={"elasticity": -1.5, "source": "fallback"}):
+            m = compute_market_metrics(db, district="ПустойРайон")
+        assert m.velocity_by_room is None
+
+    def test_velocity_by_room_none_when_no_by_room_data(self) -> None:
+        # Только grand-total строка (нет разбивки по комнатности) → sold_by_room пустой
+        # → velocity_by_room None.
+        sales_total_only = [
+            {"units_sold_window": 5, "area_sold_window": 200.0, "rooms_int": None, "is_total": 1}
+        ]
+        db = _mock_db(_FULL_STOCK, sales_total_only)
+        with patch(_ELAST, return_value={"elasticity": -1.5, "source": "fallback"}):
+            m = compute_market_metrics(db, district="Тихий")
+        assert m.velocity_by_room is None
+
+
 class TestComputeMarketMetricsThinData:
     """Graceful-on-thin-data: empty / zero → None metrics + low confidence, no crash."""
 
@@ -623,6 +666,7 @@ class TestComputeMarketMetricsThinData:
         assert m.liquidity_index is None
         assert m.overstock_index is None
         assert m.demand_concentration is None
+        assert m.velocity_by_room is None
         assert m.confidence == "low"
 
     def test_stock_none_row_no_crash(self) -> None: