diff --git a/tradein-mvp/backend/app/services/yandex_price_history.py b/tradein-mvp/backend/app/services/yandex_price_history.py index 87f74ac0..64ba3972 100644 --- a/tradein-mvp/backend/app/services/yandex_price_history.py +++ b/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) здесь НЕТ, и он бы тут не - # сработал. drop_decimal_slips требует двух свидетелей — скачка ×10 - # к предыдущей точке и подтверждения у следующей. На этом пути серия - # максимум из двух точек (последняя лежащая в БД + текущая), у - # последней точки свидетель — текущая цена лота, а она и ЕСТЬ эта - # точка: свидетель совпадает с подозреваемым, отношение всегда 1.0. - # То есть проводка была бы декорацией: ветка, которая по построению - # не может выбросить ни одной точки. Отлов ×10 у yandex требует - # другого механизма — сравнения со СЛЕДУЮЩИМ наблюдением, а значит - # DELETE уже вставленной строки. Отдельная задача: #3385. + # Сдвиг разряда у yandex ловится ТОЛЬКО на посеве (#3385). Ветка «история + # есть» на проде не записала ни одной строки (из 28 558 строк этого + # писателя на 17.09): смену listings.price_rub раньше неё пишет триггер + # record_listing_price_change (save_listings идёт первым), и сюда + # приходит latest == lot.price_rub. Все 7 скачков ×10 в yandex-истории + # — пары «посев из price.previous → текущая цена», ошибочна нижняя. 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 diff --git a/tradein-mvp/backend/data/sql/310_yandex_seed_decimal_slips.sql b/tradein-mvp/backend/data/sql/310_yandex_seed_decimal_slips.sql new file mode 100644 index 00000000..8b0979c3 --- /dev/null +++ b/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; diff --git a/tradein-mvp/backend/tests/skip_allowlist.txt b/tradein-mvp/backend/tests/skip_allowlist.txt index 02194917..69c61d4b 100644 --- a/tradein-mvp/backend/tests/skip_allowlist.txt +++ b/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 diff --git a/tradein-mvp/backend/tests/test_3385_migration_310_yandex_seed_slips.py b/tradein-mvp/backend/tests/test_3385_migration_310_yandex_seed_slips.py new file mode 100644 index 00000000..6d7decdd --- /dev/null +++ b/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 diff --git a/tradein-mvp/backend/tests/test_yandex_price_history.py b/tradein-mvp/backend/tests/test_yandex_price_history.py index 3ebd30e0..16c42d93 100644 --- a/tradein-mvp/backend/tests/test_yandex_price_history.py +++ b/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")