test(mera/b2c): гейт на сам SQL доли снижений + чистка протухших метрик

Ревью: гейт охранял не то место. Подмена знаменателя красила три теста, но
дефект «84.8% вместо 48.1%» живёт в SQL — во включении однострочных записей
истории (у domklik одна запись = «цену не менял») в знаменатель. Ревьюер
вернул дефект условием n_rows >= 2 в CTE moved, и все 25 тестов остались
зелёными: текстовые пины держали только span_days и max_abs_pct.

Новый пин держит обе половины: однострочные попадают в moved веткой CASE со
значением 0, и нигде в запросе нет фильтра по числу записей истории (ни в
WHERE, ни HAVING). Живой прогон на подготовленных строках не заведён
намеренно: DATABASE_URL в тестовой джобе — заглушка, Postgres там нет, и тест
по образцу test_purge_expired_trade_in_data.py молча скипался бы, то есть не
гейтил бы ничего. Фальсифицировано руками — с n_rows >= 2 тест красный и
называет причину.

Второе: метрика, у которой пропал вход, больше не доживает в таблице со
старым computed_at (ручка отдавала её неотличимо от свежей). Строки вне
сегодняшнего набора удаляются в той же транзакции. На ПУСТОМ наборе чистка
не ходит: разом отвалившиеся все входы — признак поломки прогона, а не пяти
одновременных «данных больше нет». Оба поведения покрыты тестами, оба
проверены на сломанном коде.

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)

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<body>.*?)\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, "исчез порог наблюдения — короткоживущие дадут ложное «не снижал»"