backend/app/services/site_finder/velocity.py

@@ -46,6 +46,7 @@ class VelocityResult:
     period_start: str  # YYYY-MM
     period_end: str  # YYYY-MM
     sample_competitors: list[dict[str, Any]]  # top-5 для UI
+    by_room_bucket: dict[str, dict[str, Any]]  # агрегат по комнатности
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -57,6 +58,7 @@ class VelocityResult:
             "months_observed": self.months_observed,
             "period": {"start": self.period_start, "end": self.period_end},
             "sample_competitors": self.sample_competitors,
+            "by_room_bucket": self.by_room_bucket,
         }
 
 
@@ -186,7 +188,7 @@ def compute_velocity(
                       FROM objective_corpus_room_month crm
                       JOIN mapped m
                         ON m.objective_complex_name = crm.project_name
-                     WHERE crm.report_month >= (CURRENT_DATE - :window_interval::interval)
+                     WHERE crm.report_month >= (CURRENT_DATE - CAST(:window_interval AS interval))
                        AND crm.deals_total_count > 0
                      GROUP BY m.obj_id
                     """
@@ -207,6 +209,75 @@ def compute_velocity(
     if not sales_rows:
         return None
 
+    # ── Step 2b: разбивка по комнатности (room_bucket) ───────────────────────
+    # Тот же маппинг domrf_obj_id → project_name. Агрегируем по room_bucket
+    # для отображения структуры спроса в UI.
+    try:
+        with db.begin_nested():
+            bucket_rows = (
+                db.execute(
+                    text(
+                        """
+                    WITH mapped AS (
+                        SELECT cm.domrf_obj_id AS obj_id,
+                               cm.objective_complex_name
+                          FROM objective_complex_mapping cm
+                         WHERE cm.domrf_obj_id = ANY(:obj_ids)
+                    )
+                    SELECT
+                        m.obj_id,
+                        crm.room_bucket,
+                        SUM(crm.deals_total_count) AS units_sold,
+                        SUM(COALESCE(crm.deals_total_vol_m2,
+                                     crm.deals_total_count * 45.0)) AS sqm_sold
+                      FROM objective_corpus_room_month crm
+                      JOIN mapped m ON m.objective_complex_name = crm.project_name
+                     WHERE crm.report_month >= (CURRENT_DATE - CAST(:window_interval AS interval))
+                       AND crm.deals_total_count > 0
+                     GROUP BY m.obj_id, crm.room_bucket
+                        """
+                    ),
+                    {
+                        "obj_ids": obj_ids,
+                        "window_interval": f"{months_window} months",
+                    },
+                )
+                .mappings()
+                .all()
+            )
+    except Exception:
+        logger.warning("velocity: bucket breakdown query failed, continuing without it")
+        bucket_rows = []
+
+    # Агрегируем по room_bucket поверх всех конкурентов.
+    by_bucket_agg: dict[str, dict[str, Any]] = {}
+    per_comp_buckets: dict[int, dict[str, int]] = {}
+
+    for row in bucket_rows:
+        bucket = str(row["room_bucket"])
+        oid = int(row["obj_id"])
+        units = int(row["units_sold"] or 0)
+        sqm = float(row["sqm_sold"] or 0.0)
+
+        if bucket not in by_bucket_agg:
+            by_bucket_agg[bucket] = {"units": 0, "sqm": 0.0, "complexes": set()}
+        by_bucket_agg[bucket]["units"] += units
+        by_bucket_agg[bucket]["sqm"] += sqm
+        by_bucket_agg[bucket]["complexes"].add(oid)
+
+        if oid not in per_comp_buckets:
+            per_comp_buckets[oid] = {}
+        per_comp_buckets[oid][bucket] = units
+
+    by_room_bucket: dict[str, dict[str, Any]] = {
+        bucket: {
+            "units": data["units"],
+            "sqm": round(data["sqm"], 0),
+            "complexes_count": len(data["complexes"]),
+        }
+        for bucket, data in by_bucket_agg.items()
+    }
+
     total_sqm = sum(float(r["total_sqm"] or 0.0) for r in sales_rows)
     months_observed = max((int(r["months_with_data"] or 0) for r in sales_rows), default=0)
     period_start_dates = [r["period_start"] for r in sales_rows if r["period_start"]]
@@ -254,6 +325,7 @@ def compute_velocity(
                 "obj_id": oid,
                 **competitor_meta[oid],
                 "total_sqm_period": round(sales_by_id.get(oid, 0.0), 0),
+                "by_room_bucket": per_comp_buckets.get(oid, {}),
             }
             for oid in obj_ids
             if oid in competitor_meta
@@ -272,6 +344,7 @@ def compute_velocity(
         period_start=period_start,
         period_end=period_end,
         sample_competitors=sample,
+        by_room_bucket=by_room_bucket,
     )
 
 
@@ -299,7 +372,7 @@ def _get_ekb_median(db: Session, months_window: int = 6) -> float | None:
                                          deals_total_count * 45.0)) AS total_sqm,
                             COUNT(DISTINCT report_month) AS months_data
                           FROM objective_corpus_room_month
-                         WHERE report_month >= (CURRENT_DATE - :window_interval::interval)
+                         WHERE report_month >= (CURRENT_DATE - CAST(:window_interval AS interval))
                            AND deals_total_count > 0
                          GROUP BY project_name
                         HAVING COUNT(DISTINCT report_month) >= 3

backend/tests/test_velocity.py

@@ -4,6 +4,7 @@ Mock-based — не требуют живой БД.
 Источник данных — objective_corpus_room_month (мигрировано с domrf_kn_sale_graph).
 Mock shape совместим: sales query возвращает те же aliases (obj_id, total_sqm,
 months_with_data, period_start, period_end) через GROUP BY domrf_obj_id.
+Третий вызов execute — bucket_rows (obj_id, room_bucket, units_sold, sqm_sold).
 """
 
 from __future__ import annotations
@@ -59,11 +60,30 @@ def _sales_row(
     return r
 
 
-def _make_db(comp_rows: list, sales_rows: list) -> MagicMock:
-    """Сконструировать mock Session с двумя последовательными вызовами execute."""
+def _bucket_row(obj_id: int, room_bucket: str, units_sold: int, sqm_sold: float) -> MagicMock:
+    r = MagicMock()
+    r.__getitem__ = lambda self, k: {
+        "obj_id": obj_id,
+        "room_bucket": room_bucket,
+        "units_sold": units_sold,
+        "sqm_sold": sqm_sold,
+    }[k]
+    return r
+
+
+def _make_db(
+    comp_rows: list,
+    sales_rows: list,
+    bucket_rows: list | None = None,
+) -> MagicMock:
+    """Сконструировать mock Session с тремя последовательными вызовами execute.
+
+    Порядок: comp_rows → sales_rows → bucket_rows.
+    bucket_rows=None → пустой список (bucket query gracefully degraded).
+    """
     db = MagicMock()
     execute_results = []
-    for rows in [comp_rows, sales_rows]:
+    for rows in [comp_rows, sales_rows, bucket_rows if bucket_rows is not None else []]:
         result = MagicMock()
         result.mappings.return_value.all.return_value = rows
         execute_results.append(result)
@@ -196,7 +216,7 @@ def test_score_zero_when_no_sales_sqm():
 
 
 def test_as_dict_structure():
-    """as_dict() содержит все ожидаемые ключи."""
+    """as_dict() содержит все ожидаемые ключи, включая by_room_bucket."""
     vr = VelocityResult(
         competitors_count=5,
         monthly_velocity_sqm=3000.0,
@@ -207,6 +227,7 @@ def test_as_dict_structure():
         period_start="2024-11",
         period_end="2025-02",
         sample_competitors=[],
+        by_room_bucket={"1": {"units": 10, "sqm": 450.0, "complexes_count": 2}},
     )
     d = vr.as_dict()
     assert "competitors_count" in d
@@ -216,6 +237,8 @@ def test_as_dict_structure():
     assert d["period"]["start"] == "2024-11"
     assert d["period"]["end"] == "2025-02"
     assert d["velocity_score"] == pytest.approx(0.333, abs=1e-3)
+    assert "by_room_bucket" in d
+    assert d["by_room_bucket"]["1"]["units"] == 10
 
 
 def test_sample_competitors_top5():
@@ -235,3 +258,80 @@ def test_sample_competitors_top5():
     assert len(result.sample_competitors) <= 5
     sqms = [c["total_sqm_period"] for c in result.sample_competitors]
     assert sqms == sorted(sqms, reverse=True)
+
+
+def test_by_room_bucket_aggregation():
+    """by_room_bucket агрегирует units/sqm поверх всех конкурентов корректно."""
+    comp_rows = [_comp_row(1), _comp_row(2)]
+    sales_rows = [
+        _sales_row(1, total_sqm=3000.0, months=3),
+        _sales_row(2, total_sqm=2000.0, months=3),
+    ]
+    bucket_rows = [
+        _bucket_row(1, "студия", units_sold=38, sqm_sold=1520.0),
+        _bucket_row(1, "1", units_sold=22, sqm_sold=990.0),
+        _bucket_row(2, "студия", units_sold=18, sqm_sold=720.0),
+        _bucket_row(2, "1", units_sold=13, sqm_sold=585.0),
+    ]
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=bucket_rows)
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    assert "студия" in result.by_room_bucket
+    assert "1" in result.by_room_bucket
+    # студия: 38+18=56 units, complexes from obj 1 and 2
+    assert result.by_room_bucket["студия"]["units"] == 56
+    assert result.by_room_bucket["студия"]["complexes_count"] == 2
+    # 1-к: 22+13=35 units
+    assert result.by_room_bucket["1"]["units"] == 35
+    # sqm rounded
+    assert result.by_room_bucket["студия"]["sqm"] == pytest.approx(2240.0)
+
+
+def test_by_room_bucket_empty_when_no_bucket_data():
+    """Если bucket query вернул пустой список — by_room_bucket пустой dict."""
+    comp_rows = [_comp_row(1)]
+    sales_rows = [_sales_row(1, total_sqm=5000.0, months=5)]
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=[])
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    assert result.by_room_bucket == {}
+
+
+def test_sample_competitors_include_by_room_bucket():
+    """sample_competitors каждого элемента содержит by_room_bucket."""
+    comp_rows = [_comp_row(1), _comp_row(2)]
+    sales_rows = [
+        _sales_row(1, total_sqm=6000.0, months=4),
+        _sales_row(2, total_sqm=4000.0, months=4),
+    ]
+    bucket_rows = [
+        _bucket_row(1, "2", units_sold=30, sqm_sold=1800.0),
+        _bucket_row(2, "2", units_sold=20, sqm_sold=1200.0),
+    ]
+    db = _make_db(comp_rows=comp_rows, sales_rows=sales_rows, bucket_rows=bucket_rows)
+
+    with patch(
+        "app.services.site_finder.velocity._get_ekb_median",
+        return_value=_EKB_MEDIAN_FALLBACK_SQM_PER_MONTH,
+    ):
+        result = compute_velocity(db, parcel_geom_wkt=_PARCEL_WKT)
+
+    assert result is not None
+    for comp in result.sample_competitors:
+        assert "by_room_bucket" in comp
+    # obj_id=1 had bucket data
+    top = result.sample_competitors[0]
+    assert top["obj_id"] == 1
+    assert top["by_room_bucket"].get("2") == 30