fix(site_finder): correct best-layouts supply fan-out + objects-first perf

domrf_kn_objects is a snapshot dimension (UNIQUE (obj_id, snapshot_date), ~8
snapshots/obj_id). _SUPPLY_BATCH_SQL joined flats to ALL object-snapshot rows
(no o.snapshot_date filter), counting each flat ~8.5x → supply_units_in_radius
inflated ~8.5x, sold_pct_of_supply deflated ~8.5x, is_oversold under-fired
(all user-facing, best_layouts.py:571-611; sold_pct=deals/supply is a raw
ratio so no canceling).

Fix: dedup objects to one row per obj_id (latest-snapshot coords) via
DISTINCT ON in an objects-first MATERIALIZED CTE, then join domrf_kn_flats via
idx_kn_flats_obj. units now = one count per flat (prod cross-check at radius
1.5km: units == count(*) == count(DISTINCT f.id) == 9612 for 65 objects;
correction factor 8.56x at 1.5km, 9.13x at 1.0km). This also aligns the supply
denominator with the deals numerator (_COMPETITORS_IN_RADIUS_SQL already uses
DISTINCT ON latest snapshot).

Perf bonus: objects-first avoids the parallel seq scan of the ~376k-row flats
snapshot. radius 1.5km / snapshot 2026-05-17: 240ms/~28k buffers/6712 disk
reads -> 49ms/1554 buffers/0 disk reads (~5x).

Tests: add SQL-text fan-out guard (DISTINCT ON + MATERIALIZED, no bare
flats->objects join); update stale EXPLAIN mirror in test_phantom_columns.

USER-FACING: best-layouts supply/sold_pct/is_oversold/sell-out-months shift
~8.5x toward correct (frontend BestLayoutsBlock only; ТЗ recommendation + PDF
unchanged — they derive from sum_deals, not supply). Deep-reviewed (APPROVE).

backend/app/services/site_finder/best_layouts.py

