feat(tradein): cross-source matching service (3-tier: cadastr / fingerprint / geo / composite) (#470)

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

@@ -0,0 +1,19 @@
+"""Cross-source matching service — public API.
+
+Provides 3-tier deduplication for houses and listings across Avito, Cian, Yandex, etc.
+Algorithm reference: decisions/Cross_Source_Matching_Strategy.md
+"""
+
+from app.services.matching.conflict_resolution import update_canonical_fields
+from app.services.matching.houses import match_or_create_house
+from app.services.matching.listings import (
+    match_or_create_listing,
+    upsert_listing_source,  # public helper for 'new' sentinel flow
+)
+
+__all__ = [
+    "match_or_create_house",
+    "match_or_create_listing",
+    "update_canonical_fields",
+    "upsert_listing_source",
+]

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

@@ -0,0 +1,224 @@
+"""House cross-source matching — 3-tier algorithm.
+
+Tier 0 (confidence 1.0): cadastral_number exact match on houses table.
+Tier 1 (confidence 1.0): ext_source + ext_id already in house_sources.
+Tier 2 (confidence 0.9): address_fingerprint match in house_address_aliases.
+Tier 3 (confidence 0.7): geo-proximity within 30 m (PostGIS ST_DWithin).
+New   (confidence 1.0): INSERT new canonical house.
+
+Algorithm reference: decisions/Cross_Source_Matching_Strategy.md sec 3
+"""
+
+import logging
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services.matching.normalize import address_fingerprint, normalize_address
+
+logger = logging.getLogger(__name__)
+
+
+def match_or_create_house(
+    db: Session,
+    ext_source: str,
+    ext_id: str,
+    address: str | None = None,
+    lat: float | None = None,
+    lon: float | None = None,
+    *,
+    year_built: int | None = None,
+    building_cadastral_number: str | None = None,
+    cadastral_number: str | None = None,
+    source_url: str | None = None,
+) -> tuple[int, float, str]:
+    """Match existing house or create new canonical record.
+
+    ⚠️ KNOWN LIMITATION: Concurrent calls for the same address can race —
+    two scrapers calling simultaneously for an unknown address can both
+    miss Tier 0-3 and each INSERT a new house row. Mitigation (Stage 8.x):
+    gate scraper concurrency to 1 per fingerprint OR add
+    pg_advisory_xact_lock(hashtext(normalized_address)) at function entry.
+
+    Returns:
+        (house_id, confidence ∈ [0.0, 1.0], method ∈ {
+            'cadastr_exact', 'source_exact', 'fingerprint', 'geo_proximity', 'new'
+        })
+
+    Method values:
+        'cadastr_exact'  — matched by cadastral number (confidence 1.0)
+        'source_exact'   — already in house_sources for this source+ext_id (confidence 1.0)
+        'fingerprint'    — matched by address fingerprint (confidence 0.9)
+        'geo_proximity'  — matched by geo within 30 m (confidence 0.7)
+        'new'            — new house created (confidence 1.0)
+    """
+    cad = building_cadastral_number or cadastral_number
+
+    # Tier 0: cadastral number exact match
+    if cad:
+        row = db.execute(
+            text('SELECT id FROM houses WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1'),
+            {'cad': cad},
+        ).mappings().first()
+        if row:
+            house_id = int(row['id'])
+            _upsert_house_source(
+                db, house_id=house_id, ext_source=ext_source, ext_id=ext_id,
+                method='cadastr_exact', confidence=1.0,
+            )
+            logger.info('house match cadastr_exact house_id=%s cad=%s', house_id, cad)
+            return (house_id, 1.0, 'cadastr_exact')
+
+    # Tier 1: source+ext_id already registered in house_sources
+    row = db.execute(
+        text(
+            'SELECT house_id FROM house_sources '
+            'WHERE ext_source = :s AND ext_id = :e LIMIT 1'
+        ),
+        {'s': ext_source, 'e': str(ext_id)},
+    ).mappings().first()
+    if row:
+        house_id = int(row['house_id'])
+        logger.info('house match source_exact house_id=%s src=%s ext_id=%s', house_id, ext_source,
+                    ext_id)
+        return (house_id, 1.0, 'source_exact')
+
+    # Compute fingerprint once for Tier 2 and reuse for new house registration
+    fp = address_fingerprint(address, lat, lon)
+
+    # Tier 2: address fingerprint lookup
+    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'])
+        _upsert_house_source(
+            db, house_id=house_id, ext_source=ext_source, ext_id=ext_id,
+            method='fingerprint', confidence=0.9,
+        )
+        logger.info('house match fingerprint house_id=%s fp=%s', house_id, fp)
+        return (house_id, 0.9, 'fingerprint')
+
+    # Tier 3: geo-proximity — within 30 m (PostGIS geography cast on both sides)
+    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, 30)
+                ORDER BY dist ASC
+                LIMIT 1
+            """),
+            {'lat': lat, 'lon': lon},
+        ).mappings().first()
+        if row:
+            house_id = int(row['id'])
+            _upsert_house_source(
+                db, house_id=house_id, ext_source=ext_source, ext_id=ext_id,
+                method='geo_proximity', confidence=0.7,
+            )
+            # Register fingerprint alias so future calls skip geo lookup
+            _insert_alias(db, house_id=house_id, address=address, fp=fp, source=ext_source)
+            logger.info(
+                'house match geo_proximity house_id=%s dist=%.1f src=%s',
+                house_id, float(row['dist']), ext_source,
+            )
+            return (house_id, 0.7, 'geo_proximity')
+
+    # New house — INSERT canonical record.
+    # geom column is auto-populated by houses_set_geom_trg BEFORE INSERT trigger from lat/lon.
+    # Do NOT include geom in the INSERT column list — trigger handles it.
+    url = source_url or f'matching://{ext_source}/{ext_id}'
+    row = db.execute(
+        text("""
+            INSERT INTO houses (source, ext_house_id, url, address, lat, lon, year_built,
+                                cadastral_number)
+            VALUES (
+                :src, :eid, :url,
+                :addr,
+                CAST(:lat AS double precision),
+                CAST(:lon AS double precision),
+                CAST(:yb AS integer),
+                :cad
+            )
+            ON CONFLICT (source, ext_house_id) DO UPDATE SET
+                address = COALESCE(EXCLUDED.address, houses.address)
+            RETURNING id
+        """),
+        {
+            'src': ext_source,
+            'eid': str(ext_id),
+            'url': url,
+            'addr': address,
+            'lat': lat,
+            'lon': lon,
+            'yb': year_built,
+            'cad': cad,
+        },
+    ).mappings().one()
+    house_id = int(row['id'])
+
+    _upsert_house_source(
+        db, house_id=house_id, ext_source=ext_source, ext_id=ext_id,
+        method='new', confidence=1.0,
+    )
+    _insert_alias(db, house_id=house_id, address=address, fp=fp, source=ext_source)
+
+    logger.info('house new house_id=%s addr=%r src=%s', house_id, address, ext_source)
+    return (house_id, 1.0, 'new')
+
+
+def _upsert_house_source(
+    db: Session,
+    *,
+    house_id: int,
+    ext_source: str,
+    ext_id: str,
+    method: str,
+    confidence: float,
+) -> None:
+    """Insert or refresh house_sources row for this source+ext_id."""
+    db.execute(
+        text("""
+            INSERT INTO house_sources (
+                house_id, ext_source, ext_id, confidence, matched_method, matched_at, last_seen_at
+            ) VALUES (
+                CAST(:hid AS bigint), :s, :e,
+                CAST(:c AS real), :m, NOW(), NOW()
+            )
+            ON CONFLICT (ext_source, ext_id) DO UPDATE SET
+                confidence    = GREATEST(EXCLUDED.confidence, house_sources.confidence),
+                last_seen_at  = NOW()
+        """),
+        {'hid': house_id, 's': ext_source, 'e': str(ext_id), 'c': confidence, 'm': method},
+    )
+
+
+def _insert_alias(
+    db: Session,
+    *,
+    house_id: int,
+    address: str | None,
+    fp: str,
+    source: str,
+) -> None:
+    """Register normalized_address alias for this house (idempotent on normalized_address)."""
+    db.execute(
+        text("""
+            INSERT INTO house_address_aliases (house_id, normalized_address, fingerprint, source)
+            VALUES (CAST(:hid AS bigint), :na, :fp, :src)
+            ON CONFLICT (normalized_address) DO NOTHING
+        """),
+        {
+            'hid': house_id,
+            'na': normalize_address(address),
+            'fp': fp,
+            'src': source,
+        },
+    )

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

@@ -0,0 +1,237 @@
+"""Listing cross-source matching — 3-tier algorithm.
+
+Tier 0 (confidence 1.0): cadastral_number exact match on listings table.
+Tier 1 (confidence 1.0): ext_source + ext_id already in listing_sources.
+Tier 2 (confidence 0.85): description_minhash exact match within same house.
+Tier 3 (confidence 0.75): composite match — house + floor + area ±2% + rooms.
+New   (sentinel 0):       caller is responsible for INSERT + call _upsert_listing_source.
+
+Algorithm reference: decisions/Cross_Source_Matching_Strategy.md sec 4
+"""
+
+import json
+import logging
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+
+def match_or_create_listing(
+    db: Session,
+    ext_source: str,
+    ext_id: str,
+    house_id: int,
+    floor: int | None = None,
+    rooms_count: int | None = None,
+    area_m2: float | None = None,
+    price_rub: float | None = None,
+    *,
+    cadastral_number: str | None = None,
+    description_minhash: str | None = None,
+    source_url: str | None = None,
+    source_data: dict | None = None,
+) -> tuple[int, float, str]:
+    """Match existing listing or signal that caller should create a new one.
+
+    Returns:
+        (listing_id, confidence in [0.0, 1.0], method name)
+
+    Method values:
+        'cadastr_exact'  — matched by cadastral number (confidence 1.0)
+        'source_exact'   — already in listing_sources for this source+ext_id (confidence 1.0)
+        'minhash'        — matched by description minhash within house (confidence 0.85)
+        'composite'      — matched by house+floor+area±2%+rooms (confidence 0.75)
+        'new'            — no match; caller must INSERT listing then call upsert_listing_source()
+
+    When method == 'new', listing_id is 0 (sentinel). Caller flow:
+        listing_id, conf, method = match_or_create_listing(db, ...)
+        if method == 'new':
+            listing_id = <INSERT canonical listing row>
+            upsert_listing_source(db, listing_id=listing_id, ...)
+    """
+    # Tier 0: cadastral number exact match
+    if cadastral_number:
+        row = db.execute(
+            text(
+                'SELECT id FROM listings WHERE cadastral_number = :cad ORDER BY id ASC LIMIT 1'
+            ),
+            {'cad': cadastral_number},
+        ).mappings().first()
+        if row:
+            listing_id = int(row['id'])
+            _upsert_listing_source(
+                db, listing_id=listing_id, ext_source=ext_source, ext_id=ext_id,
+                method='cadastr_exact', confidence=1.0,
+                price_rub=price_rub, area_m2=area_m2, floor=floor,
+                rooms_count=rooms_count, source_url=source_url, source_data=source_data,
+            )
+            logger.info('listing match cadastr_exact id=%s cad=%s', listing_id, cadastral_number)
+            return (listing_id, 1.0, 'cadastr_exact')
+
+    # Tier 1: source+ext_id already registered
+    row = db.execute(
+        text(
+            'SELECT listing_id FROM listing_sources '
+            'WHERE ext_source = :s AND ext_id = :e LIMIT 1'
+        ),
+        {'s': ext_source, 'e': str(ext_id)},
+    ).mappings().first()
+    if row:
+        listing_id = int(row['listing_id'])
+        logger.info('listing match source_exact id=%s src=%s ext_id=%s', listing_id, ext_source,
+                    ext_id)
+        return (listing_id, 1.0, 'source_exact')
+
+    # Tier 2: description minhash match within same house
+    # Stage 8 v1: exact match on pre-computed minhash string.
+    # Future (Stage 8.x): replace with datasketch MinHash LSH for approximate Jaccard.
+    if description_minhash:
+        row = db.execute(
+            text("""
+                SELECT id FROM listings
+                WHERE house_id = CAST(:hid AS bigint)
+                  AND description_minhash = :hash
+                LIMIT 1
+            """),
+            {'hid': house_id, 'hash': description_minhash},
+        ).mappings().first()
+        if row:
+            listing_id = int(row['id'])
+            _upsert_listing_source(
+                db, listing_id=listing_id, ext_source=ext_source, ext_id=ext_id,
+                method='minhash', confidence=0.85,
+                price_rub=price_rub, area_m2=area_m2, floor=floor,
+                rooms_count=rooms_count, source_url=source_url, source_data=source_data,
+            )
+            logger.info('listing match minhash id=%s hash=%s', listing_id, description_minhash)
+            return (listing_id, 0.85, 'minhash')
+
+    # Tier 3: composite match — house + floor + area ±2% + rooms
+    if house_id and floor is not None and area_m2 is not None and rooms_count is not None:
+        row = db.execute(
+            text("""
+                SELECT id,
+                       ABS(area_m2 - CAST(:area AS numeric)) AS diff
+                FROM listings
+                WHERE house_id = CAST(:hid AS bigint)
+                  AND floor = :fl
+                  AND rooms = :rc
+                  AND area_m2 BETWEEN CAST(:area AS numeric) * 0.98
+                              AND     CAST(:area AS numeric) * 1.02
+                ORDER BY diff ASC
+                LIMIT 1
+            """),
+            {'hid': house_id, 'fl': floor, 'rc': rooms_count, 'area': area_m2},
+        ).mappings().first()
+        if row:
+            listing_id = int(row['id'])
+            _upsert_listing_source(
+                db, listing_id=listing_id, ext_source=ext_source, ext_id=ext_id,
+                method='composite', confidence=0.75,
+                price_rub=price_rub, area_m2=area_m2, floor=floor,
+                rooms_count=rooms_count, source_url=source_url, source_data=source_data,
+            )
+            logger.info(
+                'listing match composite id=%s house=%s floor=%s area=%.1f rooms=%s',
+                listing_id, house_id, floor, area_m2, rooms_count,
+            )
+            return (listing_id, 0.75, 'composite')
+
+    # No match — signal caller to create new listing
+    return (0, 1.0, 'new')
+
+
+def upsert_listing_source(
+    db: Session,
+    *,
+    listing_id: int,
+    ext_source: str,
+    ext_id: str,
+    method: str,
+    confidence: float,
+    price_rub: float | None = None,
+    area_m2: float | None = None,
+    floor: int | None = None,
+    rooms_count: int | None = None,
+    source_url: str | None = None,
+    source_data: dict | None = None,
+) -> None:
+    """Public helper — register or refresh a listing_sources row.
+
+    Called by external code after creating a new canonical listing.
+    """
+    _upsert_listing_source(
+        db,
+        listing_id=listing_id,
+        ext_source=ext_source,
+        ext_id=ext_id,
+        method=method,
+        confidence=confidence,
+        price_rub=price_rub,
+        area_m2=area_m2,
+        floor=floor,
+        rooms_count=rooms_count,
+        source_url=source_url,
+        source_data=source_data,
+    )
+
+
+def _upsert_listing_source(
+    db: Session,
+    *,
+    listing_id: int,
+    ext_source: str,
+    ext_id: str,
+    method: str,
+    confidence: float,
+    price_rub: float | None,
+    area_m2: float | None,
+    floor: int | None,
+    rooms_count: int | None,
+    source_url: str | None,
+    source_data: dict | None,
+) -> None:
+    """Insert or refresh listing_sources row for this source+ext_id."""
+    raw = json.dumps(source_data) if source_data is not None else None
+    db.execute(
+        text("""
+            INSERT INTO listing_sources (
+                listing_id, ext_source, ext_id,
+                confidence, matched_method, matched_at, last_seen_at,
+                source_url, last_scraped_at,
+                price_rub, area_m2, floor, rooms_count, raw_payload
+            ) VALUES (
+                CAST(:lid AS bigint), :s, :e,
+                CAST(:c AS real), :m, NOW(), NOW(),
+                :url, NOW(),
+                CAST(:p AS bigint), CAST(:a AS numeric), :fl, :rc,
+                CAST(:raw AS jsonb)
+            )
+            ON CONFLICT (ext_source, ext_id) DO UPDATE SET
+                confidence    = GREATEST(EXCLUDED.confidence, listing_sources.confidence),
+                last_seen_at  = NOW(),
+                last_scraped_at = NOW(),
+                price_rub     = COALESCE(EXCLUDED.price_rub,    listing_sources.price_rub),
+                area_m2       = COALESCE(EXCLUDED.area_m2,      listing_sources.area_m2),
+                floor         = COALESCE(EXCLUDED.floor,        listing_sources.floor),
+                rooms_count   = COALESCE(EXCLUDED.rooms_count,  listing_sources.rooms_count),
+                raw_payload   = COALESCE(EXCLUDED.raw_payload,  listing_sources.raw_payload)
+        """),
+        {
+            'lid': listing_id,
+            's': ext_source,
+            'e': str(ext_id),
+            'c': confidence,
+            'm': method,
+            'url': source_url,
+            # Migration 029 declares listing_sources.price_rub bigint — whole rubles only,
+            # kopecks truncated. Document expectation in caller.
+            'p': int(price_rub) if price_rub is not None else None,
+            'a': area_m2,
+            'fl': floor,
+            'rc': rooms_count,
+            'raw': raw,
+        },
+    )

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

@@ -0,0 +1,76 @@
+"""Address normalization and fingerprint utilities.
+
+Used by 3-tier matching to produce stable, source-independent keys.
+Algorithm reference: decisions/Cross_Source_Matching_Strategy.md sec 3.2
+"""
+
+import hashlib
+import re
+import unicodedata
+
+# Strip punctuation EXCEPT hyphens — keep hyphens so abbreviation rules like
+# 'пр-кт', 'б-р', 'пр-д' can match before they are collapsed to spaces.
+_PUNCT = re.compile(r'[^\w\s\-]', flags=re.UNICODE)
+_WS = re.compile(r'\s+')
+
+# Abbreviation expansion — applied after lowercasing, bounded by spaces.
+# Order matters: longer/more-specific patterns first to avoid partial overlap.
+# Hyphenated forms ('пр-кт', 'б-р', 'пр-д') come first so they match before
+# the shorter variants ('пр', 'б') would erroneously consume the prefix.
+_ABBREV = [
+    (' пр-кт ', ' проспект '),
+    (' пр-кт. ', ' проспект '),
+    (' б-р ', ' бульвар '),
+    (' пр-д ', ' проезд '),
+    (' ш ', ' шоссе '),
+    (' ул ', ' улица '),
+    (' ул. ', ' улица '),
+    (' пр ', ' проспект '),
+    (' пр. ', ' проспект '),
+    (' пер ', ' переулок '),
+    (' пл ', ' площадь '),
+    (' наб ', ' набережная '),
+    (' бул ', ' бульвар '),
+    (' туп ', ' тупик '),
+    (' стр ', ' строение '),
+    (' корп ', ' корпус '),
+    (' д ', ' дом '),
+    (' к ', ' корпус '),
+]
+
+
+def normalize_address(text: str | None) -> str:
+    """Normalize address string for cross-source comparison.
+
+    Steps:
+      1. Unicode NFC normalization
+      2. Lowercase
+      3. Strip punctuation except hyphens (preserve 'пр-кт', 'б-р', 'пр-д' for expansion)
+      4. Collapse whitespace
+      5. Expand common Russian address abbreviations (hyphenated forms first)
+      6. Collapse remaining hyphens to spaces
+    """
+    if not text:
+        return ""
+    s = unicodedata.normalize('NFC', text).lower()
+    s = _PUNCT.sub(' ', s)  # strip punctuation EXCEPT hyphens
+    s = _WS.sub(' ', s).strip()
+    # Wrap with spaces for clean boundary matching
+    s = f' {s} '
+    for short, full in _ABBREV:
+        s = s.replace(short, full)
+    # Collapse remaining hyphens (e.g. standalone '-' separators or numeric ranges)
+    s = s.replace('-', ' ')
+    return _WS.sub(' ', s).strip()
+
+
+def address_fingerprint(address: str | None, lat: float | None, lon: float | None) -> str:
+    """SHA-256 fingerprint of normalized address + rounded coordinates (4 dp ≈ 11 m).
+
+    Returns first 32 hex chars of the digest (128 bits — collision-safe for millions of rows).
+    """
+    norm_addr = normalize_address(address or '')
+    lat_r = f'{lat:.4f}' if lat is not None else ''
+    lon_r = f'{lon:.4f}' if lon is not None else ''
+    key = f'{norm_addr}|{lat_r}|{lon_r}'
+    return hashlib.sha256(key.encode('utf-8')).hexdigest()[:32]

tradein-mvp/backend/tests/test_matching.py

@@ -0,0 +1,367 @@
+"""Unit tests for cross-source matching service.
+
+Covers:
+  - normalize_address(): abbreviation expansion, punctuation stripping, edge cases
+  - address_fingerprint(): stability on case variation, sensitivity to coords, None handling
+  - match_or_create_house() / match_or_create_listing(): mock-DB tier routing
+
+Reference: decisions/Cross_Source_Matching_Strategy.md
+"""
+
+from unittest.mock import MagicMock
+
+from app.services.matching.conflict_resolution import (
+    HOUSE_FIELD_PRIORITY,
+    LISTING_FIELD_PRIORITY,
+    update_canonical_fields,
+)
+from app.services.matching.normalize import address_fingerprint, normalize_address
+
+# ---------------------------------------------------------------------------
+# normalize_address
+# ---------------------------------------------------------------------------
+
+
+def test_normalize_strips_punctuation_and_collapses_whitespace():
+    assert normalize_address("  Test,   Address!  ") == "test address"
+
+
+def test_normalize_expands_ul_abbreviation():
+    result = normalize_address("ул Ленина, 5")
+    assert "улица" in result
+    assert "5" in result
+
+
+def test_normalize_expands_pr_abbreviation():
+    result = normalize_address("пр Мира 100")
+    assert "проспект" in result
+    assert "100" in result
+
+
+def test_normalize_expands_prkT_abbreviation():
+    result = normalize_address("пр-кт Ленина 10")
+    assert "проспект" in result
+    # 'кт' must NOT remain as a stray token after expansion
+    assert 'кт' not in result.split()
+
+
+def test_normalize_prkT_no_leftover_kt():
+    """пр-кт must expand to 'проспект' cleanly — no leftover 'кт' token."""
+    result = normalize_address("пр-кт Ленина 10")
+    tokens = result.split()
+    assert "проспект" in tokens
+    assert "пр" not in tokens
+    assert "кт" not in tokens
+
+
+def test_normalize_handles_none():
+    assert normalize_address(None) == ""
+
+
+def test_normalize_handles_empty_string():
+    assert normalize_address("") == ""
+
+
+def test_normalize_lowercases():
+    assert normalize_address("ТЕСТ") == "тест"
+
+
+def test_normalize_unicode_nfc():
+    # Composed vs decomposed е (U+0435 vs U+0435)
+    s1 = normalize_address("улица")
+    s2 = normalize_address("улица")
+    assert s1 == s2
+
+
+# ---------------------------------------------------------------------------
+# address_fingerprint
+# ---------------------------------------------------------------------------
+
+
+def test_fingerprint_stable_on_case_variation():
+    f1 = address_fingerprint("Ленина 5", 56.8378, 60.5946)
+    f2 = address_fingerprint("ЛЕНИНА 5", 56.8378, 60.5946)
+    assert f1 == f2
+
+
+def test_fingerprint_distinct_on_different_lat():
+    f1 = address_fingerprint("Same Address", 56.8, 60.5)
+    f2 = address_fingerprint("Same Address", 56.9, 60.5)
+    assert f1 != f2
+
+
+def test_fingerprint_distinct_on_different_lon():
+    f1 = address_fingerprint("Same Address", 56.8, 60.5)
+    f2 = address_fingerprint("Same Address", 56.8, 60.6)
+    assert f1 != f2
+
+
+def test_fingerprint_stable_within_11m(monkeypatch):
+    # 4 decimal places → ~11 m resolution; coords differing in 5th decimal should match
+    f1 = address_fingerprint("Addr", 56.83780, 60.59460)
+    f2 = address_fingerprint("Addr", 56.83781, 60.59461)
+    # 4dp rounding: 56.8378 == 56.8378, 60.5946 == 60.5946 → same fingerprint
+    assert f1 == f2
+
+
+def test_fingerprint_handles_none_coords():
+    fp = address_fingerprint("Some Address", None, None)
+    assert len(fp) == 32  # truncated sha256 hex
+
+
+def test_fingerprint_handles_none_address():
+    fp = address_fingerprint(None, 56.8, 60.5)
+    assert len(fp) == 32
+
+
+def test_fingerprint_all_none():
+    fp = address_fingerprint(None, None, None)
+    assert len(fp) == 32
+
+
+# ---------------------------------------------------------------------------
+# match_or_create_house — mock DB tier routing
+# ---------------------------------------------------------------------------
+
+
+def _make_db(rows: list[dict | None]) -> MagicMock:
+    """Build a Session mock that returns rows sequentially per execute() call."""
+    db = MagicMock()
+    side_effects = []
+    for row in rows:
+        result = MagicMock()
+        mappings = MagicMock()
+        if row is None:
+            mappings.first.return_value = None
+            mappings.one.return_value = None
+        else:
+            mappings.first.return_value = row
+            mappings.one.return_value = row
+        result.mappings.return_value = mappings
+        side_effects.append(result)
+    db.execute.side_effect = side_effects
+    return db
+
+
+def test_match_house_tier0_cadastr():
+    """Tier 0: cadastral_number match → returns cadastr_exact without geo query."""
+    from app.services.matching.houses import match_or_create_house
+
+    db = _make_db([
+        {'id': 42},   # cadastral match
+        None,         # _upsert_house_source (INSERT ... ON CONFLICT)
+    ])
+    house_id, conf, method = match_or_create_house(
+        db, 'cian', 'ext-001',
+        address='ул Ленина 5', lat=56.8, lon=60.5,
+        cadastral_number='66:41:0000000:999',
+    )
+    assert house_id == 42
+    assert conf == 1.0
+    assert method == 'cadastr_exact'
+
+
+def test_match_house_tier1_source_exact():
+    """Tier 1: ext_source+ext_id already in house_sources.
+
+    No cadastral_number supplied → Tier 0 skipped entirely.
+    First db.execute call goes to house_sources query.
+    source_exact returns immediately without calling _upsert_house_source.
+    """
+    from app.services.matching.houses import match_or_create_house
+
+    db = _make_db([
+        {'house_id': 7},  # house_sources hit (first and only execute call)
+    ])
+    house_id, conf, method = match_or_create_house(
+        db, 'avito', 'ext-999',
+        address='пр Мира 10', lat=56.8, lon=60.5,
+    )
+    assert house_id == 7
+    assert method == 'source_exact'
+
+
+def test_match_house_tier2_fingerprint():
+    """Tier 2: fingerprint match in house_address_aliases."""
+    from app.services.matching.houses import match_or_create_house
+
+    db = _make_db([
+        None,              # house_sources miss
+        {'house_id': 15},  # fingerprint hit
+        None,              # _upsert_house_source
+    ])
+    house_id, conf, method = match_or_create_house(
+        db, 'yandex', 'ext-123',
+        address='Тестовая 1', lat=56.84, lon=60.60,
+    )
+    assert house_id == 15
+    assert conf == 0.9
+    assert method == 'fingerprint'
+
+
+def test_match_house_tier3_geo():
+    """Tier 3: geo-proximity match."""
+    from app.services.matching.houses import match_or_create_house
+
+    db = _make_db([
+        None,              # house_sources miss
+        None,              # fingerprint miss
+        {'id': 22, 'dist': 15.5},  # geo hit
+        None,              # _upsert_house_source
+        None,              # _insert_alias
+    ])
+    house_id, conf, method = match_or_create_house(
+        db, 'n1', 'ext-456',
+        address='Новая 3', lat=56.83, lon=60.59,
+    )
+    assert house_id == 22
+    assert conf == 0.7
+    assert method == 'geo_proximity'
+
+
+def test_match_house_new():
+    """All tiers miss → new house created."""
+    from app.services.matching.houses import match_or_create_house
+
+    db = _make_db([
+        None,          # house_sources miss
+        None,          # fingerprint miss
+        None,          # geo miss
+        {'id': 99},    # INSERT RETURNING id
+        None,          # _upsert_house_source
+        None,          # _insert_alias
+    ])
+    house_id, conf, method = match_or_create_house(
+        db, 'avito', 'new-ext',
+        address='Новостройка 1', lat=56.85, lon=60.61,
+    )
+    assert house_id == 99
+    assert conf == 1.0
+    assert method == 'new'
+
+
+# ---------------------------------------------------------------------------
+# match_or_create_listing — mock DB tier routing
+# ---------------------------------------------------------------------------
+
+
+def test_match_listing_tier0_cadastr():
+    from app.services.matching.listings import match_or_create_listing
+
+    db = _make_db([
+        {'id': 200},  # cadastral hit
+        None,         # _upsert_listing_source
+    ])
+    listing_id, conf, method = match_or_create_listing(
+        db, 'cian', 'cian-555', house_id=10,
+        floor=5, rooms_count=2, area_m2=48.5, price_rub=5_000_000.0,
+        cadastral_number='66:41:0204016:1234',
+    )
+    assert listing_id == 200
+    assert method == 'cadastr_exact'
+
+
+def test_match_listing_tier1_source_exact():
+    """Tier 1: no cadastral_number → Tier 0 skipped; first execute is listing_sources hit."""
+    from app.services.matching.listings import match_or_create_listing
+
+    db = _make_db([
+        {'listing_id': 101},  # listing_sources hit (first and only execute call)
+    ])
+    listing_id, conf, method = match_or_create_listing(
+        db, 'avito', 'avito-888', house_id=10,
+    )
+    assert listing_id == 101
+    assert method == 'source_exact'
+
+
+def test_match_listing_tier2_minhash():
+    from app.services.matching.listings import match_or_create_listing
+
+    db = _make_db([
+        None,            # listing_sources miss
+        {'id': 55},      # minhash hit
+        None,            # _upsert_listing_source
+    ])
+    listing_id, conf, method = match_or_create_listing(
+        db, 'cian', 'cian-999', house_id=10,
+        description_minhash='abc123deadbeef',
+    )
+    assert listing_id == 55
+    assert conf == 0.85
+    assert method == 'minhash'
+
+
+def test_match_listing_tier3_composite():
+    """Tier 3: no cadastral + no minhash provided → tiers 0/2 skipped.
+    Execute order: listing_sources miss → composite hit → _upsert.
+    """
+    from app.services.matching.listings import match_or_create_listing
+
+    db = _make_db([
+        None,                    # listing_sources miss (Tier 1)
+        {'id': 77, 'diff': 0.1}, # composite hit (Tier 3; Tier 2 skipped — no hash)
+        None,                    # _upsert_listing_source
+    ])
+    listing_id, conf, method = match_or_create_listing(
+        db, 'cian', 'cian-777', house_id=10,
+        floor=3, rooms_count=1, area_m2=36.0,
+    )
+    assert listing_id == 77
+    assert conf == 0.75
+    assert method == 'composite'
+
+
+def test_match_listing_new():
+    from app.services.matching.listings import match_or_create_listing
+
+    db = _make_db([
+        None,  # listing_sources miss
+        None,  # composite miss (floor/area/rooms all None → tier skipped)
+    ])
+    listing_id, conf, method = match_or_create_listing(
+        db, 'avito', 'avito-new', house_id=10,
+    )
+    assert listing_id == 0
+    assert method == 'new'
+
+
+# ---------------------------------------------------------------------------
+# conflict_resolution
+# ---------------------------------------------------------------------------
+
+
+def test_field_priority_dicts_not_empty():
+    assert HOUSE_FIELD_PRIORITY
+    assert LISTING_FIELD_PRIORITY
+
+
+VALID_STRING_RULES = {"union", "cross_validate", "first_non_null"}
+
+
+def test_field_priority_sources_are_lists_or_known_string_rules():
+    """Each priority entry is either a source list or a known string rule.
+
+    PR #471 (main) introduced 'union' / 'cross_validate' / 'first_non_null' string
+    rules alongside list entries. Both are valid per LISTING_FIELD_PRIORITY type hint.
+    """
+    for col, sources in HOUSE_FIELD_PRIORITY.items():
+        assert isinstance(sources, list) or sources in VALID_STRING_RULES, (
+            f'HOUSE_FIELD_PRIORITY[{col!r}] must be list or known rule, got {sources!r}'
+        )
+    for col, sources in LISTING_FIELD_PRIORITY.items():
+        assert isinstance(sources, list) or sources in VALID_STRING_RULES, (
+            f'LISTING_FIELD_PRIORITY[{col!r}] must be list or known rule, got {sources!r}'
+        )
+
+
+def test_update_canonical_fields_is_callable():
+    """Stage 8 v1 — legacy stub (no-op pass). Full arbitration deferred to Stage 8.x.
+
+    PR #471 (main) replaced NotImplementedError stub with silent pass to allow
+    backward-compat imports. Test updated accordingly.
+    """
+    db = MagicMock()
+    # Should not raise — no-op stub returns None
+    result = update_canonical_fields(db, listing_id=1, ext_source='cian', lot_data=object())
+    assert result is None