fix(site-finder): per-object latest snapshot в competitors avg-price SQL (#2445-A5)

_AVG_PRICE_SQL фильтровал domrf_kn_flats по одной глобальной MAX(snapshot_date),
но таблица — per-object time series (UNIQUE(obj_id, snapshot_date)). На глобальной
дате обычно присутствует лишь 1 объект → у всех остальных конкурентов 0 строк →
avg_price_per_m2=NULL → _price_similarity падает на нейтральные 0.5, молча
деградируя relevance-ранжирование.

Портируем flats_latest DISTINCT ON (obj_id) ... ORDER BY snapshot_date DESC, id DESC
CTE из best_layouts.py (#1956) — каждый конкурент берёт СВОЙ последний снапшот.
Индекс 174_domrf_kn_flats_obj_snapshot_idx уже обслуживает CTE (новая миграция
не нужна). Objective-fallback композиция не тронута: missing_price_ids считается
после domrf-запроса → естественно сжимается.

Refs #2445

backend/app/services/site_finder/competitors.py

@@ -449,16 +449,42 @@ _COMPETITORS_SQL = text("""
 # scraper UPSERT per snapshot). Без фильтра snapshot_date AVG усреднял
 # ИСТОРИЮ цен (stale на растущем рынке). Берём ТОЛЬКО latest snapshot — зеркало
 # паттерна best_layouts._SUPPLY_BATCH_SQL и _COMPETITORS_SQL DISTINCT ON.
+#
+# === PER-OBJECT snapshot fix (#2445 A5, byte-identical bug к #1956 в best_layouts) ===
+# domrf_kn_flats — ПО-ОБЪЕКТНЫЙ time-series: каждый ЖК скрейпится в СВОЙ день,
+# UNIQUE-снимок (obj_id, snapshot_date), см. migration 174_domrf_kn_flats_obj_
+# snapshot_idx.sql. Прежний фильтр брал ОДНУ глобальную дату
+# `f.snapshot_date = MAX(snapshot_date по всей таблице)` — на этой единственной
+# глобал-max дате присутствует обычно лишь 1 объект (скрейпнутый последним) →
+# у ВСЕХ остальных конкурентов в радиусе 0 строк flats → avg_price_per_m2=NULL
+# → _price_similarity(None, ...) молча возвращает нейтраль 0.5, тихо деградируя
+# relevance-ранжирование конкурентов без единого флага.
+# Фикс (зеркало best_layouts.flats_latest, #1956): flats_latest берёт для КАЖДОГО
+# obj_id его СОБСТВЕННЫЙ последний снимок (DISTINCT ON (f.obj_id) ... ORDER BY
+# f.obj_id, snapshot_date DESC, id DESC), затем self-join к domrf_kn_flats по
+# (obj_id, snapshot_date) достаёт квартиры именно этого снимка. AVG считается
+# per-object на СВЕЖЕЙ дате объекта, а не на единственной глобал-max дате.
+# Слоение с OBJ-3 (#307) не меняется: этот запрос остаётся ПЕРВЫМ источником
+# (domrf), _OBJECTIVE_PRICE_FALLBACK_SQL — additive fallback ТОЛЬКО для obj_id,
+# у которых и после фикса нет domrf-цены (см. get_competitors шаг 4a). Ожидаемый
+# эффект: domrf возвращает реальные цены для БОЛЬШЕ конкурентов → fallback нужен
+# реже, но композиция (missing_price_ids / price_source) не меняется.
 _AVG_PRICE_SQL = text("""
+    WITH flats_latest AS (
+        SELECT DISTINCT ON (f.obj_id)
+            f.obj_id, f.snapshot_date
+        FROM domrf_kn_flats f
+        WHERE f.obj_id = ANY(:obj_ids)
+        ORDER BY f.obj_id, f.snapshot_date DESC, f.id DESC
+    )
     SELECT
         f.obj_id,
         AVG(f.price_per_m2) FILTER (WHERE f.price_per_m2 IS NOT NULL)
             AS avg_price_per_m2
-    FROM domrf_kn_flats f
-    WHERE f.obj_id = ANY(:obj_ids)
-      AND f.snapshot_date = (
-          SELECT MAX(snapshot_date) FROM domrf_kn_flats
-      )
+    FROM flats_latest fl
+    JOIN domrf_kn_flats f
+        ON f.obj_id = fl.obj_id
+       AND f.snapshot_date = fl.snapshot_date
     GROUP BY f.obj_id
 """)
 

backend/tests/api/v1/test_parcel_competitors.py

@@ -545,6 +545,52 @@ def test_competitors_no_price_anywhere_is_none() -> None:
         app.dependency_overrides.clear()
 
 
+def test_competitors_multi_object_radius_yields_per_object_prices() -> None:
+    """Регрессия #2445 A5 (byte-identical к best_layouts #1956): несколько
+    конкурентов в радиусе → КАЖДЫЙ получает свою domrf-цену, а не «1 объект на
+    global-max снимке, у остальных NULL».
+
+    Мок здесь не исполняет реальный SQL (_AVG_PRICE_SQL — строковые инварианты
+    см. test_competitors_avg_price_sql.py), но фиксирует ОЖИДАЕМЫЙ КОНТРАКТ:
+    _make_db может вернуть price_rows для ВСЕХ obj_id разом (per-object latest
+    snapshot делает это возможным на живой БД) — раньше баг гарантировал, что
+    price_rows содержал не более 1 строки на весь набор конкурентов.
+    """
+    rows = [
+        _obj_row(obj_id=1, distance_m=100.0),
+        _obj_row(obj_id=2, distance_m=200.0),
+        _obj_row(obj_id=3, distance_m=300.0),
+    ]
+    price_rows = [
+        _price_row(obj_id=1, price=140_000.0),
+        _price_row(obj_id=2, price=155_000.0),
+        _price_row(obj_id=3, price=162_000.0),
+    ]
+    db = _make_db(coord=_coord_row(), obj_rows=rows, price_rows=price_rows)
+
+    from app.core.db import get_db
+
+    app.dependency_overrides[get_db] = _override_db(db)
+    try:
+        client = TestClient(app)
+        resp = client.post(
+            "/api/v1/parcels/66:41:0303161:5/competitors",
+            json={"radius_km": 1.0, "time_window": "last_quarter"},
+        )
+        assert resp.status_code == 200, resp.text
+        comps = {c["obj_id"]: c for c in resp.json()["competitors"]}
+        assert len(comps) == 3
+        expected = {1: 140_000.0, 2: 155_000.0, 3: 162_000.0}
+        for obj_id, price in expected.items():
+            assert comps[obj_id]["avg_price_per_m2"] == pytest.approx(price), (
+                f"obj_id={obj_id}: все конкуренты должны получить СВОЮ domrf-цену, "
+                "не only-one-object-on-global-max-snapshot регрессия"
+            )
+            assert comps[obj_id]["price_source"] == "domrf"
+    finally:
+        app.dependency_overrides.clear()
+
+
 def test_competitors_is_active_flag() -> None:
     """is_active=True для русского 'Строящиеся' (реальная prod-форма #1213),
     False для 'Сданные'/null/прочих."""

backend/tests/services/site_finder/test_competitors_avg_price_sql.py

@@ -0,0 +1,76 @@
+"""Пин SQL-формы competitors._AVG_PRICE_SQL (эпик #2445, пункт A5).
+
+Byte-identical бага к best_layouts._SUPPLY_BATCH_SQL (#1956): domrf_kn_flats —
+ПО-ОБЪЕКТНЫЙ time-series (каждый ЖК скрейпится в свой день, UNIQUE(obj_id,
+snapshot_date), migration 174_domrf_kn_flats_obj_snapshot_idx.sql). Прежний
+запрос фильтровал по ОДНОЙ глобальной дате
+`f.snapshot_date = (SELECT MAX(snapshot_date) FROM domrf_kn_flats)` — на этой
+единственной дате обычно присутствует лишь 1 объект (скрейпнутый последним),
+поэтому у ВСЕХ остальных конкурентов в радиусе получалось 0 строк flats →
+avg_price_per_m2 = NULL → _price_similarity(None, ...) молча возвращал
+нейтраль 0.5, тихо деградируя relevance-ранжирование без единого флага.
+
+FIX (порт #1956 в этот файл): flats_latest CTE — DISTINCT ON (f.obj_id) ...
+ORDER BY f.obj_id, snapshot_date DESC, id DESC — берёт для КАЖДОГО obj_id его
+СОБСТВЕННЫЙ последний снимок, затем self-join к domrf_kn_flats по
+(obj_id, snapshot_date) достаёт квартиры именно этого снимка.
+
+Эти тесты — строковые инварианты на скомпилированном SQL (без живой БД), чтобы
+будущий рефактор случайно не вернул глобал-max-date регрессию.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services.site_finder.competitors import _AVG_PRICE_SQL
+
+_SQL = str(_AVG_PRICE_SQL)
+_SQL_FLAT = re.sub(r"\s+", " ", _SQL).lower()
+
+
+def test_avg_price_uses_flats_latest_per_object_cte() -> None:
+    """flats_latest CTE обязателен — без него avg_price снова считается на
+    единственной глобал-max дате (регрессия #2445 A5, byte-identical к #1956)."""
+    assert "flats_latest" in _SQL_FLAT, "нет flats_latest CTE → возврат к global-max-date багу"
+    assert "distinct on (f.obj_id)" in _SQL_FLAT, "flats не дедупятся per-object"
+
+
+def test_avg_price_orders_latest_snapshot_first() -> None:
+    """ORDER BY последним снапшотом вперёд (snapshot_date DESC, id DESC) — DISTINCT ON
+    берёт первую строку группы → собственный последний снимок объекта."""
+    assert "snapshot_date desc" in _SQL_FLAT
+    assert "id desc" in _SQL_FLAT
+    m = re.search(r"order by f\.obj_id, f\.snapshot_date desc, f\.id desc", _SQL_FLAT)
+    assert m is not None, "DISTINCT ON должен сортировать per obj_id, затем latest snapshot"
+
+
+def test_avg_price_no_global_max_snapshot_date() -> None:
+    """Регрессия #2445 A5: единственный глобальный MAX(snapshot_date) по всей
+    таблице не должен вернуться — это и есть исходный баг (1 объект на весь
+    global-max день, остальные competitors получают avg_price=NULL)."""
+    assert "select max(snapshot_date) from domrf_kn_flats" not in _SQL_FLAT
+    assert "(select max(snapshot_date)" not in _SQL_FLAT
+
+
+def test_avg_price_joins_flats_by_obj_id_and_snapshot_date() -> None:
+    """Self-join к domrf_kn_flats обязан фильтровать И по obj_id, И по
+    snapshot_date дедуплицированного flats_latest — иначе снова fan-out по
+    истории снимков объекта (см. #1210)."""
+    assert re.search(r"join\s+domrf_kn_flats\s+f\b", _SQL_FLAT)
+    assert "f.obj_id = fl.obj_id" in _SQL_FLAT
+    assert "f.snapshot_date = fl.snapshot_date" in _SQL_FLAT
+
+
+def test_avg_price_keeps_price_not_null_filter() -> None:
+    """AVG(...) FILTER (WHERE price_per_m2 IS NOT NULL) сохранён (Issue #112/227
+    — status='sold' фильтр убран, но NULL-цены по-прежнему исключаются из AVG)."""
+    assert "avg(f.price_per_m2) filter (where f.price_per_m2 is not null)" in _SQL_FLAT
+
+
+def test_avg_price_no_double_colon_cast() -> None:
+    """psycopg v3 / backend.md: никаких :name::type (игнорируется парсером)."""
+    assert not re.search(r":[a-z_]+::[a-z]", _SQL)