fix(tradein/dedup): canon cluster-key + geo/digit guard в house-merge

merge_duplicate_houses кластеризовал дома по exact lower(trim(address)) →
варианты написания одного здания (ул. Вайнера,66 vs улица Вайнера, 66) не
схлопывались, расщепляя долю в sale-share. Ключ кластера → tradein_canon_addr
(ул→улица, стрип город/район, КОРПУС сохранён). Гарды против over-merge:
- гео ≤250м (canon стрипует город; прод-дубли Мраморская 34к4 в 222м; города
  региона-66 в км+ → 250м безопасно от cross-town);
- canon обязан содержать цифру (номер дома) — исключает вырожденные
  «екатеринбург»/«сооружение».
Re-pointing/audit/dry-run/идемпотентность не тронуты. 26 passed.

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

@@ -12,10 +12,18 @@ WHAT this is:
         cluster → pick canonical → re-point children (UNIQUE-collision-safe) → delete losers
     pipeline, run inside ONE transaction so a crash leaves the table untouched.
 
-    Cluster key: EXACT `houses.address` (cadastral_number is 100% NULL on prod — confirmed
-    in migration 040 — so address is the real building key). Rows whose address is NULL/blank
-    are never clustered (cluster_key NULL → ignored). Only addresses shared by >1 house_id
-    form a cluster.
+    Cluster key: CANONICAL address via tradein_canon_addr() (cadastral_number is 100% NULL on
+    prod — confirmed in migration 040 — so address is the real building key). The canon collapses
+    spelling/район variants of the SAME building (ул→улица, strips город/район/«россия»/«м-н»/
+    «р-н» noise, removes spaces/punct, PRESERVES корпус) so e.g. «ул. Вайнера,66» and
+    «улица Вайнера, 66» share one cluster_key. Rows whose canon is NULL/blank are never clustered
+    (cluster_key NULL → ignored). Only canons shared by >1 house_id form a cluster.
+
+    GEO GUARD (anti over-merge): because the canon strips город/район, two different buildings
+    with the same street+number in different region-66 towns (e.g. «Ленина 5») would share a
+    canon. To avoid merging them, within a canon-cluster a house is a LOSER only if it is within
+    250 m of the keeper (ST_DistanceSphere on the WGS84 geom). Same-canon houses >250 m away — or
+    with NULL geom on either side — are left as separate rows (conservative; never over-merges).
 
 KEEPER RULE (one canonical row per cluster), in priority order:
     1. geom NOT NULL first      (a geocoded row is the better survivor)
@@ -109,8 +117,9 @@ _BUILD_MAPPING_SQL = text(
     WITH clustered AS (
         SELECT
             id,
-            CASE WHEN NULLIF(lower(trim(address)), '') IS NOT NULL
-                 THEN 'addr:' || lower(trim(address)) END AS cluster_key
+            CASE WHEN NULLIF(tradein_canon_addr(address), '') IS NOT NULL
+                  AND tradein_canon_addr(address) ~ '[0-9]'
+                 THEN 'addr:' || tradein_canon_addr(address) END AS cluster_key
         FROM houses
     ),
     clusters_with_count AS (
@@ -137,6 +146,7 @@ _BUILD_MAPPING_SQL = text(
             dh.id,
             dh.cluster_key,
             lower(trim(h.address)) AS norm_address,
+            h.geom AS loser_geom,
             ROW_NUMBER() OVER (
                 PARTITION BY dh.cluster_key
                 ORDER BY {_KEEPER_ORDER}
@@ -144,15 +154,30 @@ _BUILD_MAPPING_SQL = text(
             first_value(dh.id) OVER (
                 PARTITION BY dh.cluster_key
                 ORDER BY {_KEEPER_ORDER}
-            ) AS keeper_id
+            ) AS keeper_id,
+            first_value(h.geom) OVER (
+                PARTITION BY dh.cluster_key
+                ORDER BY {_KEEPER_ORDER}
+            ) AS keeper_geom
         FROM dup_houses dh
         JOIN houses h ON h.id = dh.id
         LEFT JOIN listing_counts lc ON lc.house_id = dh.id
     )
+    -- Geo guard against cross-town over-merge: tradein_canon_addr strips город/район, so two
+    -- different buildings sharing a street+number canon (e.g. «Ленина 5» in different towns)
+    -- collapse to one cluster_key. A loser is only merged when it is geographically next to the
+    -- keeper (<=250 m — покрывает geocode-разброс одного здания, прод: дубли Мраморская 34к4 в
+    -- 222 m; города региона-66 в км+ друг от друга → 250 m безопасно от cross-town). Houses in
+    -- the same canon-cluster but >250 m away, or with NULL geom on either side, are left as
+    -- separate rows (conservative — never over-merges). Cluster_key также требует цифру в canon
+    -- (номер дома) — исключает вырожденные canon вида «екатеринбург»/«сооружение».
     SELECT id AS loser_id, keeper_id, norm_address
     FROM ranked
     WHERE rn > 1
       AND id <> keeper_id
+      AND keeper_geom IS NOT NULL
+      AND loser_geom IS NOT NULL
+      AND ST_DistanceSphere(loser_geom, keeper_geom) <= 250
     """
 )
 
@@ -423,11 +448,11 @@ class DedupMergeResult:
 
 
 def merge_duplicate_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]:
-    """Cluster houses by exact address and merge duplicates onto one canonical keeper.
+    """Cluster houses by canonical address and merge duplicates onto one canonical keeper.
 
     Re-implements migration 108's proven collision-safe pipeline as a RECURRING job:
-        cluster (exact address) → pick keeper → re-point children (UNIQUE-collision-safe) →
-        delete losers → backfill sources/aliases onto keeper.
+        cluster (canon address + 250 m geo guard) → pick keeper →
+        re-point children (UNIQUE-collision-safe) → delete losers → backfill sources/aliases.
     All in ONE transaction. dry_run=True computes counts then ROLLS BACK (no writes).
     Idempotent: a clean table yields an empty mapping → every statement is a 0-row no-op.
 

tradein-mvp/backend/tests/test_house_dedup_merge.py

@@ -44,18 +44,47 @@ def _flat(sql: str) -> str:
 # ── Cluster key + keeper rule ─────────────────────────────────────────────────
 
 
-def test_cluster_key_is_exact_address_not_cadastral() -> None:
-    """Spec: cadastral is unpopulated → cluster by EXACT address only."""
+def test_cluster_key_is_canonical_address_not_cadastral() -> None:
+    """Spec: cadastral is unpopulated → cluster by CANONICAL address (tradein_canon_addr).
+
+    The canon collapses spelling/район variants of the same building (ул→улица, strips
+    город/район noise, preserves корпус) so e.g. «ул. Вайнера,66» and «улица Вайнера, 66»
+    share one cluster_key.
+    """
     flat = _flat(_MAPPING_SQL)
-    assert "'addr:' || lower(trim(address))" in flat
+    assert "'addr:' || tradein_canon_addr(address)" in flat
+    # The empty-canon guard still keeps blank-canon houses out of any cluster.
+    assert "NULLIF(tradein_canon_addr(address), '') IS NOT NULL" in flat
+    # Digit guard: canon must carry a house number — excludes degenerate canons like
+    # «екатеринбург»/«сооружение» (street-less addresses) from ever forming a cluster.
+    assert "tradein_canon_addr(address) ~ '[0-9]'" in flat
     # cadastral_number must NOT be part of the cluster key (it is 100% NULL).
     assert "cadastral_number" not in _flat(
         _MAPPING_SQL[: _MAPPING_SQL.index("ranked")]
     )  # not in cluster construction
-    # Only addresses shared by >1 house form a cluster.
+    # Only canons shared by >1 house form a cluster.
     assert "HAVING count(*) > 1" in flat
 
 
+def test_geo_guard_blocks_cross_town_over_merge() -> None:
+    """Geo guard: a loser is merged only when within 250 m of the keeper.
+
+    Because tradein_canon_addr strips город/район, two different «Ленина 5» buildings in
+    different towns share a canon. The final loser→keeper mapping is filtered on a 250 m
+    ST_DistanceSphere predicate (keeper geom exposed per partition via first_value), so
+    same-canon-but-far houses (or NULL geom on either side) are NOT merged.
+    """
+    flat = _flat(_MAPPING_SQL)
+    # keeper geom exposed per partition with the SAME keeper-priority ordering as keeper_id.
+    assert "first_value(h.geom) OVER" in flat
+    assert "AS keeper_geom" in flat
+    assert "h.geom AS loser_geom" in flat
+    # the 250 m predicate + the NULL-geom safety guards on the final mapping.
+    assert "keeper_geom IS NOT NULL" in flat
+    assert "loser_geom IS NOT NULL" in flat
+    assert "ST_DistanceSphere(loser_geom, keeper_geom) <= 250" in flat
+
+
 def test_keeper_rule_priority_order() -> None:
     """Keeper: geom NOT NULL → most linked listings → most-populated → min(id)."""
     order = _flat(hdm._KEEPER_ORDER)
@@ -584,3 +613,98 @@ def test_real_merge_repoints_dedups_deletes_and_is_idempotent() -> None:
         db.execute(_t("DELETE FROM houses WHERE id IN (900001,900002)"))
         db.commit()
         db.close()
+
+
+@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
+def test_real_canon_clusterkey_and_geo_guard_merge_semantics() -> None:
+    """End-to-end on a real DB for the canon cluster-key + 250 m geo guard:
+
+      A. «ул. ТестоваяX, 66» + «улица ТестоваяX, 66», ~10 m apart → MERGED (canon collapses
+         ул/улица), both listings re-pointed onto the single survivor.
+      B. two identical «улица ТестоваяX, 5», ~5 km apart           → NOT merged (geo guard >250 m),
+         both house rows survive.
+      C. «ТестоваяX, 34к2» + «ТестоваяX, 34к4», ~10 m apart        → NOT merged (different
+         корпус → different canon — корпус is preserved), both survive.
+
+    Synthetic street «Тестовая1772» avoids colliding with any real prod address/alias.
+    Deltas in latitude at ~56.84°: 0.00009° ≈ 10 m (geo passes), 0.045° ≈ 5 km (geo blocks).
+    """
+    from sqlalchemy import text as _t
+
+    db = _live_session()
+    assert db is not None
+    try:
+        db.execute(
+            _t(
+                "INSERT INTO houses (id, source, ext_house_id, address, lat, lon) VALUES "
+                # A — ул/улица spelling variants of the SAME building, ~10 m apart → MERGE
+                "(900010, 'avito', 'EXT-T-VK', 'улица Тестовая1772, 66', 56.84000, 60.60000),"
+                "(900011, 'cian',  'EXT-T-VL', 'ул. Тестовая1772, 66',   56.84009, 60.60000),"
+                # B — same canon (ленина-like) but ~5 km apart → geo guard BLOCKS the merge
+                "(900012, 'avito', 'EXT-T-L1', 'улица Тестовая1772, 5',  56.84000, 60.60000),"
+                "(900013, 'cian',  'EXT-T-L2', 'улица Тестовая1772, 5',  56.88500, 60.60000),"
+                # C — different корпус → different canon, ~10 m apart → NOT merged
+                "(900014, 'avito', 'EXT-T-M2', 'Тестовая1772, 34к2', 56.84000, 60.60000),"
+                "(900015, 'cian',  'EXT-T-M4', 'Тестовая1772, 34к4', 56.84009, 60.60000)"
+            )
+        )
+        db.execute(
+            _t(
+                "INSERT INTO listings "
+                "(id, source, source_url, source_id, dedup_hash, price_rub, house_id_fk) VALUES "
+                "(910010, 'avito', 'http://t/g/k', 'LG-K', 'dh-g-keep', 5000000, 900010),"
+                "(910011, 'cian',  'http://t/g/l', 'LG-L', 'dh-g-lose', 6000000, 900011)"
+            )
+        )
+        db.commit()
+
+        out = hdm.merge_duplicate_houses(db, dry_run=False)
+        assert out["losers_deleted"] >= 1
+
+        ids = [
+            r.id
+            for r in db.execute(
+                _t("SELECT id FROM houses WHERE id BETWEEN 900010 AND 900015 ORDER BY id")
+            ).all()
+        ]
+        # A: exactly the min-id keeper survives; the ул/улица spelling variant merged away.
+        assert 900010 in ids
+        assert 900011 not in ids, "ул/улица spelling variant must be merged onto the keeper"
+        # B: BOTH survive — same canon but >250 m apart → geo guard blocks cross-town over-merge.
+        assert 900012 in ids and 900013 in ids, "same-canon but >250 m apart must NOT merge"
+        # C: BOTH survive — different корпус → different canon (корпус preserved).
+        assert 900014 in ids and 900015 in ids, "different корпус must NOT merge"
+
+        # A: the merged loser's listing re-pointed onto the surviving keeper.
+        repointed = db.execute(_t("SELECT house_id_fk FROM listings WHERE id = 910011")).scalar()
+        assert repointed == 900010
+
+        # idempotent second run: our merged loser stays gone, the geo-blocked pairs untouched.
+        hdm.merge_duplicate_houses(db, dry_run=False)
+        ids2 = [
+            r.id
+            for r in db.execute(
+                _t("SELECT id FROM houses WHERE id BETWEEN 900010 AND 900015 ORDER BY id")
+            ).all()
+        ]
+        assert 900011 not in ids2
+        assert {900010, 900012, 900013, 900014, 900015}.issubset(set(ids2))
+    finally:
+        db.rollback()
+        db.execute(_t("DELETE FROM listings WHERE id IN (910010,910011)"))
+        db.execute(
+            _t(
+                "DELETE FROM house_sources WHERE ext_id IN "
+                "('EXT-T-VK','EXT-T-VL','EXT-T-L1','EXT-T-L2','EXT-T-M2','EXT-T-M4')"
+            )
+        )
+        db.execute(
+            _t(
+                "DELETE FROM house_address_aliases WHERE normalized_address IN "
+                "('улица тестовая1772, 66','ул. тестовая1772, 66','улица тестовая1772, 5',"
+                "'тестовая1772, 34к2','тестовая1772, 34к4')"
+            )
+        )
+        db.execute(_t("DELETE FROM houses WHERE id BETWEEN 900010 AND 900015"))
+        db.commit()
+        db.close()