From a0d50bd2829d20605a45cdd41084bc3091b2668d Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 24 May 2026 12:41:12 +0000 Subject: [PATCH] =?UTF-8?q?feat(tradein-estimator):=20tiered=20house-match?= =?UTF-8?q?=20S=E2=86=92H=E2=86=92W=20(#507)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/services/estimator.py | 373 +++++++++++++++--- .../backend/tests/test_extract_short_addr.py | 49 +++ 2 files changed, 367 insertions(+), 55 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_extract_short_addr.py diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 580593d0..90b43832 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/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, diff --git a/tradein-mvp/backend/tests/test_extract_short_addr.py b/tradein-mvp/backend/tests/test_extract_short_addr.py new file mode 100644 index 00000000..eb8b2b58 --- /dev/null +++ b/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