perf(site-finder): pushdown obj_pricing CTE to nearby competitors in analyze (#70, #307)

obj_pricing (OBJ-3 #1332) агрегировал objective_lots (2.1ГБ) по всем ~286
маппингам через GROUP BY domrf_obj_id, но LEFT JOIN потреблял только ~43
ближних (3км) obj_id — далёкие медианы считались и выбрасывались.

Добавлен nearby_obj CTE (3км-множество считается раз), obj_pricing scoped
через WHERE cm.domrf_obj_id IN (SELECT obj_id FROM nearby_obj), финальный
SELECT идёт FROM nearby_obj → scope == потребляемое множество. Планировщик
теперь bitmap index scan вместо full parallel seq scan.

Прод EXPLAIN (центр ЕКБ): ~700мс→~290мс, objective_lots buffer reads
212658→20561 (~10x). Байт-идентично: per-obj_id агрегаты независимы,
0 расхождений на 43 ближних группах. Цена/fallback не тронуты (scope-only).
Регрешн-гард против отката к full-table агрегации.

Refs #70

backend/app/api/v1/parcels.py

@@ -1600,7 +1600,23 @@ def analyze_parcel(
                  WHERE latitude IS NOT NULL
                  ORDER BY obj_id, snapshot_date DESC NULLS LAST
             ),
+            nearby_obj AS (
+                -- Ближние конкуренты (3 км) — единственные obj_id, чьи цены реально
+                -- попадают в выдачу. Считаем геофильтр один раз, переиспользуем в
+                -- obj_pricing (pushdown scope) и в финальном SELECT.
+                SELECT *
+                  FROM latest_obj o
+                 WHERE ST_DWithin(
+                           ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
+                           ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
+                           3000
+                       )
+            ),
             obj_pricing AS (
+                -- Pushdown: агрегируем objective_lots ТОЛЬКО для ближних obj_id.
+                -- AVG/COUNT per domrf_obj_id независимы между группами, поэтому
+                -- ограничение scope даёт байт-идентичные значения для тех obj_id,
+                -- что реально джойнятся ниже (далёкие медианы раньше считались зря).
                 SELECT
                     cm.domrf_obj_id,
                     ROUND(AVG(ol.price_per_m2_rub)::numeric, 0)  AS avg_price_per_m2_rub,
@@ -1612,6 +1628,7 @@ def analyze_parcel(
                     )                                             AS lots_with_price
                   FROM objective_complex_mapping cm
                   JOIN objective_lots ol ON ol.project_name = cm.objective_complex_name
+                 WHERE cm.domrf_obj_id IN (SELECT obj_id FROM nearby_obj)
                  GROUP BY cm.domrf_obj_id
             )
             SELECT o.obj_id,
@@ -1633,13 +1650,8 @@ def analyze_parcel(
                        ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
                        ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
                    ) AS distance_m
-              FROM latest_obj o
+              FROM nearby_obj o
               LEFT JOIN obj_pricing p ON p.domrf_obj_id = o.obj_id
-             WHERE ST_DWithin(
-                       ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
-                       ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
-                       3000
-                   )
              ORDER BY CASE o.site_status WHEN 'Строящиеся' THEN 0 ELSE 1 END,
                       distance_m ASC
              LIMIT 20

backend/tests/api/v1/test_analyze_competitors_status.py

@@ -5,13 +5,18 @@ Mock-based — не требуют живой БД.
   - поля site_status и ready_dt присутствуют в каждом элементе competitors
   - первые позиции занимают строящиеся ЖК (site_status='Строящиеся')
   - сданные ЖК идут после строящихся
+  - obj_pricing CTE ограничен ближними obj_id (pushdown perf-guard, #70/#307)
 """
 
 from __future__ import annotations
 
 import datetime
+import inspect
+import re
 from unittest.mock import MagicMock
 
+from app.api.v1 import parcels as parcels_module
+
 # ── Вспомогательные фабрики ───────────────────────────────────────────────────
 
 
@@ -120,3 +125,61 @@ class TestCompetitorsSortOrder:
         # Старая логика: первым шёл ЖК с flat_count=800 (Сданные)
         assert first_old["flat_count"] == 800
         assert first_old["site_status"] == "Сданные"
+
+
+class TestObjPricingPushdown:
+    """obj_pricing CTE считается ТОЛЬКО для ближних obj_id (perf-guard, #70/#307).
+
+    Старая версия (#1332 OBJ-3) агрегировала objective_lots (2.1 ГБ) по ВСЕМ
+    ~286 маппингам и выбрасывала далёкие медианы в LEFT JOIN. Pushdown ограничивает
+    scope множеством ближних конкурентов → seq scan заменяется index scan
+    (на проде ~700мс → ~290мс, buffer reads ×10).
+
+    Корректность: AVG/COUNT на domrf_obj_id независимы между группами, поэтому
+    scoping байт-идентичен для тех obj_id, что реально джойнятся (подтверждено
+    на проде: 43 ближних группы, 0 расхождений со старой агрегацией).
+
+    Тест защищает от тихого реверта на полнотабличную агрегацию.
+    """
+
+    @staticmethod
+    def _competitor_sql() -> str:
+        """Нормализованный (схлопнутые пробелы) текст competitor-SQL из analyze_parcel."""
+        src = inspect.getsource(parcels_module.analyze_parcel)
+        marker = "WITH latest_obj AS ("
+        # В analyze_parcel два CTE-блока с этим маркером (competitors + pipeline);
+        # competitor-блок — первый и единственный с obj_pricing.
+        start = src.index(marker)
+        block = src[start : src.index('"""', start)]
+        return re.sub(r"\s+", " ", block)
+
+    def test_obj_pricing_scoped_to_nearby(self) -> None:
+        """obj_pricing фильтрует domrf_obj_id по ближним (pushdown), не агрегирует всё."""
+        sql = self._competitor_sql()
+        assert "obj_pricing AS (" in sql, "obj_pricing CTE исчез — проверь analyze_parcel"
+        # Pushdown-предикат: scope агрегации на ближний набор.
+        assert "cm.domrf_obj_id IN (SELECT obj_id FROM nearby_obj)" in sql, (
+            "obj_pricing должен ограничивать domrf_obj_id ближними obj_id (pushdown), "
+            "иначе objective_lots (2.1ГБ) сканируется целиком — perf-регрессия #70/#307"
+        )
+
+    def test_final_select_reuses_nearby_cte(self) -> None:
+        """Финальный SELECT и obj_pricing берут один и тот же nearby_obj — scope == consumed."""
+        sql = self._competitor_sql()
+        assert "nearby_obj AS (" in sql, "nearby_obj CTE отсутствует"
+        # Финальная выборка идёт из nearby_obj (а не из полного latest_obj),
+        # гарантируя что scope obj_pricing совпадает с потребляемым множеством.
+        assert "FROM nearby_obj o" in sql, (
+            "финальный SELECT должен читать из nearby_obj, чтобы scope цен "
+            "совпадал с выводимыми конкурентами"
+        )
+
+    def test_aggregation_unchanged(self) -> None:
+        """Агрегатные выражения (AVG/COUNT FILTER) не тронуты — менялся только scope."""
+        sql = self._competitor_sql()
+        # Логика цен/fallback (#1332) неизменна: те же агрегаты per domrf_obj_id.
+        assert "ROUND(AVG(ol.price_per_m2_rub)::numeric, 0) AS avg_price_per_m2_rub" in sql
+        assert "COUNT(*) FILTER (WHERE ol.price_per_m2_rub IS NOT NULL )" in sql.replace(
+            "\n", " "
+        ) or "lots_with_price" in sql
+        assert "GROUP BY cm.domrf_obj_id" in sql