Merge pull request 'fix(tradein/data): external_valuations.house_id — резолв в write-path + идемпотентный бэкфилл (#2236)' (#2263) from fix/tradein-extval-house-id into main

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

@@ -509,11 +509,17 @@ async def _get_or_fetch_yandex_valuation_cached(
     address: str,
     offer_category: str = YANDEX_VALUATION_DEFAULT_CATEGORY,
     offer_type: str = YANDEX_VALUATION_DEFAULT_TYPE,
+    house_id: int | None = None,
 ) -> YandexValuationResult | None:
     """Cached Yandex Valuation lookup. TTL 24h via external_valuations table.
 
     Returns None on any error / cache miss + fetch failure — caller continues
     without Yandex enrichment (graceful degradation).
+
+    house_id: канонический дом, уже разрезолвленный estimate-путём через
+    match_house_readonly. Пишется в external_valuations.house_id при fresh-fetch
+    (best-effort; None → колонка остаётся NULL). ON CONFLICT сохраняет уже
+    проставленный house_id через COALESCE.
     """
     cache_key = _yandex_valuation_cache_key(address, offer_category, offer_type)
 
@@ -580,15 +586,20 @@ async def _get_or_fetch_yandex_valuation_cached(
                 """
                 INSERT INTO external_valuations (
                     source, cache_key, address,
+                    house_id,
                     raw_payload,
                     fetched_at, expires_at
                 ) VALUES (
                     'yandex_valuation', :ck, :addr,
+                    CAST(:hid AS bigint),
                     CAST(:payload AS jsonb),
                     NOW(), NOW() + (:ttl_hours || ' hours')::interval
                 )
                 ON CONFLICT (source, cache_key) DO UPDATE
                     SET raw_payload = EXCLUDED.raw_payload,
+                        house_id = COALESCE(
+                            EXCLUDED.house_id, external_valuations.house_id
+                        ),
                         fetched_at = NOW(),
                         expires_at = NOW() + (:ttl_hours || ' hours')::interval
                 """
@@ -596,6 +607,7 @@ async def _get_or_fetch_yandex_valuation_cached(
             {
                 "ck": cache_key,
                 "addr": address,
+                "hid": house_id,
                 "payload": json.dumps(result.model_dump(mode="json"), ensure_ascii=False),
                 "ttl_hours": YANDEX_VALUATION_CACHE_TTL_HOURS,
             },
@@ -2907,17 +2919,19 @@ async def estimate_quality(
     yandex_val: YandexValuationResult | None = None
     if geo is not None and geo.full_address:
         yandex_val = await _with_budget(
-            _get_or_fetch_yandex_valuation_cached(db, address=geo.full_address),
+            _get_or_fetch_yandex_valuation_cached(
+                db, address=geo.full_address, house_id=target_house_id
+            ),
             settings.estimate_yandex_valuation_timeout_s,
             label="yandex_valuation",
         )
         if yandex_val is not None:
             saved_hist = await asyncio.to_thread(_save_yandex_history_items, db, yandex_val)
             logger.info(
-                "yandex_valuation: history items processed=%d saved=%d"
+                "yandex_valuation: history items processed=%d saved=%d (ext_val house_id=%s)",
-                " (house_id=NULL — matching deferred)",
                 len(yandex_val.history_items),
                 saved_hist,
+                target_house_id,
             )
 
     # ── Stage 9: Cian Valuation as 7th source (on-demand, 24h cached) ──────
@@ -2942,6 +2956,7 @@ async def estimate_quality(
                     repair_type="cosmetic",
                     deal_type="sale",
                     use_cache=True,
+                    house_id=target_house_id,
                 ),
                 settings.estimate_cian_valuation_timeout_s,
                 label="cian_valuation",

tradein-mvp/backend/scripts/backfill_external_valuations_house_id.py

@@ -0,0 +1,366 @@
+"""Backfill external_valuations.house_id for existing rows (issue #2236).
+
+Внешние оценки (Cian Valuation Calculator, Yandex Оценка квартиры) исторически
+писались в `external_valuations` без привязки к канонической таблице `houses`:
+на проде 1579/1579 строк имели `house_id IS NULL`. Write-path теперь резолвит
+house_id тем же матчером, что и estimate (`match_house_readonly`), но уже
+накопленные строки остаются без ссылки.
+
+Эта one-shot джоба идёт по всем `external_valuations WHERE house_id IS NULL`
+и резолвит канонический дом ТЕМ ЖЕ путём, что estimate-запрос — через
+`match_house_readonly(db, address=...)`. В external_valuations нет lat/lon и
+кадастра, поэтому резолв опирается на адрес (Tier 1 fingerprint / — при пустых
+координатах — по нормализованному адресу дома). Best-effort: дом не нашёлся →
+house_id остаётся NULL, строка не трогается.
+
+Per-row SAVEPOINT (`db.begin_nested()`) per `.claude/rules/backend.md`: один
+битый матч не должен ронять весь батч.
+
+Идемпотентность / резюмируемость:
+    - Source-query: `WHERE house_id IS NULL` — уже проставленные строки
+      выпадают из выборки. Повторный прогон резолвит только оставшиеся NULL,
+      результат тот же.
+    - UPDATE guard `AND house_id IS NULL` — гонок с write-path не перетирает
+      уже проставленный house_id.
+    - Курсор `id > :after_id` двигается и по нерезолвленным строкам, поэтому
+      один прогон не зацикливается на строке, которую матчер вернул None.
+
+Никаких сетевых вызовов — матчер работает целиком в БД.
+
+Usage:
+    DATABASE_URL=postgresql+psycopg://... \\
+        python -m scripts.backfill_external_valuations_house_id --batch-size 500
+
+    # Канарейка сначала (без записи)
+    python -m scripts.backfill_external_valuations_house_id --limit 100 --dry-run
+
+    # Ограничить одним источником
+    python -m scripts.backfill_external_valuations_house_id --source cian_valuation
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from collections import defaultdict
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+# Allow running both as `python -m scripts.backfill_external_valuations_house_id`
+# (preferred) and as a stand-alone file (fallback for adhoc invocation).
+try:
+    from app.core.db import SessionLocal  # type: ignore[import-not-found]
+    from app.services.matching.houses import (  # type: ignore[import-not-found]
+        match_house_readonly,
+    )
+except ImportError:  # pragma: no cover — fallback for adhoc invocation
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    from app.core.db import SessionLocal
+    from app.services.matching.houses import match_house_readonly
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
+logger = logging.getLogger("backfill_external_valuations_house_id")
+
+# Резолвер: (db, address) -> house_id | None. Инъектируемый для тестов, по
+# умолчанию — тот же match_house_readonly, что зовёт estimate-путь.
+Matcher = Callable[[Session, str], "int | None"]
+
+
+def _default_matcher(db: Session, address: str) -> int | None:
+    """estimate-совместимый резолв дома по адресу (read-only, без создания)."""
+    match = match_house_readonly(db, address=address)
+    return match[0] if match is not None else None
+
+
+# ---------------------------------------------------------------------------
+# Domain types
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class ExtValRow:
+    """Одна строка external_valuations, которую резолвим."""
+
+    id: int
+    source: str
+    address: str
+
+
+@dataclass
+class Stats:
+    """Агрегатные счётчики — печатаются по батчу и в конце прогона."""
+
+    processed: int = 0
+    resolved: int = 0
+    unresolved: int = 0
+    errors: int = 0
+    by_source: dict[str, dict[str, int]] = field(default_factory=dict)
+
+    def bump(self, source: str, key: str) -> None:
+        bucket = self.by_source.setdefault(source, defaultdict(int))
+        bucket[key] += 1
+
+
+# ---------------------------------------------------------------------------
+# Source query — stream rows without a house_id
+# ---------------------------------------------------------------------------
+
+
+def _build_select_sql(*, source: str | None) -> str:
+    """Streaming SELECT — только строки с house_id IS NULL и непустым адресом."""
+    where_source = " AND source = :source " if source is not None else ""
+    return (
+        "SELECT id, source, address "
+        "FROM external_valuations "
+        "WHERE house_id IS NULL "
+        "  AND address IS NOT NULL "
+        "  AND id > :after_id "
+        f"  {where_source} "
+        "ORDER BY id "
+        "LIMIT CAST(:limit AS int)"
+    )
+
+
+def _fetch_batch(
+    db: Session, *, after_id: int, batch_size: int, source: str | None
+) -> list[ExtValRow]:
+    """Следующий батч нерезолвленных строк по возрастанию id."""
+    sql = _build_select_sql(source=source)
+    params: dict[str, Any] = {"after_id": after_id, "limit": batch_size}
+    if source is not None:
+        params["source"] = source
+
+    rows = db.execute(text(sql), params).mappings().all()
+    return [ExtValRow(id=int(r["id"]), source=r["source"], address=r["address"]) for r in rows]
+
+
+# ---------------------------------------------------------------------------
+# Per-row work
+# ---------------------------------------------------------------------------
+
+
+def _process_row(
+    db: Session,
+    row: ExtValRow,
+    *,
+    dry_run: bool,
+    stats: Stats,
+    matcher: Matcher,
+) -> None:
+    """Резолвит дом и (если найден) проставляет house_id одной строке.
+
+    Обёрнуто в per-row SAVEPOINT: битый матч/UPDATE не ломает батч. Best-effort —
+    matcher вернул None → строка не трогается, house_id остаётся NULL.
+    """
+    stats.processed += 1
+    stats.bump(row.source, "processed")
+
+    try:
+        # SAVEPOINT в обеих ветках: даже в dry-run matcher бьёт по БД (SELECT'ы),
+        # и его exception на реальной сессии перевёл бы транзакцию в aborted →
+        # каскад «current transaction is aborted» на всех последующих строках.
+        with db.begin_nested():
+            house_id = matcher(db, row.address)
+            if house_id is not None and not dry_run:
+                db.execute(
+                    text(
+                        "UPDATE external_valuations "
+                        "   SET house_id = CAST(:hid AS bigint) "
+                        " WHERE id = CAST(:eid AS bigint) "
+                        "   AND house_id IS NULL"
+                    ),
+                    {"hid": house_id, "eid": row.id},
+                )
+
+        if house_id is not None:
+            stats.resolved += 1
+            stats.bump(row.source, "resolved")
+        else:
+            stats.unresolved += 1
+            stats.bump(row.source, "unresolved")
+
+        logger.debug(
+            "backfill ext_val id=%s source=%s house_id=%s",
+            row.id,
+            row.source,
+            house_id,
+        )
+    except Exception as e:
+        stats.errors += 1
+        stats.bump(row.source, "errors")
+        logger.warning(
+            "backfill failed ext_val id=%s source=%s: %s",
+            row.id,
+            row.source,
+            e,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Driver
+# ---------------------------------------------------------------------------
+
+
+def run_backfill(
+    db: Session,
+    *,
+    batch_size: int,
+    limit: int | None,
+    source: str | None,
+    dry_run: bool,
+    matcher: Matcher = _default_matcher,
+) -> Stats:
+    """Главный драйвер — стримит батчи и проставляет house_id.
+
+    Args:
+        db: SQLAlchemy Session.
+        batch_size: строк за один SELECT (каждый батч коммитится).
+        limit: остановиться после стольких строк (--limit). None = до конца.
+        source: фильтр по source ('cian_valuation'/'yandex_valuation'). None = все.
+        dry_run: не писать в БД, только считать.
+        matcher: (db, address) -> house_id|None; по умолчанию match_house_readonly.
+    """
+    stats = Stats()
+    after_id = 0
+    batch_idx = 0
+
+    while True:
+        if limit is not None and stats.processed >= limit:
+            logger.info(
+                "limit reached: stopping (processed=%d, limit=%d)",
+                stats.processed,
+                limit,
+            )
+            break
+
+        effective_size = batch_size
+        if limit is not None:
+            effective_size = min(batch_size, limit - stats.processed)
+            if effective_size <= 0:
+                break
+
+        batch = _fetch_batch(db, after_id=after_id, batch_size=effective_size, source=source)
+        if not batch:
+            logger.info("no more rows — done")
+            break
+
+        batch_idx += 1
+        for row in batch:
+            _process_row(db, row, dry_run=dry_run, stats=stats, matcher=matcher)
+            after_id = max(after_id, row.id)
+
+        if not dry_run:
+            db.commit()
+
+        logger.info(
+            "batch %d done: size=%d total processed=%d resolved=%d unresolved=%d errors=%d",
+            batch_idx,
+            len(batch),
+            stats.processed,
+            stats.resolved,
+            stats.unresolved,
+            stats.errors,
+        )
+
+    return stats
+
+
+def _log_summary(stats: Stats, *, dry_run: bool) -> None:
+    """Финальная разбивка по источникам + итог."""
+    logger.info("=" * 72)
+    logger.info(
+        "backfill done (dry_run=%s): processed=%d resolved=%d unresolved=%d errors=%d",
+        dry_run,
+        stats.processed,
+        stats.resolved,
+        stats.unresolved,
+        stats.errors,
+    )
+    for src, bucket in sorted(stats.by_source.items()):
+        logger.info(
+            "  %-18s processed=%d resolved=%d unresolved=%d errors=%d",
+            src,
+            bucket.get("processed", 0),
+            bucket.get("resolved", 0),
+            bucket.get("unresolved", 0),
+            bucket.get("errors", 0),
+        )
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    """argparse setup, вынесено ради тестируемости."""
+    p = argparse.ArgumentParser(
+        description=(
+            "Backfill external_valuations.house_id — резолвит канонический дом "
+            "по адресу тем же match_house_readonly, что и estimate-путь. "
+            "Идемпотентно на re-run."
+        ),
+    )
+    p.add_argument(
+        "--batch-size",
+        type=int,
+        default=500,
+        help="Строк за SELECT (default: 500). Каждый батч коммитится.",
+    )
+    p.add_argument(
+        "--limit",
+        type=int,
+        default=None,
+        help="Потолок на общее число строк за прогон (для канарейки).",
+    )
+    p.add_argument(
+        "--source",
+        choices=("cian_valuation", "yandex_valuation"),
+        default=None,
+        help="Ограничить одним source. Default: все.",
+    )
+    p.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Логировать что было бы сделано; без записи в БД.",
+    )
+    return p.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    """CLI entry point. Возвращает число обработанных строк за прогон."""
+    args = _parse_args(argv)
+    logger.info(
+        "starting backfill: batch_size=%d limit=%s source=%s dry_run=%s",
+        args.batch_size,
+        args.limit if args.limit is not None else "all",
+        args.source or "all",
+        args.dry_run,
+    )
+
+    db = SessionLocal()
+    try:
+        stats = run_backfill(
+            db,
+            batch_size=args.batch_size,
+            limit=args.limit,
+            source=args.source,
+            dry_run=args.dry_run,
+        )
+        _log_summary(stats, dry_run=args.dry_run)
+        return stats.processed
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    sys.exit(0 if main() >= 0 else 1)

tradein-mvp/backend/tests/test_backfill_external_valuations_house_id.py

@@ -0,0 +1,214 @@
+"""Tests for the external_valuations.house_id backfill job (issue #2236).
+
+Coverage:
+  - happy path: resolvable rows get UPDATE-ed with house_id, counts add up.
+  - unresolved rows (matcher → None) stay untouched, no UPDATE, counted.
+  - per-row error isolation: one raising matcher call doesn't kill the batch.
+  - batching: batch_size < total → multiple batches, one commit each.
+  - idempotency: after a run marks rows resolved, a re-run finds nothing.
+  - dry-run: no UPDATE / no commit.
+
+No real Postgres — DB is a stateful MagicMock modelling the
+`WHERE house_id IS NULL AND id > :after_id` source query and the guarded
+UPDATE (same convention as test_backfill_house_coords).
+"""
+
+from __future__ import annotations
+
+import os
+
+# Settings требует DATABASE_URL на import.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from unittest.mock import MagicMock
+
+from scripts.backfill_external_valuations_house_id import (
+    ExtValRow,
+    _fetch_batch,
+    run_backfill,
+)
+
+
+def _make_db(pool: list[dict]):
+    """Stateful MagicMock DB.
+
+    `pool` = list of {id, source, address, house_id}. SELECT returns rows with
+    house_id IS NULL AND id > after_id (ordered, capped by limit). UPDATE sets
+    house_id on the matching pool row (guarded on house_id IS NULL).
+    Records commits.
+    """
+    db = MagicMock()
+    db.begin_nested.return_value.__enter__ = lambda self: self
+    db.begin_nested.return_value.__exit__ = lambda self, *a: False
+    commits: list[int] = []
+
+    def execute_side_effect(sql, params=None):
+        sql_str = str(sql)
+        result = MagicMock()
+        if "SELECT id, source, address" in sql_str:
+            after_id = params["after_id"]
+            limit = params["limit"]
+            src = params.get("source")
+            rows = [
+                {"id": r["id"], "source": r["source"], "address": r["address"]}
+                for r in sorted(pool, key=lambda x: x["id"])
+                if r["house_id"] is None
+                and r["id"] > after_id
+                and r["address"] is not None
+                and (src is None or r["source"] == src)
+            ][:limit]
+            result.mappings.return_value.all.return_value = rows
+        elif "UPDATE external_valuations" in sql_str:
+            for r in pool:
+                if r["id"] == params["eid"] and r["house_id"] is None:
+                    r["house_id"] = params["hid"]
+        return result
+
+    db.execute.side_effect = execute_side_effect
+    db.commit.side_effect = lambda: commits.append(1)
+    db.close = MagicMock()
+    return db, commits
+
+
+def _pool(n: int) -> list[dict]:
+    return [
+        {
+            "id": i,
+            "source": "cian_valuation",
+            "address": f"Екатеринбург, ул Тест, {i}",
+            "house_id": None,
+        }
+        for i in range(1, n + 1)
+    ]
+
+
+# ---------------------------------------------------------------------------
+# _fetch_batch
+# ---------------------------------------------------------------------------
+
+
+def test_fetch_batch_maps_rows():
+    db, _ = _make_db(_pool(3))
+    rows = _fetch_batch(db, after_id=0, batch_size=10, source=None)
+    assert [r.id for r in rows] == [1, 2, 3]
+    assert all(isinstance(r, ExtValRow) for r in rows)
+
+
+def test_fetch_batch_respects_after_id_and_limit():
+    db, _ = _make_db(_pool(5))
+    rows = _fetch_batch(db, after_id=2, batch_size=2, source=None)
+    assert [r.id for r in rows] == [3, 4]
+
+
+# ---------------------------------------------------------------------------
+# Happy path — resolvable rows updated
+# ---------------------------------------------------------------------------
+
+
+def test_run_resolves_and_updates():
+    pool = _pool(3)
+    db, commits = _make_db(pool)
+    matcher = MagicMock(return_value=555)
+
+    stats = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher)
+
+    assert stats.processed == 3
+    assert stats.resolved == 3
+    assert stats.unresolved == 0
+    assert stats.errors == 0
+    assert all(r["house_id"] == 555 for r in pool)
+    assert len(commits) == 1  # one batch → one commit
+
+
+def test_run_unresolved_rows_untouched():
+    """matcher → None: house_id остаётся NULL, UPDATE не вызывается, счётчик растёт."""
+    pool = _pool(2)
+    db, _ = _make_db(pool)
+    matcher = MagicMock(return_value=None)
+
+    stats = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher)
+
+    assert stats.processed == 2
+    assert stats.resolved == 0
+    assert stats.unresolved == 2
+    assert all(r["house_id"] is None for r in pool)
+
+
+def test_run_row_error_isolated():
+    """Матчер кидает на одной строке — батч продолжается, ошибка посчитана."""
+    pool = _pool(3)
+    db, _ = _make_db(pool)
+
+    def matcher(_db, addr):
+        if "2" in addr:
+            raise RuntimeError("boom")
+        return 999
+
+    stats = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher)
+    assert stats.processed == 3
+    assert stats.resolved == 2
+    assert stats.errors == 1
+
+
+# ---------------------------------------------------------------------------
+# Batching
+# ---------------------------------------------------------------------------
+
+
+def test_run_multiple_batches_commit_each():
+    pool = _pool(5)
+    db, commits = _make_db(pool)
+    matcher = MagicMock(return_value=7)
+
+    stats = run_backfill(db, batch_size=2, limit=None, source=None, dry_run=False, matcher=matcher)
+    # 5 rows / batch 2 → batches of 2, 2, 1 = 3 commits
+    assert stats.processed == 5
+    assert stats.resolved == 5
+    assert len(commits) == 3
+
+
+def test_run_limit_caps_total():
+    pool = _pool(10)
+    db, _ = _make_db(pool)
+    matcher = MagicMock(return_value=1)
+
+    stats = run_backfill(db, batch_size=3, limit=4, source=None, dry_run=False, matcher=matcher)
+    assert stats.processed == 4
+
+
+# ---------------------------------------------------------------------------
+# Idempotency
+# ---------------------------------------------------------------------------
+
+
+def test_run_idempotent_second_pass_noop():
+    pool = _pool(4)
+    db, _ = _make_db(pool)
+    matcher = MagicMock(return_value=42)
+
+    first = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher)
+    assert first.processed == 4 and first.resolved == 4
+
+    # Все строки теперь house_id != NULL → source-query пуст на re-run.
+    second = run_backfill(
+        db, batch_size=10, limit=None, source=None, dry_run=False, matcher=matcher
+    )
+    assert second.processed == 0
+    assert second.resolved == 0
+
+
+# ---------------------------------------------------------------------------
+# Dry-run
+# ---------------------------------------------------------------------------
+
+
+def test_run_dry_run_no_writes():
+    pool = _pool(3)
+    db, commits = _make_db(pool)
+    matcher = MagicMock(return_value=123)
+
+    stats = run_backfill(db, batch_size=10, limit=None, source=None, dry_run=True, matcher=matcher)
+    assert stats.processed == 3
+    assert stats.resolved == 3  # matcher нашёл дом
+    assert all(r["house_id"] is None for r in pool)  # но НЕ записали
+    assert commits == []

tradein-mvp/backend/tests/test_extval_house_id_write_path.py

@@ -0,0 +1,155 @@
+"""Write-path tests: external_valuations.house_id резолвится при сохранении (#2236).
+
+Cian и Yandex valuation-кэши теперь принимают house_id (уже разрезолвленный
+estimate-путём через match_house_readonly) и пишут его в external_valuations.
+Best-effort: house_id=None → колонка NULL, сохранение НЕ падает.
+
+DB — MagicMock, записываем параметры db.execute (никакого реального Postgres,
+как в test_yandex_valuation_save / test_backfill_house_coords).
+"""
+
+from __future__ import annotations
+
+import os
+
+# Settings требует DATABASE_URL на import. Ставим фиктивный DSN заранее.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from app.services.scrapers.cian_valuation import CianValuationResult, _save_to_cache
+
+# ---------------------------------------------------------------------------
+# Cian valuation — _save_to_cache прокидывает house_id в INSERT params
+# ---------------------------------------------------------------------------
+
+
+def _cian_result() -> CianValuationResult:
+    return CianValuationResult(
+        sale_price_rub=10_000_000,
+        sale_accuracy=85.0,
+        sale_price_from=9_000_000,
+        sale_price_to=11_000_000,
+        rent_price_rub=40_000,
+        rent_accuracy=70.0,
+        chart=[],
+        chart_change_pct=None,
+        chart_change_direction=None,
+        external_house_id=54016,
+        filters_hash="abc",
+        raw_state=None,
+    )
+
+
+def _cian_save(db: MagicMock, house_id: int | None) -> dict:
+    _save_to_cache(
+        db,
+        cache_key="ck",
+        address="Екатеринбург, улица Малышева, 51",
+        total_area=50.0,
+        rooms_count=2,
+        floor=3,
+        total_floors=9,
+        repair_type="cosmetic",
+        deal_type="sale",
+        result=_cian_result(),
+        house_id=house_id,
+    )
+    # _save_to_cache делает один db.execute(INSERT ...) + commit
+    return db.execute.call_args_list[0].args[1]
+
+
+def test_cian_save_persists_house_id():
+    """Резолвленный house_id попадает в INSERT-параметры."""
+    db = MagicMock()
+    params = _cian_save(db, house_id=777)
+    assert params["hid"] == 777
+    db.commit.assert_called_once()
+
+
+def test_cian_save_null_house_id_no_exception():
+    """house_id=None → hid=None в params, сохранение не падает."""
+    db = MagicMock()
+    params = _cian_save(db, house_id=None)
+    assert params["hid"] is None
+    db.commit.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Yandex valuation — _get_or_fetch_yandex_valuation_cached пишет house_id
+# ---------------------------------------------------------------------------
+
+
+def _make_db_cache_miss() -> MagicMock:
+    """MagicMock db, у которого cache-lookup даёт miss (first() → None)."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = None
+    return db
+
+
+def _yandex_result():
+    from app.services.scrapers.yandex_valuation import (
+        ValuationHouseMeta,
+        YandexValuationResult,
+    )
+
+    return YandexValuationResult(
+        address="Екатеринбург, улица Куйбышева, 106",
+        offer_category="APARTMENT",
+        offer_type="SELL",
+        page=1,
+        source_url="https://realty.yandex.ru/otsenka/?address=test&page=1",
+        house=ValuationHouseMeta(year_built=2005, total_floors=25),
+        history_items=[],
+    )
+
+
+def _insert_params(db: MagicMock) -> dict:
+    """Найти params INSERT INTO external_valuations среди вызовов db.execute."""
+    for call in db.execute.call_args_list:
+        args = call.args
+        if len(args) >= 2 and "INSERT INTO external_valuations" in str(args[0]):
+            return args[1]
+    raise AssertionError("INSERT INTO external_valuations not found in db.execute calls")
+
+
+async def test_yandex_save_persists_house_id():
+    """Fresh fetch пишет house_id в INSERT external_valuations."""
+    from app.services import estimator
+
+    db = _make_db_cache_miss()
+    scraper = AsyncMock()
+    scraper.fetch_house_history = AsyncMock(return_value=_yandex_result())
+    cm = MagicMock()
+    cm.__aenter__ = AsyncMock(return_value=scraper)
+    cm.__aexit__ = AsyncMock(return_value=False)
+
+    with patch.object(estimator, "YandexValuationScraper", return_value=cm):
+        res = await estimator._get_or_fetch_yandex_valuation_cached(
+            db, address="Екатеринбург, улица Куйбышева, 106", house_id=4242
+        )
+
+    assert res is not None
+    params = _insert_params(db)
+    assert params["hid"] == 4242
+
+
+async def test_yandex_save_null_house_id_no_exception():
+    """house_id=None → hid=None, INSERT проходит без исключения."""
+    from app.services import estimator
+
+    db = _make_db_cache_miss()
+    scraper = AsyncMock()
+    scraper.fetch_house_history = AsyncMock(return_value=_yandex_result())
+    cm = MagicMock()
+    cm.__aenter__ = AsyncMock(return_value=scraper)
+    cm.__aexit__ = AsyncMock(return_value=False)
+
+    with patch.object(estimator, "YandexValuationScraper", return_value=cm):
+        res = await estimator._get_or_fetch_yandex_valuation_cached(
+            db, address="Екатеринбург, улица Куйбышева, 106"
+        )
+
+    assert res is not None
+    params = _insert_params(db)
+    assert params["hid"] is None