tradein-mvp/backend/app/services/house_dedup_merge.py

@@ -13,8 +13,10 @@ WHAT this is:
     pipeline, run inside ONE transaction so a crash leaves the table untouched.
 
     Cluster key: CANONICAL address via tradein_canon_addr() over the CLEAN address
-    COALESCE(short_address, full_address, address) (cadastral_number is 100% NULL on prod —
-    confirmed in migration 040 — so address is the real building key). The clean source matters:
+    COALESCE(short_address, full_address, address) — the address is the only building key we
+    have (why: the KEY section below; the older claim here, «cadastral_number is 100% NULL on
+    prod», is no longer true — 2 648 of 9 179 rows carry one — and the conclusion no longer
+    rests on it). The clean source matters:
     `address` can carry район-noise the canon does not strip (e.g. «улица Вайнера, 66 · р-н Центр»
     → canon «вайнера66рнцентр»), while `short_address` holds the clean «улица Вайнера, 66»
     (→ «вайнера66») — preferring the clean field lets such a row cluster with its twin. The canon
@@ -116,6 +118,43 @@ MERGE JOURNAL — the merge is REVERSIBLE (#2690, migration 230):
     asymmetry — merge allowed without a proximity check — was invisible in data before; now
     «how many merges happened beyond N metres, on which key» is one query.
 
+KEY — there is no second, address-independent observation. Measured on prod 2026-08-10 (#2690):
+    #2690 asked for a cluster key that does not come from the normalized address, so that two
+    rows merge on two independent statements of identity rather than one restated twice. Every
+    field `houses` carries was checked against the live table. None qualifies:
+
+      cadastral_number  2 648 filled, ALL 2 648 values DISTINCT → collapses nothing. Provenance:
+                        all 2 648 also carry dadata_enriched_at and house_fias_id, i.e. they are
+                        DaData's answer to our address string, not a second observation of the
+                        building. (The other cadastre we hold, listings.building_cadastral_number,
+                        is the KNN geo-nearest hint — 20.1% of its values cover >1 ГАР building;
+                        #2674 refused it as an identity key and that stands.)
+      house_fias_id     3 678 filled, ALL DISTINCT → the FIAS pass merges 0 rows today. Same
+                        DaData provenance.
+      gar_house_guid    the key #2690 rejected, re-measured: of 458 same-guid pairs, 441 share
+                        the canon (the guid restates it), 17 do not — and 5 of those 17 are
+                        >250 m apart, worst 5 064 km. Still circular, still noisy.
+      zhkh_house_guid   looks independent (ГИС ЖКХ is an external registry) and is not: the
+                        loader sets it WHERE gar_house_guid = <guid>, i.e. it IS the ГАР guid for
+                        4 268 of 4 663 rows. The 395 that differ come from the cadastre fallback
+                        — keyed by that same KNN hint. Of its 194 pairs with a DIFFERENT canon,
+                        193 come through the fallback, and 30 of the 31 pairs >250 m apart do too.
+      source+ext_house_id, cian_internal_house_id, yandex_jk_id
+                        distinct by construction / 39 / 0 rows → nothing to cluster.
+      coordinates       a real independent observation, but not an IDENTITY: neighbours share a
+                        yard. It is already used the only way it can be — as the guard.
+      year_built+total_floors
+                        a FALSE witness, not a corroborator: of the 391 same-canon pairs the
+                        guard cannot judge, only 18 agree on both fields (357 have a NULL), while
+                        306 pairs the guard rejected at >250 m DO agree — it would confirm merges
+                        that are provably wrong.
+
+    Conclusion: do NOT strengthen the key, and do not read the leftover as a backlog. What the
+    canon key + 250 m guard reach IS the ceiling; what is left is counted, not queued — see the
+    residual census (`_RESIDUAL_SQL`), whose buckets keep «the guard was silent» apart from «the
+    guard rejected on the merits». Prod 2026-08-10, 963 excess rows: 568 of them are >250 m apart
+    (median 1 084 m) — those are not duplicates at all, the canon key is wrong about them.
+
 IDEMPOTENCY:
     Every UPDATE/DELETE keys off a temp mapping of (loser→keeper). On a clean table the
     mapping is empty → every statement touches 0 rows → no-op. Re-running is safe.
@@ -165,6 +204,14 @@ _COMPLETENESS_EXPR = """
 # правилу. Последствие не косметическое: объявления проигравшего переезжают на запись, на которую
 # корпус никогда не ссылался, а COALESCE-перенос полей неполон (год постройки / тип дома /
 # этажность / застройщик не переносятся) — данные богатого проигравшего удаляются безвозвратно.
+#
+# ПРОВЕРЕНО ЗАДНИМ ЧИСЛОМ (#2690 п.3, 2026-08-10): первый прогон на исправленном правиле —
+# 08.08, 821 слияние — разобран по house_merge_log (у проигравшего число объявлений = длина
+# children_repointed['listings.house_id_fk'], у победителя — что висело на нём до слияния).
+# Слияний, где победитель беднее проигравшего по объявлениям: 0 из 821. Контрфактика старого
+# правила на тех же кластерах: 6 из 762 забрали бы пустого победителя (8 объявлений). Мерить
+# «победителя до слияния» по listings.scraped_at НЕЛЬЗЯ — #2206 двигает его при каждом
+# ре-подтверждении, отчего появляются 207 несуществующих «худших победителей».
 _KEEPER_ORDER = f"""
     (h.geom IS NOT NULL) DESC,
     listing_cnt DESC NULLS LAST,
@@ -199,37 +246,17 @@ _CANON_KEY_EXPR = """
 """
 
 
-def _mapping_sql(cluster_key_case: str, *, apply_geo_guard: bool = True) -> str:
-    """Render the loser→keeper mapping SQL for one pass, given its cluster-key CASE expression.
+def _ranked_cte(cluster_key_case: str) -> str:
+    """Render the `WITH … ranked AS (…)` prelude: cluster → rank → expose the keeper per row.
 
-    Only cluster keys shared by >1 house_id form a cluster; the keeper is rn=1 per cluster, losers
-    are rn>1. The CROSS-FIAS guard always applies (a no-op for the fias pass, where every clustered
-    row shares one fias by construction).
-
-    apply_geo_guard (#2187): the 250 m ST_DistanceSphere guard is emitted ONLY when True.
-      - CANON pass → True: the canon strips город/район, so same-street-number buildings in
-        different region-66 towns share a canon; the guard stops the cross-town over-merge.
-      - FIAS pass → False: a shared ФИАС/ГАР UUID IS the building identity and strictly outranks
-        proximity, so same-fias rows merge even with NULL geom on a side or >250 m apart (the
-        geom-first keeper rule simultaneously repairs the broken coordinate).
+    Shared verbatim by the merge mapping (`_mapping_sql`) and the residual census
+    (`_RESIDUAL_SQL`) so the census counts EXACTLY the rows the merge reasons about — a census
+    built from its own copy of the clustering would drift from the pass it describes and the
+    drift would be invisible (it is the same class of error as #2690's cluster key: two
+    expressions that look alike and are not).
     `cluster_key_case` is a STATIC module constant (never runtime data) — no value injection.
     """
-    geo_guard = (
-        """
-      -- GEO GUARD (canon pass only — #2187). tradein_canon_addr strips город/район, so two
-      -- different buildings sharing a street+number canon («Ленина 5» in different region-66
-      -- towns) collapse to one cluster_key. A loser merges only when geographically next to the
-      -- keeper (<=250 m — covers one building's geocode spread, prod: Мраморская 34к4 dupes at
-      -- 222 m; region-66 towns are km+ apart → 250 m is safe from cross-town). >250 m, or NULL
-      -- geom on either side, → left as separate rows (conservative — never over-merges).
-      AND keeper_geom IS NOT NULL
-      AND loser_geom IS NOT NULL
-      AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250"""
-        if apply_geo_guard
-        else ""
-    )
     return f"""
-    CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS
     WITH clustered AS (
         SELECT
             id,
@@ -281,7 +308,41 @@ def _mapping_sql(cluster_key_case: str, *, apply_geo_guard: bool = True) -> str:
         FROM dup_houses dh
         JOIN houses h ON h.id = dh.id
         LEFT JOIN listing_counts lc ON lc.house_id = dh.id
+    )"""
+
+
+def _mapping_sql(cluster_key_case: str, *, apply_geo_guard: bool = True) -> str:
+    """Render the loser→keeper mapping SQL for one pass, given its cluster-key CASE expression.
+
+    Only cluster keys shared by >1 house_id form a cluster; the keeper is rn=1 per cluster, losers
+    are rn>1. The CROSS-FIAS guard always applies (a no-op for the fias pass, where every clustered
+    row shares one fias by construction).
+
+    apply_geo_guard (#2187): the 250 m ST_DistanceSphere guard is emitted ONLY when True.
+      - CANON pass → True: the canon strips город/район, so same-street-number buildings in
+        different region-66 towns share a canon; the guard stops the cross-town over-merge.
+      - FIAS pass → False: a shared ФИАС/ГАР UUID IS the building identity and strictly outranks
+        proximity, so same-fias rows merge even with NULL geom on a side or >250 m apart (the
+        geom-first keeper rule simultaneously repairs the broken coordinate).
+    `cluster_key_case` is a STATIC module constant (never runtime data) — no value injection.
+    """
+    geo_guard = (
+        """
+      -- GEO GUARD (canon pass only — #2187). tradein_canon_addr strips город/район, so two
+      -- different buildings sharing a street+number canon («Ленина 5» in different region-66
+      -- towns) collapse to one cluster_key. A loser merges only when geographically next to the
+      -- keeper (<=250 m — covers one building's geocode spread, prod: Мраморская 34к4 dupes at
+      -- 222 m; region-66 towns are km+ apart → 250 m is safe from cross-town). >250 m, or NULL
+      -- geom on either side, → left as separate rows (conservative — never over-merges).
+      AND keeper_geom IS NOT NULL
+      AND loser_geom IS NOT NULL
+      AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250"""
+        if apply_geo_guard
+        else ""
     )
+    return f"""
+    CREATE TEMP TABLE _1772_dup_mapping ON COMMIT DROP AS
+    {_ranked_cte(cluster_key_case)}
     -- CROSS-FIAS guard (#1772 follow-up): never merge two rows that BOTH carry a non-null but
     -- DIFFERENT house_fias_id — provably different buildings the cluster key collapsed (canon
     -- slash-collapse «Сулимова, 32»/«Сулимова, 3/2»). No-op for the fias pass (one fias per
@@ -314,6 +375,54 @@ _BUILD_MAPPING_SQL = text(_mapping_sql(_CANON_KEY_EXPR))
 # merge even with NULL geom or >250 m apart (the geom-first keeper rule fixes broken coords).
 _BUILD_MAPPING_SQL_FIAS = text(_mapping_sql(_FIAS_KEY_EXPR, apply_geo_guard=False))
 
+# ── RESIDUAL CENSUS (#2690 п.2/п.4) ───────────────────────────────────────────
+#
+# Read-only, run AFTER both passes: how many same-canon rows the merge LEFT BEHIND, and WHY.
+# Same `ranked` prelude as the canon mapping, minus the guard — so every row the guard filtered
+# out is counted here, bucketed by the reason it survived.
+#
+# WHY this exists. #2690 asked for a second, address-independent key; measured 2026-08-10, there
+# is none (see the KEY section in the module docstring), so the remainder is a CEILING, not a
+# backlog — and a ceiling has to be a live number, not a one-off. The one-off rots fast: the
+# issue's own census (781 excess rows, 06.08) was 963 four days later, after a run deleted 821.
+#
+# The buckets are deliberately NOT summed into one «остаток». «Guard was silent» and «guard
+# rejected» are opposite facts:
+#   residual_no_geom    — one side has no coordinates: the guard could not speak. UNKNOWN.
+#   residual_far        — both geocoded, >250 m apart: the guard spoke on the merits. These are
+#                         NOT duplicates — the canon key is wrong about them (prod 2026-08-10:
+#                         568 rows, median 1084 m). Counting them as «дубли» inflates the debt.
+#   residual_cross_fias — provably different buildings (two different ФИАС UUIDs).
+#   residual_mergeable  — passes every guard and STILL was not merged. Must be 0 after a real
+#                         run; non-zero is a tripwire on the pass itself, not a census entry.
+# residual_listings is the user-visible size of the remainder (listings hanging on those rows).
+_RESIDUAL_SQL = text(
+    f"""
+    {_ranked_cte(_CANON_KEY_EXPR)}
+    SELECT
+        count(*) FILTER (WHERE rn > 1)                                   AS residual_rows,
+        COALESCE(sum(lcnt) FILTER (WHERE rn > 1), 0)                     AS residual_listings,
+        count(*) FILTER (WHERE rn > 1 AND cross_fias)                    AS residual_cross_fias,
+        count(*) FILTER (WHERE rn > 1 AND NOT cross_fias AND dist IS NULL)
+                                                                         AS residual_no_geom,
+        count(*) FILTER (WHERE rn > 1 AND NOT cross_fias AND dist > 250) AS residual_far,
+        count(*) FILTER (WHERE rn > 1 AND NOT cross_fias AND dist <= 250)
+                                                                         AS residual_mergeable
+    FROM (
+        SELECT rn,
+               COALESCE(lc.listing_cnt, 0) AS lcnt,
+               CASE WHEN keeper_geom IS NOT NULL AND loser_geom IS NOT NULL
+                    THEN ST_DistanceSphere(loser_geom, keeper_geom)
+                    END AS dist,
+               (NULLIF(loser_fias, '') IS NOT NULL
+                AND NULLIF(keeper_fias, '') IS NOT NULL
+                AND lower(loser_fias) <> lower(keeper_fias)) AS cross_fias
+        FROM ranked
+        LEFT JOIN listing_counts lc ON lc.house_id = ranked.id
+    ) r
+    """
+)
+
 # Each step keys off _1772_dup_mapping → empty mapping ⇒ 0 rows touched ⇒ idempotent no-op.
 _STEPS: list[tuple[str, str]] = [
     # ── Plain re-point (no UNIQUE on the FK column) ───────────────────────────
@@ -726,6 +835,15 @@ class DedupMergeResult:
     listings_repointed: int = 0  # listings.house_id_fk moved loser→keeper
     children_deleted: int = 0  # collision/dedup deletions across all UNIQUE children
     children_repointed: int = 0  # survivor child rows moved loser→keeper
+    # Residual census (#2690): same-canon rows STILL in the table after this run, by reason.
+    # Not a backlog — measured 2026-08-10 there is no address-independent key to shrink it with,
+    # so this is the ceiling of what this pass can reach. See _RESIDUAL_SQL.
+    residual_rows: int = 0  # excess same-canon rows left behind (sum of the three buckets)
+    residual_listings: int = 0  # listings hanging on them (the user-visible size)
+    residual_no_geom: int = 0  # guard was SILENT — one side has no coordinates
+    residual_far: int = 0  # guard SPOKE — >250 m apart, i.e. not the same building
+    residual_cross_fias: int = 0  # two different ФИАС UUIDs — provably different buildings
+    residual_mergeable: int = 0  # passed every guard and still unmerged — TRIPWIRE, expect 0
     dry_run: bool = False
     duration_sec: float = field(default=0.0)
 
@@ -736,6 +854,12 @@ class DedupMergeResult:
             "listings_repointed": self.listings_repointed,
             "children_deleted": self.children_deleted,
             "children_repointed": self.children_repointed,
+            "residual_rows": self.residual_rows,
+            "residual_listings": self.residual_listings,
+            "residual_no_geom": self.residual_no_geom,
+            "residual_far": self.residual_far,
+            "residual_cross_fias": self.residual_cross_fias,
+            "residual_mergeable": self.residual_mergeable,
             "dry_run": int(self.dry_run),
             "duration_sec": int(self.duration_sec),
         }
@@ -853,6 +977,49 @@ def _run_merge_pass(
     db.execute(_BACKFILL_ALIASES_SQL)
 
 
+def _measure_residual(db: Session, result: DedupMergeResult) -> None:
+    """Count the same-canon rows this run did NOT merge, bucketed by the reason (#2690).
+
+    Read-only; runs after both passes, so it describes the table as the run leaves it (under
+    dry_run it sees the not-yet-rolled-back state, which is the correct preview). Kept out of
+    `_run_merge_pass` because the census is about the CANON key only and must be taken once per
+    call, not once per pass.
+
+    Never fails the merge: the merge itself is the product, the census is instrumentation, and a
+    census that can abort a committed-by-now transaction would be worse than a missing number.
+    """
+    try:
+        rows = db.execute(_RESIDUAL_SQL).all()
+    except Exception:
+        logger.exception("merge_duplicate_houses: residual census failed — counters left at 0")
+        return
+    if not rows:
+        return
+    r = rows[0]
+    result.residual_rows = int(r.residual_rows or 0)
+    result.residual_listings = int(r.residual_listings or 0)
+    result.residual_no_geom = int(r.residual_no_geom or 0)
+    result.residual_far = int(r.residual_far or 0)
+    result.residual_cross_fias = int(r.residual_cross_fias or 0)
+    result.residual_mergeable = int(r.residual_mergeable or 0)
+    logger.info(
+        "merge_duplicate_houses: residual rows=%d listings=%d "
+        "(страж молчит=%d · страж отверг >250м=%d · cross-fias=%d · сливаемых=%d)",
+        result.residual_rows,
+        result.residual_listings,
+        result.residual_no_geom,
+        result.residual_far,
+        result.residual_cross_fias,
+        result.residual_mergeable,
+    )
+    if result.residual_mergeable:
+        logger.warning(
+            "merge_duplicate_houses: %d rows pass every guard yet were NOT merged — the pass "
+            "left work on the table (expected 0)",
+            result.residual_mergeable,
+        )
+
+
 def merge_duplicate_houses(
     db: Session,
     *,
@@ -908,6 +1075,10 @@ def merge_duplicate_houses(
             result=result,
         )
 
+        # Census of what is LEFT (read-only). Runs before the no-op early return on purpose:
+        # a run that merged nothing is exactly the run whose remainder needs a number.
+        _measure_residual(db, result)
+
         if result.losers_deleted == 0:
             # Clean table — both passes empty. Roll back (we only opened temp tables).
             db.rollback()

tradein-mvp/backend/tests/test_house_dedup_merge.py

@@ -18,6 +18,7 @@ import inspect
 import os
 import re
 from pathlib import Path
+from types import SimpleNamespace
 from typing import Any
 
 import pytest
@@ -486,9 +487,13 @@ class _FakeDB:
         mapping_rows: list[_Row],
         step_rowcount: int = 1,
         fk_children: dict[str, str] | None = None,
+        residual: dict[str, int] | None = None,
+        residual_raises: bool = False,
     ):
         self._mapping_rows = mapping_rows
         self._step_rowcount = step_rowcount
+        self._residual = residual
+        self._residual_raises = residual_raises
         self._mapping_served = False
         # The catalog the FK-child guard reads; defaults to the real live set.
         self._fk_children = _FK_CHILDREN if fk_children is None else fk_children
@@ -503,6 +508,12 @@ class _FakeDB:
             return _FakeResult()
         if "FROM pg_constraint" in sql:
             return _FakeResult(rows=[_FKChild(t, c) for t, c in self._fk_children.items()])
+        if "AS residual_rows" in sql:  # residual census (#2690) — read-only, after both passes
+            if self._residual_raises:
+                raise RuntimeError("census exploded")
+            if self._residual is None:
+                return _FakeResult(rows=[])
+            return _FakeResult(rows=[SimpleNamespace(**self._residual)])
         if "SELECT loser_id, keeper_id, norm_address" in sql:
             # The service now runs TWO passes (fias, then canon). Model «fias pass found the
             # duplicates, canon pass is clean»: serve the scripted mapping once, empty afterwards.
@@ -550,6 +561,85 @@ def test_dry_run_computes_counts_but_rolls_back() -> None:
     assert db.rollbacks == 1
 
 
+# ── Residual census (#2690 п.2/п.4): остаток фиксируется числом, а не оценкой ──
+
+
+def test_residual_census_reuses_the_canon_mapping_prelude() -> None:
+    """Перепись остатка обязана считать РОВНО те строки, о которых рассуждает канон-проход.
+
+    Общий `_ranked_cte` — не косметика: собственная копия кластеризации разъехалась бы с
+    проходом, который она описывает, и разъезд был бы невидим (это тот же класс ошибки, что и
+    ключ в #2690 — два выражения, похожие друг на друга и не равные). RED до фикса: переписи
+    нет вовсе.
+    """
+    residual = _flat(str(hdm._RESIDUAL_SQL.text))
+    prelude = _flat(hdm._ranked_cte(hdm._CANON_KEY_EXPR))
+    assert prelude in residual
+    assert prelude in _flat(hdm._mapping_sql(hdm._CANON_KEY_EXPR))
+
+
+def test_residual_census_keeps_silent_guard_apart_from_rejecting_guard() -> None:
+    """«Координат нет» и «дальше 250 м» — противоположные факты, в одну сумму их нельзя.
+
+    Первое означает, что страж не смог высказаться (остаток неизвестен), второе — что он
+    высказался по существу и дома РАЗНЫЕ (это вообще не дубли). Сумма из этих двух читается как
+    «долг», которого нет. Плюс: перепись НЕ применяет гео-фильтр — иначе она считала бы только
+    то, что и так слилось.
+    """
+    residual = _flat(str(hdm._RESIDUAL_SQL.text))
+    for bucket in (
+        "AS residual_rows",
+        "AS residual_listings",
+        "AS residual_no_geom",
+        "AS residual_far",
+        "AS residual_cross_fias",
+        "AS residual_mergeable",
+    ):
+        assert bucket in residual, bucket
+    # Гео-страж 250 м здесь — РАЗДЕЛИТЕЛЬ корзин, а не фильтр строк.
+    assert "AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250" not in residual
+    assert set(hdm.DedupMergeResult().to_counters()) >= {
+        "residual_rows",
+        "residual_listings",
+        "residual_no_geom",
+        "residual_far",
+        "residual_cross_fias",
+        "residual_mergeable",
+    }
+
+
+def test_residual_counters_reported_even_when_nothing_merged() -> None:
+    """Прогон, который не слил ничего, — ровно тот, чьему остатку нужно число."""
+    db = _FakeDB(
+        mapping_rows=[],
+        residual={
+            "residual_rows": 963,
+            "residual_listings": 1765,
+            "residual_no_geom": 326,
+            "residual_far": 568,
+            "residual_cross_fias": 8,
+            "residual_mergeable": 61,
+        },
+    )
+    out = hdm.merge_duplicate_houses(db, dry_run=False)  # type: ignore[arg-type]
+    assert out["losers_deleted"] == 0
+    assert out["residual_rows"] == 963
+    assert out["residual_listings"] == 1765
+    assert out["residual_no_geom"] == 326
+    assert out["residual_far"] == 568
+    assert out["residual_cross_fias"] == 8
+    assert out["residual_mergeable"] == 61
+
+
+def test_residual_census_failure_never_breaks_the_merge() -> None:
+    """Перепись — приборы, слияние — продукт: упавший счётчик не отменяет коммит."""
+    db = _FakeDB(mapping_rows=[_Row(2, 1, "ул. мира, 10")], residual_raises=True)
+    out = hdm.merge_duplicate_houses(db, dry_run=False)  # type: ignore[arg-type]
+    assert out["losers_deleted"] == 1
+    assert out["residual_rows"] == 0  # неизвестно — и это видно, а не выдумано
+    assert db.commits == 1
+
+
 def test_real_merge_commits() -> None:
     """dry_run=False with dups → exactly one commit, no rollback."""
     rows = [_Row(2, 1, "ул. мира, 10")]