@@ -210,7 +210,52 @@ _INLINE_VELOCITY_SQL = text("""
 # становились dead lookup. SF-08 euro-биннинг откладываем до момента, когда его
 # начнёт отдавать velocity-сторона.
 
+# Perf + CORRECTNESS fix (perf/best-layouts-objects-first):
+# objects-first MATERIALIZED CTE + устранение fan-out по снимкам объекта.
+#
+# === Корректность (главное) ===
+# domrf_kn_objects — это snapshot-таблица: UNIQUE (obj_id, snapshot_date)
+# [uq_kn_objects_id_snap], ~8 снимков на obj_id, координаты могут меняться между
+# снимками. Прежний запрос джойнил flats к domrf_kn_objects по obj_id БЕЗ фильтра
+# o.snapshot_date → каждая квартира считалась по разу на КАЖДУЮ строку-объект,
+# прошедшую ST_DWithin (fan-out ~×8.5 на проде, радиус 1.5км: 82251 → 9612).
+# Последствие в выдаче (best_layouts.py): supply_units_in_radius завышен ~8.5×,
+# sold_pct_of_supply (= deals/supply) занижен ~8.5×, is_oversold недосрабатывает.
+# Фикс: latest_obj дедуплицирует до ОДНОЙ строки на obj_id (последний снимок —
+# самые свежие координаты) через DISTINCT ON; ST_DWithin применяется к этим
+# каноническим координатам. nearby содержит уникальные obj_id → JOIN с flats не
+# фанаутит → units = ровно один счёт на квартиру (cross-check на проде:
+# units == count(DISTINCT f.id) == 9612 для 65 объектов в радиусе).
+#
+# === Перформанс ===
+# Селективный предикат — ST_DWithin по маленькой таблице объектов (~1.5k уникальных
+# obj_id), поэтому фильтруем объекты ПЕРВЫМИ, затем джойним большую flats
+# (последний полный снимок ~376k строк) через idx_kn_flats_obj. Прежний flats-first
+# план параллельно seq-сканировал весь снимок flats → ~28k buffers / ~240-360ms
+# (cold ~549ms). Этот план: objects-first + меньше строк после дедупа.
+#
+# MATERIALIZED у nearby — обязателен: без него планировщик инлайнит CTE и
+# возвращается к flats-first seq scan (domrf_kn_objects не имеет GIST-индекса).
 _SUPPLY_BATCH_SQL = text("""
+    WITH latest_obj AS (
+        SELECT DISTINCT ON (o.obj_id)
+            o.obj_id, o.latitude, o.longitude
+        FROM domrf_kn_objects o
+        WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
+        ORDER BY o.obj_id, o.snapshot_date DESC
+    ),
+    nearby AS MATERIALIZED (
+        SELECT lo.obj_id
+        FROM latest_obj lo
+        WHERE ST_DWithin(
+            ST_SetSRID(ST_MakePoint(lo.longitude, lo.latitude), 4326)::geography,
+            ST_SetSRID(
+                ST_MakePoint(CAST(:center_lon AS float), CAST(:center_lat AS float)),
+                4326
+            )::geography,
+            CAST(:radius_m AS float)
+        )
+    )
     SELECT
         CASE
             WHEN f.is_studio = TRUE OR f.flat_type = 'Квартира-студия' THEN 'studio'
@@ -228,18 +273,10 @@ _SUPPLY_BATCH_SQL = text("""
             ELSE '100+'
         END AS ab,
         COUNT(*) AS units
-    FROM domrf_kn_flats f
-    JOIN domrf_kn_objects o ON f.obj_id = o.obj_id
-    WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
-      AND ST_DWithin(
-          ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
-          ST_SetSRID(
-              ST_MakePoint(CAST(:center_lon AS float), CAST(:center_lat AS float)),
-              4326
-          )::geography,
-          CAST(:radius_m AS float)
-      )
-      AND f.snapshot_date = CAST(:latest_snap AS date)
+    FROM nearby n
+    JOIN domrf_kn_flats f
+        ON f.obj_id = n.obj_id
+       AND f.snapshot_date = CAST(:latest_snap AS date)
     GROUP BY rb, ab
 """)
 

backend/tests/integration/test_phantom_columns.py

@@ -261,10 +261,33 @@ class TestDomrfKnFlats:
     """Проверяем таблицу domrf_kn_flats и её колонки."""
 
     def test_supply_batch_query(self, phantom_check_session: Session) -> None:
-        """best_layouts._SUPPLY_BATCH_SQL — total_area, rooms, is_studio, flat_type."""
+        """best_layouts._SUPPLY_BATCH_SQL — total_area, rooms, is_studio, flat_type.
+
+        Зеркалит objects-first MATERIALIZED CTE (latest-snapshot dedup, anti-fan-out)
+        с хардкод-значениями вместо bind-параметров — EXPLAIN-гейт на phantom-колонки
+        остаётся осмысленным после перехода с flats-first на objects-first.
+        """
         _explain(
             phantom_check_session,
             """
+            WITH latest_obj AS (
+                SELECT DISTINCT ON (o.obj_id)
+                    o.obj_id, o.latitude, o.longitude
+                FROM domrf_kn_objects o
+                WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
+                ORDER BY o.obj_id, o.snapshot_date DESC
+            ),
+            nearby AS MATERIALIZED (
+                SELECT lo.obj_id
+                FROM latest_obj lo
+                WHERE ST_DWithin(
+                    ST_SetSRID(ST_MakePoint(lo.longitude, lo.latitude), 4326)::geography,
+                    ST_SetSRID(
+                        ST_MakePoint(CAST(60.6 AS float), CAST(56.84 AS float)), 4326
+                    )::geography,
+                    CAST(1500 AS float)
+                )
+            )
             SELECT
                 CASE
                     WHEN f.is_studio = TRUE OR f.flat_type = 'Квартира-студия' THEN 'studio'
@@ -282,10 +305,10 @@ class TestDomrfKnFlats:
                     ELSE '100+'
                 END AS ab,
                 COUNT(*) AS units
-            FROM domrf_kn_flats f
-            JOIN domrf_kn_objects o ON f.obj_id = o.obj_id
-            WHERE o.latitude IS NOT NULL AND o.longitude IS NOT NULL
-              AND f.snapshot_date = CAST('2026-01-01' AS date)
+            FROM nearby n
+            JOIN domrf_kn_flats f
+                ON f.obj_id = n.obj_id
+               AND f.snapshot_date = CAST('2026-01-01' AS date)
             GROUP BY rb, ab
             """,
         )

backend/tests/services/site_finder/test_best_layouts.py

@@ -393,6 +393,29 @@ def test_supply_does_not_emit_euro_buckets() -> None:
     assert "'euro-2'" not in sql_text, "_SUPPLY_BATCH_SQL вернул euro-2 — see #1229"
 
 
+def test_supply_dedups_objects_to_latest_snapshot() -> None:
+    """Regression guard против object-snapshot fan-out (supply units ~8.5x inflation).
+
+    domrf_kn_objects — snapshot-dimension (UNIQUE (obj_id, snapshot_date), ~8
+    снимков/obj_id). Если supply снова заджойнит flats НАПРЯМУЮ к objects без
+    дедупа до одного снимка на obj_id, каждый flat посчитается ~8.5x →
+    supply_units_in_radius завышен, sold_pct_of_supply занижен, is_oversold
+    недо-фаерит (все три — user-facing). Фикс: objects-first MATERIALIZED CTE
+    с DISTINCT ON (obj_id) по последнему снимку, flats джойнятся к нему.
+    """
+    from app.services.site_finder.best_layouts import _SUPPLY_BATCH_SQL
+
+    sql_text = str(_SUPPLY_BATCH_SQL.text)
+    # objects дедуплицируются до одного (последнего) снимка на obj_id
+    assert "DISTINCT ON (o.obj_id)" in sql_text, "supply не дедупит objects → fan-out вернётся"
+    assert "snapshot_date DESC" in sql_text, "DISTINCT ON должен брать ПОСЛЕДНИЙ снимок"
+    # objects-first CTE материализован — иначе планнер инлайнит → flats-first seq scan
+    assert "MATERIALIZED" in sql_text, "nearby CTE должен быть MATERIALIZED (план objects-first)"
+    # flats джойнятся к дедуплицированному nearby, НЕ напрямую к raw objects
+    assert "FROM nearby" in sql_text
+    assert "JOIN domrf_kn_objects" not in sql_text, "прямой flats→objects join = fan-out по снимкам"
+
+
 # ── Тесты _cap_and_redistribute (Fix SF-09 review) ───────────────────────────