feat(tradein-estimator): tiered house-match S→H→W (same-house, class, wide)

Currently year/house_type only influence relevance sort. For a 17-эт. 1975
building we returned 5-9 эт. брежневки as analogs.

Three tiers:
- S (same building): address ILIKE prefix, ≥3 → return only these
- H (same class): year ±15, total_floors ±30%, ≥5 → return
- W (wide): current logic without year/floors WHERE filter

Tier propagated to confidence_explanation so user sees why analogs may differ.

Source: estimate a0a0b820-e8a8-4eee-aa73-0ab3b98ac233.

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

@@ -22,6 +22,7 @@ from __future__ import annotations
 import hashlib
 import json
 import logging
+import math
 import re
 from datetime import UTC, datetime, timedelta
 from typing import Any
@@ -493,39 +494,45 @@ async def estimate_quality(
             if target_house_type is None:
                 target_house_type = house_meta.house_type
 
-    # 3. Three-tier fallback:
+    # 3. House-match: S → H → W tiered lookup (see _fetch_analogs docstring).
-    #    a) 1km + ±15% area
+    #    Radius fallback still applies when W tier has < 5 results at 1km.
-    #    b) 2km + ±15% area  (fallback_used = True)
+    listings, fallback_used, analog_tier = _fetch_analogs(
-    #    c) 2km + ±25% area  (fallback_used = True, area_widened = True)
-    listings, fallback_used = _fetch_analogs(
         db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
         radius_m=DEFAULT_RADIUS_M,
+        full_address=geo.full_address,
         year_built=target_year, house_type=target_house_type,
+        total_floors=payload.total_floors,
     )
     area_widened = False
 
     if len(listings) < 5:
-        listings_wide, _ = _fetch_analogs(
+        listings_wide, _, analog_tier_wide = _fetch_analogs(
             db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
             radius_m=FALLBACK_RADIUS_M,
+            full_address=geo.full_address,
             year_built=target_year, house_type=target_house_type,
+            total_floors=payload.total_floors,
         )
         if len(listings_wide) > len(listings):
             listings = listings_wide
             fallback_used = True
+            analog_tier = analog_tier_wide
 
     # Tier C: если даже на 2км мало — расширяем area tolerance до ±25%
     # (актуально для отдалённых районов / новостроек с нестандартной планировкой)
     if len(listings) < 3:
-        listings_widearea, _ = _fetch_analogs(
+        listings_widearea, _, analog_tier_wa = _fetch_analogs(
             db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
             radius_m=FALLBACK_RADIUS_M, area_tolerance=0.25,
+            full_address=geo.full_address,
             year_built=target_year, house_type=target_house_type,
+            total_floors=payload.total_floors,
         )
         if len(listings_widearea) > len(listings):
             listings = listings_widearea
             fallback_used = True
             area_widened = True
+            analog_tier = analog_tier_wa
 
     # 3. Outlier filter
     listings_clean = _filter_outliers(listings)
@@ -567,7 +574,20 @@ async def estimate_quality(
         q3_ppm2 if listings_clean else 0, fallback_used, area_widened,
         listings=listings_clean,
     )
-    explanation = (explanation or "") + repair_note
+
+    # Tier note — информируем пользователя о качестве house-match
+    tier_note = ""
+    if analog_tier == "S":
+        tier_note = " (аналоги из того же дома)"
+    elif analog_tier == "H":
+        tf_str = f"{payload.total_floors}-эт." if payload.total_floors else ""
+        yr_str = f"{target_year}±15 г." if target_year else ""
+        parts_str = ", ".join(p for p in [yr_str, tf_str] if p)
+        tier_note = f" (аналоги из домов того же класса: {parts_str})" if parts_str else ""
+    else:
+        tier_note = " (нет аналогов в том же доме/классе — расширили поиск)"
+
+    explanation = (explanation or "") + tier_note + repair_note
 
     # ── Stage 3: Avito IMV evaluation as 5-th source (on-demand cached) ──
     imv_eval: IMVEvaluation | None = None
@@ -841,17 +861,119 @@ def _compute_freshness_minutes(lots: list[dict[str, Any]]) -> int | None:
 
 
 # ── Internals ────────────────────────────────────────────────────────────────
+
+
+def _extract_short_addr(full_address: str) -> str | None:
+    """Извлекает «улица + номер дома» из полного адреса для поиска в том же доме.
+
+    Примеры:
+      "г. Екатеринбург, ул. Крауля, 48/2, кв. 5"  → "ул. Крауля, 48/2"
+      "Екатеринбург, Ленина, 36, корп. 2, кв. 10" → "Ленина, 36"
+      "Свердловская обл., г. Екатеринбург, пр-т Ленина, 36 к2" → "пр-т Ленина, 36"
+
+    Алгоритм:
+    1. Разбиваем по запятой.
+    2. Отбрасываем сегменты, которые выглядят как «г.», «обл.», «р-н» (city/region prefix).
+    3. Берём первые 2 оставшихся токена (улица + дом), strip кв/корп/к-суффикс из последнего.
+    4. Возвращаем None, если результат слишком короткий (< 3 символов) — не с чем матчить.
+    """
+    if not full_address:
+        return None
+
+    parts = [p.strip() for p in full_address.split(",")]
+
+    # Паттерн для «административных» сегментов: г., обл., р-н, с., д. (населённый пункт)
+    admin_re = re.compile(
+        r"^(г\.?|обл\.?|р-н\.?|пгт\.?|с\.?|д\.?|мкр\.?)\s",
+        re.IGNORECASE,
+    )
+
+    meaningful = [p for p in parts if not admin_re.match(p) and len(p) > 1]
+
+    if len(meaningful) < 2:
+        return None
+
+    street = meaningful[0]
+    house_raw = meaningful[1]
+
+    # Убираем «кв. N», «корп. N», «к2», «к 2» из номера дома
+    house = re.sub(r"\s*(кв\.?|корп\.?|к\.?)\s*\d+.*$", "", house_raw, flags=re.IGNORECASE).strip()
+
+    result = f"{street}, {house}"
+    return result if len(result) >= 3 else None
+
+
+def _stratify_candidates(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Стратифицированная выборка Approach B — гарантирует MIN_ANALOGS_PER_SOURCE слотов.
+
+    Candidates должны быть уже отсортированы по relevance_score (ASC).
+    """
+    guaranteed: list[dict[str, Any]] = []
+    guaranteed_ids: set[int] = set()
+    by_source: dict[str, list[dict[str, Any]]] = {}
+    for row in candidates:
+        src = row.get("source") or "unknown"
+        by_source.setdefault(src, []).append(row)
+
+    for _src, src_rows in by_source.items():
+        quota = min(len(src_rows), MIN_ANALOGS_PER_SOURCE)
+        for row in src_rows[:quota]:
+            if id(row) not in guaranteed_ids:
+                guaranteed.append(row)
+                guaranteed_ids.add(id(row))
+
+    remaining_slots = 50 - len(guaranteed)
+    remainder: list[dict[str, Any]] = []
+    if remaining_slots > 0:
+        for row in candidates:
+            if id(row) not in guaranteed_ids:
+                remainder.append(row)
+                if len(remainder) >= remaining_slots:
+                    break
+
+    result = guaranteed + remainder
+    result.sort(key=lambda r: r.get("relevance_score") or 0.0)
+    return result[:50]
+
+
+_ANALOG_SELECT_COLS = """
+    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
+"""
+
+_COMMON_WHERE = """
+    AND rooms = :rooms
+    AND area_m2 BETWEEN :area_min AND :area_max
+    AND is_active = true
+    AND scraped_at > NOW() - (:fresh_days || ' days')::interval
+    AND price_rub > 0
+"""
+
+
 def _fetch_analogs(
     db: Session, *, lat: float, lon: float, rooms: int, area: float, radius_m: int,
+    full_address: str | None = None,
     area_tolerance: float = AREA_TOLERANCE,
     year_built: int | None = None, house_type: str | None = None,
-) -> tuple[list[dict[str, Any]], bool]:
+    total_floors: int | None = None,
-    """SELECT аналогов с PostGIS distance + house-match relevance.
+    ext_house_id: str | None = None,
+) -> tuple[list[dict[str, Any]], bool, str]:
+    """SELECT аналогов — трёхуровневый house-match (S → H → W).
 
-    House-match (встреча Птицы — «соразмерные квартиры»): сортировка не просто
+    **Tier S (same building):** address ILIKE prefix OR ext_house_id match.
-    по расстоянию, а по relevance-скору, где учитывается близость года постройки
+    Если ≥3 результатов → возвращаем только их; tier='S'.
-    и совпадение типа дома. Так аналог «рядом + та же эпоха дома» побеждает
+
-    аналог «чуть ближе, но дом на 30 лет старше».
+    **Tier H (same class):** PostGIS + rooms + area + year ±15 + total_floors ±30%.
+    Если ≥5 результатов → возвращаем; tier='H'.
+    Пропускается если year_built или total_floors неизвестны.
+
+    **Tier W (wide / current):** текущая логика без year/floors WHERE фильтра.
+    tier='W'.
+
+    House-match relevance_score используется для сортировки в Tier H и W.
 
     Стратифицированная выборка (Approach B):
     1. SQL вытягивает до 300 кандидатов с per-address row_number (cap MAX_ANALOGS_PER_ADDRESS).
@@ -860,9 +982,177 @@ def _fetch_analogs(
     4. Итоговый список отсортирован по relevance, LIMIT 50.
 
     Returns:
-        (list_of_listings_as_dicts, fallback_radius_used_flag)
+        (list_of_listings_as_dicts, fallback_radius_used_flag, tier)
+        tier: 'S' | 'H' | 'W'
     """
-    rows = db.execute(
+    area_min = area * (1 - area_tolerance)
+    area_max = area * (1 + area_tolerance)
+    base_params: dict[str, Any] = {
+        "rooms": rooms,
+        "area_min": area_min,
+        "area_max": area_max,
+        "fresh_days": LISTINGS_FRESH_DAYS,
+        "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
+    }
+
+    # ── Tier S: same building ─────────────────────────────────────────────────
+    short_addr = _extract_short_addr(full_address) if full_address else None
+    has_ext_id = ext_house_id is not None
+
+    if short_addr or has_ext_id:
+        addr_clause = ""
+        if short_addr and has_ext_id:
+            addr_clause = (
+                "(address ILIKE :short_addr_prefix"
+                " OR (ext_house_id IS NOT NULL AND ext_house_id = :ext_house_id))"
+            )
+        elif short_addr:
+            addr_clause = "address ILIKE :short_addr_prefix"
+        else:
+            addr_clause = "ext_house_id IS NOT NULL AND ext_house_id = :ext_house_id"
+
+        tier_s_params = {**base_params}
+        if short_addr:
+            tier_s_params["short_addr_prefix"] = short_addr + "%"
+        if has_ext_id:
+            tier_s_params["ext_house_id"] = ext_house_id
+
+        tier_s_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 {addr_clause}
+                    {_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 relevance_score
+                LIMIT 300
+                """
+            ),
+            tier_s_params,
+        ).mappings().all()
+
+        tier_s = [dict(r) for r in tier_s_rows]
+        if len(tier_s) >= 3:
+            logger.info(
+                "analogs tier=S addr_prefix=%r → %d results",
+                short_addr,
+                len(tier_s),
+            )
+            return _stratify_candidates(tier_s), radius_m > DEFAULT_RADIUS_M, "S"
+
+    # ── Tier H: same class (year ±15, total_floors ±30%) ─────────────────────
+    if year_built is not None and total_floors is not None:
+        year_min = year_built - 15
+        year_max = year_built + 15
+        tf_min = math.floor(total_floors * 0.7)
+        tf_max = math.ceil(total_floors * 1.3)
+
+        tier_h_rows = db.execute(
+            text(
+                f"""
+                WITH base AS (
+                    SELECT
+                        {_ANALOG_SELECT_COLS},
+                        ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
+                            AS distance_m,
+                        (
+                            ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
+                                / 1000.0
+                            + CASE
+                                WHEN year_built IS NOT NULL
+                                THEN abs(year_built - CAST(:target_year AS integer)) / 12.0
+                                ELSE 0
+                              END
+                            + CASE
+                                WHEN CAST(:target_house_type AS text) IS NOT NULL
+                                     AND house_type IS NOT NULL
+                                     AND house_type <> CAST(:target_house_type AS text)
+                                THEN 1.5
+                                ELSE 0
+                              END
+                        ) AS relevance_score,
+                        row_number() OVER (
+                            PARTITION BY address
+                            ORDER BY (
+                                ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
+                                    / 1000.0
+                                + CASE
+                                    WHEN year_built IS NOT NULL
+                                    THEN abs(year_built - CAST(:target_year AS integer)) / 12.0
+                                    ELSE 0
+                                  END
+                                + CASE
+                                    WHEN CAST(:target_house_type AS text) IS NOT NULL
+                                         AND house_type IS NOT NULL
+                                         AND house_type <> CAST(:target_house_type AS text)
+                                    THEN 1.5
+                                    ELSE 0
+                                  END
+                            )
+                        ) AS rn_addr
+                    FROM listings
+                    WHERE ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
+                    {_COMMON_WHERE}
+                      AND total_floors BETWEEN CAST(:tf_min AS integer)
+                                           AND CAST(:tf_max AS integer)
+                      AND year_built BETWEEN CAST(:year_min AS integer)
+                                        AND CAST(:year_max AS integer)
+                )
+                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 relevance_score
+                LIMIT 300
+                """
+            ),
+            {
+                **base_params,
+                "lat": lat,
+                "lon": lon,
+                "radius": radius_m,
+                "target_year": year_built,
+                "target_house_type": house_type,
+                "tf_min": tf_min,
+                "tf_max": tf_max,
+                "year_min": year_min,
+                "year_max": year_max,
+            },
+        ).mappings().all()
+
+        tier_h = [dict(r) for r in tier_h_rows]
+        if len(tier_h) >= 5:
+            logger.info(
+                "analogs tier=H year=%d±15 tf=%d-%d → %d results",
+                year_built, tf_min, tf_max, len(tier_h),
+            )
+            return _stratify_candidates(tier_h), radius_m > DEFAULT_RADIUS_M, "H"
+
+        logger.info(
+            "analogs tier=H year=%d±15 tf=%d-%d → only %d (fallthrough to W)",
+            year_built, tf_min, tf_max, len(tier_h),
+        )
+
+    # ── Tier W: wide (current logic, year/floors only in relevance sort) ──────
+    tier_w_rows = db.execute(
         text(
             """
             WITH base AS (
@@ -945,8 +1235,8 @@ def _fetch_analogs(
             "lon": lon,
             "radius": radius_m,
             "rooms": rooms,
-            "area_min": area * (1 - area_tolerance),
+            "area_min": area_min,
-            "area_max": area * (1 + area_tolerance),
+            "area_max": area_max,
             "fresh_days": LISTINGS_FRESH_DAYS,
             "target_year": year_built,
             "target_house_type": house_type,
@@ -954,41 +1244,9 @@ def _fetch_analogs(
         },
     ).mappings().all()
 
-    candidates: list[dict[str, Any]] = [dict(r) for r in rows]
+    candidates: list[dict[str, Any]] = [dict(r) for r in tier_w_rows]
-
+    logger.info("analogs tier=W radius=%dm → %d candidates", radius_m, len(candidates))
-    # Stratified quota: гарантируем MIN_ANALOGS_PER_SOURCE слотов каждому source.
+    return _stratify_candidates(candidates), radius_m > DEFAULT_RADIUS_M, "W"
-    # Candidates уже отсортированы по relevance_score (лучшие первые) из SQL.
-    guaranteed: list[dict[str, Any]] = []
-    guaranteed_ids: set[int] = set()  # по object id, не по внешнему ключу
-    by_source: dict[str, list[dict[str, Any]]] = {}
-    for row in candidates:
-        src = row.get("source") or "unknown"
-        by_source.setdefault(src, []).append(row)
-
-    for _src, src_rows in by_source.items():
-        quota = min(len(src_rows), MIN_ANALOGS_PER_SOURCE)
-        for row in src_rows[:quota]:
-            if id(row) not in guaranteed_ids:
-                guaranteed.append(row)
-                guaranteed_ids.add(id(row))
-
-    # Оставшиеся слоты из candidates, которые ещё не попали в guaranteed.
-    remaining_slots = 50 - len(guaranteed)
-    remainder: list[dict[str, Any]] = []
-    if remaining_slots > 0:
-        for row in candidates:
-            if id(row) not in guaranteed_ids:
-                remainder.append(row)
-                if len(remainder) >= remaining_slots:
-                    break
-
-    result = guaranteed + remainder
-    # Финальная сортировка по relevance (candidates из SQL уже отсортированы,
-    # но guaranteed + remainder смешиваются). relevance_score присутствует в каждом dict.
-    result.sort(key=lambda r: r.get("relevance_score") or 0.0)
-    result = result[:50]
-
-    return result, radius_m > DEFAULT_RADIUS_M
 
 
 def _fetch_deals(