tradein-mvp/backend/scripts/geocode_deals_nominatim.py

@@ -0,0 +1,564 @@
+"""Geocode deals still `lat IS NULL` after the house-centroid pass, via geocoder.
+
+Issue #569, Step 3 — the Nominatim (cache → Cadastral FDW → Yandex → Nominatim)
+fallback for the ~14% of deals the centroid join can never cover.
+
+Background
+----------
+Step 2 (`geocode_deals_from_houses.py`) backfilled `deals.lat/lon` by matching
+each deal's normalized street to a centroid computed from already-geocoded
+`houses`. On prod that covered 86% (42,840 / 49,791 rows). The remaining ~14%
+(~6,951 rows) sit on streets with NO geocoded house at all, so the centroid join
+has nothing to match against — those need a real geocoder call.
+
+This script picks up exactly those leftovers (`deals WHERE lat IS NULL`) and runs
+each through `app.services.geocoder.geocode(address, db)`, which already does
+cache → Cadastral FDW → Yandex → Nominatim with rate-limiting and writes the
+result into `geocode_cache`. We do NOT call Nominatim directly — that keeps a
+single source of truth for provider order, ЕКБ bbox filtering, and the 1 req/sec
+policy.
+
+Why dedup by address (not one call per row)
+-------------------------------------------
+`deals.address` is street-only ('Екатеринбург, <street>'), so thousands of rows
+share the same address. The geocoder caches by normalized address, but we also
+GROUP BY address up front so the real call count is driven by DISTINCT streets,
+not the ~6,951 row backlog. One geocode call → UPDATE every deal on that street.
+
+Street-level precision is accepted: the estimator's comparable search uses a
+1000-2000 m radius, so a street-level point lands every deal on that street in
+the same comparable window. The `deals_set_geom_trg` BEFORE UPDATE OF lat, lon
+trigger (002_core_tables.sql) fills `geom` from lat/lon, so we set lat/lon ONLY,
+never geom.
+
+Resumable + bounded retries (critical — Nominatim is ~1 req/s)
+--------------------------------------------------------------
+We stamp `geocode_tried_at = NOW()` on EVERY attempt — success AND failure — so
+an un-geocodable address (Nominatim returns nothing) drops out of the candidate
+set and is NOT re-scanned every run. Without this, each run would re-try the
+full ~7k failures at 1 req/s. The candidate query combines:
+  - `lat IS NULL` (the `deals_geocode_pending_idx` partial index, 005), and
+  - `geocode_tried_at IS NULL OR geocode_tried_at < NOW() - interval '30 days'`
+so failed addresses are retried at most once every 30 days, not every run.
+
+Per-address `db.begin_nested()` SAVEPOINT (backend.md rule) isolates one bad
+UPDATE from the rest of the batch; a per-address commit means a crash mid-run
+leaves already-geocoded streets persisted and resume picks up the rest.
+
+Usage:
+    DATABASE_URL=postgresql+psycopg://... \\
+    YANDEX_GEOCODER_API_KEY=...   # optional — geocoder falls back to Nominatim \\
+    python -m scripts.geocode_deals_nominatim --dry-run
+
+    # real run, default cap 2000 deal rows / run
+    python -m scripts.geocode_deals_nominatim --batch 2026-05-28
+
+Flags:
+    --limit N      max deal ROWS to fan out over this run (default 2000). The
+                   number of geocode CALLS is the distinct-address count within
+                   that slice, which is far smaller.
+    --dry-run      report how many rows / distinct streets would be attempted,
+                   plus the geocode hit/miss split. No DB writes.
+    --batch LABEL  log label (default `deals_nominatim_YYYY-MM-DD`).
+    --stale-days N retry addresses last tried more than N days ago (default 30).
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+# Allow running both as `python -m scripts.geocode_deals_nominatim` (preferred,
+# matches backfill_houses_dadata.py) and as a 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.geocoder import (  # type: ignore[import-not-found]
+        GeocodeResult,
+        geocode,
+    )
+except ImportError:  # pragma: no cover
+    import sys
+
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    from app.services.geocoder import GeocodeResult, geocode
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
+logger = logging.getLogger("geocode_deals_nominatim")
+
+# Default per-run row cap. Nominatim is ~1 req/sec; the number of CALLS is the
+# distinct-address count inside this slice (far below the row count thanks to
+# dedup + cache), but we still cap rows so a single run can't block for hours.
+_DEFAULT_LIMIT = 2000
+
+# Retry an address last tried more than this many days ago. Bounds retries on
+# un-geocodable addresses (failed today → not re-scanned until 30 days pass).
+_DEFAULT_STALE_DAYS = 30
+
+# Progress log cadence (per distinct address processed).
+_LOG_EVERY = 25
+
+
+# ---------------------------------------------------------------------------
+# Domain types
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class AddressGroup:
+    """One distinct deals.address and how many lat-IS-NULL rows share it."""
+
+    address: str
+    deals_count: int
+
+
+@dataclass
+class Stats:
+    """Final-summary counters.
+
+    - processed     distinct addresses fed to the geocoder this run.
+    - geocoded      addresses the geocoder resolved to coords.
+    - geocode_failed addresses the geocoder returned None for (still stamped).
+    - skipped       distinct addresses skipped before any geocode call
+                    (too short to be a usable query).
+    - deals_updated total deal ROWS that received lat/lon (sum over geocoded
+                    addresses).
+    - cache_hits / cache_misses  provider == 'cache' vs a real provider call.
+    """
+
+    processed: int = 0
+    geocoded: int = 0
+    geocode_failed: int = 0
+    skipped: int = 0
+    deals_updated: int = 0
+    cache_hits: int = 0
+    cache_misses: int = 0
+
+
+# ---------------------------------------------------------------------------
+# Source query — distinct addresses among the leftover lat-IS-NULL deals
+# ---------------------------------------------------------------------------
+
+
+def _select_pending_addresses(
+    db: Session, *, limit: int, stale_days: int
+) -> list[AddressGroup]:
+    """Distinct deals.address still needing coords — resume-safe candidate set.
+
+    Combines the `deals_geocode_pending_idx` partial index predicate
+    (`lat IS NULL`) with a staleness filter so failed/un-geocodable addresses
+    are retried at most once every `stale_days`, never every run.
+
+    The GROUP BY collapses the ~6,951-row backlog into its distinct streets;
+    `:limit` caps the deal ROWS fanned out, computed from the running SUM of
+    per-address counts so a single huge street can't blow past the cap. We
+    order by occurrence DESC (biggest ROI per geocode call first) then address
+    for a deterministic resume order.
+    """
+    rows = db.execute(
+        text(
+            "SELECT address, deals_count FROM ("
+            "  SELECT address, "
+            "         COUNT(*) AS deals_count, "
+            "         SUM(COUNT(*)) OVER ("
+            "             ORDER BY COUNT(*) DESC, address ASC"
+            "         ) AS running_rows "
+            "  FROM deals "
+            "  WHERE lat IS NULL "
+            "    AND address IS NOT NULL "
+            "    AND length(trim(address)) >= 3 "
+            "    AND (geocode_tried_at IS NULL "
+            "         OR geocode_tried_at < NOW() "
+            "            - make_interval(days => CAST(:stale_days AS int))) "
+            "  GROUP BY address "
+            ") g "
+            "WHERE running_rows - deals_count < CAST(:limit AS int) "
+            "ORDER BY deals_count DESC, address ASC"
+        ),
+        {"limit": limit, "stale_days": stale_days},
+    ).mappings().all()
+    return [AddressGroup(address=r["address"], deals_count=r["deals_count"]) for r in rows]
+
+
+def _count_pending_total(db: Session, *, stale_days: int) -> tuple[int, int]:
+    """Full backlog: (distinct addresses, total rows) eligible this pass.
+
+    Denominators for the dry-run projection — counts every lat-IS-NULL deal
+    that passes the staleness filter, ignoring --limit.
+    """
+    row = db.execute(
+        text(
+            "SELECT COUNT(DISTINCT address) AS streets, COUNT(*) AS rows "
+            "FROM deals "
+            "WHERE lat IS NULL "
+            "  AND address IS NOT NULL "
+            "  AND length(trim(address)) >= 3 "
+            "  AND (geocode_tried_at IS NULL "
+            "       OR geocode_tried_at < NOW() "
+            "          - make_interval(days => CAST(:stale_days AS int)))"
+        ),
+        {"stale_days": stale_days},
+    ).first()
+    if row is None:
+        return (0, 0)
+    return (int(row[0]), int(row[1]))
+
+
+# ---------------------------------------------------------------------------
+# DB writers
+# ---------------------------------------------------------------------------
+
+
+def _update_deals_geocoded(db: Session, *, address: str, lat: float, lon: float) -> int:
+    """UPDATE every lat-IS-NULL deal on `address`; geom auto-fills via trigger.
+
+    The `deals_set_geom_trg` BEFORE UPDATE OF lat, lon trigger
+    (002_core_tables.sql, reuses listings_set_geom()) populates geom from the
+    new lat/lon, so we never touch geom here. Stamps `geocode_tried_at = NOW()`
+    so the row drops out of the candidate set. The `AND lat IS NULL` guard
+    keeps this idempotent and avoids clobbering rows another pass already set.
+
+    Returns the number of deal rows updated.
+    """
+    result = db.execute(
+        text(
+            "UPDATE deals "
+            "   SET lat = CAST(:lat AS double precision), "
+            "       lon = CAST(:lon AS double precision), "
+            "       geocode_tried_at = NOW() "
+            " WHERE address = CAST(:addr AS text) "
+            "   AND lat IS NULL"
+        ),
+        {"addr": address, "lat": lat, "lon": lon},
+    )
+    return result.rowcount
+
+
+def _mark_deals_tried(db: Session, *, address: str) -> int:
+    """Stamp `geocode_tried_at = NOW()` WITHOUT touching lat/lon (geocode miss).
+
+    Critical for resume: an address the geocoder can't resolve must still drop
+    out of the candidate set, or the next run re-scans it at ~1 req/s. We leave
+    lat/lon NULL (still "pending coords") but the staleness filter now excludes
+    it for `stale_days`. The `AND lat IS NULL` guard means a concurrent success
+    can't be downgraded.
+
+    Returns the number of deal rows stamped.
+    """
+    result = db.execute(
+        text(
+            "UPDATE deals "
+            "   SET geocode_tried_at = NOW() "
+            " WHERE address = CAST(:addr AS text) "
+            "   AND lat IS NULL"
+        ),
+        {"addr": address},
+    )
+    return result.rowcount
+
+
+# ---------------------------------------------------------------------------
+# Main loop
+# ---------------------------------------------------------------------------
+
+
+async def _run_backfill(
+    db: Session,
+    groups: list[AddressGroup],
+    *,
+    batch: str,
+    dry_run: bool,
+) -> Stats:
+    """For each distinct address: geocode once, then UPDATE all its deals.
+
+    Per-address SAVEPOINT (`db.begin_nested()`) so one bad UPDATE can't abort
+    the batch (backend.md SAVEPOINT rule). Per-address commit on success → a
+    crash mid-run leaves already-geocoded streets persisted, and resume picks
+    up the rest (lat IS NULL + staleness filter).
+
+    Rate-limiting lives inside `geocode()` (Nominatim 1 req/s sleep, Yandex
+    retry/backoff), so we don't add our own sleep here — that would double the
+    walltime. A geocode that raises is treated as a failure for THIS run but is
+    NOT stamped (left for the next pass to retry sooner than a clean miss).
+    """
+    stats = Stats()
+
+    for i, group in enumerate(groups, start=1):
+        address = group.address
+
+        # The geocoder itself rejects <3 chars, but skip here too so the dry-run
+        # report and counters stay honest (no phantom "processed" address).
+        if len(address.strip()) < 3:
+            stats.skipped += 1
+            continue
+
+        result: GeocodeResult | None = None
+        try:
+            result = await geocode(address, db)
+        except Exception as exc:  # defensive — one geocode error must not kill batch
+            stats.geocode_failed += 1
+            stats.processed += 1
+            logger.warning(
+                "geocode raised for addr=%r (%d deals): %s",
+                address[:60],
+                group.deals_count,
+                exc,
+            )
+            # Do NOT stamp on a raised exception — a transient network/HTTP error
+            # should be retried on the next run, not deferred for stale_days.
+            continue
+
+        if result is None:
+            # Clean miss — geocoder exhausted all providers. Stamp so we don't
+            # re-scan this un-geocodable address every run.
+            stats.geocode_failed += 1
+            stats.processed += 1
+            if dry_run:
+                logger.info(
+                    "DRY-RUN addr=%r (%d deals) → NOT FOUND (would stamp tried)",
+                    address[:60],
+                    group.deals_count,
+                )
+            else:
+                try:
+                    with db.begin_nested():
+                        _mark_deals_tried(db, address=address)
+                    db.commit()
+                except Exception as exc:  # defensive — isolate one bad UPDATE
+                    db.rollback()
+                    logger.warning("mark_tried failed for addr=%r: %s", address[:60], exc)
+            _maybe_log_progress(i, groups, batch, stats)
+            continue
+
+        # Hit — record provider split for the summary.
+        if result.provider == "cache":
+            stats.cache_hits += 1
+        else:
+            stats.cache_misses += 1
+        stats.geocoded += 1
+        stats.processed += 1
+
+        if dry_run:
+            logger.info(
+                "DRY-RUN addr=%r → (%.5f, %.5f) provider=%s would update %d deals",
+                address[:60],
+                result.lat,
+                result.lon,
+                result.provider,
+                group.deals_count,
+            )
+            _maybe_log_progress(i, groups, batch, stats)
+            continue
+
+        try:
+            with db.begin_nested():
+                n = _update_deals_geocoded(
+                    db, address=address, lat=result.lat, lon=result.lon
+                )
+            # Per-address commit so resume picks up exactly where we crashed.
+            db.commit()
+            stats.deals_updated += n
+        except Exception as exc:  # defensive — isolate one bad UPDATE
+            db.rollback()
+            # The geocode itself succeeded (and is cached); only the write
+            # failed. Count the address as geocoded but log the write failure.
+            logger.warning("db_write failed for addr=%r: %s", address[:60], exc)
+
+        _maybe_log_progress(i, groups, batch, stats)
+
+    return stats
+
+
+def _maybe_log_progress(i: int, groups: list[AddressGroup], batch: str, stats: Stats) -> None:
+    """Emit a progress line every `_LOG_EVERY` distinct addresses."""
+    if i % _LOG_EVERY == 0:
+        logger.info(
+            "batch=%s progress %d/%d geocoded=%d failed=%d deals_updated=%d "
+            "cache=(hit=%d miss=%d)",
+            batch,
+            i,
+            len(groups),
+            stats.geocoded,
+            stats.geocode_failed,
+            stats.deals_updated,
+            stats.cache_hits,
+            stats.cache_misses,
+        )
+
+
+# ---------------------------------------------------------------------------
+# Dry-run reporting
+# ---------------------------------------------------------------------------
+
+
+def _report_dry_run(
+    stats: Stats,
+    *,
+    total_streets: int,
+    total_rows: int,
+    scanned_streets: int,
+) -> None:
+    """Log the would-attempt counts + the geocode hit/miss split.
+
+    `scanned_streets` is how many distinct addresses we actually fed to the
+    geocoder this run (capped by --limit fan-out); `total_streets` / `total_rows`
+    are the full eligible backlog so the projection extrapolates honestly.
+    """
+    hit_rate = (stats.geocoded / stats.processed) if stats.processed else 0.0
+    projected_streets = round(hit_rate * total_streets)
+
+    logger.info("─" * 60)
+    logger.info("DRY-RUN SUMMARY (no DB writes)")
+    logger.info(
+        "full eligible backlog: %d distinct streets across %d deal rows "
+        "(lat IS NULL, not tried within stale window)",
+        total_streets,
+        total_rows,
+    )
+    logger.info("distinct streets attempted this run (capped by --limit): %d", scanned_streets)
+    logger.info("  → geocoded:      %d", stats.geocoded)
+    logger.info("  → not found:     %d", stats.geocode_failed)
+    logger.info("  → skipped (short): %d", stats.skipped)
+    logger.info("  → cache hit/miss:  %d / %d", stats.cache_hits, stats.cache_misses)
+    logger.info("hit rate on attempted streets: %.1f%%", hit_rate * 100.0)
+    logger.info(
+        "projected over full backlog: ≈ %d / %d streets resolvable (%.1f%%)",
+        projected_streets,
+        total_streets,
+        hit_rate * 100.0,
+    )
+    logger.info("─" * 60)
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    """argparse setup, factored out for testability."""
+    p = argparse.ArgumentParser(
+        description=(
+            "Issue #569 Step 3 — geocode deals still lat IS NULL after the "
+            "house-centroid pass, via app.services.geocoder.geocode "
+            "(cache → Cadastral FDW → Yandex → Nominatim)."
+        ),
+    )
+    p.add_argument(
+        "--limit",
+        type=int,
+        default=_DEFAULT_LIMIT,
+        help=(
+            f"Max deal ROWS to fan out over this run (default {_DEFAULT_LIMIT}). "
+            "Geocode CALLS = distinct-address count within that slice (smaller)."
+        ),
+    )
+    p.add_argument(
+        "--batch",
+        default=f"deals_nominatim_{date.today().isoformat()}",
+        help="Log batch label (does not affect DB filters — logs only).",
+    )
+    p.add_argument(
+        "--dry-run",
+        action="store_true",
+        help=(
+            "Report how many rows / distinct streets would be attempted and the "
+            "geocode hit/miss split. No DB writes."
+        ),
+    )
+    p.add_argument(
+        "--stale-days",
+        type=int,
+        default=_DEFAULT_STALE_DAYS,
+        help=(
+            f"Retry addresses last tried more than N days ago (default "
+            f"{_DEFAULT_STALE_DAYS}). Bounds retries on un-geocodable addresses."
+        ),
+    )
+    return p.parse_args(argv)
+
+
+async def main(argv: list[str] | None = None) -> int:
+    """CLI entry point. Returns the number of distinct addresses geocoded.
+
+    Async because `geocode()` is a coroutine; run via `asyncio.run(main())`
+    below, mirroring backfill_houses_dadata.py / backfill_house_coords.py.
+    No external creds are required — the geocoder degrades to Nominatim when
+    YANDEX_GEOCODER_API_KEY is unset, so we don't fail-fast on a missing key.
+    """
+    args = _parse_args(argv)
+    logger.info(
+        "starting batch=%s limit=%s stale_days=%s dry_run=%s",
+        args.batch,
+        args.limit,
+        args.stale_days,
+        args.dry_run,
+    )
+
+    db = SessionLocal()
+    try:
+        groups = _select_pending_addresses(
+            db, limit=args.limit, stale_days=args.stale_days
+        )
+        total_rows = sum(g.deals_count for g in groups)
+        logger.info(
+            "loaded %d distinct addresses (%d deal rows) needing coords",
+            len(groups),
+            total_rows,
+        )
+        if not groups:
+            logger.info(
+                "nothing to do — no deals with lat IS NULL eligible (all tried "
+                "within the last %d days, or no addressable rows)",
+                args.stale_days,
+            )
+            return 0
+
+        stats = await _run_backfill(db, groups, batch=args.batch, dry_run=args.dry_run)
+
+        if args.dry_run:
+            total_streets, backlog_rows = _count_pending_total(
+                db, stale_days=args.stale_days
+            )
+            _report_dry_run(
+                stats,
+                total_streets=total_streets,
+                total_rows=backlog_rows,
+                scanned_streets=len(groups),
+            )
+
+        logger.info(
+            "done: batch=%s processed=%d geocoded=%d geocode_failed=%d "
+            "skipped=%d deals_updated=%d cache=(hit=%d miss=%d)",
+            args.batch,
+            stats.processed,
+            stats.geocoded,
+            stats.geocode_failed,
+            stats.skipped,
+            stats.deals_updated,
+            stats.cache_hits,
+            stats.cache_misses,
+        )
+        return stats.geocoded
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(asyncio.run(main()))

tradein-mvp/backend/tests/scripts/test_geocode_deals_nominatim.py

@@ -0,0 +1,456 @@
+"""Unit tests for issue #569 Step 3 — geocode_deals_nominatim.py.
+
+Coverage (per the issue's test plan):
+    - a NULL-coords deal gets geocoded → UPDATE deals issued with lat/lon +
+      geocode_tried_at, counted as geocoded, deals_updated reflects rowcount.
+    - geocode() returns None → geocode_tried_at still stamped (a "mark tried"
+      UPDATE with no lat/lon), no coords UPDATE, counted as geocode_failed.
+    - --dry-run issues no UPDATE / no commit, counters still move.
+    - the resume filter excludes already-tried-recently rows (SQL carries the
+      lat IS NULL + staleness predicate, and binds --stale-days through).
+    - per-address SAVEPOINT isolation: one failing UPDATE doesn't abort the
+      batch.
+    - dedup: distinct addresses drive geocode call count (one call per address).
+    - main() wiring: SessionLocal, --limit bind, returns geocoded count.
+
+No real Postgres. `geocode()` is async → patched with AsyncMock (mirrors
+tests/tasks/test_geocode_missing.py). The Session is a MagicMock that routes
+SELECT side-effects by SQL substring and records UPDATE binds (mirrors
+tests/scripts/test_geocode_deals_from_houses.py).
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import AsyncMock, MagicMock, patch
+
+# Settings needs a DSN at import time — set a dummy before any app.* import.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from app.services.geocoder import GeocodeResult
+from scripts.geocode_deals_nominatim import (
+    AddressGroup,
+    Stats,
+    _mark_deals_tried,
+    _run_backfill,
+    _select_pending_addresses,
+    _update_deals_geocoded,
+    main,
+)
+
+_GEOCODE_PATH = "scripts.geocode_deals_nominatim.geocode"
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _result(provider: str = "nominatim", lat: float = 56.838, lon: float = 60.605) -> GeocodeResult:
+    return GeocodeResult(
+        lat=lat,
+        lon=lon,
+        full_address="Екатеринбург, ул. Тестовая",
+        provider=provider,  # type: ignore[arg-type]
+        confidence="approximate",
+    )
+
+
+def _make_db_mock(
+    *,
+    address_rows: list[dict] | None = None,
+    total_streets: int = 0,
+    total_rows: int = 0,
+    update_rowcount: int = 1,
+) -> tuple[MagicMock, list[dict], list[dict]]:
+    """MagicMock Session that:
+
+    - returns `address_rows` for the candidate SELECT (`FROM deals` GROUP BY)
+    - returns (total_streets, total_rows) for the dry-run COUNT
+    - records coords-UPDATE binds into `coord_updates`
+    - records mark-tried-UPDATE binds into `tried_updates`
+    - every UPDATE returns a result with `.rowcount == update_rowcount`
+    - supports `db.begin_nested()` as a context manager
+
+    Returns (db, coord_updates, tried_updates).
+    """
+    address_rows = address_rows or []
+    coord_updates: list[dict] = []
+    tried_updates: 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 "UPDATE deals" in sql_str:
+            # Distinguish the coords UPDATE (sets lat) from the mark-tried UPDATE.
+            if "SET lat" in sql_str:
+                coord_updates.append(dict(params) if params else {})
+            else:
+                tried_updates.append(dict(params) if params else {})
+            result.rowcount = update_rowcount
+            return result
+        if "COUNT(DISTINCT address)" in sql_str:
+            result.first.return_value = (total_streets, total_rows)
+            return result
+        if "FROM deals" in sql_str:
+            result.mappings.return_value.all.return_value = address_rows
+            return result
+        return result
+
+    db.execute.side_effect = execute_side_effect
+    db.commit = MagicMock()
+    db.rollback = MagicMock()
+    db.close = MagicMock()
+    return db, coord_updates, tried_updates
+
+
+# ---------------------------------------------------------------------------
+# _run_backfill — happy path: geocoded → coords UPDATE + tried_at stamped
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_geocoded_issues_coords_update():
+    groups = [AddressGroup(address="Екатеринбург, Тестовая", deals_count=5)]
+    db, coord_updates, tried_updates = _make_db_mock(update_rowcount=5)
+
+    with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")):
+        stats = await _run_backfill(db, groups, batch="b1", dry_run=False)
+
+    assert stats.geocoded == 1
+    assert stats.geocode_failed == 0
+    assert stats.processed == 1
+    assert stats.deals_updated == 5
+    assert stats.cache_misses == 1
+    assert stats.cache_hits == 0
+    # Exactly one coords UPDATE, bound with lat/lon/addr; no mark-tried UPDATE.
+    assert len(coord_updates) == 1
+    assert coord_updates[0]["addr"] == "Екатеринбург, Тестовая"
+    assert coord_updates[0]["lat"] == 56.838
+    assert coord_updates[0]["lon"] == 60.605
+    assert tried_updates == []
+    assert db.commit.call_count == 1
+
+
+async def test_run_backfill_cache_hit_counted_separately():
+    groups = [AddressGroup(address="Екатеринбург, Кэшевая", deals_count=2)]
+    db, coord_updates, _ = _make_db_mock(update_rowcount=2)
+
+    with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("cache")):
+        stats = await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    assert stats.geocoded == 1
+    assert stats.cache_hits == 1
+    assert stats.cache_misses == 0
+    assert len(coord_updates) == 1
+
+
+# ---------------------------------------------------------------------------
+# geocode None → geocode_tried_at stamped, NO coords UPDATE, counted failed
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_geocode_none_stamps_tried_no_coords():
+    groups = [AddressGroup(address="Екатеринбург, Несуществующая", deals_count=3)]
+    db, coord_updates, tried_updates = _make_db_mock(update_rowcount=3)
+
+    with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=None):
+        stats = await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    assert stats.geocode_failed == 1
+    assert stats.geocoded == 0
+    assert stats.processed == 1
+    assert stats.deals_updated == 0
+    # No coords UPDATE…
+    assert coord_updates == []
+    # …but geocode_tried_at IS stamped via the mark-tried UPDATE.
+    assert len(tried_updates) == 1
+    assert tried_updates[0]["addr"] == "Екатеринбург, Несуществующая"
+    assert db.commit.call_count == 1
+
+
+async def test_mark_deals_tried_sql_stamps_tried_at_only():
+    """The mark-tried writer sets geocode_tried_at and nothing else (no lat/lon)."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock(rowcount=4)
+    n = _mark_deals_tried(db, address="Екатеринбург, X")
+    args, _kw = db.execute.call_args
+    sql_str = str(args[0])
+    assert "UPDATE deals" in sql_str
+    assert "geocode_tried_at = NOW()" in sql_str
+    assert "SET lat" not in sql_str
+    assert "lat IS NULL" in sql_str  # idempotency / no-clobber guard
+    assert n == 4
+
+
+async def test_run_backfill_geocode_raises_not_stamped():
+    """A raised geocode error → counted failed but NOT stamped (retry sooner)."""
+    groups = [AddressGroup(address="Екатеринбург, Сетевая", deals_count=1)]
+    db, coord_updates, tried_updates = _make_db_mock()
+
+    with patch(_GEOCODE_PATH, new_callable=AsyncMock, side_effect=RuntimeError("network")):
+        stats = await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    assert stats.geocode_failed == 1
+    assert stats.geocoded == 0
+    # Neither a coords UPDATE nor a mark-tried UPDATE — leave it for next run.
+    assert coord_updates == []
+    assert tried_updates == []
+    assert db.commit.call_count == 0
+
+
+# ---------------------------------------------------------------------------
+# --dry-run — no DB writes
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_dry_run_issues_no_update():
+    groups = [
+        AddressGroup(address="Екатеринбург, Тестовая", deals_count=4),
+        AddressGroup(address="Екатеринбург, Пустая", deals_count=1),
+    ]
+    db, coord_updates, tried_updates = _make_db_mock()
+
+    # First address resolves, second is a miss — neither must write in dry-run.
+    with patch(
+        _GEOCODE_PATH,
+        new_callable=AsyncMock,
+        side_effect=[_result("nominatim"), None],
+    ):
+        stats = await _run_backfill(db, groups, batch="dry", dry_run=True)
+
+    assert stats.geocoded == 1
+    assert stats.geocode_failed == 1
+    assert stats.processed == 2
+    assert coord_updates == []
+    assert tried_updates == []
+    assert db.commit.call_count == 0
+
+
+# ---------------------------------------------------------------------------
+# dedup — one geocode call per distinct address
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_one_call_per_distinct_address():
+    groups = [
+        AddressGroup(address="Екатеринбург, А", deals_count=10),
+        AddressGroup(address="Екатеринбург, Б", deals_count=2),
+    ]
+    db, coord_updates, _ = _make_db_mock(update_rowcount=1)
+
+    with patch(
+        _GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")
+    ) as mock_geo:
+        stats = await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    # 2 distinct addresses → exactly 2 geocode calls (not 12 = sum of deals).
+    assert mock_geo.call_count == 2
+    assert stats.geocoded == 2
+    assert len(coord_updates) == 2
+
+
+async def test_run_backfill_skips_too_short_address_without_geocoding():
+    groups = [AddressGroup(address="ек", deals_count=1)]  # < 3 chars
+    db, coord_updates, _ = _make_db_mock()
+
+    with patch(_GEOCODE_PATH, new_callable=AsyncMock) as mock_geo:
+        stats = await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    assert stats.skipped == 1
+    assert stats.processed == 0
+    mock_geo.assert_not_called()
+    assert coord_updates == []
+
+
+# ---------------------------------------------------------------------------
+# per-address SAVEPOINT — one bad UPDATE doesn't abort the batch
+# ---------------------------------------------------------------------------
+
+
+async def test_run_backfill_db_write_failure_isolated_to_address():
+    groups = [
+        AddressGroup(address="Екатеринбург, Первая", deals_count=1),
+        AddressGroup(address="Екатеринбург, Вторая", deals_count=1),
+    ]
+    db, coord_updates, _ = _make_db_mock(update_rowcount=1)
+
+    # Make the FIRST coords UPDATE raise, the rest succeed.
+    call = {"n": 0}
+    real_side_effect = db.execute.side_effect
+
+    def failing_execute(sql, params=None):
+        sql_str = str(sql)
+        if "UPDATE deals" in sql_str and "SET lat" in sql_str:
+            call["n"] += 1
+            if call["n"] == 1:
+                raise RuntimeError("constraint blew up")
+        return real_side_effect(sql, params)
+
+    db.execute.side_effect = failing_execute
+
+    with patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")):
+        stats = await _run_backfill(db, groups, batch="b", dry_run=False)
+
+    # Both addresses geocoded; one write failed but was isolated via SAVEPOINT.
+    assert stats.geocoded == 2
+    assert db.rollback.call_count == 1
+    # Only the second address's coords UPDATE was recorded.
+    assert [u["addr"] for u in coord_updates] == ["Екатеринбург, Вторая"]
+    # deals_updated only counts the successful write.
+    assert stats.deals_updated == 1
+
+
+# ---------------------------------------------------------------------------
+# _update_deals_geocoded — bind shape + geom NOT set manually
+# ---------------------------------------------------------------------------
+
+
+def test_update_deals_geocoded_sets_lat_lon_tried_at_not_geom():
+    db = MagicMock()
+    db.execute.return_value = MagicMock(rowcount=7)
+    n = _update_deals_geocoded(db, address="Екатеринбург, Y", lat=56.1, lon=60.2)
+    args, _kw = db.execute.call_args
+    sql_str = str(args[0])
+    binds = args[1]
+    assert "UPDATE deals" in sql_str
+    assert "geocode_tried_at = NOW()" in sql_str
+    # geom must NOT be set manually — the deals_set_geom_trg trigger fills it.
+    assert "geom" not in sql_str
+    assert "lat IS NULL" in sql_str  # no-clobber guard
+    assert binds == {"addr": "Екатеринбург, Y", "lat": 56.1, "lon": 60.2}
+    assert n == 7
+
+
+# ---------------------------------------------------------------------------
+# resume filter — candidate SQL carries lat IS NULL + staleness predicate
+# ---------------------------------------------------------------------------
+
+
+def test_select_pending_addresses_filters_null_and_stale():
+    """The candidate query must combine lat IS NULL with the staleness window
+    and bind --stale-days, so already-tried-recently rows are excluded."""
+    db, _, _ = _make_db_mock(
+        address_rows=[{"address": "Екатеринбург, Z", "deals_count": 3}]
+    )
+
+    groups = _select_pending_addresses(db, limit=2000, stale_days=30)
+
+    assert groups == [AddressGroup(address="Екатеринбург, Z", deals_count=3)]
+    # Inspect the SQL + binds of the SELECT.
+    select_call = next(
+        c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
+    )
+    sql_str = str(select_call[0][0])
+    binds = select_call[0][1]
+    assert "lat IS NULL" in sql_str
+    assert "geocode_tried_at IS NULL" in sql_str
+    assert "make_interval(days => CAST(:stale_days AS int))" in sql_str
+    assert "GROUP BY address" in sql_str
+    assert binds["stale_days"] == 30
+    assert binds["limit"] == 2000
+
+
+def test_select_pending_addresses_passes_custom_stale_days():
+    db, _, _ = _make_db_mock(address_rows=[])
+    _select_pending_addresses(db, limit=500, stale_days=7)
+    select_call = next(
+        c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
+    )
+    binds = select_call[0][1]
+    assert binds["stale_days"] == 7
+    assert binds["limit"] == 500
+
+
+# ---------------------------------------------------------------------------
+# main() — wiring, --limit, --dry-run, return value
+# ---------------------------------------------------------------------------
+
+
+async def test_main_geocodes_and_returns_count():
+    address_rows = [
+        {"address": "Екатеринбург, Малышева", "deals_count": 4},
+        {"address": "Екатеринбург, Ленина", "deals_count": 2},
+    ]
+    db, coord_updates, _ = _make_db_mock(address_rows=address_rows, update_rowcount=4)
+
+    with (
+        patch("scripts.geocode_deals_nominatim.SessionLocal", return_value=db),
+        patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")),
+    ):
+        n = await main(["--batch", "test_main"])
+
+    assert n == 2  # both distinct addresses geocoded
+    assert len(coord_updates) == 2
+    assert {u["addr"] for u in coord_updates} == {
+        "Екатеринбург, Малышева",
+        "Екатеринбург, Ленина",
+    }
+
+
+async def test_main_dry_run_writes_nothing():
+    address_rows = [{"address": "Екатеринбург, Малышева", "deals_count": 3}]
+    db, coord_updates, tried_updates = _make_db_mock(
+        address_rows=address_rows, total_streets=1, total_rows=3
+    )
+
+    with (
+        patch("scripts.geocode_deals_nominatim.SessionLocal", return_value=db),
+        patch(_GEOCODE_PATH, new_callable=AsyncMock, return_value=_result("nominatim")),
+    ):
+        n = await main(["--dry-run"])
+
+    # dry-run returns the would-geocode count, writes nothing.
+    assert n == 1
+    assert coord_updates == []
+    assert tried_updates == []
+    assert db.commit.call_count == 0
+
+
+async def test_main_respects_limit_bind():
+    db, _, _ = _make_db_mock(address_rows=[])
+
+    with (
+        patch("scripts.geocode_deals_nominatim.SessionLocal", return_value=db),
+        patch(_GEOCODE_PATH, new_callable=AsyncMock),
+    ):
+        await main(["--limit", "50"])
+
+    select_call = next(
+        c for c in db.execute.call_args_list if "FROM deals" in str(c[0][0])
+    )
+    assert select_call[0][1]["limit"] == 50
+
+
+async def test_main_no_pending_returns_zero():
+    db, coord_updates, _ = _make_db_mock(address_rows=[])
+
+    with (
+        patch("scripts.geocode_deals_nominatim.SessionLocal", return_value=db),
+        patch(_GEOCODE_PATH, new_callable=AsyncMock) as mock_geo,
+    ):
+        n = await main([])
+
+    assert n == 0
+    assert coord_updates == []
+    mock_geo.assert_not_called()
+    db.close.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Stats dataclass
+# ---------------------------------------------------------------------------
+
+
+def test_stats_defaults_zero():
+    s = Stats()
+    assert s.processed == 0
+    assert s.geocoded == 0
+    assert s.geocode_failed == 0
+    assert s.skipped == 0
+    assert s.deals_updated == 0
+    assert s.cache_hits == 0
+    assert s.cache_misses == 0