diff --git a/tradein-mvp/backend/data/sql/070_houses_dadata_enrichment.sql b/tradein-mvp/backend/data/sql/070_houses_dadata_enrichment.sql
new file mode 100644
index 00000000..5b6dca20
--- /dev/null
+++ b/tradein-mvp/backend/data/sql/070_houses_dadata_enrichment.sql
@@ -0,0 +1,62 @@
+-- 070_houses_dadata_enrichment.sql
+-- PR Q3: DaData backfill полей на таблице `houses`.
+--
+-- После PR Q1 (#604) сервис `dadata.clean_address()` живёт на проде и обогащает
+-- адреса estimator-flow on-demand. Этот скрипт (`scripts/backfill_houses_dadata.py`)
+-- проходит по `houses` daily 100/day (DaData demo quota) и заполняет:
+--   - lat / lon (3801 rows без координат)
+--   - cadastral_number (12437 rows — 100% NULL)
+--   - house_fias_id — НОВАЯ колонка, симметрия с trade_in_estimates.house_fias_id (PR Q1)
+--
+-- Колонки dadata_qc_geo / dadata_qc_house / dadata_enriched_at — tracking:
+-- помечают что попытка enrichment была сделана, даже если результат не записан
+-- (quality codes 2/3/4 — слишком низкое разрешение, не overwrite coords).
+-- Это позволяет:
+--   1) resume-safe backfill: `WHERE dadata_enriched_at IS NULL` пропускает обработанные
+--   2) re-enrich stale: при необходимости фильтр `dadata_enriched_at < NOW() - 30 days`
+--
+-- Все поля nullable + IF NOT EXISTS → миграция идемпотентна.
+--
+-- Dependencies: 009_houses.sql + 010_houses_alter.sql + 029_extend_matching_valuation_dynamics.sql
+-- Apply after: 069_trade_in_estimates_dadata_fields.sql
+
+BEGIN;
+
+ALTER TABLE houses
+    ADD COLUMN IF NOT EXISTS house_fias_id text,
+    ADD COLUMN IF NOT EXISTS dadata_qc_geo integer,
+    ADD COLUMN IF NOT EXISTS dadata_qc_house integer,
+    ADD COLUMN IF NOT EXISTS dadata_enriched_at timestamptz;
+
+-- Индекс на ФИАС для будущих JOIN'ов между houses ↔ trade_in_estimates (PR Q1).
+CREATE INDEX IF NOT EXISTS houses_fias_id_idx
+    ON houses (house_fias_id)
+    WHERE house_fias_id IS NOT NULL;
+
+-- Cadastral_number индекс уже создан в 029_extend_matching_valuation_dynamics.sql
+-- (имя houses_cadastral_number_idx) и в 050_search_optimization.sql (houses_kadastr_idx2).
+-- IF NOT EXISTS → no-op при повторном создании. Оставляем для idempotent docs.
+CREATE INDEX IF NOT EXISTS houses_cadastral_number_idx
+    ON houses (cadastral_number)
+    WHERE cadastral_number IS NOT NULL;
+
+-- Partial index для быстрого выбора кандидатов backfill'а:
+-- `WHERE dadata_enriched_at IS NULL ORDER BY id`. После полного прохода
+-- (12437 row → ~124 дня walltime) индекс схлопнется до пустого set.
+CREATE INDEX IF NOT EXISTS houses_dadata_not_enriched_idx
+    ON houses (id)
+    WHERE dadata_enriched_at IS NULL;
+
+COMMENT ON COLUMN houses.house_fias_id IS
+    'DaData UUID ФИАС дома (PR Q3, #604+#605 follow-up). '
+    'Заполняется backfill-скриптом scripts/backfill_houses_dadata.py.';
+COMMENT ON COLUMN houses.dadata_qc_geo IS
+    'DaData qc_geo: 0=exact, 1=street, 2=settlement, 3=city. '
+    'Только qc_geo IN (0,1) триггерит overwrite lat/lon в backfill.';
+COMMENT ON COLUMN houses.dadata_qc_house IS
+    'DaData qc_house: 2=в ФИАС, 10=на картах, иначе не найден.';
+COMMENT ON COLUMN houses.dadata_enriched_at IS
+    'Когда последний раз пробовали enrich через DaData (NULL = ни разу). '
+    'Используется для resume-safe backfill — `WHERE dadata_enriched_at IS NULL`.';
+
+COMMIT;
diff --git a/tradein-mvp/backend/scripts/backfill_houses_dadata.py b/tradein-mvp/backend/scripts/backfill_houses_dadata.py
new file mode 100644
index 00000000..65e144da
--- /dev/null
+++ b/tradein-mvp/backend/scripts/backfill_houses_dadata.py
@@ -0,0 +1,518 @@
+"""Backfill houses table via DaData /clean/address.
+
+PR Q3 follow-up to PR Q1 (#604) — теперь, когда сервис `dadata.clean_address()`
+живёт на проде, проходим по таблице `houses` daily 100/day (demo quota) и
+дополняем недостающие координаты + кадастровые номера + ФИАС.
+
+DB state on prod (verified 2026-05-27):
+  - houses: 12437 rows total
+  - lat IS NULL OR lon IS NULL: 3801 (нужны координаты)
+  - cadastral_number IS NULL: 12437 (100% — все нужно дополнить)
+  - house_fias_id IS NULL: 12437 (колонка добавлена в migration 070)
+
+Walltime: 12437 / 100 = ~124 дня. Acceptable — это batch backfill, не realtime.
+
+Priority order (--priority both, default):
+  1. NULL lat OR NULL lon (3801 rows) — нужны coords для spatial match
+  2. NULL cadastral_number AND lat IS NOT NULL (~8636 rows) — есть coords,
+     дополняем только cadnum / fias / qc-codes
+
+`--priority coords` — только #1, `--priority cadnum` — только #2.
+
+Per-row outcome rules (после `clean_address` ответа):
+  - result.qc_geo IN (0, 1) AND result.lat IS NOT NULL → точное / уличное совпадение,
+    UPDATE houses SET lat=, lon=, cadastral_number=, house_fias_id=,
+        dadata_qc_geo=, dadata_qc_house=, dadata_enriched_at=NOW()
+  - result.qc_geo IN (2, 3, 4) OR result IS None → не overwrite координаты,
+    только UPDATE houses SET dadata_qc_geo=, dadata_qc_house=, dadata_enriched_at=NOW()
+    (фиксируем факт попытки → resume пропустит row при следующем запуске)
+
+Resume-safe: `WHERE dadata_enriched_at IS NULL` фильтрует уже обработанные.
+
+Design choices:
+- **Per-row SAVEPOINT** (`db.begin_nested()`) — одна DaData / DB ошибка не должна
+  завалить весь batch (тот же паттерн что и в backfill_house_coords.py).
+- **Rate limit**: 0.5s между requests (DaData demo allows ~5 req/sec, мы при limit=100
+  делаем 50 секунд непрерывной работы — далеко от quota burst-detection).
+- **Daily budget**: `--limit 100` default matches demo tier (100 req/day).
+
+Usage:
+    DATABASE_URL=postgresql+psycopg://... \\
+    DADATA_API_TOKEN=xxx DADATA_API_SECRET=xxx \\
+    python -m scripts.backfill_houses_dadata --limit 100
+
+Flags:
+    --limit N        max rows to process (default 100)
+    --batch LABEL    log label (default `dadata_YYYY-MM-DD`)
+    --dry-run        log what would happen, no DB writes
+    --priority X     coords | cadnum | both (default both)
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import os
+from dataclasses import dataclass, field
+from datetime import date
+from pathlib import Path
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+# Allow running both as `python -m scripts.backfill_houses_dadata` (preferred,
+# матчит pattern из backfill_house_coords.py) и как stand-alone file.
+try:
+    from app.core.db import SessionLocal  # type: ignore[import-not-found]
+except ImportError:  # pragma: no cover — fallback for adhoc invocation
+    import sys
+
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    from app.core.db import SessionLocal
+
+try:
+    from app.services.dadata import (  # type: ignore[import-not-found]
+        DadataAddressResult,
+        clean_address,
+    )
+except ImportError:  # pragma: no cover
+    import sys
+
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    from app.services.dadata import DadataAddressResult, clean_address
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
+logger = logging.getLogger("backfill_houses_dadata")
+
+# DaData demo tier: 100 requests/day, no burst limit documented but
+# best practice — sleep между requests, чтобы не триггерить anti-flood.
+# 0.5s = 2 req/sec → 50s walltime для full 100/day batch.
+_REQUEST_DELAY_S = 0.5
+
+# qc_geo: 0=exact, 1=street, 2=settlement, 3=city, 4=region, 5=unknown.
+# Только 0/1 точны достаточно для overwrite coords дома.
+_QC_GEO_ACCEPT = frozenset({0, 1})
+
+# Логировать каждые N rows
+_LOG_EVERY = 10
+
+
+# ---------------------------------------------------------------------------
+# Domain types
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class HouseRow:
+    """Минимальные поля из houses для enrichment loop'а.
+
+    `address` — сырой адрес (приоритет houses.address, fallback на full_address).
+    """
+
+    id: int
+    address: str
+    lat: float | None
+    lon: float | None
+    cadastral_number: str | None
+
+
+@dataclass
+class Stats:
+    """Счётчики для финального summary log'а."""
+
+    processed: int = 0
+    enriched: int = 0  # qc_geo IN (0,1) — coords/cadnum/fias записаны
+    no_change: int = 0  # qc_geo IN (2,3,4) или canonical_address None — только tracking
+    failed: int = 0  # exception в clean_address или DB write
+    skipped: int = 0  # пустой address (нечего geocode'ить)
+    by_priority: dict[str, int] = field(default_factory=dict)
+
+    def bump(self, priority: str, outcome: str) -> None:
+        """Внутренний учёт по priority bucket'у."""
+        bucket = self.by_priority.setdefault(
+            priority, {"enriched": 0, "no_change": 0, "failed": 0, "skipped": 0}
+        )
+        bucket[outcome] = bucket.get(outcome, 0) + 1
+
+
+# ---------------------------------------------------------------------------
+# Source-row query — приоритет: NULL coords → NULL cadnum
+# ---------------------------------------------------------------------------
+
+
+def _select_candidates(
+    db: Session, *, priority: str, limit: int
+) -> list[tuple[HouseRow, str]]:
+    """Возвращает список (HouseRow, priority_bucket) для enrichment.
+
+    priority:
+      - 'coords' → только rows с NULL coords (max бизнес-value)
+      - 'cadnum' → только rows с NOT NULL coords + NULL cadnum
+      - 'both'   → coords first, потом cadnum (FIFO в пределах buckets)
+
+    Фильтр `dadata_enriched_at IS NULL` — resume-safe (не повторяем уже
+    обработанные rows). После full enrichment колонка станет NOT NULL у всех
+    rows которые мы успели тронуть, и партишен-индекс
+    `houses_dadata_not_enriched_idx` схлопнется.
+    """
+    # Skip rows без какого-либо адреса — geocode'ить нечего.
+    address_filter = (
+        "(COALESCE(NULLIF(trim(address), ''), NULLIF(trim(full_address), '')) IS NOT NULL)"
+    )
+
+    out: list[tuple[HouseRow, str]] = []
+    remaining = limit
+
+    def _fetch(where: str, bucket: str, lim: int) -> list[tuple[HouseRow, str]]:
+        sql = text(
+            "SELECT id, "
+            "       COALESCE(NULLIF(trim(address), ''), NULLIF(trim(full_address), '')) "
+            "         AS address, "
+            "       lat, lon, cadastral_number "
+            "FROM houses "
+            f"WHERE {where} "
+            "  AND dadata_enriched_at IS NULL "
+            f"  AND {address_filter} "
+            "ORDER BY id "
+            "LIMIT CAST(:lim AS int)"
+        )
+        rows = db.execute(sql, {"lim": lim}).mappings().all()
+        return [
+            (
+                HouseRow(
+                    id=r["id"],
+                    address=r["address"],
+                    lat=r["lat"],
+                    lon=r["lon"],
+                    cadastral_number=r["cadastral_number"],
+                ),
+                bucket,
+            )
+            for r in rows
+        ]
+
+    if priority in ("coords", "both"):
+        out.extend(_fetch("(lat IS NULL OR lon IS NULL)", "coords", remaining))
+        remaining = limit - len(out)
+    if priority in ("cadnum", "both") and remaining > 0:
+        out.extend(
+            _fetch(
+                "(lat IS NOT NULL AND lon IS NOT NULL AND cadastral_number IS NULL)",
+                "cadnum",
+                remaining,
+            )
+        )
+
+    return out[:limit]
+
+
+# ---------------------------------------------------------------------------
+# DB writers
+# ---------------------------------------------------------------------------
+
+
+def _update_house_enriched(
+    db: Session,
+    *,
+    house_id: int,
+    result: DadataAddressResult,
+) -> None:
+    """UPDATE houses с полным набором полей (qc_geo IN (0,1) → точное матч).
+
+    Не overwrite'ит lat/lon если result.lat/lon None (DaData может вернуть
+    qc_geo=0 для города но без точки — edge case). COALESCE'им через
+    `CASE WHEN :lat IS NOT NULL THEN :lat ELSE lat END` чтобы не ломать
+    существующие значения.
+
+    cadastral_number / house_fias_id — overwrite только если DaData вернул
+    непустое значение (COALESCE с existing).
+    """
+    db.execute(
+        text(
+            "UPDATE houses "
+            "   SET lat = CASE WHEN CAST(:lat AS double precision) IS NOT NULL "
+            "                  THEN CAST(:lat AS double precision) "
+            "                  ELSE lat END, "
+            "       lon = CASE WHEN CAST(:lon AS double precision) IS NOT NULL "
+            "                  THEN CAST(:lon AS double precision) "
+            "                  ELSE lon END, "
+            "       cadastral_number = COALESCE(CAST(:cadnum AS text), cadastral_number), "
+            "       house_fias_id    = COALESCE(CAST(:fias AS text), house_fias_id), "
+            "       dadata_qc_geo    = CAST(:qc_geo AS integer), "
+            "       dadata_qc_house  = CAST(:qc_house AS integer), "
+            "       dadata_enriched_at = NOW() "
+            " WHERE id = CAST(:id AS bigint)"
+        ),
+        {
+            "id": house_id,
+            "lat": result.lat,
+            "lon": result.lon,
+            "cadnum": result.house_cadnum,
+            "fias": result.house_fias_id,
+            "qc_geo": result.qc_geo,
+            "qc_house": result.qc_house,
+        },
+    )
+
+
+def _update_house_attempt(
+    db: Session,
+    *,
+    house_id: int,
+    qc_geo: int | None,
+    qc_house: int | None,
+) -> None:
+    """UPDATE только tracking-поля (qc_geo вне ACCEPT set, либо None result).
+
+    НЕ трогаем lat/lon/cadnum/fias — DaData либо не нашёл точно (qc=2,3,4),
+    либо вообще не распознал адрес. Фиксируем `dadata_enriched_at` чтобы
+    resume не повторял эту row при следующем запуске.
+    """
+    db.execute(
+        text(
+            "UPDATE houses "
+            "   SET dadata_qc_geo    = CAST(:qc_geo AS integer), "
+            "       dadata_qc_house  = CAST(:qc_house AS integer), "
+            "       dadata_enriched_at = NOW() "
+            " WHERE id = CAST(:id AS bigint)"
+        ),
+        {"id": house_id, "qc_geo": qc_geo, "qc_house": qc_house},
+    )
+
+
+# ---------------------------------------------------------------------------
+# Main loop
+# ---------------------------------------------------------------------------
+
+
+def _is_enriched_result(result: DadataAddressResult | None) -> bool:
+    """Решает, можно ли overwrite'ить координаты + cadnum по этому result'у.
+
+    Условие: qc_geo IN (0, 1) AND lat / lon заданы.
+    Возврат False означает «записать только tracking, не overwrite payload».
+    """
+    if result is None:
+        return False
+    if result.qc_geo not in _QC_GEO_ACCEPT:
+        return False
+    if result.lat is None or result.lon is None:
+        return False
+    return True
+
+
+async def _run_backfill(
+    db: Session,
+    candidates: list[tuple[HouseRow, str]],
+    *,
+    batch: str,
+    dry_run: bool,
+) -> Stats:
+    """Главный цикл — для каждого candidate зовём DaData + UPDATE houses.
+
+    Per-row SAVEPOINT защищает batch от падений на одной row. Per-row commit
+    после успешного nested block → resume picks up exactly где упали.
+    """
+    stats = Stats()
+
+    for i, (row, priority) in enumerate(candidates, start=1):
+        result: DadataAddressResult | None = None
+        outcome = "no_change"
+
+        try:
+            # `clean_address` сам обрабатывает HTTP errors / quota / timeout
+            # и возвращает None — exception будет только при unexpected error.
+            result = await clean_address(row.address)
+        except Exception as exc:  # defensive — одна ошибка не должна валить batch
+            stats.failed += 1
+            stats.bump(priority, "failed")
+            logger.warning(
+                "dadata_call failed for house_id=%s addr=%r: %s",
+                row.id,
+                row.address[:60],
+                exc,
+            )
+            # Не пишем в DB при exception — оставляем row для следующей попытки.
+            continue
+
+        if dry_run:
+            if _is_enriched_result(result):
+                outcome = "enriched"
+                logger.info(
+                    "DRY-RUN house_id=%s [%s] WOULD enrich → "
+                    "qc_geo=%s qc_house=%s lat=%s lon=%s cadnum=%s fias=%s",
+                    row.id,
+                    priority,
+                    result.qc_geo if result else None,
+                    result.qc_house if result else None,
+                    result.lat if result else None,
+                    result.lon if result else None,
+                    result.house_cadnum if result else None,
+                    result.house_fias_id if result else None,
+                )
+            else:
+                outcome = "no_change"
+                logger.info(
+                    "DRY-RUN house_id=%s [%s] WOULD record attempt only → "
+                    "qc_geo=%s qc_house=%s (result_present=%s)",
+                    row.id,
+                    priority,
+                    result.qc_geo if result else None,
+                    result.qc_house if result else None,
+                    result is not None,
+                )
+            stats.processed += 1
+            if outcome == "enriched":
+                stats.enriched += 1
+            else:
+                stats.no_change += 1
+            stats.bump(priority, outcome)
+        else:
+            try:
+                with db.begin_nested():
+                    if _is_enriched_result(result):
+                        assert result is not None  # narrowed by _is_enriched_result
+                        _update_house_enriched(db, house_id=row.id, result=result)
+                        outcome = "enriched"
+                        stats.enriched += 1
+                    else:
+                        _update_house_attempt(
+                            db,
+                            house_id=row.id,
+                            qc_geo=result.qc_geo if result else None,
+                            qc_house=result.qc_house if result else None,
+                        )
+                        outcome = "no_change"
+                        stats.no_change += 1
+                # Per-row commit чтобы resume picks up exactly where we crashed.
+                db.commit()
+                stats.processed += 1
+                stats.bump(priority, outcome)
+            except Exception as exc:  # defensive — DB write failure isolation
+                db.rollback()
+                stats.failed += 1
+                stats.bump(priority, "failed")
+                logger.warning(
+                    "db_write failed for house_id=%s: %s", row.id, exc
+                )
+
+        if i % _LOG_EVERY == 0:
+            logger.info(
+                "batch=%s progress %d/%d enriched=%d no_change=%d failed=%d",
+                batch,
+                i,
+                len(candidates),
+                stats.enriched,
+                stats.no_change,
+                stats.failed,
+            )
+
+        # Rate-limit: пауза только если ещё есть rows впереди.
+        if i < len(candidates):
+            await asyncio.sleep(_REQUEST_DELAY_S)
+
+    return stats
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    """argparse setup, выделено отдельно для testability."""
+    p = argparse.ArgumentParser(
+        description=(
+            "PR Q3 — backfill houses.lat/lon/cadastral_number/house_fias_id "
+            "через DaData /clean/address (demo tier, 100/day)."
+        ),
+    )
+    p.add_argument(
+        "--limit",
+        type=int,
+        default=100,
+        help="Max rows to process (default 100 — matches DaData demo daily quota).",
+    )
+    p.add_argument(
+        "--batch",
+        default=f"dadata_{date.today().isoformat()}",
+        help="Log batch label. Не влияет на DB фильтры — только для логов.",
+    )
+    p.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Логирует что бы произошло, но не пишет в DB.",
+    )
+    p.add_argument(
+        "--priority",
+        choices=("coords", "cadnum", "both"),
+        default="both",
+        help=(
+            "Какой набор кандидатов обрабатывать. coords = только NULL coords, "
+            "cadnum = NOT NULL coords + NULL cadnum, both (default) = первый "
+            "приоритет coords."
+        ),
+    )
+    return p.parse_args(argv)
+
+
+async def main(argv: list[str] | None = None) -> int:
+    """CLI entry point. Возвращает кол-во processed rows."""
+    args = _parse_args(argv)
+
+    # Сервис graceful'но disable'ится без creds, но script без них бессмысленен —
+    # лучше fail fast, чтобы cron-runs не съели DB-locks впустую.
+    token = os.environ.get("DADATA_API_TOKEN", "").strip()
+    secret = os.environ.get("DADATA_API_SECRET", "").strip()
+    if not token or not secret:
+        raise SystemExit(
+            "DADATA_API_TOKEN + DADATA_API_SECRET required — "
+            "DaData /clean/address без secret не работает."
+        )
+
+    logger.info(
+        "starting batch=%s limit=%s priority=%s dry_run=%s",
+        args.batch,
+        args.limit,
+        args.priority,
+        args.dry_run,
+    )
+
+    db = SessionLocal()
+    try:
+        candidates = _select_candidates(db, priority=args.priority, limit=args.limit)
+        logger.info(
+            "loaded candidates: %d (priority=%s)", len(candidates), args.priority
+        )
+        if not candidates:
+            logger.info(
+                "nothing to do — нет rows с dadata_enriched_at IS NULL для priority=%s",
+                args.priority,
+            )
+            return 0
+
+        stats = await _run_backfill(
+            db, candidates, batch=args.batch, dry_run=args.dry_run
+        )
+
+        logger.info(
+            "done: batch=%s processed=%d enriched=%d no_change=%d "
+            "failed=%d skipped=%d by_priority=%s",
+            args.batch,
+            stats.processed,
+            stats.enriched,
+            stats.no_change,
+            stats.failed,
+            stats.skipped,
+            stats.by_priority,
+        )
+        return stats.processed
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    asyncio.run(main())
diff --git a/tradein-mvp/backend/tests/scripts/test_backfill_houses_dadata.py b/tradein-mvp/backend/tests/scripts/test_backfill_houses_dadata.py
new file mode 100644
index 00000000..661f1e38
--- /dev/null
+++ b/tradein-mvp/backend/tests/scripts/test_backfill_houses_dadata.py
@@ -0,0 +1,464 @@
+"""Unit tests for PR Q3 — backfill_houses_dadata.py.
+
+Coverage:
+    - Happy path: row с NULL coords + qc_geo=0 → UPDATE с полным набором полей.
+    - qc_geo=3 (city-level) → tracking only, lat/lon не overwrite'аются.
+    - DaData returns None → tracking only (qc_geo / qc_house = NULL).
+    - Per-row SAVEPOINT: exception от clean_address не валит batch.
+    - --dry-run: nothing written в DB.
+    - --limit N: батч обрезается до N даже если кандидатов больше.
+    - --priority coords / cadnum / both — фильтр source-rows работает.
+    - _is_enriched_result — accept / reject правила.
+    - main() с отсутствующими DADATA_API_TOKEN / SECRET → SystemExit.
+
+No real Postgres. Session — MagicMock что записывает UPDATE и роутит SELECT
+side-effects (тот же pattern что в test_backfill_house_coords.py /
+test_backfill_listing_sources.py).
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import AsyncMock, MagicMock, patch
+
+# Settings требует DSN на import — set dummy DSN до любого app.* import.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+import pytest
+
+from app.services.dadata import DadataAddressResult
+from scripts.backfill_houses_dadata import (
+    HouseRow,
+    Stats,
+    _is_enriched_result,
+    _run_backfill,
+    _select_candidates,
+    _update_house_attempt,
+    _update_house_enriched,
+    main,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_dadata_result(
+    *,
+    qc_geo: int | None = 0,
+    qc_house: int | None = 2,
+    lat: float | None = 56.838,
+    lon: float | None = 60.586,
+    cadnum: str | None = "66:41:0704045:350",
+    fias: str | None = "abcdef00-aaaa-bbbb-cccc-1234567890ab",
+    canonical: str | None = "г Екатеринбург, ул Малышева, д 51",
+) -> DadataAddressResult:
+    """DadataAddressResult fixture с sensible defaults для tests."""
+    return DadataAddressResult(
+        canonical_address=canonical,
+        house_cadnum=cadnum,
+        house_fias_id=fias,
+        lat=lat,
+        lon=lon,
+        qc_geo=qc_geo,
+        qc_house=qc_house,
+        kladr_id=None,
+        okato=None,
+        oktmo=None,
+        metro=[],
+        raw={"test": True},
+    )
+
+
+def _make_db_mock(
+    coords_sample: list[dict] | None = None,
+    cadnum_sample: list[dict] | None = None,
+) -> tuple[MagicMock, list[dict]]:
+    """MagicMock Session что:
+      - возвращает coords_sample для SELECT с `lat IS NULL OR lon IS NULL`
+      - возвращает cadnum_sample для SELECT с `cadastral_number IS NULL`
+      - записывает UPDATE houses в `updated` список (полные binds dict'ы)
+      - корректно работает с `db.begin_nested()` контекст-менеджером
+    """
+    coords_sample = coords_sample or []
+    cadnum_sample = cadnum_sample or []
+    updated: list[dict] = []
+
+    db = MagicMock()
+    db.begin_nested.return_value.__enter__ = lambda self: self
+    db.begin_nested.return_value.__exit__ = lambda self, *a: False
+
+    def execute_side_effect(sql, params=None):
+        sql_str = str(sql)
+        result = MagicMock()
+        if "FROM houses" in sql_str and "lat IS NULL OR lon IS NULL" in sql_str:
+            lim = params.get("lim", len(coords_sample)) if params else len(coords_sample)
+            result.mappings.return_value.all.return_value = coords_sample[:lim]
+        elif "FROM houses" in sql_str and "cadastral_number IS NULL" in sql_str:
+            lim = params.get("lim", len(cadnum_sample)) if params else len(cadnum_sample)
+            result.mappings.return_value.all.return_value = cadnum_sample[:lim]
+        elif "UPDATE houses" in sql_str:
+            updated.append(dict(params) if params else {})
+        return result
+
+    db.execute.side_effect = execute_side_effect
+    db.commit = MagicMock()
+    db.rollback = MagicMock()
+    db.close = MagicMock()
+    return db, updated
+
+
+# ---------------------------------------------------------------------------
+# _is_enriched_result — accept / reject правила
+# ---------------------------------------------------------------------------
+
+
+def test_is_enriched_result_qc_geo_0_lat_lon_present_accepts():
+    res = _make_dadata_result(qc_geo=0, lat=56.838, lon=60.586)
+    assert _is_enriched_result(res) is True
+
+
+def test_is_enriched_result_qc_geo_1_accepts():
+    """qc_geo=1 (street-level) — также accept'аем для overwrite."""
+    res = _make_dadata_result(qc_geo=1, lat=56.838, lon=60.586)
+    assert _is_enriched_result(res) is True
+
+
+def test_is_enriched_result_qc_geo_2_rejects():
+    """qc_geo=2 (settlement) — слишком грубо для дома."""
+    res = _make_dadata_result(qc_geo=2)
+    assert _is_enriched_result(res) is False
+
+
+def test_is_enriched_result_qc_geo_3_rejects():
+    """qc_geo=3 (city) — точно не дом."""
+    res = _make_dadata_result(qc_geo=3)
+    assert _is_enriched_result(res) is False
+
+
+def test_is_enriched_result_none_input_rejects():
+    assert _is_enriched_result(None) is False
+
+
+def test_is_enriched_result_qc_geo_ok_but_no_coords_rejects():
+    """Defensive: qc_geo=0 но lat/lon отсутствуют → не overwrite."""
+    res = _make_dadata_result(qc_geo=0, lat=None, lon=None)
+    assert _is_enriched_result(res) is False
+
+
+# ---------------------------------------------------------------------------
+# _update_house_enriched — bind shape
+# ---------------------------------------------------------------------------
+
+
+def test_update_house_enriched_passes_full_payload():
+    """UPDATE houses содержит lat/lon/cadnum/fias/qc-codes и dadata_enriched_at = NOW()."""
+    db = MagicMock()
+    res = _make_dadata_result(
+        qc_geo=0, qc_house=2, lat=56.838, lon=60.586,
+        cadnum="66:41:0704045:350", fias="fias-uuid-here",
+    )
+    _update_house_enriched(db, house_id=42, result=res)
+
+    args, _kw = db.execute.call_args
+    sql_str = str(args[0])
+    binds = args[1]
+    assert "UPDATE houses" in sql_str
+    assert "dadata_enriched_at = NOW()" in sql_str
+    assert "cadastral_number" in sql_str
+    assert "house_fias_id" in sql_str
+    assert binds["id"] == 42
+    assert binds["lat"] == 56.838
+    assert binds["lon"] == 60.586
+    assert binds["cadnum"] == "66:41:0704045:350"
+    assert binds["fias"] == "fias-uuid-here"
+    assert binds["qc_geo"] == 0
+    assert binds["qc_house"] == 2
+
+
+def test_update_house_attempt_writes_only_tracking_fields():
+    """attempt-write не должен затрагивать lat/lon/cadnum/fias columns."""
+    db = MagicMock()
+    _update_house_attempt(db, house_id=7, qc_geo=3, qc_house=10)
+    args, _kw = db.execute.call_args
+    sql_str = str(args[0])
+    binds = args[1]
+    assert "UPDATE houses" in sql_str
+    assert "dadata_enriched_at = NOW()" in sql_str
+    # Не должно быть UPDATE на coord-payload поля
+    assert "lat =" not in sql_str
+    assert "lon =" not in sql_str
+    assert "cadastral_number =" not in sql_str
+    assert "house_fias_id =" not in sql_str
+    assert binds == {"id": 7, "qc_geo": 3, "qc_house": 10}
+
+
+# ---------------------------------------------------------------------------
+# _run_backfill — happy path qc_geo=0
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_qc_geo_0_writes_enriched_update():
+    """Happy path — qc_geo=0 → UPDATE с полным payload, stats.enriched=1."""
+    row = HouseRow(id=1, address="ул Малышева 51", lat=None, lon=None, cadastral_number=None)
+    db, updated = _make_db_mock()
+    fake = _make_dadata_result(qc_geo=0)
+
+    with patch(
+        "scripts.backfill_houses_dadata.clean_address",
+        new=AsyncMock(return_value=fake),
+    ):
+        stats = await _run_backfill(
+            db, [(row, "coords")], batch="b1", dry_run=False
+        )
+
+    assert stats.enriched == 1
+    assert stats.no_change == 0
+    assert stats.failed == 0
+    assert stats.processed == 1
+    assert len(updated) == 1
+    assert updated[0]["lat"] == 56.838
+    assert updated[0]["lon"] == 60.586
+    assert updated[0]["cadnum"] == "66:41:0704045:350"
+    assert updated[0]["qc_geo"] == 0
+    assert db.commit.call_count == 1
+    assert stats.by_priority["coords"]["enriched"] == 1
+
+
+# ---------------------------------------------------------------------------
+# _run_backfill — qc_geo=3 city-level → tracking-only
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_qc_geo_3_records_attempt_only():
+    """qc_geo=3 → НЕ overwrite coords, только tracking UPDATE."""
+    row = HouseRow(
+        id=2, address="Екатеринбург (без улицы)", lat=None, lon=None, cadastral_number=None
+    )
+    db, updated = _make_db_mock()
+    fake = _make_dadata_result(qc_geo=3, qc_house=None, lat=None, lon=None, cadnum=None, fias=None)
+
+    with patch(
+        "scripts.backfill_houses_dadata.clean_address",
+        new=AsyncMock(return_value=fake),
+    ):
+        stats = await _run_backfill(
+            db, [(row, "coords")], batch="b2", dry_run=False
+        )
+
+    assert stats.enriched == 0
+    assert stats.no_change == 1
+    assert len(updated) == 1
+    # Tracking-only UPDATE — нет lat/lon/cadnum binds
+    assert "lat" not in updated[0]
+    assert "lon" not in updated[0]
+    assert updated[0]["qc_geo"] == 3
+    assert stats.by_priority["coords"]["no_change"] == 1
+
+
+# ---------------------------------------------------------------------------
+# _run_backfill — DaData returns None (graceful)
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_dadata_returns_none_records_attempt():
+    """clean_address возвращает None (HTTP error / unrecognized) → tracking only с NULL qc."""
+    row = HouseRow(id=3, address="несуществующая 99", lat=None, lon=None, cadastral_number=None)
+    db, updated = _make_db_mock()
+
+    with patch(
+        "scripts.backfill_houses_dadata.clean_address",
+        new=AsyncMock(return_value=None),
+    ):
+        stats = await _run_backfill(
+            db, [(row, "coords")], batch="b3", dry_run=False
+        )
+
+    assert stats.no_change == 1
+    assert stats.enriched == 0
+    assert len(updated) == 1
+    assert updated[0]["qc_geo"] is None
+    assert updated[0]["qc_house"] is None
+
+
+# ---------------------------------------------------------------------------
+# Per-row SAVEPOINT — exception в clean_address не валит batch
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_clean_address_exception_isolated_to_row():
+    """Один clean_address raise → этот row failed, остальные processed normally."""
+    rows = [
+        (HouseRow(id=10, address="ул Bad", lat=None, lon=None, cadastral_number=None), "coords"),
+        (HouseRow(id=11, address="ул Good 5", lat=None, lon=None, cadastral_number=None), "coords"),
+    ]
+    db, updated = _make_db_mock()
+    good_result = _make_dadata_result(qc_geo=0)
+
+    call_count = {"n": 0}
+
+    async def fake_clean(addr):
+        call_count["n"] += 1
+        if call_count["n"] == 1:
+            raise RuntimeError("dadata blew up")
+        return good_result
+
+    with patch("scripts.backfill_houses_dadata.clean_address", side_effect=fake_clean):
+        stats = await _run_backfill(db, rows, batch="b4", dry_run=False)
+
+    assert stats.processed == 1  # только 2-й row дошёл до DB-write success
+    assert stats.failed == 1
+    assert stats.enriched == 1
+    # Только один UPDATE — для второго row
+    assert len(updated) == 1
+    assert updated[0]["id"] == 11
+
+
+# ---------------------------------------------------------------------------
+# --dry-run — no DB writes
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_dry_run_skips_db_writes():
+    """dry_run=True → counters обновляются, но db.execute UPDATE не вызывается."""
+    row = HouseRow(id=20, address="ул Test 1", lat=None, lon=None, cadastral_number=None)
+    db, updated = _make_db_mock()
+    fake = _make_dadata_result(qc_geo=0)
+
+    with patch(
+        "scripts.backfill_houses_dadata.clean_address",
+        new=AsyncMock(return_value=fake),
+    ):
+        stats = await _run_backfill(
+            db, [(row, "coords")], batch="dry", dry_run=True
+        )
+
+    assert stats.processed == 1
+    assert stats.enriched == 1
+    assert updated == []  # никаких UPDATE'ов
+    assert db.commit.call_count == 0  # никаких commit'ов
+
+
+# ---------------------------------------------------------------------------
+# _select_candidates — priority filter работает
+# ---------------------------------------------------------------------------
+
+
+def test_select_candidates_coords_priority_only_fetches_null_coords():
+    """priority=coords → запрашивается только `lat IS NULL OR lon IS NULL` SELECT."""
+    coords_rows = [
+        {"id": 1, "address": "ул A 1", "lat": None, "lon": None, "cadastral_number": None},
+        {"id": 2, "address": "ул B 2", "lat": None, "lon": None, "cadastral_number": None},
+    ]
+    db, _updated = _make_db_mock(coords_sample=coords_rows)
+
+    out = _select_candidates(db, priority="coords", limit=10)
+
+    assert len(out) == 2
+    assert all(p == "coords" for _row, p in out)
+    assert out[0][0].id == 1
+    assert out[1][0].id == 2
+
+
+def test_select_candidates_cadnum_priority_only_fetches_null_cadnum():
+    """priority=cadnum → только NOT NULL coords + NULL cadastral_number rows."""
+    cadnum_rows = [
+        {"id": 50, "address": "ул C 3", "lat": 56.8, "lon": 60.6, "cadastral_number": None},
+    ]
+    db, _updated = _make_db_mock(cadnum_sample=cadnum_rows)
+
+    out = _select_candidates(db, priority="cadnum", limit=10)
+
+    assert len(out) == 1
+    assert out[0][1] == "cadnum"
+    assert out[0][0].id == 50
+
+
+def test_select_candidates_both_priority_fetches_coords_first_then_cadnum():
+    """priority=both → сначала NULL coords, потом NULL cadnum (в пределах limit)."""
+    coords_rows = [
+        {"id": 1, "address": "ул A 1", "lat": None, "lon": None, "cadastral_number": None},
+    ]
+    cadnum_rows = [
+        {"id": 100, "address": "ул C 3", "lat": 56.8, "lon": 60.6, "cadastral_number": None},
+    ]
+    db, _updated = _make_db_mock(coords_sample=coords_rows, cadnum_sample=cadnum_rows)
+
+    out = _select_candidates(db, priority="both", limit=10)
+
+    assert len(out) == 2
+    # coords bucket первый
+    assert out[0][1] == "coords"
+    assert out[0][0].id == 1
+    assert out[1][1] == "cadnum"
+    assert out[1][0].id == 100
+
+
+# ---------------------------------------------------------------------------
+# --limit caps batch size
+# ---------------------------------------------------------------------------
+
+
+async def test_main_respects_limit(monkeypatch):
+    """main(--limit 1) при 3 кандидатах → только 1 row processed."""
+    coords_rows = [
+        {"id": 1, "address": "ул A 1", "lat": None, "lon": None, "cadastral_number": None},
+        {"id": 2, "address": "ул B 2", "lat": None, "lon": None, "cadastral_number": None},
+        {"id": 3, "address": "ул C 3", "lat": None, "lon": None, "cadastral_number": None},
+    ]
+    db, updated = _make_db_mock(coords_sample=coords_rows)
+    monkeypatch.setenv("DADATA_API_TOKEN", "TEST_TOKEN")
+    monkeypatch.setenv("DADATA_API_SECRET", "TEST_SECRET")
+    fake = _make_dadata_result(qc_geo=0)
+
+    with (
+        patch("scripts.backfill_houses_dadata.SessionLocal", return_value=db),
+        patch(
+            "scripts.backfill_houses_dadata.clean_address",
+            new=AsyncMock(return_value=fake),
+        ),
+        # Уберём sleep чтобы test не висел 0.5s
+        patch("scripts.backfill_houses_dadata.asyncio.sleep", new=AsyncMock()),
+    ):
+        n = await main(["--limit", "1", "--batch", "test_limit"])
+
+    assert n == 1
+    assert len(updated) == 1
+
+
+# ---------------------------------------------------------------------------
+# Missing creds → SystemExit
+# ---------------------------------------------------------------------------
+
+
+async def test_main_requires_dadata_creds(monkeypatch):
+    """Без DADATA_API_TOKEN + DADATA_API_SECRET → SystemExit ещё до DB connect."""
+    monkeypatch.delenv("DADATA_API_TOKEN", raising=False)
+    monkeypatch.delenv("DADATA_API_SECRET", raising=False)
+    with pytest.raises(SystemExit):
+        await main(["--limit", "1"])
+
+
+async def test_main_requires_secret_not_just_token(monkeypatch):
+    """Только TOKEN (без SECRET) → /clean/address не сработает → fail fast."""
+    monkeypatch.setenv("DADATA_API_TOKEN", "TEST_TOKEN")
+    monkeypatch.delenv("DADATA_API_SECRET", raising=False)
+    with pytest.raises(SystemExit):
+        await main(["--limit", "1"])
+
+
+# ---------------------------------------------------------------------------
+# Stats summary contains by_priority breakdown
+# ---------------------------------------------------------------------------
+
+
+def test_stats_bump_buckets_by_priority():
+    """Stats.bump поддерживает несколько priority bucket'ов одновременно."""
+    s = Stats()
+    s.bump("coords", "enriched")
+    s.bump("coords", "enriched")
+    s.bump("cadnum", "no_change")
+    assert s.by_priority["coords"]["enriched"] == 2
+    assert s.by_priority["cadnum"]["no_change"] == 1
+    # Не существующие outcome'ы должны быть 0
+    assert s.by_priority["coords"]["failed"] == 0