diff --git a/tradein-mvp/backend/app/tasks/landing_stats.py b/tradein-mvp/backend/app/tasks/landing_stats.py index 2767035b..b6821e93 100644 --- a/tradein-mvp/backend/app/tasks/landing_stats.py +++ b/tradein-mvp/backend/app/tasks/landing_stats.py @@ -14,7 +14,9 @@ (нет оценок, нет истории цен, нет сделок) — строка в landing_stats просто не появляется, ручка её не отдаёт, фронт не рисует блок. Ноль здесь читался бы как измеренный ноль («ни одно объявление не снижало цену»), а это враньё другого -рода, чем отсутствие данных. +рода, чем отсутствие данных. Правило действует и на ВТОРОМ прогоне: пропавшая +метрика удаляется из таблицы (см. refresh_landing_stats), иначе она осталась бы +на витрине со старым computed_at и читалась бы как измеренная сегодня. ЧЕГО ЗДЕСЬ НЕТ И НЕ БУДЕТ ------------------------- @@ -188,6 +190,16 @@ _UPSERT_SQL = text(""" computed_at = EXCLUDED.computed_at """) +# Строки метрик, которых в СЕГОДНЯШНЕМ наборе нет, удаляются. Метрика исчезает +# из набора ровно тогда, когда у неё пропал вход (см. «нет входа — нет строки»), +# и оставленная строка продолжала бы отдаваться ручкой как обычная — со старым +# computed_at, который витрина не обязана читать. Удалённая метрика — блок, +# которого на странице нет; протухшая — блок с враньём. +_PRUNE_SQL = text(""" + DELETE FROM landing_stats + WHERE metric <> ALL(CAST(:kept AS text[])) +""") + def _num(value: Any) -> float | None: """Привести значение агрегата к float; None остаётся None. @@ -332,19 +344,32 @@ def refresh_landing_stats( Sync (вызывается scheduler-триггером в executor, как check_deals_freshness). `params` не используется — принимается ради единой сигнатуры обработчиков. - Пустой результат — НЕ ошибка прогона: на свежей базе метрик может не быть - ни одной, и падать в failed из-за этого значит завести шумный алерт там, где - система работает штатно. Строки при этом не трогаются: старый срез лучше - отсутствующего, а его возраст виден по computed_at. + Метрика, у которой пропал вход, СНИМАЕТСЯ с витрины, а не доживает со старым + computed_at: строки, которых нет в сегодняшнем наборе, удаляются в той же + транзакции. Иначе «нет входа — нет строки» действует только на первом + прогоне, а дальше отсутствие данных выглядит как данные — ручка отдаёт такую + строку неотличимо от свежей, и отличить её можно только сравнив computed_at с + соседями, чего фронт не делает. + + Пустой результат — НЕ ошибка прогона: на свежей базе метрик может не быть ни + одной, и падать в failed из-за этого значит завести шумный алерт там, где + система работает штатно. Но и чистка в этом случае НЕ выполняется: разом + отвалившиеся все входы — это признак поломки самого прогона (пустая/недоступная + база), а не пяти одновременных «данных больше нет», и стирать по такому + признаку всю витрину нельзя. Чистка ходит только с непустым набором, где + пропажу конкретной метрики видно на фоне посчитавшихся соседей. """ del params - counters: dict[str, int] = {"metrics_written": 0} + counters: dict[str, int] = {"metrics_written": 0, "metrics_removed": 0} try: runs_mod.update_heartbeat(db, run_id, counters) metrics = collect_landing_metrics(db) for row in metrics: db.execute(_UPSERT_SQL, row) + if metrics: + removed = db.execute(_PRUNE_SQL, {"kept": [m["metric"] for m in metrics]}) + counters["metrics_removed"] = int(removed.rowcount or 0) db.commit() counters["metrics_written"] = len(metrics) diff --git a/tradein-mvp/backend/tests/test_landing_stats.py b/tradein-mvp/backend/tests/test_landing_stats.py index 001635a1..f6b71772 100644 --- a/tradein-mvp/backend/tests/test_landing_stats.py +++ b/tradein-mvp/backend/tests/test_landing_stats.py @@ -61,9 +61,13 @@ PREFIX = "/api/public/mera" # контракта collect_landing_metrics (он же порядок метрик на витрине), и его # перестановка должна быть заметна. class _FakeSession: - def __init__(self, rows: list[Any]) -> None: + def __init__(self, rows: list[Any], *, prune_rowcount: int = 0) -> None: self._rows = list(rows) self.upserts: list[dict[str, Any]] = [] + # Чистка протухших метрик: пишем сюда параметры каждого DELETE, чтобы + # тест видел И факт вызова, И список оставляемых метрик. + self.prunes: list[dict[str, Any] | None] = [] + self._prune_rowcount = prune_rowcount self.committed = 0 # Считываем ТОЛЬКО запросы самой витрины: по этой же сессии ходит @@ -83,6 +87,9 @@ class _FakeSession: assert params is not None self.upserts.append(params) return MagicMock() + if "DELETE FROM landing_stats" in sql: + self.prunes.append(params) + return SimpleNamespace(rowcount=self._prune_rowcount) if not any(marker in sql for marker in self._METRIC_SQL_MARKERS): return MagicMock() assert self._rows, f"неожиданный лишний SELECT: {sql[:80]}" @@ -228,6 +235,44 @@ def test_refresh_upserts_every_metric_and_commits() -> None: } +def test_metric_that_stopped_computing_is_deleted_not_left_stale() -> None: + """Пропал вход у метрики — строка УДАЛЯЕТСЯ, а не доживает со старым + computed_at: иначе ручка отдаёт её неотличимо от посчитанной сегодня. + + Здесь сделок нет (`deals.n = 0`), значит `deals_total_12m` в наборе не + появляется — и именно её обязан вынести DELETE, оставив ровно посчитанные. + """ + rows = _rows(deals=SimpleNamespace(n=0)) + db = _FakeSession(rows, prune_rowcount=1) + counters = ls.refresh_landing_stats(db, run_id=1) # type: ignore[arg-type] + + assert len(db.prunes) == 1, "чистка протухших метрик не выполнена" + kept = set(db.prunes[0]["kept"]) # type: ignore[index] + assert kept == {u["metric"] for u in db.upserts} + assert "deals_total_12m" not in kept, "метрика без входа осталась бы на витрине" + assert counters["metrics_removed"] == 1 + + +def test_totally_empty_run_keeps_the_showcase_instead_of_wiping_it() -> None: + """Разом пропали ВСЕ входы — это похоже на поломку прогона (пустая или + недоступная база), а не на пять одновременных «данных больше нет». По такому + признаку витрина не стирается: DELETE не выполняется вовсе.""" + empty = _rows( + estimates=SimpleNamespace(total=0, period_days=None), + analogs=SimpleNamespace(n=0, median=None), + listing_age=SimpleNamespace(n=0, median=None), + price=SimpleNamespace(n=0, n_cut=0, median_pct_per_month=None), + deals=SimpleNamespace(n=0), + ) + db = _FakeSession(empty) + counters = ls.refresh_landing_stats(db, run_id=1) # type: ignore[arg-type] + + assert db.upserts == [] + assert db.prunes == [], "пустой прогон стёр бы всю витрину" + assert counters["metrics_written"] == 0 + assert counters["metrics_removed"] == 0 + + # ── 3. Границы выборки, которые исполняет Postgres ─────────────────────────── @@ -239,6 +284,44 @@ def test_price_sql_takes_domklik_only() -> None: assert "avito" not in sql and "yandex" not in sql +def test_price_sql_keeps_single_row_listings_in_denominator() -> None: + """Знаменатель доли снижений включает объявления с ОДНОЙ записью истории. + + Это тот самый дефект, из-за которого на проде получалось бы 84.8% вместо + 48.1%: у domklik триггер пишет стартовую цену, поэтому одна запись означает + «цену не менял» — наблюдение, а не отсутствие данных. Выкинув такие строки, + считаешь долю снижавших ТОЛЬКО среди менявших цену, то есть почти единицу. + + Гейт текстовый, а не прогон на живой базе: DATABASE_URL в CI — + заглушка (deploy-tradein.yml: `test:` job), Postgres в тестовой джобе нет, + и живой тест по образцу test_purge_expired_trade_in_data.py тут молча + скипался бы — то есть не гейтил бы ничего. Пин проверяет две половины + дефекта: (1) однострочные попадают в `moved` через ветку CASE со значением + 0 («не снижал»), а не отбрасываются; (2) нигде в запросе нет фильтра по + числу записей, который бы их отсёк. + """ + sql = str(ls._PRICE_MOVES_SQL) + + case = re.search(r"CASE\b(?P.*?)\bEND\b", sql, re.S | re.I) + assert case is not None, "исчезла ветка для однострочных — они больше не «не снижал»" + body = case.group("body") + assert "n_rows" in body, "ветка перестала различать однострочные записи истории" + assert re.search(r"\b(THEN|ELSE)\s+0\b", body), ( + "однострочным объявлениям больше не приписывается изменение 0% — " + "они либо выпали из выборки, либо получили выдуманное значение" + ) + + rest = sql.replace(case.group(0), "") + leftover = re.search(r"n_rows\s*(>=|>|<|<>|=|!=)", rest) + assert leftover is None, ( + f"появился фильтр по числу записей истории вне ветки CASE ({leftover.group(0)!r}) — " + "он выкидывает не менявших цену из знаменателя, доля вырастет с ~48% до ~85%" + ) + assert not re.search(r"\bHAVING\b", rest, re.I), ( + "HAVING в агрегате истории отсекает однострочные ещё до знаменателя" + ) + + def test_price_sql_keeps_span_and_outlier_gates() -> None: sql = str(ls._PRICE_MOVES_SQL) assert "span_days" in sql, "исчез порог наблюдения — короткоживущие дадут ложное «не снижал»"