fix(site-finder): align competitor flats_sold to objective_lots.contract_date (#1926) (#1927)

backend/app/services/site_finder/competitors.py

@@ -445,26 +445,15 @@ _COMPETITORS_SQL = text("""
 # (99.8% NULL) — фильтр давал 0 строк и avg_price_per_m2 всегда None (Issue #112/227).
 # AVG по всем квартирам с price_per_m2 IS NOT NULL даёт корректную среднюю цену ЖК.
 #
-# #949 PR B: добавлена колонка flats_sold (COUNT квартир со status LIKE '%прод%')
-# — нужна для stage_at_horizon (available = flat_count - flats_sold). status
-# заполнен слабо (~0.2%), поэтому flats_sold чаще будет 0/мало → stage_at_horizon
-# падает в нейтраль 0.5 (graceful, см. helper). Колонка добавлена в ТОТ ЖЕ запрос
-# (без 4-го db.execute), чтобы не ломать существующие callers/тесты.
 # #1210: domrf_kn_flats версионируется (UNIQUE(id, snapshot_date) — м.50,
 # scraper UPSERT per snapshot). Без фильтра snapshot_date AVG усреднял
-# ИСТОРИЮ цен (stale на растущем рынке), а COUNT '%прод%' множил sold ×N
+# ИСТОРИЮ цен (stale на растущем рынке). Берём ТОЛЬКО latest snapshot — зеркало
-# снапшотов (raw_sold/flat_count кратно завышен → попадал в гард-нейтраль
-# 0.5 или искажал stage_at_horizon). Берём ТОЛЬКО latest snapshot — зеркало
 # паттерна best_layouts._SUPPLY_BATCH_SQL и _COMPETITORS_SQL DISTINCT ON.
 _AVG_PRICE_SQL = text("""
     SELECT
         f.obj_id,
         AVG(f.price_per_m2) FILTER (WHERE f.price_per_m2 IS NOT NULL)
-            AS avg_price_per_m2,
+            AS avg_price_per_m2
-        COUNT(*) FILTER (
-            WHERE LOWER(COALESCE(f.status, '')) LIKE '%прод%'
-               OR LOWER(COALESCE(f.status, '')) = 'sold'
-        ) AS flats_sold
     FROM domrf_kn_flats f
     WHERE f.obj_id = ANY(:obj_ids)
       AND f.snapshot_date = (
@@ -473,6 +462,59 @@ _AVG_PRICE_SQL = text("""
     GROUP BY f.obj_id
 """)
 
+# Кол-во проданных квартир (flats_sold) для набора obj_id — нужно для
+# stage_at_horizon (available = flat_count - flats_sold).
+#
+# #1926 (аудит #1871): РАНЬШЕ flats_sold считался из domrf_kn_flats.status
+# (LIKE '%прод%'/'sold'). status заполнен в ~0.2% строк (99.8% NULL) → массовый
+# UNDERCOUNT (в репортнутом кейсе ~1600 ед. недосчёта): почти все конкуренты
+# получали flats_sold≈0 → stage_at_horizon вырождался в нейтраль/«ничего не
+# продано», и блок конкурентов РАСХОДИЛСЯ с MarketMetrics, который считает sold
+# по objective_lots.
+#
+# FIX: считаем sold ИЗ ТОГО ЖЕ ИСТОЧНИКА И ПО ТОМУ ЖЕ ПРЕДИКАТУ, что
+# MarketMetrics (_STOCK_SQL): лот продан, если
+#   is_sold IS TRUE OR contract_date IS NOT NULL OR LOWER(status)='продан'
+# (Объектив заполняет эти поля неконсистентно, поэтому берём ИХ ИЛИ; см.
+# market_metrics._STOCK_SQL:285-289). Так flats_sold побайтово зеркалит n_sold.
+#
+# Мост domrf-obj → лоты — тот же, что в _OBJECTIVE_PRICE_FALLBACK_SQL
+# (primary_price CTE): domrf_kn_objects.obj_id == objective_complex_mapping
+# .domrf_obj_id → objective_complex_mapping.objective_complex_name ==
+# objective_lots.project_name.
+#
+# ВАЖНО (code-review #1927): objective_complex_mapping UNIQUE — это
+# (objective_complex_name, objective_group), а НЕ domrf_obj_id (68_schema:321).
+# Значит ОДИН obj_id может иметь НЕСКОЛЬКО mapping-строк (то же имя, разный
+# objective_group) → наивный COUNT(*) по cm JOIN ol задвоил бы лоты (GROUP BY
+# схлопывает выходные строки, но не дубли ВНУТРИ агрегата). Поэтому
+# COUNT(DISTINCT ol.objective_lot_id) — дедуп на уровне лота, fan-out-safe.
+# (_OBJECTIVE_PRICE_FALLBACK не прецедент: percentile/median dup-нечувствительны.)
+#
+# premise_kind='квартира' — как в _STOCK_SQL: без него проданные паркинги/
+# кладовые/нежилое считались бы как проданные квартиры (over-count).
+# objective_lots — current-state UPSERT (ON CONFLICT (objective_lot_id) DO
+# UPDATE, ровно одна строка на лот) → snapshot-дедуп НЕ нужен (как в _STOCK_SQL).
+#
+# Дефолт premise_kind зеркалит market_metrics.compute_market_metrics
+# (premise_kind="квартира") — оба считают продажи жилых лотов.
+_SOLD_PREMISE_KIND = "квартира"
+_SOLD_COUNT_SQL = text("""
+    SELECT
+        cm.domrf_obj_id AS obj_id,
+        COUNT(DISTINCT ol.objective_lot_id) FILTER (
+            WHERE ol.is_sold IS TRUE
+               OR ol.contract_date IS NOT NULL
+               OR LOWER(COALESCE(ol.status, '')) = 'продан'
+        ) AS flats_sold
+    FROM objective_complex_mapping cm
+    JOIN objective_lots ol
+        ON ol.project_name = cm.objective_complex_name
+    WHERE cm.domrf_obj_id = ANY(:obj_ids)
+      AND ol.premise_kind = :premise_kind
+    GROUP BY cm.domrf_obj_id
+""")
+
 # OBJ-3 (#307): additive fallback цены м² из objective_lots, когда domrf-цены нет.
 # Замер прода 2026-06: на ПОСЛЕДНЕМ snapshot domrf_kn_flats.price_per_m2 = 0 строк
 # (скрапер перестал отдавать per-flat цены), поэтому _AVG_PRICE_SQL выше даёт
@@ -687,13 +729,9 @@ def get_competitors(
 
     obj_ids: list[int] = [int(r["obj_id"]) for r in rows]
 
-    # ── 4. Средняя цена м² + sold-count (graceful — таблица может быть пуста) ─
+    # ── 4. Средняя цена м² (graceful — таблица может быть пуста) ──────────────
-    # flats_sold нужен для stage_at_horizon (available = flat_count - flats_sold).
-    # Читаем через _row_get: существующие тесты мокируют только avg_price_per_m2,
-    # отсутствующий ключ flats_sold → None (не KeyError) → нейтральный stage.
     avg_price_map: dict[int, float] = {}
     price_source_map: dict[int, str] = {}
-    sold_count_map: dict[int, int] = {}
     try:
         price_rows = db.execute(_AVG_PRICE_SQL, {"obj_ids": obj_ids}).mappings().all()
         for r in price_rows:
@@ -702,11 +740,30 @@ def get_competitors(
             if price is not None:
                 avg_price_map[oid] = float(price)
                 price_source_map[oid] = "domrf"
+    except Exception:
+        logger.warning("competitors: avg_price query failed, continuing without prices")
+
+    # ── 4·sold. flats_sold из objective_lots (#1926) ─────────────────────────
+    # flats_sold нужен для stage_at_horizon (available = flat_count - flats_sold).
+    # Источник И предикат выровнены с MarketMetrics._STOCK_SQL (is_sold/
+    # contract_date/status='продан', premise_kind='квартира'), см. _SOLD_COUNT_SQL
+    # — ушли от domrf_kn_flats.status (0.2%, недосчёт). Отдельный db.execute (не
+    # ломает мок-тесты _AVG_PRICE_SQL); _row_get graceful: отсутствующий ключ
+    # flats_sold → None → нейтральный stage.
+    sold_count_map: dict[int, int] = {}
+    try:
+        sold_rows = (
+            db.execute(_SOLD_COUNT_SQL, {"obj_ids": obj_ids, "premise_kind": _SOLD_PREMISE_KIND})
+            .mappings()
+            .all()
+        )
+        for r in sold_rows:
+            oid = int(r["obj_id"])
             sold = _row_get(r, "flats_sold")
             if sold is not None:
                 sold_count_map[oid] = int(sold)
     except Exception:
-        logger.warning("competitors: avg_price query failed, continuing without prices")
+        logger.warning("competitors: sold-count query failed, continuing without flats_sold")
 
     # ── 4a. OBJ-3 (#307) additive fallback: где domrf-цены НЕТ → objective_lots ──
     # Заполняем ТОЛЬКО пробелы (obj_id без domrf-цены). Существующие domrf-выводы
@@ -743,9 +800,10 @@ def get_competitors(
     dominant_class = _dominant_class([r["obj_class"] for r in rows])
 
     # ── 5. Сборка результата + relevance_weight (ТЗ §9.1) ────────────────────
-    # flats_sold: COUNT из domrf_kn_flats по status LIKE '%прод%' (заполнен слабо
+    # flats_sold: COUNT проданных лотов из objective_lots.contract_date (#1926,
-    # ~0.2%, поэтому чаще 0/None). При known flats_sold считаем sold_pct и
+    # выровнено с MarketMetrics; раньше domrf_kn_flats.status ~0.2% → недосчёт).
-    # available для stage_at_horizon; иначе оба поля None и stage → нейтраль 0.5.
+    # При known flats_sold считаем sold_pct и available для stage_at_horizon;
+    # иначе оба поля None и stage → нейтраль 0.5.
     competitors: list[Competitor] = []
     for r in rows:
         obj_id = int(r["obj_id"])

backend/tests/api/v1/test_parcel_competitors.py

@@ -53,6 +53,18 @@ def _price_row(obj_id: int, price: float) -> MagicMock:
     return r
 
 
+def _sold_row(obj_id: int, flats_sold: int) -> MagicMock:
+    """Строка sold-count (_SOLD_COUNT_SQL, #1926).
+
+    flats_sold теперь COUNT проданных лотов из objective_lots.contract_date
+    (выровнено с MarketMetrics), а не COUNT по domrf_kn_flats.status — последний
+    заполнен ~0.2% и давал массовый недосчёт.
+    """
+    r = MagicMock()
+    r.__getitem__ = lambda self, k: {"obj_id": obj_id, "flats_sold": flats_sold}[k]
+    return r
+
+
 def _obj_price_row(obj_id: int, price: float) -> MagicMock:
     """Строка objective-fallback цены (_OBJECTIVE_PRICE_FALLBACK_SQL, OBJ-3 #307)."""
     r = MagicMock()
@@ -67,15 +79,18 @@ def _make_db(
     coord: MagicMock | None = None,
     obj_rows: list[MagicMock] | None = None,
     price_rows: list[MagicMock] | None = None,
+    sold_rows: list[MagicMock] | None = None,
     obj_price_rows: list[MagicMock] | None = None,
 ) -> MagicMock:
     """Сконструировать mock Session.
 
-    Порядок вызовов execute:
+    Порядок вызовов execute (#1926: добавлен отдельный sold-count запрос):
     1. centroid query → coord
     2. competitors query → obj_rows
     3. domrf avg_price query → price_rows
-    4. objective price fallback (OBJ-3 #307) → obj_price_rows
+    4. sold-count query (_SOLD_COUNT_SQL, objective_lots.contract_date) → sold_rows
+       (по умолчанию [] = sold неизвестен → flats_sold None, нейтральный stage)
+    5. objective price fallback (OBJ-3 #307) → obj_price_rows
        (вызывается ТОЛЬКО если у части obj_id нет domrf-цены — почти всегда,
        т.к. domrf на последнем snapshot пуст; по умолчанию [] = fallback пуст)
     """
@@ -86,6 +101,7 @@ def _make_db(
         (coord, True),
         (obj_rows or [], False),
         (price_rows or [], False),
+        (sold_rows or [], False),
         (obj_price_rows or [], False),
     ]:
         result = MagicMock()
@@ -351,13 +367,14 @@ def test_competitors_empty_radius() -> None:
 
 
 def test_competitors_sold_pct_null() -> None:
-    """sold_pct и flats_sold — None (MVP: данные недоступны из domrf_kn_objects).
+    """sold_pct и flats_sold — None, когда sold-count запрос пуст (graceful).
 
-    Полный расчёт продаж требует JOIN с domrf_kn_flats COUNT WHERE status='sold'
+    #1926: flats_sold берётся из objective_lots.contract_date (_SOLD_COUNT_SQL).
-    — отложен за пределы текущего PR.
+    Если у obj_id нет маппинга в objective_complex_mapping (coverage ~2.5%) →
+    sold-count строки нет → flats_sold/sold_pct None, stage → нейтраль 0.5.
     """
     rows = [_obj_row(obj_id=1, flat_count=200)]
-    db = _make_db(coord=_coord_row(), obj_rows=rows)
+    db = _make_db(coord=_coord_row(), obj_rows=rows)  # sold_rows default [] → None
 
     from app.core.db import get_db
 
@@ -377,6 +394,39 @@ def test_competitors_sold_pct_null() -> None:
         app.dependency_overrides.clear()
 
 
+def test_competitors_flats_sold_from_objective_lots() -> None:
+    """flats_sold/sold_pct заполняются из objective_lots (#1926).
+
+    Регрессия аудита #1871/#1926: РАНЬШЕ flats_sold считался из
+    domrf_kn_flats.status (LIKE '%прод%'), заполненного ~0.2% → массовый
+    недосчёт и расхождение с MarketMetrics. Теперь источник И предикат — как
+    в market_metrics._STOCK_SQL (is_sold OR contract_date OR status='продан',
+    premise_kind='квартира'). Здесь sold-count запрос возвращает 120 проданных
+    из 200 → sold_pct 60.0%. (SQL-структуру предиката/fan-out/premise_kind
+    проверяют test_sold_count_sql_* ниже — мок тут SQL не исполняет.)
+    """
+    rows = [_obj_row(obj_id=1, flat_count=200)]
+    sold_rows = [_sold_row(obj_id=1, flats_sold=120)]
+    db = _make_db(coord=_coord_row(), obj_rows=rows, sold_rows=sold_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
+        comp = resp.json()["competitors"][0]
+        assert comp["flats_sold"] == 120
+        assert comp["sold_pct"] == pytest.approx(60.0)
+        assert comp["flats_total"] == 200
+    finally:
+        app.dependency_overrides.clear()
+
+
 def test_competitors_avg_price_populated() -> None:
     """avg_price_per_m2 не None если domrf_kn_flats возвращает строки с ценой.
 
@@ -523,3 +573,49 @@ def test_competitors_is_active_flag() -> None:
         assert comps[3]["is_active"] is False
     finally:
         app.dependency_overrides.clear()
+
+
+# ── _SOLD_COUNT_SQL структура (#1926 / code-review #1927) ───────────────────────
+# Mock-тесты выше НЕ исполняют реальный SQL (db.execute замокан) → не ловят
+# fan-out маппинга, отсутствие premise_kind-фильтра и рассинхрон sold-предиката
+# с MarketMetrics. Эти проверки — статический guard на текст _SOLD_COUNT_SQL:
+# регресс к COUNT(*) / потеря premise_kind / усечение предиката упадёт здесь.
+# (Полноценный fan-out-тест с двумя mapping-строками на один obj_id требует
+# живой БД — он в tests/integration/test_phantom_columns.py, skip без TEST_DB.)
+
+
+def _sold_sql_text() -> str:
+    from app.services.site_finder.competitors import _SOLD_COUNT_SQL
+
+    return str(_SOLD_COUNT_SQL)
+
+
+def test_sold_count_sql_is_fanout_safe() -> None:
+    """COUNT(DISTINCT objective_lot_id) — маппинг 1:N по obj_id не задваивает.
+
+    objective_complex_mapping UNIQUE = (objective_complex_name, objective_group),
+    НЕ domrf_obj_id → один obj_id может иметь несколько mapping-строк (то же имя,
+    разный objective_group) → JOIN по project_name задвоил бы лоты. DISTINCT по
+    objective_lot_id дедуплицирует на уровне лота. Регресс к COUNT(*) → fail.
+    """
+    sql = _sold_sql_text()
+    assert "COUNT(DISTINCT ol.objective_lot_id)" in sql, (
+        "fan-out guard: маппинг не unique по domrf_obj_id — нужен COUNT(DISTINCT lot)"
+    )
+    assert "COUNT(*)" not in sql, "COUNT(*) задвоит лоты при 1:N маппинге"
+
+
+def test_sold_count_sql_filters_premise_kind() -> None:
+    """premise_kind-фильтр (как в MarketMetrics._STOCK_SQL) — без него проданные
+    паркинги/кладовые/нежилое считались бы проданными квартирами (over-count)."""
+    sql = _sold_sql_text()
+    assert "ol.premise_kind = :premise_kind" in sql
+
+
+def test_sold_count_sql_predicate_matches_market_metrics() -> None:
+    """sold-предикат побайтово совпадает с MarketMetrics: is_sold OR contract_date
+    OR status='продан' (Объектив заполняет поля неконсистентно → берём их ИЛИ)."""
+    sql = _sold_sql_text()
+    assert "ol.is_sold IS TRUE" in sql
+    assert "ol.contract_date IS NOT NULL" in sql
+    assert "LOWER(COALESCE(ol.status, '')) = 'продан'" in sql

backend/tests/integration/test_phantom_columns.py

@@ -290,18 +290,25 @@ class TestDomrfKnFlats:
             """,
         )
 
-    def test_avg_price_sold_query(self, phantom_check_session: Session) -> None:
+    def test_avg_price_query(self, phantom_check_session: Session) -> None:
-        """competitors._AVG_PRICE_SQL — price_per_m2, status='sold'."""
+        """competitors._AVG_PRICE_SQL — price_per_m2 на последнем snapshot.
+
+        #1926: status больше НЕ используется (был ~0.2%-заполнен). avg_price
+        считается по всем квартирам с known price_per_m2 на latest snapshot;
+        sold-count вынесен в _SOLD_COUNT_SQL (objective_lots.contract_date).
+        """
         _explain(
             phantom_check_session,
             """
             SELECT
                 f.obj_id,
-                AVG(f.price_per_m2) AS avg_price_per_m2
+                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(ARRAY[1, 2, 3])
-              AND f.price_per_m2 IS NOT NULL
+              AND f.snapshot_date = (
-              AND f.status = 'sold'
+                  SELECT MAX(snapshot_date) FROM domrf_kn_flats
+              )
             GROUP BY f.obj_id
             """,
         )
@@ -325,6 +332,34 @@ class TestObjectiveTables:
             """,
         )
 
+    def test_sold_count_query(self, phantom_check_session: Session) -> None:
+        """competitors._SOLD_COUNT_SQL (#1926, code-review #1927) — mapping → lots.
+
+        flats_sold выровнен с MarketMetrics._STOCK_SQL: предикат
+        is_sold/contract_date/status='продан' + premise_kind='квартира'.
+        COUNT(DISTINCT objective_lot_id) — fan-out-safe (mapping НЕ unique по
+        domrf_obj_id). Мост objective_complex_mapping.domrf_obj_id →
+        objective_complex_name == objective_lots.project_name.
+        """
+        _explain(
+            phantom_check_session,
+            """
+            SELECT
+                cm.domrf_obj_id AS obj_id,
+                COUNT(DISTINCT ol.objective_lot_id) FILTER (
+                    WHERE ol.is_sold IS TRUE
+                       OR ol.contract_date IS NOT NULL
+                       OR LOWER(COALESCE(ol.status, '')) = 'продан'
+                ) AS flats_sold
+            FROM objective_complex_mapping cm
+            JOIN objective_lots ol
+                ON ol.project_name = cm.objective_complex_name
+            WHERE cm.domrf_obj_id = ANY(ARRAY[1, 2, 3])
+              AND ol.premise_kind = 'квартира'
+            GROUP BY cm.domrf_obj_id
+            """,
+        )
+
     def test_objective_corpus_room_month_columns(self, phantom_check_session: Session) -> None:
         """velocity._VELOCITY_BY_ROOM_SQL — deals_total_count, deals_total_vol_m2."""
         _explain(