tradein-mvp/backend/app/services/scrapers/base.py

@@ -30,6 +30,7 @@ from sqlalchemy.orm import Session
 from tenacity import retry, stop_after_attempt, wait_exponential
 
 from app.services.matching import match_or_create_house, upsert_listing_source
+from app.services.scrapers.snapshot_writer import upsert_listing_snapshot
 
 logger = logging.getLogger(__name__)
 
@@ -201,9 +202,22 @@ class BaseScraper(ABC):
 
 
 # ── Запись пачки результатов в Postgres ─────────────────────────────────────
-def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
+def save_listings(
+    db: Session,
+    lots: list[ScrapedLot],
+    *,
+    run_id: int | None = None,
+) -> tuple[int, int]:
     """Пишем list[ScrapedLot] в `listings` с upsert по dedup_hash.
 
+    Дополнительно пишет point-in-time snapshot в listings_snapshots (ON CONFLICT DO UPDATE
+    — берётся последний за день). Ошибка записи snapshot НЕ прерывает основной INSERT.
+
+    Args:
+        db: SQLAlchemy session.
+        lots: список распарсенных объявлений.
+        run_id: FK scrape_runs.id текущего run'а (опционально; None при ad-hoc вызовах).
+
     Returns:
         (inserted, updated) — counters для логов.
     """
@@ -326,6 +340,28 @@ def save_listings(db: Session, lots: list[ScrapedLot]) -> tuple[int, int]:
             else:
                 updated += 1
 
+        # ── Snapshot: point-in-time observation in listings_snapshots ───
+        # Fault-tolerant: failure here MUST NOT abort the listings batch.
+        # ON CONFLICT DO UPDATE — самый последний snapshot за день.
+        if listing_id is not None:
+            try:
+                with db.begin_nested():
+                    upsert_listing_snapshot(
+                        db,
+                        listing_id=listing_id,
+                        price_rub=lot.price_rub,
+                        price_per_m2=ppm2,
+                        run_id=run_id,
+                        status="active",
+                    )
+            except Exception as e:
+                logger.warning(
+                    "save_listings:snapshot_failed source=%s listing_id=%s: %s",
+                    lot.source,
+                    listing_id,
+                    e,
+                )
+
         # ── Hook: link listing → house via matching service ─────────────
         # Idempotent (UPSERT on listing_sources UNIQUE(ext_source, ext_id))
         # and fault-tolerant: failure here MUST NOT abort the listings batch.

tradein-mvp/backend/app/services/scrapers/cian_detail.py

@@ -21,6 +21,7 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.services.scrapers.cian_state_parser import extract_all_states, extract_state
+from app.services.scrapers.snapshot_writer import upsert_listing_snapshot
 
 logger = logging.getLogger(__name__)
 
@@ -248,6 +249,37 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
         },
     )
 
+    # Snapshot: point-in-time observation в listings_snapshots.
+    # Берём price_rub / price_per_m2 из listings строки (только что обновлённой).
+    # Используем COALESCE — на случай если listings.price_rub NULL (не должно быть, но safe).
+    _snap_row = db.execute(
+        text("""
+            SELECT price_rub, price_per_m2
+            FROM listings
+            WHERE id = CAST(:lid AS bigint)
+        """),
+        {"lid": listing_id},
+    ).fetchone()
+    _snap_written = False
+    if _snap_row is not None and _snap_row.price_rub is not None:
+        try:
+            with db.begin_nested():
+                upsert_listing_snapshot(
+                    db,
+                    listing_id=listing_id,
+                    price_rub=int(_snap_row.price_rub),
+                    price_per_m2=int(_snap_row.price_per_m2) if _snap_row.price_per_m2 else None,
+                    run_id=None,
+                    status="active",
+                )
+            _snap_written = True
+        except Exception as exc:
+            logger.warning(
+                "save_detail_enrichment: snapshot write failed listing_id=%s: %s",
+                listing_id,
+                exc,
+            )
+
     # Price changes — INSERT new rows, de-dup by listing_id + change_time
     for change in enrichment.price_changes:
         if not change.get("change_time") or not change.get("price_rub"):
@@ -263,7 +295,7 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
                     CAST(:diff AS numeric),
                     'cian'
                 )
