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

@@ -41,7 +41,7 @@ from app.services.dadata import DadataAddressResult
 from app.services.dadata import clean_address as dadata_clean_address
 from app.services.geocoder import GeocodeResult, geocode
 from app.services.house_metadata import get_house_metadata
-from app.services.matching.houses import match_or_create_house
+from app.services.matching.houses import match_house_readonly, match_or_create_house
 from app.services.scrapers.avito_imv import (
     IMVAddressNotFoundError,
     IMVAuthError,
@@ -606,6 +606,28 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
     except Exception as exc:  # pragma: no cover — defensive
         logger.warning("dadata: unexpected error (graceful): %s", exc)
 
+    # 1c. #6 House-match: резолвим target в КАНОНИЧЕСКИЙ house_id (read-only, без
+    #     создания записи). Это даёт детерминированный Tier S «тот же дом» через
+    #     listings.house_id_fk (99% покрытие), точнее хрупкого address-string match.
+    #     cadastr от DaData → cadastr_exact tier заработает по мере backfill houses.
+    #     Best-effort: None при любой ошибке, estimator продолжает на гео-tier'ах.
+    target_house_id: int | None = None
+    try:
+        match = match_house_readonly(
+            db,
+            address=(dadata.canonical_address if dadata else None) or geo.full_address,
+            lat=geo.lat,
+            lon=geo.lon,
+            cadastral_number=(dadata.house_cadnum if dadata else None),
+        )
+        if match is not None:
+            target_house_id = match[0]
+            logger.info(
+                "estimate target → house_id=%s via %s (conf=%.2f)", match[0], match[2], match[1]
+            )
+    except Exception as exc:  # pragma: no cover — defensive
+        logger.warning("target house match failed (graceful): %s", exc)
+
     # 2. #392: обогащаем год / тип дома из картографии (OSM Overpass), если
     #    пользователь их не указал — это улучшает house-match аналогов (#6).
     #    Best-effort: при недоступности OSM target_* остаются None.
@@ -636,6 +658,7 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
             area=payload.area_m2,
             radius_m=DEFAULT_RADIUS_M,
             full_address=geo.full_address,
+            target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
             total_floors=payload.total_floors,
@@ -659,6 +682,7 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
             area=payload.area_m2,
             radius_m=DEFAULT_RADIUS_M,
             full_address=geo.full_address,
+            target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
             total_floors=payload.total_floors,
@@ -674,6 +698,7 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
             area=payload.area_m2,
             radius_m=FALLBACK_RADIUS_M,
             full_address=geo.full_address,
+            target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
             total_floors=payload.total_floors,
@@ -695,6 +720,7 @@ async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> Aggreg
             radius_m=FALLBACK_RADIUS_M,
             area_tolerance=0.25,
             full_address=geo.full_address,
+            target_house_id=target_house_id,
             year_built=target_year,
             house_type=target_house_type,
             total_floors=payload.total_floors,
@@ -1292,12 +1318,13 @@ def _fetch_analogs(
     total_floors: int | None = None,
     cohort_year_min: int | None = None,  # NEW: lower bound year_built inclusive
     cohort_year_max: int | None = None,  # NEW: upper bound year_built inclusive
-    # TODO: когда listings получит колонку house_id_fk — добавить ext_house_id JOIN для Tier S.
+    target_house_id: int | None = None,  # #6: canonical house for same-building Tier S
 ) -> tuple[list[dict[str, Any]], bool, str]:
     """SELECT аналогов — трёхуровневый house-match (S → H → W).
 
-    **Tier S (same building):** address ILIKE prefix-match по short_addr.
-    Если ≥3 результатов → возвращаем только их; tier='S'.
+    **Tier S (same building):** сначала канонический match по house_id_fk
+    (если задан target_house_id) — детерминированно «тот же дом»; fallback —
+    address ILIKE prefix-match по short_addr. Если ≥3 результатов → tier='S'.
 
     **Tier H (same class):** PostGIS + rooms + area + year ±15 + total_floors ±30%.
     Если ≥5 результатов → возвращаем; tier='H'.
@@ -1335,7 +1362,56 @@ def _fetch_analogs(
         "cohort_year_max": cohort_year_max,
     }
 
-    # ── Tier S: same building ─────────────────────────────────────────────────
+    # ── Tier S (canonical): same building via house_id_fk ─────────────────────
+    # #6: детерминированный «тот же дом» через канонический house-граф. 99%
+    # listings слинкованы (listings.house_id_fk → houses.id), что несравнимо
+    # надёжнее address-string match'а между разнородными источниками
+    # (Avito/Cian/Yandex форматируют адрес по-разному). Если target_house_id
+    # неизвестен или результатов <3 — падаем на address-prefix Tier S ниже.
+    if target_house_id is not None:
+        tier_sc_params = {**base_params, "target_house_id": target_house_id}
+        tier_sc_rows = (
+            db.execute(
+                text(
+                    f"""
+                WITH base AS (
+                    SELECT
+                        {_ANALOG_SELECT_COLS},
+                        0.0 AS distance_m,
+                        0.0 AS relevance_score,
+                        row_number() OVER (PARTITION BY address ORDER BY scraped_at DESC) AS rn_addr
+                    FROM listings
+                    WHERE house_id_fk = CAST(:target_house_id AS bigint)
+                    {_COMMON_WHERE}
+                )
+                SELECT
+                    source, source_url, address, lat, lon,
+                    rooms, area_m2, floor, total_floors,
+                    price_rub, price_per_m2,
+                    listing_date, days_on_market, photo_urls,
+                    scraped_at, distance_m, relevance_score
+                FROM base
+                WHERE rn_addr <= :max_per_addr
+                ORDER BY scraped_at DESC
+                LIMIT 300
+                """
+                ),
+                tier_sc_params,
+            )
+            .mappings()
+            .all()
+        )
+
+        tier_sc = [dict(r) for r in tier_sc_rows]
+        if len(tier_sc) >= 3:
+            logger.info(
+                "analogs tier=S(canonical) house_id=%s → %d results",
+                target_house_id,
+                len(tier_sc),
+            )
+            return _stratify_candidates(tier_sc), radius_m > DEFAULT_RADIUS_M, "S"
+
+    # ── Tier S (fallback): same building via address prefix ───────────────────
     short_addr = _extract_short_addr(full_address)
 
     if short_addr:

tradein-mvp/backend/app/services/matching/houses.py

@@ -183,6 +183,80 @@ def match_or_create_house(
     return (house_id, 1.0, 'new')
 
 
+def match_house_readonly(
+    db: Session,
+    *,
+    address: str | None = None,
+    lat: float | None = None,
+    lon: float | None = None,
+    cadastral_number: str | None = None,
+) -> tuple[int, float, str] | None:
+    """Match a target address to an EXISTING canonical house WITHOUT creating one.
+
+    Read-only counterpart of match_or_create_house() for the estimate-target flow:
+    we must NOT pollute `houses` with user-queried addresses, so this never INSERTs,
+    never writes house_sources / aliases, and takes no advisory lock (no write race
+    to guard).
+
+    Tiers (same lookups as match_or_create_house, minus source_exact which needs an
+    ext_id):
+      1. cadastr_exact  — houses.cadastral_number = :cad      (confidence 1.0)
+      2. fingerprint    — house_address_aliases.fingerprint   (confidence 0.9)
+      3. geo_proximity  — within 50 m of an existing house     (confidence 0.7)
+
+    Returns (house_id, confidence, method) or None if nothing confident matches.
+    Uses 50 m for geo (vs 30 m in match_or_create_house) to absorb geocoder jitter
+    on the user-entered target address.
+    """
+    # Tier 0: cadastral number exact match
+    if cadastral_number:
+        row = db.execute(
+            text('SELECT id FROM houses WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1'),
+            {'cad': cadastral_number},
+        ).mappings().first()
+        if row:
+            house_id = int(row['id'])
+            logger.info(
+                'house readonly match cadastr_exact house_id=%s cad=%s', house_id, cadastral_number
+            )
+            return (house_id, 1.0, 'cadastr_exact')
+
+    # Tier 1: address fingerprint lookup
+    fp = address_fingerprint(address, lat, lon)
+    row = db.execute(
+        text('SELECT house_id FROM house_address_aliases WHERE fingerprint = :fp LIMIT 1'),
+        {'fp': fp},
+    ).mappings().first()
+    if row:
+        house_id = int(row['house_id'])
+        logger.info('house readonly match fingerprint house_id=%s fp=%s', house_id, fp)
+        return (house_id, 0.9, 'fingerprint')
+
+    # Tier 2: geo-proximity — within 50 m
+    if lat is not None and lon is not None:
+        row = db.execute(
+            text("""
+                SELECT id,
+                       ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS dist
+                FROM houses
+                WHERE geom IS NOT NULL
+                  AND ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, 50)
+                ORDER BY dist ASC
+                LIMIT 1
+            """),
+            {'lat': lat, 'lon': lon},
+        ).mappings().first()
+        if row:
+            house_id = int(row['id'])
+            logger.info(
+                'house readonly match geo_proximity house_id=%s dist=%.1f',
+                house_id, float(row['dist']),
+            )
+            return (house_id, 0.7, 'geo_proximity')
+
+    return None
+
+
 def _upsert_house_source(
     db: Session,
     *,

tradein-mvp/backend/tests/test_same_building_match.py

@@ -0,0 +1,153 @@
+"""Tests for #6 same-building analog matching via canonical house_id.
+
+Two pieces:
+  1. matching.houses.match_house_readonly — read-only target→house resolver
+     (cadastr → fingerprint → geo, никогда не INSERT'ит).
+  2. estimator._fetch_analogs canonical Tier S — when target_house_id задан,
+     матчит listings по house_id_fk; иначе fallback на address-prefix Tier S.
+"""
+import os
+
+# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
+os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
+
+from datetime import UTC, datetime
+from typing import Any
+from unittest.mock import MagicMock
+
+from app.services.estimator import _fetch_analogs
+from app.services.matching.houses import match_house_readonly
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+def _listing(**over: Any) -> dict[str, Any]:
+    base = {
+        "source": "cian",
+        "source_url": "https://cian.ru/offer/1",
+        "address": "Екатеринбург, ул Малышева, д 125",
+        "lat": 56.8408,
+        "lon": 60.6540,
+        "rooms": 2,
+        "area_m2": 50.0,
+        "floor": 3,
+        "total_floors": 9,
+        "price_rub": 6_000_000.0,
+        "price_per_m2": 120_000.0,
+        "listing_date": datetime(2026, 5, 1),
+        "days_on_market": 10,
+        "photo_urls": [],
+        "scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
+        "distance_m": 0.0,
+        "relevance_score": 0.0,
+    }
+    base.update(over)
+    return base
+
+
+def _db_returning(rows: list[dict[str, Any]]) -> MagicMock:
+    """Session mock where every db.execute().mappings().all() returns rows."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = rows
+    return db
+
+
+def _executed_sqls(db: MagicMock) -> list[str]:
+    return [str(call.args[0]) for call in db.execute.call_args_list if call.args]
+
+
+# ── match_house_readonly ────────────────────────────────────────────────────
+def test_readonly_cadastr_exact_hit():
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.side_effect = [{"id": 42}]
+    result = match_house_readonly(
+        db, address="ул Малышева 125", cadastral_number="66:41:0704045:350"
+    )
+    assert result == (42, 1.0, "cadastr_exact")
+
+
+def test_readonly_fingerprint_hit():
+    db = MagicMock()
+    # cadastral_number=None → cadastr tier skipped; first .first() is fingerprint
+    db.execute.return_value.mappings.return_value.first.side_effect = [{"house_id": 7}]
+    result = match_house_readonly(db, address="Екатеринбург, ул Малышева, д 125")
+    assert result == (7, 0.9, "fingerprint")
+
+
+def test_readonly_geo_proximity_hit():
+    db = MagicMock()
+    # fingerprint miss → geo hit
+    db.execute.return_value.mappings.return_value.first.side_effect = [
+        None, {"id": 9, "dist": 12.3}
+    ]
+    result = match_house_readonly(db, address="неизвестный адрес", lat=56.84, lon=60.65)
+    assert result == (9, 0.7, "geo_proximity")
+
+
+def test_readonly_no_match_returns_none():
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.side_effect = [None, None]
+    result = match_house_readonly(db, address="нигде", lat=56.84, lon=60.65)
+    assert result is None
+
+
+def test_readonly_never_inserts_or_writes():
+    """Критично: read-only matcher не должен ничего создавать/писать."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.side_effect = [None, None]
+    match_house_readonly(db, address="нигде", lat=56.84, lon=60.65)
+    for sql in _executed_sqls(db):
+        upper = sql.upper()
+        assert "INSERT" not in upper
+        assert "UPDATE" not in upper
+        assert "ADVISORY" not in upper  # no pg_advisory_xact_lock
+
+
+def test_readonly_no_coords_skips_geo_tier():
+    """Без lat/lon geo-tier не запускается; при fingerprint-miss → None."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.side_effect = [None]
+    result = match_house_readonly(db, address="только адрес без координат")
+    assert result is None
+
+
+# ── _fetch_analogs canonical Tier S ──────────────────────────────────────────
+def test_canonical_tier_s_fires_with_house_id():
+    rows = [_listing(source=s) for s in ("cian", "avito", "yandex", "n1")]
+    db = _db_returning(rows)
+    listings, _fallback, tier = _fetch_analogs(
+        db, lat=56.84, lon=60.65, rooms=2, area=50.0, radius_m=1000,
+        full_address="Екатеринбург, ул Малышева, д 125", target_house_id=555,
+    )
+    assert tier == "S"
+    assert len(listings) >= 3
+    # Канонический Tier S — первый запрос должен матчить по house_id_fk
+    assert "house_id_fk" in _executed_sqls(db)[0]
+
+
+def test_falls_back_to_address_tier_s_without_house_id():
+    rows = [_listing(source=s) for s in ("cian", "avito", "yandex", "n1")]
+    db = _db_returning(rows)
+    _listings, _fallback, tier = _fetch_analogs(
+        db, lat=56.84, lon=60.65, rooms=2, area=50.0, radius_m=1000,
+        full_address="Екатеринбург, ул Малышева, д 125", target_house_id=None,
+    )
+    assert tier == "S"
+    # Без target_house_id канонический tier пропущен — первый запрос по address ILIKE
+    first_sql = _executed_sqls(db)[0]
+    assert "house_id_fk" not in first_sql
+    assert "ILIKE" in first_sql
+
+
+def test_canonical_tier_s_too_few_falls_through():
+    """<3 в том же доме → не возвращаем Tier S(canonical), идём дальше."""
+    rows = [_listing()]  # только 1 в доме
+    db = _db_returning(rows)
+    # full_address без распознаваемого short_addr И year/floors None →
+    # address-Tier S и Tier H пропустятся, упадём в Tier W (тоже вернёт rows mock).
+    _listings, _fallback, tier = _fetch_analogs(
+        db, lat=56.84, lon=60.65, rooms=2, area=50.0, radius_m=1000,
+        full_address=None, target_house_id=555,
+    )
+    # canonical Tier S не сработал (1 < 3); tier — не S(canonical) early-return.
+    # Mock возвращает те же rows для Tier W, так что итог не пустой, tier='W'.
+    assert tier == "W"