ПТИЦА конкуренты: gap-fill берёт скорость и цену своего ЖК, а не всех ЖК под complex_id (#2962)

Что было. Для конкурентов вне objective_complex_mapping мост шёл
«ближайший complex → objective_lots.complex_id → все project_name».
complex_id проставлен один раз миграцией 76 на загрузке 10.05, а
еженедельный 70_parse_objective_raw.py UPSERT'ом по objective_lot_id
переписывает project_name и не трогает complex_id. Прод 17.09: из
303 677 строк с complex_id у 236 354 проект чужой (все вставлены 10.05,
переписаны 17.05–15.09). У 185 gap-fill конкурентов своих лотов 23 %,
скорость в медиане завышена в 39 раз (сумма 65 798 против 1 609 сделок/мес).

Что сделано. В _COMPETITORS_SQL и _OBJECTIVE_PRICE_FALLBACK_SQL complex
связывается с проектом через complex_sources (source='objective', 1:1),
лоты и сделки берутся по project_name. Связь в complex_sources почти вся
fuzzy и не проверена (у «ЖК VEER PARK» стоит 'Clever Park', у «ЖК Графит» —
'Гранит'), поэтому имя проекта дополнительно сверяется с именем объекта
ДОМ.РФ без регистра и пунктуации; сверка стоит после DISTINCT ON, внутри
join планировщик гонял regexp по 383k пар (3 с).

Замер на проде (все 1556 объектов, окно 3 мес): явный маппинг 308 и
остальные 1063 — изменилось 0; gap-fill 185 — изменились все, у 9 связь
отвергнута (7 чужих проектов + 2 «Традиции»/«Традиция»), доля своих лотов
23 % → 96 %. На 500 участках: 4317 прежних пар участок-конкурент — 0
изменений, 598 gap-fill пар — изменились все. Время запроса конкурентов
393 → 65 мс, ценового fallback 23 → 54 мс.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/services/site_finder/competitors.py

@@ -344,21 +344,23 @@ _COMPETITORS_SQL = text("""
         -- важна — velocity питает forecast). Спатиальный join дёшев, т.к.
         -- complexes мал (~1.5k строк с geom) — НЕ потому что latest_obj
         -- пре-фильтрован (радиус участка применяется ниже, в финальном SELECT).
+        --
+        -- #2962: complex → проект Объектива берём из complex_sources (source=
+        -- 'objective', строго 1:1), а НЕ из objective_lots.complex_id. Тот проставлен
+        -- один раз миграцией 76 на загрузке 10.05, а еженедельный
+        -- 70_parse_objective_raw.py UPSERT'ом по objective_lot_id переписывает
+        -- project_name и не трогает complex_id → под одним complex_id лежат чужие ЖК
+        -- (прод 17.09: 236 354 из 303 677 строк, у «ЖК Мичуринский» 129 проектов).
         SELECT DISTINCT ON (o.obj_id)
                o.obj_id,
-               c.id AS complex_id
+               cs.source_id AS project_name,
+               regexp_replace(lower(o.comm_name), '[^0-9a-zа-яё]', '', 'g') AS comm_key,
+               regexp_replace(lower(cs.source_id), '[^0-9a-zа-яё]', '', 'g') AS project_key
           FROM latest_obj o
           JOIN complexes c
             ON c.latitude IS NOT NULL
            AND c.longitude IS NOT NULL
            AND c.canonical_name IS NOT NULL
-           -- #968: только complexes, у которых ЕСТЬ objective velocity-данные →
-           -- «ближайший» выбирается среди data-bearing. Иначе ближайший complex
-           -- без objective_lots съедал бы матч (prod: +37 наивно → +181 так).
-           AND EXISTS (
-                   SELECT 1 FROM objective_lots ol
-                    WHERE ol.complex_id = c.id AND ol.project_name IS NOT NULL
-               )
            AND ST_DWithin(
                    ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
                    ST_SetSRID(ST_MakePoint(c.longitude, c.latitude), 4326)::geography,
@@ -369,6 +371,16 @@ _COMPETITORS_SQL = text("""
                 OR lower(btrim(c.canonical_name)) LIKE '%' || lower(btrim(o.comm_name)) || '%'
                 OR lower(btrim(o.comm_name)) LIKE '%' || lower(btrim(c.canonical_name)) || '%'
                )
+          JOIN complex_sources cs
+            ON cs.complex_id = c.id
+           AND cs.source = 'objective'
+           -- #968: только complexes, у которых ЕСТЬ objective-данные →
+           -- «ближайший» выбирается среди data-bearing. Иначе ближайший complex
+           -- без objective_lots съедал бы матч (prod: +37 наивно → +181 так).
+           AND EXISTS (
+                   SELECT 1 FROM objective_lots ol
+                    WHERE ol.project_name = cs.source_id
+               )
          WHERE o.comm_name IS NOT NULL
            AND btrim(o.comm_name) <> ''  -- иначе LIKE '%%' матчит всё → spatial-only
            AND o.obj_id NOT IN (SELECT domrf_obj_id FROM objective_complex_mapping)
@@ -384,15 +396,19 @@ _COMPETITORS_SQL = text("""
                cm.objective_complex_name
           FROM objective_complex_mapping cm
         UNION
-        -- FALLBACK (gap-fill only): все project_name ОДНОГО ближайшего complex'а
+        -- FALLBACK (gap-fill only): ровно один project_name ближайшего complex'а
-        -- (у комплекса может быть несколько корпус-project_name → velocity
+        -- (корпуса у Объектива — corpus_name внутри проекта, не отдельные имена).
-        -- легитимно суммируется по ним, но НЕ по нескольким комплексам).
         SELECT nc.obj_id,
-               ol.project_name AS objective_complex_name
+               nc.project_name AS objective_complex_name
           FROM nearest_cx nc
-          JOIN objective_lots ol
+         -- #2962: связь complex → проект в complex_sources почти вся fuzzy и не
-            ON ol.complex_id = nc.complex_id
+         -- проверена (у «ЖК VEER PARK» стоит 'Clever Park', у «ЖК Графит» —
-           AND ol.project_name IS NOT NULL
+         -- 'Гранит'), поэтому имя проекта сверяем с именем объекта ДОМ.РФ ещё раз,
+         -- без регистра и пунктуации («СтудияПарк» = «Студия Парк»). Фильтр стоит
+         -- ПОСЛЕ DISTINCT ON: внутри join планировщик считал regexp на всех парах
+         -- объект × проект (383k) раньше гео-фильтра — 3 с против 0.4 с.
+         WHERE nc.comm_key LIKE '%' || nc.project_key || '%'
+            OR nc.project_key LIKE '%' || nc.comm_key || '%'
     ),
     velocity AS (
         SELECT
@@ -605,9 +621,11 @@ _SOLD_COUNT_SQL = text("""
 # fallback должен покрывать ОБА, иначе конкурент с velocity>0 из spatial-матча получает
 # avg_price=None и price_similarity падает в нейтраль. Зеркалим тот же мост obj→lots:
 #   PRIMARY:  objective_complex_mapping.objective_complex_name == objective_lots.project_name
-#   GAP-FILL: nearest_cx (≤ :velocity_match_radius_m м + tolerant-name) → complex_id →
+#   GAP-FILL: nearest_cx (≤ :velocity_match_radius_m м + tolerant-name) → complex_sources
-#             objective_lots по complex_id (тот же DISTINCT ON ближайший complex, что и
+#             (source='objective', 1:1) → objective_lots по project_name (тот же DISTINCT ON
-#             velocity gap-fill — обязан совпадать, чтобы цена и velocity были про ОДИН ЖК).
+#             ближайший complex, что и velocity gap-fill — обязан совпадать, чтобы цена и
+#             velocity были про ОДИН ЖК). НЕ по objective_lots.complex_id: он устарел и под
+#             ним лежат чужие ЖК (#2962, см. _COMPETITORS_SQL nearest_cx).
 # obj_id мапится в РОВНО один источник (mapping 1:1; gap-fill — только для obj_id ВНЕ
 # mapping, см. NOT IN ниже), поэтому пересечения нет и UNION ALL безопасен.
 _OBJECTIVE_PRICE_FALLBACK_SQL = text("""
@@ -624,16 +642,14 @@ _OBJECTIVE_PRICE_FALLBACK_SQL = text("""
     nearest_cx AS (
         SELECT DISTINCT ON (o.obj_id)
                o.obj_id,
-               c.id AS complex_id
+               cs.source_id AS project_name,
+               regexp_replace(lower(o.comm_name), '[^0-9a-zа-яё]', '', 'g') AS comm_key,
+               regexp_replace(lower(cs.source_id), '[^0-9a-zа-яё]', '', 'g') AS project_key
           FROM domrf_kn_objects o
           JOIN complexes c
             ON c.latitude IS NOT NULL
            AND c.longitude IS NOT NULL
            AND c.canonical_name IS NOT NULL
-           AND EXISTS (
-                   SELECT 1 FROM objective_lots ol
-                    WHERE ol.complex_id = c.id AND ol.project_name IS NOT NULL
-               )
            AND ST_DWithin(
                    ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
                    ST_SetSRID(ST_MakePoint(c.longitude, c.latitude), 4326)::geography,
@@ -644,6 +660,13 @@ _OBJECTIVE_PRICE_FALLBACK_SQL = text("""
                 OR lower(btrim(c.canonical_name)) LIKE '%' || lower(btrim(o.comm_name)) || '%'
                 OR lower(btrim(o.comm_name)) LIKE '%' || lower(btrim(c.canonical_name)) || '%'
                )
+          JOIN complex_sources cs
+            ON cs.complex_id = c.id
+           AND cs.source = 'objective'
+           AND EXISTS (
+                   SELECT 1 FROM objective_lots ol
+                    WHERE ol.project_name = cs.source_id
+               )
          WHERE o.obj_id = ANY(:obj_ids)
            AND o.latitude IS NOT NULL
            AND o.longitude IS NOT NULL
@@ -662,8 +685,11 @@ _OBJECTIVE_PRICE_FALLBACK_SQL = text("""
             ol.price_per_m2_rub AS price_per_m2_rub
         FROM nearest_cx nc
         JOIN objective_lots ol
-            ON ol.complex_id = nc.complex_id
+            ON ol.project_name = nc.project_name
            AND ol.price_per_m2_rub IS NOT NULL
+        -- та же сверка имени, что в _COMPETITORS_SQL mapped (#2962)
+        WHERE nc.comm_key LIKE '%' || nc.project_key || '%'
+           OR nc.project_key LIKE '%' || nc.comm_key || '%'
     )
     SELECT
         p.obj_id,

backend/tests/services/site_finder/test_2962_competitors_gapfill_bridge.py

@@ -0,0 +1,173 @@
+"""Gap-fill конкурентов берёт скорость и цену своего ЖК, а не всех ЖК под complex_id (#2962).
+
+`objective_lots.complex_id` проставлен один раз миграцией 76, а еженедельный
+`70_parse_objective_raw.py` UPSERT'ом по objective_lot_id переписывает project_name и
+не трогает complex_id. Прод 17.09.2026: из 303 677 строк с complex_id у 236 354
+проект чужой; у 185 gap-fill конкурентов своих лотов 23 %, скорость завышена в
+медиане в 39 раз.
+
+Тест герметичный и прогоняет НАСТОЯЩИЕ `_COMPETITORS_SQL` и
+`_OBJECTIVE_PRICE_FALLBACK_SQL`: временные таблицы затеняют боевые в пределах сессии.
+Нужен Postgres с PostGIS, иначе skip.
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+
+def _dsn() -> str:
+    raw = os.environ.get("TEST_DATABASE_URL") or os.environ["DATABASE_URL"]
+    return (
+        raw
+        if raw.startswith("postgresql+")
+        else raw.replace("postgresql://", "postgresql+psycopg://")
+    )
+
+
+def _postgis_reachable() -> tuple[bool, str]:
+    try:
+        eng = create_engine(_dsn(), connect_args={"connect_timeout": 3})
+        with eng.connect() as c:
+            if c.execute(text("SELECT 1 FROM pg_extension WHERE extname = 'postgis'")).first():
+                return True, ""
+            return False, "нет расширения postgis"
+    except Exception as exc:
+        return False, str(exc)
+
+
+_DB_OK, _DB_ERR = _postgis_reachable()
+pytestmark = pytest.mark.skipif(not _DB_OK, reason=f"Postgres/PostGIS недоступен: {_DB_ERR}")
+
+_SCHEMA = [
+    """CREATE TEMP TABLE domrf_kn_objects (
+        obj_id bigint, comm_name text, dev_name text, obj_class text,
+        obj_class_fallback text, latitude double precision, longitude double precision,
+        flat_count int, site_status text, snapshot_date date) ON COMMIT DROP""",
+    """CREATE TEMP TABLE objective_complex_mapping (
+        domrf_obj_id bigint, objective_complex_name text) ON COMMIT DROP""",
+    """CREATE TEMP TABLE complexes (
+        id bigint, canonical_name text, latitude double precision,
+        longitude double precision) ON COMMIT DROP""",
+    """CREATE TEMP TABLE complex_sources (
+        complex_id bigint, source text, source_id text) ON COMMIT DROP""",
+    """CREATE TEMP TABLE objective_lots (
+        project_name text, complex_id bigint, price_per_m2_rub numeric) ON COMMIT DROP""",
+    """CREATE TEMP TABLE objective_corpus_room_month (
+        project_name text, report_month date, deals_total_count int) ON COMMIT DROP""",
+]
+
+MICH, VEER, STUDIO, EXPLICIT = 1, 2, 3, 4
+
+_DATA = [
+    # obj_id, comm_name, …, latitude, longitude — объекты дальше 200 м друг от друга
+    """INSERT INTO domrf_kn_objects VALUES
+        (1, 'ЖК Мичуринский', NULL, NULL, NULL, 56.840, 60.600, 500, 'Строящиеся', CURRENT_DATE),
+        (2, 'ЖК VEER PARK', NULL, NULL, NULL, 56.850, 60.620, 500, 'Строящиеся', CURRENT_DATE),
+        (3, 'Жилой комплекс "СтудияПарк"', NULL, NULL, NULL, 56.830, 60.580, 500,
+            'Строящиеся', CURRENT_DATE),
+        (4, 'ЖК Явный', NULL, NULL, NULL, 56.845, 60.640, 500, 'Строящиеся', CURRENT_DATE)""",
+    "INSERT INTO objective_complex_mapping VALUES (4, 'Явный')",
+    """INSERT INTO complexes VALUES
+        (10, 'ЖК Мичуринский', 56.840, 60.600),
+        (20, 'ЖК VEER PARK', 56.850, 60.620),
+        (30, 'СтудияПарк', 56.830, 60.580)""",
+    # 20 → 'Clever Park': неверная fuzzy-связь, как на проде (complexes.id=1493)
+    """INSERT INTO complex_sources VALUES
+        (10, 'objective', 'Мичуринский'),
+        (20, 'objective', 'Clever Park'),
+        (30, 'objective', 'Студия Парк')""",
+    # Под complex_id=10 лежит и свой лот, и три лота чужого «Малахита» с устаревшим
+    # complex_id; два своих лота новые — complex_id у них NULL.
+    """INSERT INTO objective_lots VALUES
+        ('Мичуринский', 10, 100000), ('Мичуринский', NULL, 110000),
+        ('Мичуринский', NULL, 120000),
+        ('Малахит', 10, 300000), ('Малахит', 10, 300000), ('Малахит', 10, 300000),
+        ('Clever Park', NULL, 150000),
+        ('Студия Парк', NULL, 90000),
+        ('Явный', NULL, 200000)""",
+    """INSERT INTO objective_corpus_room_month VALUES
+        ('Мичуринский', date_trunc('month', now())::date, 30),
+        ('Малахит', date_trunc('month', now())::date, 300),
+        ('Clever Park', date_trunc('month', now())::date, 90),
+        ('Студия Парк', date_trunc('month', now())::date, 15),
+        ('Явный', date_trunc('month', now())::date, 60)""",
+]
+
+
+@pytest.fixture(scope="module")
+def result() -> tuple[dict[int, float], dict[int, float]]:
+    """(velocity по obj_id, медианная цена objective-fallback по obj_id)."""
+    from app.services.site_finder.competitors import (
+        _COMPETITORS_SQL,
+        _OBJECTIVE_PRICE_FALLBACK_SQL,
+        _VELOCITY_MATCH_RADIUS_M,
+    )
+
+    session = sessionmaker(bind=create_engine(_dsn()))()
+    try:
+        for stmt in _SCHEMA + _DATA:
+            session.execute(text(stmt))
+        vel_rows = session.execute(
+            _COMPETITORS_SQL,
+            {
+                "center_lon": 60.6,
+                "center_lat": 56.84,
+                "radius_m": 5000.0,
+                "time_window_months": 3.0,
+                "window_interval": "3 months",
+                "velocity_match_radius_m": _VELOCITY_MATCH_RADIUS_M,
+            },
+        ).mappings()
+        velocity = {int(r["obj_id"]): float(r["velocity_per_month"]) for r in vel_rows}
+        price_rows = session.execute(
+            _OBJECTIVE_PRICE_FALLBACK_SQL,
+            {
+                "obj_ids": [MICH, VEER, STUDIO, EXPLICIT],
+                "velocity_match_radius_m": _VELOCITY_MATCH_RADIUS_M,
+            },
+        ).mappings()
+        price = {int(r["obj_id"]): float(r["median_price_per_m2"]) for r in price_rows}
+        return velocity, price
+    finally:
+        session.rollback()
+        session.close()
+
+
+def test_gapfill_velocity_counts_only_own_project(result) -> None:
+    """30 сделок своего ЖК за 3 месяца = 10/мес; с чужим «Малахитом» было бы 110."""
+    velocity, _ = result
+    assert velocity[MICH] == pytest.approx(10.0), velocity
+
+
+def test_gapfill_price_is_median_of_own_lots(result) -> None:
+    """Медиана своих лотов 100/110/120 тыс. = 110 тыс.; по complex_id было бы 300 тыс."""
+    _, price = result
+    assert price[MICH] == pytest.approx(110000.0), price
+
+
+def test_wrong_fuzzy_link_gives_no_numbers(result) -> None:
+    """«ЖК VEER PARK» связан в complex_sources с 'Clever Park' — чужие цифры не берём."""
+    velocity, price = result
+    assert velocity[VEER] == 0.0, velocity
+    assert VEER not in price, price
+
+
+def test_punctuation_difference_still_matches(result) -> None:
+    """«СтудияПарк» и «Студия Парк» — один ЖК: 15 сделок / 3 = 5, цена 90 тыс."""
+    velocity, price = result
+    assert velocity[STUDIO] == pytest.approx(5.0), velocity
+    assert price[STUDIO] == pytest.approx(90000.0), price
+
+
+def test_explicit_mapping_unchanged(result) -> None:
+    """Явный маппинг правка не трогает: 60 / 3 = 20, цена 200 тыс."""
+    velocity, price = result
+    assert velocity[EXPLICIT] == pytest.approx(20.0), velocity
+    assert price[EXPLICIT] == pytest.approx(200000.0), price