-                ON CONFLICT DO NOTHING
+                ON CONFLICT ON CONSTRAINT offer_price_history_listing_change_uq DO NOTHING
             """),
             {
                 "lid": listing_id,
@@ -317,8 +349,10 @@ def save_detail_enrichment(db: Session, listing_id: int, enrichment: DetailEnric
 
     db.commit()
     logger.info(
-        "Cian detail enrichment saved for listing_id=%s (price_changes=%d, agent=%s)",
+        "Cian detail enrichment saved for listing_id=%s "
+        "(price_changes=%d, agent=%s, snapshot=%s)",
         listing_id,
         len(enrichment.price_changes),
         bool(enrichment.agent_profile),
+        _snap_written,
     )

tradein-mvp/backend/app/services/scrapers/snapshot_writer.py

@@ -0,0 +1,92 @@
+"""listings_snapshots writer — point-in-time observation per listing per day.
+
+Таблица listings_snapshots (016_listings_snapshots.sql):
+  PRIMARY KEY (listing_id, snapshot_date) — максимум 1 snapshot в сутки.
+  ON CONFLICT DO UPDATE — берём последний за день (перезаписываем при повторном run'е).
+
+Используется двумя путями:
+  1. SERP scrape (save_listings в base.py) — записывает цену + позицию в выдаче
+     каждый раз когда listing появляется в поиске.
+  2. Detail backfill (save_detail_enrichment в cian_detail.py) — записывает цену
+     из detail-страницы для listings которые раньше не имели snapshot'а.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import date
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+
+def upsert_listing_snapshot(
+    db: Session,
+    *,
+    listing_id: int,
+    price_rub: int,
+    price_per_m2: int | None = None,
+    run_id: int | None = None,
+    snapshot_date: date | None = None,
+    position_in_serp: int | None = None,
+    status: str | None = "active",
+) -> None:
+    """Записать / обновить snapshot для listing за текущий (или указанный) день.
+
+    Идемпотентно: повторный вызов с теми же listing_id + snapshot_date перезаписывает
+    строку (берём последние данные за день — ON CONFLICT DO UPDATE).
+
+    Args:
+        db: открытая SQLAlchemy session. Commit делает CALLER.
+        listing_id: PK из таблицы listings.
+        price_rub: текущая цена в рублях.
+        price_per_m2: цена за кв.м (вычисляется в caller'е, None если нет).
+        run_id: FK scrape_runs.id — с каким run'ом связан snapshot (None если backfill
+                запускается вне run'а).
+        snapshot_date: дата наблюдения; если None — используется CURRENT_DATE (по БД).
+        position_in_serp: позиция в SERP (1..60+), только при SERP-scrape.
+        status: 'active' / 'closed' / None. Default 'active' при обычном scrape.
+    """
+    db.execute(
+        text("""
+            INSERT INTO listings_snapshots
+                (listing_id, snapshot_date, run_id,
+                 price_rub, price_per_m2, position_in_serp, status, observed_at)
+            VALUES (
+                CAST(:lid AS bigint),
+                COALESCE(CAST(:snap_date AS date), CURRENT_DATE),
+                CAST(:run_id AS bigint),
+                CAST(:price AS bigint),
+                CAST(:ppm2 AS int),
+                CAST(:pos AS int),
+                :status,
+                NOW()
+            )
+            ON CONFLICT (listing_id, snapshot_date) DO UPDATE SET
+                run_id           = COALESCE(EXCLUDED.run_id, listings_snapshots.run_id),
+                price_rub        = EXCLUDED.price_rub,
+                price_per_m2     = COALESCE(EXCLUDED.price_per_m2, listings_snapshots.price_per_m2),
+                position_in_serp = COALESCE(
+                    EXCLUDED.position_in_serp, listings_snapshots.position_in_serp
+                ),
+                status           = COALESCE(EXCLUDED.status, listings_snapshots.status),
+                observed_at      = EXCLUDED.observed_at
+        """),
+        {
+            "lid": listing_id,
+            "snap_date": snapshot_date,
+            "run_id": run_id,
+            "price": price_rub,
+            "ppm2": price_per_m2,
+            "pos": position_in_serp,
+            "status": status,
+        },
+    )
+    logger.debug(
+        "upsert_listing_snapshot: listing_id=%s date=%s price=%s run_id=%s",
+        listing_id,
+        snapshot_date or "CURRENT_DATE",
+        price_rub,
+        run_id,
+    )

tradein-mvp/backend/tests/test_snapshot_writer.py

@@ -0,0 +1,470 @@
+"""Тесты для snapshot_writer.py и интеграции с save_detail_enrichment / save_listings.
+
+Цель: доказать что:
+1. upsert_listing_snapshot строит корректный INSERT INTO listings_snapshots.
+2. Повторный вызов с теми же listing_id + snapshot_date НЕ дублирует строку
+   (ON CONFLICT DO UPDATE обновляет observed_at).
+3. save_detail_enrichment вызывает snapshot-write после SELECT price из listings.
+4. save_listings вызывает snapshot-write для каждого сохранённого listing.
+5. Ошибка snapshot не прерывает основной INSERT в save_listings.
+
+Используем unittest.mock — без реальной БД.
+"""
+from __future__ import annotations
+
+from contextlib import contextmanager
+from datetime import date
+from unittest.mock import MagicMock, patch
+
+from app.services.scrapers.cian_detail import DetailEnrichment, save_detail_enrichment
+from app.services.scrapers.snapshot_writer import upsert_listing_snapshot
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+def _mock_db_simple() -> MagicMock:
+    """Минимальный mock db.execute — scalar None для всех вызовов."""
+    db = MagicMock()
+    db.execute.return_value = MagicMock()
+    db.execute.return_value.fetchone.return_value = None
+    db.execute.return_value.scalar_one_or_none.return_value = None
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+    return db
+
+
+def _get_all_sqls(db: MagicMock) -> list[str]:
+    """Собрать все SQL-строки из всех db.execute() вызовов."""
+    sqls = []
+    for call_args in db.execute.call_args_list:
+        if call_args.args:
+            sqls.append(str(call_args.args[0]))
+    return sqls
+
+
+def _get_all_params(db: MagicMock) -> list[dict]:
+    """Собрать все params dict из db.execute() вызовов."""
+    params_list = []
+    for call_args in db.execute.call_args_list:
+        if len(call_args.args) >= 2:
+            params_list.append(call_args.args[1])
+    return params_list
+
+
+def _find_snapshot_call(db: MagicMock) -> tuple[str, dict] | None:
+    """Найти первый db.execute вызов содержащий 'listings_snapshots'."""
+    for call_args in db.execute.call_args_list:
+        if not call_args.args:
+            continue
+        sql = str(call_args.args[0])
+        if "listings_snapshots" in sql:
+            params = call_args.args[1] if len(call_args.args) >= 2 else {}
+            return sql, params
+    return None
+
+
+# ── upsert_listing_snapshot — SQL shape ──────────────────────────────────────
+
+
+def test_upsert_snapshot_sql_contains_insert_and_on_conflict():
+    """SQL содержит INSERT INTO listings_snapshots и ON CONFLICT (listing_id, snapshot_date)."""
+    db = _mock_db_simple()
+    upsert_listing_snapshot(db, listing_id=42, price_rub=5_000_000)
+
+    sqls = _get_all_sqls(db)
+    assert len(sqls) == 1, "ровно один db.execute вызов"
+    sql = sqls[0]
+    assert "listings_snapshots" in sql
+    assert "ON CONFLICT" in sql
+    assert "DO UPDATE" in sql
+
+
+def test_upsert_snapshot_params_minimal():
+    """Минимальный вызов: только listing_id + price_rub — остальные None."""
+    db = _mock_db_simple()
+    upsert_listing_snapshot(db, listing_id=7, price_rub=3_000_000)
+
+    params = _get_all_params(db)
+    assert len(params) == 1
+    p = params[0]
+    assert p["lid"] == 7
+    assert p["price"] == 3_000_000
+    assert p["run_id"] is None
+    assert p["snap_date"] is None
+    assert p["ppm2"] is None
+    assert p["pos"] is None
+    assert p["status"] == "active"
+
+
+def test_upsert_snapshot_params_full():
+    """Полный вызов — все поля правильно попадают в params."""
+    db = _mock_db_simple()
+    snap_date = date(2026, 5, 1)
+    upsert_listing_snapshot(
+        db,
+        listing_id=100,
+        price_rub=7_200_000,
+        price_per_m2=130_000,
+        run_id=99,
+        snapshot_date=snap_date,
+        position_in_serp=3,
+        status="active",
+    )
+
+    params = _get_all_params(db)[0]
+    assert params["lid"] == 100
+    assert params["price"] == 7_200_000
+    assert params["ppm2"] == 130_000
+    assert params["run_id"] == 99
+    assert params["snap_date"] == snap_date
+    assert params["pos"] == 3
+    assert params["status"] == "active"
+
+
+def test_upsert_snapshot_no_commit():
+    """upsert_listing_snapshot НЕ вызывает db.commit() — commit делает caller."""
+    db = _mock_db_simple()
+    upsert_listing_snapshot(db, listing_id=1, price_rub=1_000_000)
+    db.commit.assert_not_called()
+
+
+# ── save_detail_enrichment интеграция ────────────────────────────────────────
+
+
+def _mock_db_detail(price_rub: int = 5_500_000, price_per_m2: int | None = 110_000) -> MagicMock:
+    """Mock для save_detail_enrichment: SELECT listings возвращает price_rub/price_per_m2."""
+    db = MagicMock()
+
+    # SELECT price_rub, price_per_m2 FROM listings
+    price_row = MagicMock()
+    price_row.price_rub = price_rub
+    price_row.price_per_m2 = price_per_m2
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+
+    # Разные вызовы db.execute() возвращают разное:
+    # 1. UPDATE listings → result.fetchone() = None (не используется)
+    # 2. SELECT price → result.fetchone() = price_row
+    # 3+. INSERT offer_price_history / agents → не важно
+    execute_results = []
+
+    def _execute_side_effect(sql, params=None):
+        sql_str = str(sql)
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = None
+        if "SELECT price_rub" in sql_str:
+            mock_result.fetchone.return_value = price_row
+        else:
+            mock_result.fetchone.return_value = None
+        execute_results.append((sql_str, params, mock_result))
+        return mock_result
+
+    db.execute.side_effect = _execute_side_effect
+    db._execute_log = execute_results
+    return db
+
+
+def test_save_detail_enrichment_writes_snapshot():
+    """save_detail_enrichment вызывает upsert_listing_snapshot через SELECT listings."""
+    db = _mock_db_detail(price_rub=6_000_000, price_per_m2=120_000)
+    enrichment = DetailEnrichment(ceiling_height=2.7)
+
+    save_detail_enrichment(db, 55, enrichment)
+
+    snap = _find_snapshot_call(db)
+    assert snap is not None, "должен быть INSERT INTO listings_snapshots"
+    _sql, params = snap
+    assert params["lid"] == 55
+    assert params["price"] == 6_000_000
+    assert params["ppm2"] == 120_000
+    assert params["status"] == "active"
+
+
+def test_save_detail_enrichment_snapshot_skipped_when_no_price_row():
+    """Если listings строка не найдена (None) — snapshot не записывается, ошибки нет."""
+    db = MagicMock()
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+
+    def _exec(sql, params=None):
+        mock_result = MagicMock()
+        mock_result.fetchone.return_value = None
+        mock_result.scalar_one_or_none.return_value = None
+        return mock_result
+
+    db.execute.side_effect = _exec
+
+    enrichment = DetailEnrichment()
+    # Должно завершиться без исключений
+    save_detail_enrichment(db, 99, enrichment)
+
+    # Убедимся что listings_snapshots не упоминается
+    sqls = _get_all_sqls(db)
+    assert not any("listings_snapshots" in s for s in sqls)
+
+
+def test_save_detail_enrichment_snapshot_failure_does_not_abort():
+    """Если upsert_listing_snapshot падает — save_detail_enrichment всё равно commit'ит."""
+    db = _mock_db_detail(price_rub=4_000_000)
+
+    with patch(
+        "app.services.scrapers.cian_detail.upsert_listing_snapshot",
+        side_effect=RuntimeError("db snapshot blip"),
+    ):
+        enrichment = DetailEnrichment(ceiling_height=2.5)
+        # Должно завершиться без исключений — snapshot error логируется, не пробрасывается
+        save_detail_enrichment(db, 10, enrichment)
+
+    db.commit.assert_called_once()
+
+
+def test_save_detail_enrichment_writes_offer_price_history_rows():
+    """price_changes из enrichment записываются в offer_price_history."""
+    db = _mock_db_detail(price_rub=5_000_000)
+    enrichment = DetailEnrichment(
+        price_changes=[
+            {"change_time": "2026-04-01T00:00:00Z", "price_rub": 5_200_000, "diff_percent": None},
+            {"change_time": "2026-04-10T00:00:00Z", "price_rub": 5_000_000, "diff_percent": -3.8},
+        ]
+    )
+
+    save_detail_enrichment(db, 20, enrichment)
+
+    sqls = _get_all_sqls(db)
+    oph_inserts = [s for s in sqls if "offer_price_history" in s]
+    assert len(oph_inserts) == 2, "по одному INSERT для каждой price_change"
+
+
+def test_save_detail_enrichment_oph_on_conflict_constraint():
+    """INSERT offer_price_history использует именованный ON CONFLICT ON CONSTRAINT."""
+    db = _mock_db_detail(price_rub=5_000_000)
+    enrichment = DetailEnrichment(
+        price_changes=[
+            {"change_time": "2026-05-01T12:00:00Z", "price_rub": 5_000_000, "diff_percent": 0},
+        ]
+    )
+
+    save_detail_enrichment(db, 21, enrichment)
+
+    sqls = _get_all_sqls(db)
+    oph_sqls = [s for s in sqls if "offer_price_history" in s]
+    assert oph_sqls, "INSERT offer_price_history не найден"
+    assert "offer_price_history_listing_change_uq" in oph_sqls[0], (
+        "ON CONFLICT должен ссылаться на UNIQUE constraint"
+    )
+
+
+def test_save_detail_enrichment_skips_price_change_without_change_time():
+    """price_changes без change_time пропускаются, не вызывают INSERT."""
+    db = _mock_db_detail(price_rub=5_000_000)
+    enrichment = DetailEnrichment(
+        price_changes=[
+            {"change_time": None, "price_rub": 5_000_000},       # пропустить
+            {"change_time": "2026-04-01T00:00:00Z", "price_rub": None},  # пропустить
+            {"change_time": "2026-04-01T00:00:00Z", "price_rub": 5_000_000},  # записать
+        ]
+    )
+
+    save_detail_enrichment(db, 22, enrichment)
+
+    sqls = _get_all_sqls(db)
+    oph_inserts = [s for s in sqls if "offer_price_history" in s]
+    assert len(oph_inserts) == 1
+
+
+# ── save_listings интеграция ─────────────────────────────────────────────────
+
+
+def _mock_db_for_save_listings(
+    inserted: bool = True,
+    listing_id: int = 42,
+) -> MagicMock:
+    """Mock db для save_listings с правильным RETURNING + begin_nested."""
+    row = MagicMock()
+    row.id = listing_id
+    row.inserted = inserted
+
+    ins_result = MagicMock()
+    ins_result.fetchone.return_value = row
+
+    db = MagicMock()
+    db.execute.return_value = ins_result
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+    return db
+
+
+def test_save_listings_writes_snapshot_per_lot():
+    """save_listings вызывает upsert_listing_snapshot для каждого сохранённого lot."""
+    from app.services.scrapers.base import ScrapedLot, save_listings
+
+    db = _mock_db_for_save_listings(inserted=True, listing_id=77)
+    lots = [
+        ScrapedLot(
+            source="cian",
+            source_url="https://ekb.cian.ru/sale/flat/111/",
+            source_id="111",
+            price_rub=5_000_000,
+        ),
+        ScrapedLot(
+            source="cian",
+            source_url="https://ekb.cian.ru/sale/flat/222/",
+            source_id="222",
+            price_rub=6_000_000,
+        ),
+    ]
+
+    with (
+        patch("app.services.scrapers.base.match_or_create_house", return_value=(1, 1.0, "new")),
+        patch("app.services.scrapers.base.upsert_listing_source"),
+        patch("app.services.scrapers.base.upsert_listing_snapshot") as mock_snap,
+    ):
+        save_listings(db, lots, run_id=10)
+
+    assert mock_snap.call_count == 2, "snapshot должен вызваться для каждого lot"
+    # Первый вызов
+    kwargs0 = mock_snap.call_args_list[0].kwargs
+    assert kwargs0["listing_id"] == 77
+    assert kwargs0["price_rub"] == 5_000_000
+    assert kwargs0["run_id"] == 10
+    assert kwargs0["status"] == "active"
+
+
+def test_save_listings_snapshot_receives_run_id():
+    """Если передан run_id — он пробрасывается в upsert_listing_snapshot."""
+    from app.services.scrapers.base import ScrapedLot, save_listings
+
+    db = _mock_db_for_save_listings(listing_id=88)
+    lot = ScrapedLot(
+        source="avito",
+        source_url="https://avito.ru/1",
+        source_id="1",
+        price_rub=3_000_000,
+    )
+
+    with (
+        patch("app.services.scrapers.base.match_or_create_house", return_value=(1, 1.0, "new")),
+        patch("app.services.scrapers.base.upsert_listing_source"),
+        patch("app.services.scrapers.base.upsert_listing_snapshot") as mock_snap,
+    ):
+        save_listings(db, [lot], run_id=55)
+
+    kwargs = mock_snap.call_args.kwargs
+    assert kwargs["run_id"] == 55
+
+
+def test_save_listings_snapshot_none_run_id_by_default():
+    """Без run_id — upsert_listing_snapshot получает run_id=None."""
+    from app.services.scrapers.base import ScrapedLot, save_listings
+
+    db = _mock_db_for_save_listings(listing_id=9)
+    lot = ScrapedLot(
+        source="cian",
+        source_url="https://ekb.cian.ru/sale/flat/9/",
+        source_id="9",
+        price_rub=2_000_000,
+    )
+
+    with (
+        patch("app.services.scrapers.base.match_or_create_house", return_value=(1, 1.0, "new")),
+        patch("app.services.scrapers.base.upsert_listing_source"),
+        patch("app.services.scrapers.base.upsert_listing_snapshot") as mock_snap,
+    ):
+        save_listings(db, [lot])  # no run_id
+
+    kwargs = mock_snap.call_args.kwargs
+    assert kwargs["run_id"] is None
+
+
+def test_save_listings_snapshot_failure_does_not_abort_batch():
+    """Ошибка snapshot не прерывает основной INSERT; все lots сохраняются."""
+    from app.services.scrapers.base import ScrapedLot, save_listings
+
+    db = _mock_db_for_save_listings(inserted=True, listing_id=5)
+    lots = [
+        ScrapedLot(
+            source="cian",
+            source_url=f"https://ekb.cian.ru/sale/flat/{i}/",
+            source_id=str(i),
+            price_rub=1_000_000 * i,
+        )
+        for i in range(1, 4)
+    ]
+
+    with (
+        patch("app.services.scrapers.base.match_or_create_house", return_value=(1, 1.0, "new")),
+        patch("app.services.scrapers.base.upsert_listing_source"),
+        patch(
+            "app.services.scrapers.base.upsert_listing_snapshot",
+            side_effect=RuntimeError("snapshot fail"),
+        ),
+    ):
+        inserted, _updated = save_listings(db, lots)
+
+    # INSERT listings всё равно прошёл для всех 3 lots
+    assert inserted == 3
+
+
+def test_save_listings_empty_no_snapshot_call():
+    """Пустой список → upsert_listing_snapshot не вызывается."""
+    from app.services.scrapers.base import save_listings
+
+    db = _mock_db_for_save_listings()
+
+    with patch("app.services.scrapers.base.upsert_listing_snapshot") as mock_snap:
+        save_listings(db, [])
+
+    mock_snap.assert_not_called()
+
+
+# ── Идемпотентность: повторный вызов тот же listing_id + дата ────────────────
+
+
+def test_upsert_snapshot_idempotent_same_params_no_error():
+    """Два одинаковых вызова (listing_id + date совпадают) — нет ошибок.
+
+    Реальная де-дупликация — ON CONFLICT DO UPDATE в БД.
+    Здесь проверяем что Python-код не падает при повторном вызове с теми же данными.
+    """
+    db = _mock_db_simple()
+    snap_date = date(2026, 5, 28)
+
+    upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=snap_date)
+    upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=snap_date)
+
+    # Два вызова db.execute — каждый попытается INSERT ON CONFLICT DO UPDATE
+    assert db.execute.call_count == 2
+    # Убедимся что оба SQL содержат ON CONFLICT DO UPDATE
+    sqls = _get_all_sqls(db)
+    assert all("ON CONFLICT" in s for s in sqls)
+
+
+def test_upsert_snapshot_different_dates_no_conflict():
+    """Разные даты для одного listing → два отдельных snapshot (нет конфликта по PK)."""
+    db = _mock_db_simple()
+
+    upsert_listing_snapshot(db, listing_id=1, price_rub=5_000_000, snapshot_date=date(2026, 5, 1))
+    upsert_listing_snapshot(db, listing_id=1, price_rub=4_800_000, snapshot_date=date(2026, 5, 2))
+
+    # Каждый вызов делает INSERT: разные snapshot_date → разные PRIMARY KEY → нет конфликта
+    params_list = _get_all_params(db)
+    assert params_list[0]["snap_date"] == date(2026, 5, 1)
+    assert params_list[1]["snap_date"] == date(2026, 5, 2)
+    assert params_list[0]["price"] == 5_000_000
+    assert params_list[1]["price"] == 4_800_000