feat(tradein-estimator): tiered house-match S→H→W (#507)

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:
-    #    a) 1km + ±15% area
-    #    b) 2km + ±15% area  (fallback_used = True)
-    #    c) 2km + ±25% area  (fallback_used = True, area_widened = True)
-    listings, fallback_used = _fetch_analogs(
+    # 3. House-match: S → H → W tiered lookup (see _fetch_analogs docstring).
+    #    Radius fallback still applies when W tier has < 5 results at 1km.
+    listings, fallback_used, analog_tier = _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)
@@ -556,7 +563,7 @@ async def estimate_quality(
         range_low = int(range_low * repair_coef)
         range_high = int(range_high * repair_coef)
         median_ppm2 = median_ppm2 * repair_coef
-        pct = int(round((repair_coef - 1.0) * 100))
+        pct = round((repair_coef - 1.0) * 100)
         repair_note = (
             f" Цена скорректирована на состояние ремонта "
             f"({_REPAIR_LABEL.get(payload.repair_state, '')} {pct:+d}%)."
@@ -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,137 @@ def _compute_freshness_minutes(lots: list[dict[str, Any]]) -> int | None:
 
 
 # ── Internals ────────────────────────────────────────────────────────────────
+
+# Compiled regexes for _extract_short_addr — module-level for performance.
+
+# Strips leading admin prefixes: «Россия», «Свердловская область», «г. Екатеринбург» etc.
+_ADMIN_PREFIX_RE = re.compile(
+    r"^(?:"
+    r"\s*(?:Россия|РФ|Российская\s+Федерация)\s*,?\s*|"
+    r"\s*[А-Яа-яёЁ][А-Яа-яёЁ\s-]+(?:\s+(?:обл(?:асть)?|р-н|район|округ|край|республика))\.?\s*,?\s*|"
+    r"\s*(?:г(?:ород)?|гор)\.?\s*[А-Яа-яёЁ][А-Яа-яёЁ\s-]+\s*,?\s*|"
+    r"\s*[А-Я][а-яё-]+(?:\s+[А-Я][а-яё-]+)?\s*,?\s*"
+    r")+",
+    flags=re.UNICODE,
+)
+
+# Recognizes start of a street keyword.
+_STREET_START_RE = re.compile(
+    r"(?:ул\.|улица|пр\.|пр-т|проспект|пер\.|переулок|"
+    r"б-р|бульвар|ш\.|шоссе|наб\.|набережная|проезд|тракт|пл\.|площадь|"
+    r"мкр\.?|микрорайон)\s+",
+    flags=re.IGNORECASE | re.UNICODE,
+)
+
+# Drops trailing apartment / office / corpus noise from the end.
+_TRAILING_NOISE_RE = re.compile(
+    r"\s*,\s*(?:кв\.?\s*\d+|корп\.?\s*\w+|оф\.?\s*\d+|пом\.?\s*\d+|подъезд\s*\d+).*$",
+    flags=re.IGNORECASE | re.UNICODE,
+)
+
+
+def _extract_short_addr(full_address: str | None) -> str | None:
+    """Извлекает «улица + номер дома» из полного адреса для поиска в том же доме.
+
+    Примеры:
+      "Свердловская область, г. Екатеринбург, ул. Заводская, д. 44-а" → "ул. Заводская, д. 44-а"
+      "Россия, Екатеринбург, ул. Малышева, 1"                          → "ул. Малышева, 1"
+      "РФ, Свердловская обл., Екатеринбург, ул. Ленина, 5, кв. 12"    → "ул. Ленина, 5"
+      "г. Екатеринбург, проспект Ленина, 50"                           → "проспект Ленина, 50"
+      "Екатеринбург, ул. Крауля, 48/2"                                 → "ул. Крауля, 48/2"
+
+    Алгоритм:
+    1. Отрезаем trailing кв./корп./оф. noise.
+    2. Ищем первый street-keyword токен (ул./пр./пер. и т.д.) — возвращаем с него.
+    3. Fallback: агрессивно strip admin-prefix regex, вернуть остаток.
+    4. None если строка пустая или нечего возвращать.
+    """
+    if not full_address:
+        return None
+    s = full_address.strip()
+    s = _TRAILING_NOISE_RE.sub("", s)
+
+    # Find first street-keyword position and return from there.
+    m = _STREET_START_RE.search(s)
+    if m:
+        return s[m.start():].strip(" ,.")
+
+    # Fallback: strip known admin prefixes, return whatever remains.
+    s = _ADMIN_PREFIX_RE.sub("", s)
+    return s.strip(" ,.") or 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]:
-    """SELECT аналогов с PostGIS distance + house-match relevance.
+    total_floors: int | None = None,
+    # TODO: когда listings получит колонку house_id_fk — добавить ext_house_id JOIN для Tier S.
+) -> tuple[list[dict[str, Any]], bool, str]:
+    """SELECT аналогов — трёхуровневый house-match (S → H → W).
 
-    House-match (встреча Птицы — «соразмерные квартиры»): сортировка не просто
-    по расстоянию, а по relevance-скору, где учитывается близость года постройки
-    и совпадение типа дома. Так аналог «рядом + та же эпоха дома» побеждает
-    аналог «чуть ближе, но дом на 30 лет старше».
+    **Tier S (same building):** address ILIKE prefix-match по short_addr.
+    Если ≥3 результатов → возвращаем только их; tier='S'.
+
+    **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 +1000,164 @@ 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 short_addr:
+        tier_s_params = {
+            **base_params,
+            "short_addr_prefix": short_addr + "%",
+        }
+
+        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 address ILIKE :short_addr_prefix
+                    {_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_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 +1240,8 @@ def _fetch_analogs(
             "lon": lon,
             "radius": radius_m,
             "rooms": rooms,
-            "area_min": area * (1 - area_tolerance),
-            "area_max": area * (1 + area_tolerance),
+            "area_min": area_min,
+            "area_max": area_max,
             "fresh_days": LISTINGS_FRESH_DAYS,
             "target_year": year_built,
             "target_house_type": house_type,
@@ -954,41 +1249,9 @@ def _fetch_analogs(
         },
     ).mappings().all()
 
-    candidates: list[dict[str, Any]] = [dict(r) for r in rows]
-
-    # Stratified quota: гарантируем MIN_ANALOGS_PER_SOURCE слотов каждому source.
-    # 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
+    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))
+    return _stratify_candidates(candidates), radius_m > DEFAULT_RADIUS_M, "W"
 
 
 def _fetch_deals(
@@ -1145,7 +1408,7 @@ def _listing_to_analog(row: dict[str, Any]) -> AnalogLot:
         price_per_m2=int(row.get("price_per_m2") or 0),
         listing_date=row.get("listing_date"),
         days_on_market=row.get("days_on_market"),
-        photo_url=(row.get("photo_urls") or [None])[0] if isinstance(row.get("photo_urls"), list) else None,
+        photo_url=(row["photo_urls"] or [None])[0] if row.get("photo_urls") else None,
         source=row.get("source"),
         source_url=row.get("source_url"),
         distance_m=int(row["distance_m"]) if row.get("distance_m") is not None else None,

tradein-mvp/backend/tests/test_extract_short_addr.py

@@ -0,0 +1,49 @@
+"""Unit tests for _extract_short_addr in app.services.estimator.
+
+Covers real Ekaterinburg address formats: full admin chain, Россия/РФ prefix,
+apartment stripped, prospect keyword, korpus stripped, non-street fallback, None passthrough.
+"""
+
+from app.services.estimator import _extract_short_addr
+
+
+def test_full_admin_chain() -> None:
+    assert _extract_short_addr(
+        "Свердловская область, г. Екатеринбург, Склад, ул. Заводская, д. 44-а"
+    ) == "ул. Заводская, д. 44-а"
+
+
+def test_russia_prefix() -> None:
+    assert _extract_short_addr(
+        "Россия, Екатеринбург, ул. Малышева, 1"
+    ) == "ул. Малышева, 1"
+
+
+def test_apt_stripped() -> None:
+    out = _extract_short_addr("Екатеринбург, ул. Ленина, 5, кв. 12")
+    assert out is not None
+    assert "кв" not in out
+    assert out.startswith("ул. Ленина")
+
+
+def test_prospekt_keyword() -> None:
+    out = _extract_short_addr("г. Екатеринбург, проспект Ленина, 50")
+    assert out is not None
+    assert out.startswith("проспект Ленина")
+
+
+def test_korp_stripped() -> None:
+    out = _extract_short_addr("Екатеринбург, ул. Крауля, 48, корп. 2")
+    assert out is not None
+    assert "корп" not in out.lower()
+
+
+def test_no_street_keyword_returns_fallback() -> None:
+    # Should not crash; either returns stripped str or None.
+    out = _extract_short_addr("Industrial zone X")
+    assert out is None or "Industrial" in out
+
+
+def test_none_passthrough() -> None:
+    assert _extract_short_addr(None) is None
+    assert _extract_short_addr("") is None