diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 4e2fc03e..c840cc4d 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) @@ -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]: - """SELECT аналогов с PostGIS distance + house-match relevance. + total_floors: int | None = None, + ext_house_id: str | None = None, +) -> tuple[list[dict[str, Any]], bool, str]: + """SELECT аналогов — трёхуровневый house-match (S → H → W). - House-match (встреча Птицы — «соразмерные квартиры»): сортировка не просто - по расстоянию, а по relevance-скору, где учитывается близость года постройки - и совпадение типа дома. Так аналог «рядом + та же эпоха дома» побеждает - аналог «чуть ближе, но дом на 30 лет старше». + **Tier S (same building):** address ILIKE prefix OR ext_house_id match. + Если ≥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 +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_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 +1244,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(