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

@@ -14,6 +14,7 @@ Idempotent-логика (на каждый yandex-lot):
   - история ПУСТА:
       * price_previous задан И != price_rub → seed previous-точку (change_time = now-1d,
         timestamp синтетический — gate реального времени смены не несёт), затем current-точку;
+        НО previous, отличающаяся от текущей цены на порядок и больше, не сеется (#3385);
       * иначе → только current-точку (now).
   - история ЕСТЬ и latest_price != price_rub → одна current-точка (now) — зафиксировали смену.
   - история ЕСТЬ и latest_price == price_rub → skip (цена не менялась).
@@ -41,6 +42,14 @@ logger = logging.getLogger(__name__)
 
 _SOURCE = "yandex"
 
+# #3385: price.previous, отличающаяся от текущей цены в 9.5 раза и больше (в любую
+# сторону), — не рынок, а потерянный разряд или цена в тысячах. Прод 17.09.2026, все
+# 8313 посевов yandex: самый большой честный скачок ×2.13, дальше пусто до ×9.74; выше
+# порога 11 посевов (×10 — 7, ×175…×1000 — 4), и у всех нижняя цена 0.1–37 тыс/м².
+# Посев синтетический (now-1d, реального времени смены gate не несёт), без него
+# теряется только фиктивная точка — текущая цена пишется всё равно.
+_SEED_IMPLAUSIBLE_RATIO = 9.5
+
 # Резолв listing_id по стабильному (source, source_id) — у Yandex source_id = offer_id.
 _RESOLVE_SQL = text(
     """
@@ -96,6 +105,12 @@ def _insert_point(
     )
 
 
+def _implausible_seed(previous: float, current: float) -> bool:
+    """previous и current разнятся на порядок и больше (или одна из цен не > 0)."""
+    low, high = sorted((float(previous), float(current)))
+    return low <= 0 or high / low >= _SEED_IMPLAUSIBLE_RATIO
+
+
 def record_yandex_price_history(db: Session, lots: list[ScrapedLot]) -> int:
     """Дописать offer_price_history по yandex-lots из gate price.previous/trend.
 
@@ -115,6 +130,7 @@ def record_yandex_price_history(db: Session, lots: list[ScrapedLot]) -> int:
 
     inserted = 0
     skipped = 0
+    seed_rejected = 0
     errors = 0
 
     for lot in lots:
@@ -140,20 +156,29 @@ def record_yandex_price_history(db: Session, lots: list[ScrapedLot]) -> int:
                     {"listing_id": listing_id},
                 ).fetchone()
 
-                # ПОТОЛОК: гейта сдвига разряда (#3376) здесь НЕТ, и он бы тут не
+                # Сдвиг разряда у yandex ловится ТОЛЬКО на посеве (#3385). Ветка «история
-                # сработал. drop_decimal_slips требует двух свидетелей — скачка ×10
+                # есть» на проде не записала ни одной строки (из 28 558 строк этого
-                # к предыдущей точке и подтверждения у следующей. На этом пути серия
+                # писателя на 17.09): смену listings.price_rub раньше неё пишет триггер
-                # максимум из двух точек (последняя лежащая в БД + текущая), у
+                # record_listing_price_change (save_listings идёт первым), и сюда
-                # последней точки свидетель — текущая цена лота, а она и ЕСТЬ эта
+                # приходит latest == lot.price_rub. Все 7 скачков ×10 в yandex-истории
-                # точка: свидетель совпадает с подозреваемым, отношение всегда 1.0.
+                # — пары «посев из price.previous → текущая цена», ошибочна нижняя.
-                # То есть проводка была бы декорацией: ветка, которая по построению
-                # не может выбросить ни одной точки. Отлов ×10 у yandex требует
-                # другого механизма — сравнения со СЛЕДУЮЩИМ наблюдением, а значит
-                # DELETE уже вставленной строки. Отдельная задача: #3385.
                 if latest is None:
                     # История пуста — seed (+ опционально previous-точка).
                     prev = lot.price_previous_rub
-                    if prev is not None and prev != lot.price_rub:
+                    if (
+                        prev is not None
+                        and prev != lot.price_rub
+                        and _implausible_seed(prev, lot.price_rub)
+                    ):
+                        seed_rejected += 1
+                        logger.warning(
+                            "yandex_price_history: price.previous=%s не сеется — на порядок "
+                            "от текущей %s (source_id=%s)",
+                            prev,
+                            lot.price_rub,
+                            lot.source_id,
+                        )
+                    elif prev is not None and prev != lot.price_rub:
                         _insert_point(
                             db,
                             listing_id=listing_id,
@@ -191,10 +216,11 @@ def record_yandex_price_history(db: Session, lots: list[ScrapedLot]) -> int:
 
     db.commit()
     logger.info(
-        "yandex_price_history: lots=%d inserted=%d skipped=%d errors=%d",
+        "yandex_price_history: lots=%d inserted=%d skipped=%d seed_rejected=%d errors=%d",
         len(lots),
         inserted,
         skipped,
+        seed_rejected,
         errors,
     )
     return inserted

tradein-mvp/backend/data/sql/310_yandex_seed_decimal_slips.sql

@@ -0,0 +1,70 @@
+-- 310_yandex_seed_decimal_slips.sql
+-- Удалить из offer_price_history yandex-посевы, отличающиеся от следующей точки на
+-- порядок и больше (#3385).
+--
+-- Apply after: 307_domclick_msk_window_no_collision.sql
+--
+-- ЧТО НЕ ТАК
+--   services/yandex_price_history.py при пустой истории сеет точку из gate-поля
+--   price.previous: change_time = now − 1 сутки (синтетическое время), следом текущая
+--   цена лота с change_time = now; обе строки одним INSERT-батчем, поэтому recorded_at у
+--   них совпадает. У 11 объявлений price.previous — не рынок: 1 370 000 → 13 700 000
+--   (87.8 м², нижняя цена 15.6 тыс/м²), 10 900 → 10 900 000 (цена в тысячах). Замер
+--   прода 17.09.2026 по всем 8313 yandex-посевам: ×10 — 7 (одно из них ×9.74), ×175…×1000
+--   — 4, «вниз» — 0; самый большой скачок ниже порога — ×2.13, между ×2.13 и ×9.74 нет
+--   ни одного. Ошибочна именно НИЖНЯЯ, посевная точка: у нижних 0.1–37 тыс/м², у верхних
+--   118–375 тыс/м²; верхняя совпадает с listings.price_rub у 10 из 11, а 10387563 дальше
+--   дешевеет от неё же (14M → 13M → 12M, триггерные строки).
+--
+--   Код починен в том же PR: посев на порядок и больше больше не пишется. Эта миграция —
+--   задним числом по уже записанным строкам.
+--
+-- КРИТЕРИЙ — ПОДПИСЬ ПОСЕВА, А НЕ ТОЛЬКО ОТНОШЕНИЕ
+--   Удаляется строка o, только если у того же объявления есть строка s, записанная тем же
+--   батчем (s.recorded_at = o.recorded_at) ровно через 24 часа (s.change_time =
+--   o.change_time + 24 hours), и цены разнятся в 9.5 раза и больше. Так посевы пишет
+--   только этот писатель; триггерные строки (change_time = recorded_at) и строки других
+--   источников условием не задеваются. Интервал '24 hours', а не '1 day': прибавление
+--   суток к timestamptz зависит от таймзоны сессии, а Python считает ровно 24 часа.
+--   diff_percent у второй строки не трогаем: писатель его не заполняет, у всех 11 он NULL,
+--   и после удаления посева вторая строка становится первой — NULL там и есть верное.
+--
+-- СКОЛЬКО ЖДЁМ
+--   11 строк на 17.09.2026 (7 × 10 и 4 × 175…1000). До деплоя кода новые посевы ещё
+--   могут появиться (06.09 → 17.09 прибавился один). Порог остановки 30: поймали больше —
+--   критерий ловит не то, миграция падает, а не молча чистит историю.
+--
+-- ИДЕМПОТЕНТНОСТЬ
+--   Удалённый посев второй раз не находится, у оставшейся строки пары больше нет →
+--   повторный прогон удаляет 0. Новых объектов схемы нет.
+BEGIN;
+-- Конвенция проекта (#2752): DELETE берёт блокировки строк offer_price_history; без
+-- lock_timeout деплой встанет в очередь за чужой сессией.
+SET LOCAL lock_timeout = '5s';
+
+DO $$
+DECLARE
+    deleted_rows bigint;
+BEGIN
+    DELETE FROM offer_price_history o
+     USING offer_price_history s
+     WHERE o.source = 'yandex'
+       AND s.source = 'yandex'
+       AND s.listing_id = o.listing_id
+       AND s.recorded_at = o.recorded_at
+       AND s.change_time = o.change_time + interval '24 hours'
+       AND o.price_rub > 0
+       AND s.price_rub > 0
+       AND greatest(o.price_rub, s.price_rub) / least(o.price_rub, s.price_rub) >= 9.5;
+    GET DIAGNOSTICS deleted_rows = ROW_COUNT;
+    RAISE NOTICE 'offer_price_history: удалено yandex-посевов на порядок = %', deleted_rows;
+
+    -- Исключение откатывает всю транзакцию вместе с DELETE выше.
+    IF deleted_rows > 30 THEN
+        RAISE EXCEPTION
+            'offer_price_history: к удалению % yandex-посевов при замере 17.09.2026 = 11 — '
+            'критерий ловит не то, миграция остановлена', deleted_rows;
+    END IF;
+END $$;
+
+COMMIT;

tradein-mvp/backend/tests/skip_allowlist.txt

@@ -146,3 +146,9 @@ tests/test_3463_db_timeouts.py::test_statement_over_ceiling_is_cancelled_not_hun
 tests/test_3463_db_timeouts.py::test_lock_wait_over_ceiling_is_aborted
 tests/test_3463_db_timeouts.py::test_set_local_statement_timeout_overrides_session_ceiling
 tests/test_3463_db_timeouts.py::test_set_local_is_scoped_to_its_transaction
+
+# Миграция 310 (#3385): настоящий SQL-файл с DELETE и RAISE-остановкой исполняет только
+# Postgres. В ci-tradein.yml бегут по-настоящему (postgres-сервис, #2745); краснеют от
+# снятия подписи батча, сужения окна до ×10 и снятия порога — проверено вручную 17.09.
+tests/test_3385_migration_310_yandex_seed_slips.py::test_deletes_only_seed_off_by_order_and_is_idempotent
+tests/test_3385_migration_310_yandex_seed_slips.py::test_stops_and_rolls_back_when_too_many

tradein-mvp/backend/tests/test_3385_migration_310_yandex_seed_slips.py

@@ -0,0 +1,150 @@
+"""Миграция 310 (#3385): yandex-посев на порядок от следующей точки удаляется, остальное нет.
+
+Прогон НАСТОЯЩЕГО файла миграции на живом Postgres. Реальную таблицу не трогаем:
+временная offer_price_history той же формы затеняет её в этой сессии (pg_temp первым
+в search_path), а миграция обращается к таблице без схемы. Без БД — skip (в CI Trade-In
+есть postgres-сервис).
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+_MIGRATION = (
+    Path(__file__).resolve().parents[1] / "data" / "sql" / "310_yandex_seed_decimal_slips.sql"
+)
+
+_T0 = datetime(2026, 7, 4, 16, 48, 35, 71486, tzinfo=UTC)
+_DAY = timedelta(hours=24)
+
+
+def _live_conn() -> Any | None:
+    dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
+    if not dsn or "localhost:5432/test" in dsn:
+        return None
+    try:
+        import psycopg
+
+        return psycopg.connect(
+            dsn.replace("postgresql+psycopg://", "postgresql://"),
+            autocommit=True,
+            connect_timeout=3,
+        )
+    except Exception:
+        return None
+
+
+@pytest.fixture()
+def conn() -> Any:
+    c = _live_conn()
+    if c is None:
+        pytest.skip("no reachable Postgres test DB")
+    c.execute(
+        """
+        CREATE TEMP TABLE offer_price_history (
+            id           bigserial PRIMARY KEY,
+            listing_id   bigint NOT NULL,
+            change_time  timestamptz NOT NULL,
+            price_rub    numeric NOT NULL,
+            diff_percent numeric(8, 2),
+            source       text,
+            recorded_at  timestamptz NOT NULL DEFAULT now()
+        )
+        """
+    )
+    try:
+        yield c
+    finally:
+        c.close()
+
+
+def _add(
+    c: Any,
+    listing_id: int,
+    change_time: datetime,
+    price: int,
+    *,
+    recorded_at: datetime,
+    source: str = "yandex",
+) -> None:
+    c.execute(
+        "INSERT INTO offer_price_history (listing_id, change_time, price_rub, source, recorded_at) "
+        "VALUES (%s, %s, %s, %s, %s)",
+        (listing_id, change_time, price, source, recorded_at),
+    )
+
+
+def _seed_pair(
+    c: Any, listing_id: int, previous: int, current: int, *, source: str = "yandex"
+) -> None:
+    """Ровно так пишет посев yandex_price_history: now−24ч и now, recorded_at один на батч."""
+    recorded = _T0 + _DAY + timedelta(microseconds=1404)
+    _add(c, listing_id, _T0, previous, recorded_at=recorded, source=source)
+    _add(c, listing_id, _T0 + _DAY, current, recorded_at=recorded, source=source)
+
+
+def _rows(c: Any) -> list[tuple[int, int]]:
+    return [
+        (int(lid), int(price))
+        for lid, price in c.execute(
+            "SELECT listing_id, price_rub FROM offer_price_history "
+            "ORDER BY listing_id, change_time, id"
+        ).fetchall()
+    ]
+
+
+def _migrate(c: Any) -> None:
+    c.execute(_MIGRATION.read_text(encoding="utf-8"))
+
+
+def test_deletes_only_seed_off_by_order_and_is_idempotent(conn: Any) -> None:
+    _seed_pair(conn, 1, 1_370_000, 13_700_000)  # прод 3380975: ×10
+    later = _T0 + 20 * _DAY
+    _add(conn, 1, later, 13_000_000, recorded_at=later)  # триггерная строка того же объявления
+    _seed_pair(conn, 2, 10_900, 10_900_000)  # прод 1879742: ×1000
+    _seed_pair(conn, 3, 13_700_000, 1_370_000)  # ÷10
+    _seed_pair(conn, 4, 8_225_550, 17_500_000)  # прод 10359546: ×2.13, честный посев
+    _seed_pair(conn, 5, 1_450_000, 13_700_000)  # ×9.45 — ниже порога
+    _seed_pair(conn, 6, 1_370_000, 13_700_000, source="cian")  # не yandex
+    # ×10, но триггерные строки (change_time = recorded_at) — это наблюдения, не посев.
+    _add(conn, 7, _T0, 1_000_000, recorded_at=_T0)
+    _add(conn, 7, _T0 + _DAY, 10_000_000, recorded_at=_T0 + _DAY)
+
+    expected = [
+        (1, 13_700_000),
+        (1, 13_000_000),
+        (2, 10_900_000),
+        (3, 1_370_000),
+        (4, 8_225_550),
+        (4, 17_500_000),
+        (5, 1_450_000),
+        (5, 13_700_000),
+        (6, 1_370_000),
+        (6, 13_700_000),
+        (7, 1_000_000),
+        (7, 10_000_000),
+    ]
+    _migrate(conn)
+    assert _rows(conn) == expected
+    _migrate(conn)
+    assert _rows(conn) == expected
+
+
+def test_stops_and_rolls_back_when_too_many(conn: Any) -> None:
+    for listing_id in range(1, 32):
+        _seed_pair(conn, listing_id, 1_000_000, 10_000_000)
+    before = _rows(conn)
+    assert len(before) == 62
+
+    with pytest.raises(Exception, match="миграция остановлена"):
+        _migrate(conn)
+    conn.execute("ROLLBACK")
+    assert _rows(conn) == before

tradein-mvp/backend/tests/test_yandex_price_history.py

@@ -163,6 +163,46 @@ def test_first_capture_previous_equals_current_inserts_one_point(db: Session) ->
     assert len(_history(db, lid)) == 1
 
 
+@pytest.mark.parametrize(
+    ("previous", "current"),
+    [
+        (1_370_000, 13_700_000),  # прод 3380975: ×10, 15.6 тыс/м² у нижней
+        (972_130, 9_472_130),  # прод 10365196: ×9.74 — сдвиг вместе с мелкой сменой
+        (10_900, 10_900_000),  # прод 1879742: ×1000, цена в тысячах
+        (13_700_000, 1_370_000),  # ÷10 — синтетическую точку на порядок не сеем и вниз
+    ],
+)
+def test_first_capture_previous_off_by_order_is_not_seeded(
+    db: Session, previous: int, current: int
+) -> None:
+    """#3385: price.previous на порядок от текущей → в истории одна текущая точка."""
+    lid = _seed_listing(db, source="yandex", source_id="3385")
+    n = yph.record_yandex_price_history(
+        db, [_lot(source_id="3385", price_rub=current, price_previous_rub=previous)]
+    )
+    assert n == 1
+    assert [price for price, _ in _history(db, lid)] == [float(current)]
+
+
+@pytest.mark.parametrize(
+    ("previous", "current"),
+    [
+        (12_900_000, 13_700_000),
+        (8_225_550, 17_500_000),  # прод 10359546: ×2.13 — самый большой честный посев
+        (1_450_000, 13_700_000),  # ×9.45 — ниже порога, посев остаётся
+    ],
+)
+def test_first_capture_previous_within_order_is_seeded(
+    db: Session, previous: int, current: int
+) -> None:
+    lid = _seed_listing(db, source="yandex", source_id="3385b")
+    n = yph.record_yandex_price_history(
+        db, [_lot(source_id="3385b", price_rub=current, price_previous_rub=previous)]
+    )
+    assert n == 2
+    assert [price for price, _ in _history(db, lid)] == [float(previous), float(current)]
+
+
 def test_recapture_same_price_is_noop(db: Session) -> None:
     """Повторный захват с той же ценой → 0 строк (idempotent)."""
     _seed_listing(db, source="yandex", source_id="333")