МЕРА: в истории цен Яндекса больше нет выдуманных «прошлых цен» в 10 и 1000 раз ниже #3564
5 changed files with 304 additions and 12 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue