feat(yandex-valuation): link history rows to houses via match_or_create_house (#531)

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

@@ -39,6 +39,7 @@ from app.schemas.trade_in import (
 )
 from app.services.geocoder import GeocodeResult, geocode
 from app.services.house_metadata import get_house_metadata
+from app.services.matching.houses import match_or_create_house
 from app.services.scrapers.avito_imv import (
     IMVAddressNotFoundError,
     IMVAuthError,
@@ -61,24 +62,28 @@ logger = logging.getLogger(__name__)
 
 
 # ── Constants ────────────────────────────────────────────────────────────────
-DEFAULT_RADIUS_M = 1000   # ПО ВСТРЕЧЕ ПТИЦЫ: «локация не дальше 800-1000 м»
+DEFAULT_RADIUS_M = 1000  # ПО ВСТРЕЧЕ ПТИЦЫ: «локация не дальше 800-1000 м»
 FALLBACK_RADIUS_M = 2000
-AREA_TOLERANCE = 0.15        # ±15% площади
+AREA_TOLERANCE = 0.15  # ±15% площади
 MAX_ANALOGS_PER_ADDRESS = 5  # анти-bias: не больше 5 лотов с одного адреса
-MIN_ANALOGS_PER_SOURCE = 5   # гарантированный минимум на live source
-LISTINGS_FRESH_DAYS = 14     # объявления не старше 14 дней
-DEALS_PERIOD_MONTHS = 12     # сделки за последний год
+MIN_ANALOGS_PER_SOURCE = 5  # гарантированный минимум на live source
+LISTINGS_FRESH_DAYS = 14  # объявления не старше 14 дней
+DEALS_PERIOD_MONTHS = 12  # сделки за последний год
 
 # Когорта по году постройки — типизация массовой застройки РФ.
 # Используется как hard-filter в Tier 0 _fetch_analogs (PR 9, 2026-05-24).
 # Если target_year не задан — cohort = None → фильтр отключён, Tier 0 пропускается.
 COHORTS = (
     # (cohort_name, year_min_inclusive, year_max_inclusive)
-    ('khrushchev',  1955, 1969),  # Хрущёвки 5-эт
-    ('brezhnev',    1970, 1989),  # Брежневка кирпич/панель 9–12-эт
-    ('late_soviet', 1990, 1999),  # Поздний СССР (no overlap; first-match would never pick old range)
-    ('2000s',       2000, 2010),  # Ранние новостройки
-    ('modern',      2011, 2100),  # Современные ЖК
+    ("khrushchev", 1955, 1969),  # Хрущёвки 5-эт
+    ("brezhnev", 1970, 1989),  # Брежневка кирпич/панель 9–12-эт
+    (
+        "late_soviet",
+        1990,
+        1999,
+    ),  # Поздний СССР (no overlap; first-match would never pick old range)
+    ("2000s", 2000, 2010),  # Ранние новостройки
+    ("modern", 2011, 2100),  # Современные ЖК
 )
 # Минимум аналогов чтобы остаться на Tier 0 (с cohort); ниже — fallback на Tier A.
 MIN_ANALOGS_TIER_0 = 5
@@ -122,16 +127,16 @@ _IMV_REPAIR_MAP: dict[str | None, str | None] = {
 }
 
 _REPAIR_COEF: dict[str, float] = {
-    "needs_repair": 0.92,   # требует ремонта — ниже рынка
-    "standard":     0.98,
-    "good":         1.03,
-    "excellent":    1.08,   # евроремонт — выше рынка
+    "needs_repair": 0.92,  # требует ремонта — ниже рынка
+    "standard": 0.98,
+    "good": 1.03,
+    "excellent": 1.08,  # евроремонт — выше рынка
 }
 _REPAIR_LABEL: dict[str | None, str] = {
     "needs_repair": "требует ремонта",
-    "standard":     "стандартный ремонт",
-    "good":         "хороший ремонт",
-    "excellent":    "евроремонт",
+    "standard": "стандартный ремонт",
+    "good": "хороший ремонт",
+    "excellent": "евроремонт",
 }
 
 
@@ -182,13 +187,21 @@ async def _get_or_fetch_imv_cached(
     """
     try:
         cache_key = compute_imv_cache_key(
-            address, rooms, area_m2, floor, floor_at_home,
-            house_type, renovation_type, has_balcony, has_loggia,
+            address,
+            rooms,
+            area_m2,
+            floor,
+            floor_at_home,
+            house_type,
+            renovation_type,
+            has_balcony,
+            has_loggia,
         )
 
-        existing = db.execute(
-            text(
-                """
+        existing = (
+            db.execute(
+                text(
+                    """
                 SELECT id, cache_key, address, rooms, area_m2, floor, floor_at_home,
                        house_type, renovation_type, has_balcony, has_loggia,
                        lat, lon, geo_hash, avito_address_id, avito_location_id,
@@ -201,16 +214,21 @@ async def _get_or_fetch_imv_cached(
                 ORDER BY fetched_at DESC
                 LIMIT 1
                 """
-            ),
-            {"ck": cache_key, "ttl_hours": IMV_CACHE_TTL_HOURS},
-        ).mappings().first()
+                ),
+                {"ck": cache_key, "ttl_hours": IMV_CACHE_TTL_HOURS},
+            )
+            .mappings()
+            .first()
+        )
 
         if existing is not None:
             logger.info(
                 "imv: cache HIT key=%s recommended=%d",
-                cache_key[:8], existing["recommended_price"],
+                cache_key[:8],
+                existing["recommended_price"],
             )
             from app.services.scrapers.avito_imv import IMVGeo
+
             return IMVEvaluation(
                 cache_key=existing["cache_key"],
                 address=existing["address"],
@@ -241,15 +259,22 @@ async def _get_or_fetch_imv_cached(
         # Cache miss — fresh fetch
         logger.info("imv: cache MISS key=%s — fetching fresh", cache_key[:8])
         result = await evaluate_via_imv(
-            address=address, rooms=rooms, area_m2=area_m2,
-            floor=floor, floor_at_home=floor_at_home,
-            house_type=house_type, renovation_type=renovation_type,
-            has_balcony=has_balcony, has_loggia=has_loggia,
+            address=address,
+            rooms=rooms,
+            area_m2=area_m2,
+            floor=floor,
+            floor_at_home=floor_at_home,
+            house_type=house_type,
+            renovation_type=renovation_type,
+            has_balcony=has_balcony,
+            has_loggia=has_loggia,
         )
         save_imv_evaluation(db, result, estimate_id=estimate_id_for_link)
         logger.info(
             "imv: fresh recommended=%d range=(%d, %d) count=%d",
-            result.recommended_price, result.lower_price, result.higher_price,
+            result.recommended_price,
+            result.lower_price,
+            result.higher_price,
             result.market_count or 0,
         )
         return result
@@ -292,7 +317,8 @@ async def _get_or_fetch_imv_cached(
         return None
     except IMVAuthError as e:
         logger.error(
-            "imv: auth/quota error — manual action required: %s", e,
+            "imv: auth/quota error — manual action required: %s",
+            e,
         )
         return None
     except IMVTransientError as e:
@@ -305,9 +331,8 @@ async def _get_or_fetch_imv_cached(
 
 # ── Yandex Valuation cache lookup (Stage 8) ─────────────────────────────────
 
-def _yandex_valuation_cache_key(
-    address: str, offer_category: str, offer_type: str
-) -> str:
+
+def _yandex_valuation_cache_key(address: str, offer_category: str, offer_type: str) -> str:
     """SHA256 cache key for Yandex Valuation lookups."""
     payload = f"{address}|{offer_category}|{offer_type}"
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()
@@ -329,9 +354,10 @@ async def _get_or_fetch_yandex_valuation_cached(
 
     # Cache lookup
     try:
-        cached = db.execute(
-            text(
-                """
+        cached = (
+            db.execute(
+                text(
+                    """
                 SELECT raw_payload, fetched_at
                 FROM external_valuations
                 WHERE source = 'yandex_valuation'
@@ -340,9 +366,12 @@ async def _get_or_fetch_yandex_valuation_cached(
                 ORDER BY fetched_at DESC
                 LIMIT 1
                 """
-            ),
-            {"ck": cache_key},
-        ).mappings().first()
+                ),
+                {"ck": cache_key},
+            )
+            .mappings()
+            .first()
+        )
     except Exception as e:
         logger.warning("yandex_valuation: cache lookup failed: %s", e)
         cached = None
@@ -372,9 +401,7 @@ async def _get_or_fetch_yandex_valuation_cached(
                 offer_type=offer_type,
             )
     except Exception as e:
-        logger.warning(
-            "yandex_valuation: fetch failed — estimator продолжает без Yandex: %s", e
-        )
+        logger.warning("yandex_valuation: fetch failed — estimator продолжает без Yandex: %s", e)
         return None
 
     if result is None:
@@ -427,53 +454,102 @@ def _save_yandex_history_items(
 ) -> int:
     """Persist history items to house_placement_history. Returns saved count.
 
-    house_id stays NULL — estimator doesn't compute target_house_id yet.
-    Idempotent via UNIQUE (source, ext_item_id); we synthesize ext_item_id from
-    (address|date|area|floor) hash since Yandex history items don't carry an
-    explicit ID.
+    Resolves house_id ONCE per result via match_or_create_house() using the
+    valuation page's address + meta (year_built/total_floors). All items from
+    the same page share that house_id.
 
-    Batch semantics (closes finding #5 from 2026-05-24 audit): all items
-    inserted under a single try/except; on any failure the whole batch is
-    rolled back as a unit (prior per-item rollback() destroyed the parent tx).
+    Confidence pipeline:
+      method_confidence  <- match_or_create_house (1.0 cadastr/source, 0.9 fp, 0.7 geo, 1.0 new)
+      final_confidence   = method_confidence
+
+    Idempotent via UNIQUE (source, ext_item_id); ext_item_id synthesized from
+    (address|publish_date|area|floor|prices) hash.
+
+    Batch semantics: single try/except; on any failure the batch rolls back.
     """
     if not result.history_items:
         return 0
 
+    # Resolve house ONCE per page. Synthetic ext_id = sha256(address)[:16]
+    # — stable across re-runs, distinguishes pages for different addresses.
+    address_seed = (result.address or "").strip().lower()
+    house_ext_id = (
+        hashlib.sha256(address_seed.encode("utf-8")).hexdigest()[:16] if address_seed else "unknown"
+    )
+
+    try:
+        house_id, method_confidence, method = match_or_create_house(
+            db,
+            ext_source="yandex_valuation",
+            ext_id=house_ext_id,
+            address=result.address,
+            year_built=result.house.year_built,
+        )
+    except Exception as e:
+        logger.warning(
+            "yandex_valuation: house resolution failed for address=%r: %s"
+            " — saving with house_id=NULL",
+            result.address,
+            e,
+        )
+        db.rollback()
+        house_id = None
+        method_confidence = 0.0
+        method = "fail"
+
+    logger.info(
+        "yandex_valuation: house resolved house_id=%s method=%s confidence=%.2f addr=%r",
+        house_id,
+        method,
+        method_confidence,
+        result.address,
+    )
+
     rows = []
     for item in result.history_items:
-        # Synthesize stable ext_item_id (no native ID in valuation page)
         ext_seed = (
             f"{result.address}|{item.publish_date}|{item.area_m2}|{item.floor}|"
             f"{item.start_price}|{item.last_price}"
         )
         ext_item_id = hashlib.sha256(ext_seed.encode("utf-8")).hexdigest()[:32]
-        rows.append({
-            "ext_id": ext_item_id,
-            "rooms": item.rooms,
-            "area": item.area_m2,
-            "floor": item.floor,
-            "start_price": item.start_price,
-            "last_price": item.last_price,
-            "publish_date": item.publish_date,
-            "exposure": item.exposure_days,
-            "raw": json.dumps(item.model_dump(mode="json"), ensure_ascii=False),
-        })
+        rows.append(
+            {
+                "ext_id": ext_item_id,
+                "house_id": house_id,
+                "rooms": item.rooms,
+                "area": item.area_m2,
+                "floor": item.floor,
+                "total_floors": result.house.total_floors,
+                "start_price": item.start_price,
+                "last_price": item.last_price,
+                "publish_date": item.publish_date,
+                "removed_date": item.removed_date,
+                "exposure": item.exposure_days,
+                "confidence": float(method_confidence),
+                "notes": f"match_method={method}" if method != "fail" else None,
+                "raw": json.dumps(item.model_dump(mode="json"), ensure_ascii=False),
+            }
+        )
 
     sql = text(
         """
         INSERT INTO house_placement_history (
-            source, ext_item_id,
-            rooms, area_m2, floor,
+            source, ext_item_id, house_id,
+            rooms, area_m2, floor, total_floors,
             start_price, start_price_date,
             last_price, last_price_date,
+            removed_date,
             exposure_days,
+            source_confidence, notes,
             raw_payload
         ) VALUES (
-            'yandex_valuation', :ext_id,
-            :rooms, :area, :floor,
+            'yandex_valuation', :ext_id, :house_id,
+            :rooms, :area, :floor, :total_floors,
             :start_price, :publish_date,
             :last_price, :publish_date,
+            :removed_date,
             :exposure,
+            :confidence, :notes,
             CAST(:raw AS jsonb)
         )
         ON CONFLICT (source, ext_item_id) DO NOTHING
@@ -488,16 +564,15 @@ def _save_yandex_history_items(
     except Exception as e:
         logger.warning(
             "yandex_valuation: failed to save history batch (%d items): %s",
-            len(rows), e,
+            len(rows),
+            e,
         )
         db.rollback()
         return 0
 
 
 # ── Public ───────────────────────────────────────────────────────────────────
-async def estimate_quality(
-    payload: TradeInEstimateInput, db: Session
-) -> AggregatedEstimate:
+async def estimate_quality(payload: TradeInEstimateInput, db: Session) -> AggregatedEstimate:
     """Главная функция — оценка квартиры по реальным данным.
 
     NOTE 2026-05-24: rosreestr_deals temporarily NOT included in actual_deals
@@ -540,16 +615,22 @@ async def estimate_quality(
     if cohort_range is not None:
         cy_min, cy_max = cohort_range
         listings_tier0, _, analog_tier = _fetch_analogs(
-            db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
+            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,
+            year_built=target_year,
+            house_type=target_house_type,
             total_floors=payload.total_floors,
-            cohort_year_min=cy_min, cohort_year_max=cy_max,
+            cohort_year_min=cy_min,
+            cohort_year_max=cy_max,
         )
     else:
         listings_tier0 = []
-        analog_tier = 'W'
+        analog_tier = "W"
 
     if len(listings_tier0) >= MIN_ANALOGS_TIER_0:
         listings = listings_tier0
@@ -557,20 +638,30 @@ async def estimate_quality(
     else:
         # Tier 0 пуст/мал — graceful fallback на Tier A без cohort
         listings, fallback_used, analog_tier = _fetch_analogs(
-            db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
+            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,
+            year_built=target_year,
+            house_type=target_house_type,
             total_floors=payload.total_floors,
         )
     area_widened = False
 
     if len(listings) < 5:
         listings_wide, _, analog_tier_wide = _fetch_analogs(
-            db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
+            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,
+            year_built=target_year,
+            house_type=target_house_type,
             total_floors=payload.total_floors,
         )
         if len(listings_wide) > len(listings):
@@ -582,10 +673,16 @@ async def estimate_quality(
     # (актуально для отдалённых районов / новостроек с нестандартной планировкой)
     if len(listings) < 3:
         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,
+            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,
+            year_built=target_year,
+            house_type=target_house_type,
             total_floors=payload.total_floors,
         )
         if len(listings_widearea) > len(listings):
@@ -630,8 +727,12 @@ async def estimate_quality(
         )
 
     confidence, explanation = _compute_confidence(
-        n_analogs, median_ppm2, q1_ppm2 if listings_clean else 0,
-        q3_ppm2 if listings_clean else 0, fallback_used, area_widened,
+        n_analogs,
+        median_ppm2,
+        q1_ppm2 if listings_clean else 0,
+        q3_ppm2 if listings_clean else 0,
+        fallback_used,
+        area_widened,
         listings=listings_clean,
     )
 
@@ -687,7 +788,8 @@ async def estimate_quality(
     yandex_val: YandexValuationResult | None = None
     if geo is not None and geo.full_address:
         yandex_val = await _get_or_fetch_yandex_valuation_cached(
-            db, address=geo.full_address,
+            db,
+            address=geo.full_address,
         )
         if yandex_val is not None:
             sources_used_pre = sorted(set(sources_used_pre) | {"yandex_valuation"})
@@ -695,7 +797,8 @@ async def estimate_quality(
             logger.info(
                 "yandex_valuation: history items processed=%d saved=%d"
                 " (house_id=NULL — matching deferred)",
-                len(yandex_val.history_items), saved_hist,
+                len(yandex_val.history_items),
+                saved_hist,
             )
 
     # ── Stage 9: Cian Valuation as 7th source (on-demand, 24h cached, graceful if no cookies) ──
@@ -985,7 +1088,7 @@ def _extract_short_addr(full_address: str | None) -> str | None:
     # Find first street-keyword position and return from there.
     m = _STREET_START_RE.search(s)
     if m:
-        return s[m.start():].strip(" ,.")
+        return s[m.start() :].strip(" ,.")
 
     # Fallback: strip known admin prefixes, return whatever remains.
     s = _ADMIN_PREFIX_RE.sub("", s)
@@ -1056,13 +1159,20 @@ _COMMON_WHERE = """
 
 
 def _fetch_analogs(
-    db: Session, *, lat: float, lon: float, rooms: int, area: float, radius_m: int,
+    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,
+    year_built: int | None = None,
+    house_type: str | None = None,
     total_floors: int | None = None,
-    cohort_year_min: int | None = None,   # NEW: lower bound year_built inclusive
-    cohort_year_max: int | None = None,   # NEW: upper bound year_built inclusive
+    cohort_year_min: int | None = None,  # NEW: lower bound year_built inclusive
+    cohort_year_max: int | None = None,  # NEW: upper bound year_built inclusive
     # TODO: когда listings получит колонку house_id_fk — добавить ext_house_id JOIN для Tier S.
 ) -> tuple[list[dict[str, Any]], bool, str]:
     """SELECT аналогов — трёхуровневый house-match (S → H → W).
@@ -1115,9 +1225,10 @@ def _fetch_analogs(
             "short_addr_prefix": short_addr + "%",
         }
 
-        tier_s_rows = db.execute(
-            text(
-                f"""
+        tier_s_rows = (
+            db.execute(
+                text(
+                    f"""
                 WITH base AS (
                     SELECT
                         {_ANALOG_SELECT_COLS},
@@ -1139,9 +1250,12 @@ def _fetch_analogs(
                 ORDER BY scraped_at DESC
                 LIMIT 300
                 """
-            ),
-            tier_s_params,
-        ).mappings().all()
+                ),
+                tier_s_params,
+            )
+            .mappings()
+            .all()
+        )
 
         tier_s = [dict(r) for r in tier_s_rows]
         if len(tier_s) >= 3:
@@ -1159,9 +1273,10 @@ def _fetch_analogs(
         tf_min = math.floor(total_floors * 0.7)
         tf_max = math.ceil(total_floors * 1.3)
 
-        tier_h_rows = db.execute(
-            text(
-                f"""
+        tier_h_rows = (
+            db.execute(
+                text(
+                    f"""
                 WITH base AS (
                     SELECT
                         {_ANALOG_SELECT_COLS},
@@ -1221,38 +1336,48 @@ def _fetch_analogs(
                 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()
+                ),
+                {
+                    **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),
+                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),
+            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(
-            """
+    tier_w_rows = (
+        db.execute(
+            text(
+                """
             WITH base AS (
                 SELECT
                     source, source_url, address, lat, lon,
@@ -1337,22 +1462,25 @@ def _fetch_analogs(
             ORDER BY relevance_score
             LIMIT 300
             """
-        ),
-        {
-            "lat": lat,
-            "lon": lon,
-            "radius": radius_m,
-            "rooms": rooms,
-            "area_min": area_min,
-            "area_max": area_max,
-            "fresh_days": LISTINGS_FRESH_DAYS,
-            "target_year": year_built,
-            "target_house_type": house_type,
-            "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
-            "cohort_year_min": cohort_year_min,   # NEW
-            "cohort_year_max": cohort_year_max,   # NEW
-        },
-    ).mappings().all()
+            ),
+            {
+                "lat": lat,
+                "lon": lon,
+                "radius": radius_m,
+                "rooms": rooms,
+                "area_min": area_min,
+                "area_max": area_max,
+                "fresh_days": LISTINGS_FRESH_DAYS,
+                "target_year": year_built,
+                "target_house_type": house_type,
+                "max_per_addr": MAX_ANALOGS_PER_ADDRESS,
+                "cohort_year_min": cohort_year_min,  # NEW
+                "cohort_year_max": cohort_year_max,  # NEW
+            },
+        )
+        .mappings()
+        .all()
+    )
 
     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))
@@ -1362,9 +1490,10 @@ def _fetch_analogs(
 def _fetch_deals(
     db: Session, *, lat: float, lon: float, rooms: int, area: float, radius_m: int
 ) -> list[dict[str, Any]]:
-    rows = db.execute(
-        text(
-            """
+    rows = (
+        db.execute(
+            text(
+                """
             SELECT
                 source, address, lat, lon,
                 rooms, area_m2, floor, total_floors,
@@ -1380,17 +1509,20 @@ def _fetch_deals(
             ORDER BY deal_date DESC
             LIMIT 30
             """
-        ),
-        {
-            "lat": lat,
-            "lon": lon,
-            "radius": radius_m,
-            "rooms": rooms,
-            "area_min": area * (1 - AREA_TOLERANCE),
-            "area_max": area * (1 + AREA_TOLERANCE),
-            "months": DEALS_PERIOD_MONTHS,
-        },
-    ).mappings().all()
+            ),
+            {
+                "lat": lat,
+                "lon": lon,
+                "radius": radius_m,
+                "rooms": rooms,
+                "area_min": area * (1 - AREA_TOLERANCE),
+                "area_max": area * (1 + AREA_TOLERANCE),
+                "months": DEALS_PERIOD_MONTHS,
+            },
+        )
+        .mappings()
+        .all()
+    )
 
     return [dict(r) for r in rows]
 

tradein-mvp/backend/tests/test_estimator_yandex_integration.py

@@ -1,4 +1,5 @@
 """Tests for Yandex Valuation integration in estimator.py."""
+
 import os
 
 # Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
@@ -151,9 +152,13 @@ async def test_fetch_returns_none_propagates_none():
 def test_save_history_items_inserts_each():
     db = MagicMock()
     result = _sample_result()
-    saved = _save_yandex_history_items(db, result)
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        return_value=(1, 0.9, "fingerprint"),
+    ):
+        saved = _save_yandex_history_items(db, result)
     assert saved == 2
-    # 2 INSERTs + 1 commit
+    # 2 INSERTs (one per item) + 1 commit
     assert db.execute.call_count == 2
     db.commit.assert_called_once()
 
@@ -169,8 +174,11 @@ def test_save_history_items_empty_no_commit():
         house=ValuationHouseMeta(),
         history_items=[],
     )
-    saved = _save_yandex_history_items(db, result)
+    # match_or_create_house must NOT be called when there are no items (early return)
+    with patch("app.services.estimator.match_or_create_house") as m:
+        saved = _save_yandex_history_items(db, result)
     assert saved == 0
+    m.assert_not_called()
     db.execute.assert_not_called()
     db.commit.assert_not_called()
 
@@ -180,8 +188,12 @@ def test_save_history_items_ext_id_stable_across_calls():
     db1 = MagicMock()
     db2 = MagicMock()
     result = _sample_result()
-    _save_yandex_history_items(db1, result)
-    _save_yandex_history_items(db2, result)
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        return_value=(1, 0.9, "fingerprint"),
+    ):
+        _save_yandex_history_items(db1, result)
+        _save_yandex_history_items(db2, result)
     # Both got same call params (ext_id derived from same seed)
     ext_ids_1 = [c.args[1]["ext_id"] for c in db1.execute.call_args_list]
     ext_ids_2 = [c.args[1]["ext_id"] for c in db2.execute.call_args_list]
@@ -193,7 +205,11 @@ def test_save_history_items_db_error_rolls_back_batch():
     db = MagicMock()
     db.execute.side_effect = [RuntimeError("first row fails"), None]
     result = _sample_result()
-    saved = _save_yandex_history_items(db, result)
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        return_value=(1, 0.9, "fingerprint"),
+    ):
+        saved = _save_yandex_history_items(db, result)
     assert saved == 0  # whole batch rolled back
     db.rollback.assert_called_once()
     db.commit.assert_not_called()

tradein-mvp/backend/tests/test_yandex_valuation_save.py

@@ -0,0 +1,194 @@
+"""Tests for _save_yandex_history_items — house_id linking + new columns."""
+
+from __future__ import annotations
+
+import os
+
+# Settings requires DATABASE_URL at init time. Set dummy DSN before any app import.
+os.environ.setdefault("DATABASE_URL", "postgresql://test:test@localhost/test_db")
+
+from datetime import date
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.services.estimator import _save_yandex_history_items
+from app.services.scrapers.yandex_valuation import (
+    ValuationHistoryItem,
+    ValuationHouseMeta,
+    YandexValuationResult,
+)
+
+
+def _make_result(items: list[ValuationHistoryItem]) -> YandexValuationResult:
+    return YandexValuationResult(
+        address="Россия, Свердловская область, Екатеринбург, улица Куйбышева, 106",
+        offer_category="APARTMENT",
+        offer_type="SELL",
+        page=1,
+        source_url=(
+            "https://realty.yandex.ru/otsenka-kvartiry-po-adresu-onlayn/?address=test&page=1"
+        ),
+        house=ValuationHouseMeta(year_built=2005, total_floors=25, has_lift=True),
+        history_items=items,
+    )
+
+
+def _two_items() -> list[ValuationHistoryItem]:
+    return [
+        ValuationHistoryItem(
+            area_m2=50.0,
+            rooms=2,
+            floor=3,
+            start_price=9_000_000,
+            last_price=9_000_000,
+            publish_date=date(2024, 5, 10),
+            removed_date=None,
+            exposure_days=30,
+            status="В продаже",
+        ),
+        ValuationHistoryItem(
+            area_m2=55.0,
+            rooms=2,
+            floor=5,
+            start_price=10_000_000,
+            last_price=9_500_000,
+            publish_date=date(2024, 3, 1),
+            removed_date=date(2024, 6, 1),
+            exposure_days=92,
+            status="Снято",
+        ),
+    ]
+
+
+def test_save_resolves_house_once_per_page():
+    """match_or_create_house must be called exactly once per page (not per item)."""
+    db = MagicMock()
+    result = _make_result(_two_items())
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        return_value=(12345, 0.9, "fingerprint"),
+    ) as m:
+        saved = _save_yandex_history_items(db, result)
+    assert m.call_count == 1, "match_or_create_house must be called once per page"
+    assert saved == 2
+
+
+def test_save_row_contains_house_id_and_confidence():
+    """Each INSERT row dict carries house_id and source_confidence."""
+    db = MagicMock()
+    result = _make_result(_two_items())
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        return_value=(54321, 0.9, "fingerprint"),
+    ):
+        _save_yandex_history_items(db, result)
+
+    # db.execute called twice (one per item)
+    assert db.execute.call_count == 2
+    for c in db.execute.call_args_list:
+        row = c.args[1]
+        assert row["house_id"] == 54321
+        assert row["confidence"] == pytest.approx(0.9)
+        assert row["notes"] == "match_method=fingerprint"
+        assert row["total_floors"] == 25
+
+
+def test_save_row_contains_removed_date():
+    """Item with removed_date set → row dict has removed_date populated."""
+    db = MagicMock()
+    items = [
+        ValuationHistoryItem(
+            area_m2=60.0,
+            rooms=2,
+            floor=4,
+            start_price=11_000_000,
+            last_price=10_500_000,
+            publish_date=date(2024, 1, 15),
+            removed_date=date(2024, 5, 20),
+            exposure_days=125,
+            status="Снято",
+        ),
+    ]
+    result = _make_result(items)
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        return_value=(99, 1.0, "new"),
+    ):
+        _save_yandex_history_items(db, result)
+
+    row = db.execute.call_args_list[0].args[1]
+    assert row["removed_date"] == date(2024, 5, 20)
+
+
+def test_save_row_removed_date_none_when_active():
+    """Active listing (status В продаже) → removed_date=None in row."""
+    db = MagicMock()
+    items = [
+        ValuationHistoryItem(
+            area_m2=42.0,
+            rooms=1,
+            floor=2,
+            start_price=7_500_000,
+            last_price=7_500_000,
+            publish_date=date(2024, 6, 1),
+            removed_date=None,
+            exposure_days=10,
+            status="В продаже",
+        ),
+    ]
+    result = _make_result(items)
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        return_value=(7, 1.0, "new"),
+    ):
+        _save_yandex_history_items(db, result)
+
+    row = db.execute.call_args_list[0].args[1]
+    assert row["removed_date"] is None
+
+
+def test_save_handles_match_failure_gracefully():
+    """If match_or_create_house raises, save continues with house_id=None, confidence=0."""
+    db = MagicMock()
+    items = [
+        ValuationHistoryItem(
+            area_m2=42.0,
+            rooms=1,
+            floor=2,
+            start_price=7_500_000,
+            last_price=7_500_000,
+            publish_date=date(2024, 6, 1),
+            removed_date=None,
+            exposure_days=10,
+            status="В продаже",
+        ),
+    ]
+    result = _make_result(items)
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        side_effect=RuntimeError("simulated lookup failure"),
+    ):
+        saved = _save_yandex_history_items(db, result)
+
+    assert saved == 1
+    row = db.execute.call_args_list[0].args[1]
+    assert row["house_id"] is None
+    assert row["confidence"] == pytest.approx(0.0)
+    assert row["notes"] is None
+
+
+def test_save_match_called_with_address_and_year():
+    """match_or_create_house receives correct address and year_built from result.house."""
+    db = MagicMock()
+    result = _make_result(_two_items())
+    with patch(
+        "app.services.estimator.match_or_create_house",
+        return_value=(1, 1.0, "new"),
+    ) as m:
+        _save_yandex_history_items(db, result)
+
+    kwargs = m.call_args.kwargs
+    assert kwargs["address"] == result.address
+    assert kwargs["year_built"] == result.house.year_built
+    assert kwargs["ext_source"] == "yandex_valuation"