fix(tradein/deactivate): make TTL-cap multiplier configurable per source
Review of 3a1e29a7 found CAP_MULT=2 is uniform across sources with wildly
different ttl_days, so it produces a different ABSOLUTE ceiling per source:
cian/yandex (ttl=30) -> 60d, avito (ttl=10) -> 20d, domklik (ttl=14) -> 28d.
That breaks exactly where the crawl's revisit tail doesn't scale with
ttl_days: avito's measured p99 revisit gap is 42.1d (_REVISIT_TAIL) --
above its own default cap of 20d -- so a legitimately slow-but-alive avito
crawl cycle would get its floor cut below the very tail the floor exists
to protect (the false-kill scenario #2659 was filed for). cian/yandex/
domklik aren't affected: their default ceilings (60/60/28) already sit
comfortably above their own measured tails (26.6/43.0/3.1).
Fix: cap_mult is now a function parameter (same pattern as
revisit_floor_quantile/min_confirmations) with the module constant CAP_MULT
as its default, wired through product_handlers via default_params["cap_mult"]
so a schedule can override it without touching the shared default. Also
closes the ttl_days<=0 edge case flagged in the same review: before the cap,
max(ttl_days, floor) tolerated a misconfigured ttl_days<=0 as long as the
floor was positive; with the cap, min(floor, ttl_days*cap_mult<=0) would
silently defeat that protection and match nearly the whole active pool.
ttl_days<=0 now raises ValueError before any SQL, same contract as the
existing staleness_column whitelist check.
Also verified (read-only, postgres-tradein) the review's core "no-op"
claim: false. scrape_runs.counters for the 6 days since the revisit-floor
went live (08-10..08-15) show the cap DID bind on 3 of 6 runs for avito
(floor 52 vs cap 20) and 3 of 6 for yandex (floor 75 vs cap 60) -- the
reviewer's "no source hits the cap" read a single-day trough right after
a natural recovery, not the whole observation window. See PR discussion
for the full counter history and refutation detail.
Refs #2659
This commit is contained in:
bot-backend
parent
3a1e29a7da
commit
cb79c67bfc
3 changed files with 183 additions and 13 deletions
|
|
@ -217,6 +217,7 @@ async def _job_deactivate_stale(
|
|||
) -> None:
|
||||
from app.core.config import settings as _settings
|
||||
from app.tasks.deactivate_stale_avito import (
|
||||
CAP_MULT,
|
||||
DEFAULT_MIN_CONFIRMATIONS,
|
||||
DEFAULT_REVISIT_FLOOR_QUANTILE,
|
||||
deactivate_stale_listings,
|
||||
|
|
@ -236,6 +237,11 @@ async def _job_deactivate_stale(
|
|||
revisit_floor_quantile: float = params.get(
|
||||
"revisit_floor_quantile", DEFAULT_REVISIT_FLOOR_QUANTILE
|
||||
)
|
||||
# Потолок эффективного TTL (см. CAP_MULT в deactivate_stale_avito.py) — множитель,
|
||||
# а не голая константа: источник с непропорционально длинным хвостом переобхода
|
||||
# относительно своего ttl_days переопределяет его через default_params (ключ
|
||||
# "cap_mult"), не трогая дефолт для остальных источников.
|
||||
cap_mult: float = params.get("cap_mult", CAP_MULT)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
await loop.run_in_executor(
|
||||
|
|
@ -249,6 +255,7 @@ async def _job_deactivate_stale(
|
|||
staleness_column=staleness_column,
|
||||
min_confirmations=min_confirmations,
|
||||
revisit_floor_quantile=revisit_floor_quantile,
|
||||
cap_mult=cap_mult,
|
||||
),
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -213,6 +213,24 @@ _REVISIT_FLOOR_SEGMENT_FILTER = "\n AND l.listing_segment = ANY(CAST(:s
|
|||
# отдельный гейт по здоровью (min_confirmations) выше в этой же функции, который
|
||||
# выключает деактивацию целиком, а не растягивает TTL до бесконечности. Калибровочная
|
||||
# ручка, не догма -- при новом замере можно пересмотреть, как и revisit_floor_quantile.
|
||||
#
|
||||
# ПОЧЕМУ MULT, А НЕ ФИКСИРОВАННОЕ ЧИСЛО СУТОК -- И ГДЕ ЭТА ФОРМА ЛОМАЕТСЯ. Множитель
|
||||
# от ttl_days даёт разный АБСОЛЮТНЫЙ потолок на разных источниках: cian/yandex
|
||||
# (ttl=30) -> 60 суток, avito (ttl=10) -> 20 суток, domklik (ttl=14) -> 28 суток. Это
|
||||
# ломается ровно там, где абсолютный хвост переобхода источника НЕ пропорционален его
|
||||
# ttl_days. Замер (_REVISIT_TAIL, 40 суток): avito p99 = 42.1 сут -- ВЫШЕ его же
|
||||
# потолка 20. То есть для avito дефолтный CAP_MULT=2 может резать ttl ниже
|
||||
# собственного хвоста обхода -- ровно тот false-kill, ради которого пол вообще
|
||||
# заведён (см. комментарий выше). У cian/yandex (потолок 60) и domklik (потолок 28
|
||||
# при хвосте 3.1) такого разрыва нет -- множитель 2 для них калиброван верно.
|
||||
#
|
||||
# ПОЭТОМУ cap_mult -- параметр функции (как revisit_floor_quantile, min_confirmations),
|
||||
# не голая константа: default = CAP_MULT для источников, где 2x достаточно, но
|
||||
# расписание может переопределить через default_params (JSON-колонка scrape_schedules,
|
||||
# ключ "cap_mult") для источника с непропорционально длинным хвостом -- см. миграцию
|
||||
# для avito, поднимающую cap_mult до 6 (потолок 60 суток, тот же порядок, что у
|
||||
# cian/yandex, и с запасом выше и статического p99=42.1, и живого прод-пика 52,
|
||||
# замеренного 2026-08-10..12).
|
||||
CAP_MULT = 2
|
||||
|
||||
|
||||
|
|
@ -338,6 +356,7 @@ def deactivate_stale_listings(
|
|||
min_confirmations: int = 0,
|
||||
health_window_days: int = _HEALTH_WINDOW_DAYS,
|
||||
revisit_floor_quantile: float = 0.0,
|
||||
cap_mult: float = CAP_MULT,
|
||||
) -> dict[str, int]:
|
||||
"""Пометить is_active=false объявления, чья свежесть старше ttl_days дней.
|
||||
|
||||
|
|
@ -360,10 +379,17 @@ def deactivate_stale_listings(
|
|||
health_window_days: окно подтверждений для гейта, суток. Дефолт 3.
|
||||
revisit_floor_quantile: пол TTL по измеренному циклу переобхода (#2659).
|
||||
Квантиль возраста, при котором свип за окно ДОКАЗАЛ строку живой;
|
||||
эффективный TTL = min(max(ttl_days, этот пол), ttl_days * CAP_MULT) --
|
||||
эффективный TTL = min(max(ttl_days, этот пол), ttl_days * cap_mult) --
|
||||
пол поднимает TTL, но не выше потолка. 0 -> пол выключен (так
|
||||
вызывают старые тесты и совместимая обёртка), рабочее значение —
|
||||
DEFAULT_REVISIT_FLOOR_QUANTILE, см. комментарий выше.
|
||||
cap_mult: множитель потолка эффективного TTL (см. комментарий у модульной
|
||||
константы CAP_MULT). Дефолт -- сама CAP_MULT=2, но параметр, а НЕ голая
|
||||
константа: источник с непропорционально длинным хвостом переобхода
|
||||
относительно своего ttl_days (avito: p99=42.1 при ttl=10 -> дефолтный
|
||||
потолок 20 режет ниже хвоста) может переопределить его через
|
||||
default_params расписания (ключ "cap_mult"), не трогая остальные
|
||||
источники. Итоговый потолок = ttl_days * cap_mult.
|
||||
|
||||
Sync (вызывается scheduler-триггером в executor, как snapshot_listing_sources).
|
||||
Один statement в транзакции: UPDATE флага + снимок 'stale' в listings_snapshots
|
||||
|
|
@ -373,15 +399,22 @@ def deactivate_stale_listings(
|
|||
Если гейт не пропустил прогон: {"deactivated": 0, "confirmations": N,
|
||||
"skipped_unhealthy": 1} и НИ ОДНА строка не тронута. Если пол переобхода поднял
|
||||
TTL: дополнительно {"revisit_floor_days": N, "ttl_days_effective": N}. Если пол
|
||||
упёрся в потолок CAP_MULT: дополнительно {"ttl_floor_capped": 1,
|
||||
упёрся в потолок cap_mult: дополнительно {"ttl_floor_capped": 1,
|
||||
"ttl_days_floor_raw": N} -- N это то, во что пол поднял бы TTL БЕЗ потолка.
|
||||
|
||||
Raises:
|
||||
ValueError: если staleness_column не входит в whitelist (проверка ДО SQL,
|
||||
никакой интерполяции пользовательского ввода в запрос).
|
||||
ValueError: если staleness_column не входит в whitelist, ИЛИ ttl_days <= 0
|
||||
(проверка ДО SQL, никакой интерполяции пользовательского ввода в запрос;
|
||||
ttl_days<=0 в WHERE-условии last_seen_at < NOW() - INTERVAL 'N days'
|
||||
матчит практически весь активный пул -- без явного guard'а потолок
|
||||
(ttl_days * cap_mult <= 0) к тому же перебивал бы пол в формуле min(),
|
||||
снимая защиту, которую max(ttl_days, floor) давал раньше).
|
||||
"""
|
||||
counters: dict[str, int] = {"deactivated": 0}
|
||||
try:
|
||||
if ttl_days <= 0:
|
||||
raise ValueError(f"ttl_days must be positive, got {ttl_days!r}")
|
||||
|
||||
# Whitelist-проверка ДО построения/выполнения SQL: только после неё имя колонки
|
||||
# интерполируется f-string'ом. Значения по-прежнему идут через param-binding.
|
||||
# Внутри try -> невалидная колонка финализирует run как failed (mark_failed),
|
||||
|
|
@ -431,8 +464,9 @@ def deactivate_stale_listings(
|
|||
return counters
|
||||
|
||||
# Пол TTL по измеренному циклу переобхода (#2659) — тоже ДО UPDATE и по тому же
|
||||
# срезу. Поднимает порог (max), но не выше потолка CAP_MULT * ttl_days (min) —
|
||||
# см. комментарий у CAP_MULT про петлю с положительной обратной связью.
|
||||
# срезу. Поднимает порог (max), но не выше потолка cap_mult * ttl_days (min) —
|
||||
# см. комментарий у CAP_MULT про петлю с положительной обратной связью и про
|
||||
# то, почему cap_mult -- параметр, а не голая константа.
|
||||
effective_ttl_days = ttl_days
|
||||
if revisit_floor_quantile > 0:
|
||||
floor_params: dict[str, Any] = {
|
||||
|
|
@ -450,12 +484,14 @@ def deactivate_stale_listings(
|
|||
# Тогда пола нет и TTL остаётся как задан: выдумывать пол не из чего.
|
||||
if floor_days is not None:
|
||||
counters["revisit_floor_days"] = ceil(float(floor_days))
|
||||
# Пол поднимает TTL (max), потолок CAP_MULT его не пускает выше
|
||||
# ttl_days * CAP_MULT (min) — без этого пол растёт без ограничения
|
||||
# (см. комментарий у CAP_MULT).
|
||||
# Пол поднимает TTL (max), потолок cap_mult его не пускает выше
|
||||
# ttl_days * cap_mult (min) — без этого пол растёт без ограничения
|
||||
# (см. комментарий у CAP_MULT). capped_ttl_days может быть float,
|
||||
# если cap_mult переопределён нецелым значением из default_params —
|
||||
# effective_ttl_days приводим к int (UPDATE ждёт целые сутки).
|
||||
raw_effective_ttl_days = max(ttl_days, counters["revisit_floor_days"])
|
||||
capped_ttl_days = ttl_days * CAP_MULT
|
||||
effective_ttl_days = min(raw_effective_ttl_days, capped_ttl_days)
|
||||
capped_ttl_days = ttl_days * cap_mult
|
||||
effective_ttl_days = int(min(raw_effective_ttl_days, capped_ttl_days))
|
||||
counters["ttl_days_effective"] = effective_ttl_days
|
||||
|
||||
if raw_effective_ttl_days > capped_ttl_days:
|
||||
|
|
@ -467,13 +503,13 @@ def deactivate_stale_listings(
|
|||
counters["ttl_days_floor_raw"] = raw_effective_ttl_days
|
||||
logger.warning(
|
||||
"deactivate_stale source=%s run_id=%d TTL пол упёрся в потолок "
|
||||
"CAP_MULT=%d: пол поднял бы TTL до %d сут, потолок ограничивает "
|
||||
"cap_mult=%s: пол поднял бы TTL до %d сут, потолок ограничивает "
|
||||
"заданные %d сут значением %d (квантиль %.3f, segments=%r) — "
|
||||
"растущий без ограничения пол это петля с положительной обратной "
|
||||
"связью, см. комментарий у CAP_MULT",
|
||||
listing_source,
|
||||
run_id,
|
||||
CAP_MULT,
|
||||
cap_mult,
|
||||
raw_effective_ttl_days,
|
||||
ttl_days,
|
||||
effective_ttl_days,
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ ttl_days * CAP_MULT, независимо от того, насколько вы
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
|
@ -196,3 +197,129 @@ def test_avito_floor_above_ttl_but_under_cap_passes_through_uncapped(
|
|||
out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
|
||||
assert out["ttl_days_effective"] == 18
|
||||
assert "ttl_floor_capped" not in out
|
||||
|
||||
|
||||
# ── cap_mult конфигурируем per-source (найдено ревью 2026-08-15) ────────────────
|
||||
# Дефолтный CAP_MULT=2 даёт разный АБСОЛЮТНЫЙ потолок на разных источниках
|
||||
# (cian/yandex 60 сут, avito 20 сут), а хвост переобхода не пропорционален
|
||||
# ttl_days: avito p99=42.1 -- выше его же дефолтного потолка 20. cap_mult -- ручка
|
||||
# для конкретно такого источника, без изменения дефолта для остальных.
|
||||
|
||||
|
||||
def test_cap_mult_defaults_to_module_constant_when_not_overridden(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Без явного cap_mult поведение не меняется: потолок = ttl_days * CAP_MULT (2)."""
|
||||
db = _FakeDB(floor_days=75.0)
|
||||
out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
|
||||
assert out["ttl_days_effective"] == 10 * task_mod.CAP_MULT
|
||||
|
||||
|
||||
def test_cap_mult_override_raises_the_ceiling_for_a_long_tailed_source(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""avito p99=42.1: cap_mult=6 (потолок 60) больше не режет пол ниже хвоста обхода,
|
||||
в отличие от дефолтного cap_mult=2 (потолок 20)."""
|
||||
db = _FakeDB(floor_days=45.0)
|
||||
out = _run(
|
||||
db,
|
||||
monkeypatch,
|
||||
listing_source="avito",
|
||||
ttl_days=10,
|
||||
revisit_floor_quantile=0.99,
|
||||
cap_mult=6,
|
||||
)
|
||||
assert out["ttl_days_effective"] == 45
|
||||
assert "ttl_floor_capped" not in out
|
||||
|
||||
|
||||
def test_cap_mult_override_still_caps_when_floor_exceeds_the_wider_ceiling(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""cap_mult поднимает потолок, но не убирает его -- пол выше 60 всё равно срезается."""
|
||||
db = _FakeDB(floor_days=90.0)
|
||||
out = _run(
|
||||
db,
|
||||
monkeypatch,
|
||||
listing_source="avito",
|
||||
ttl_days=10,
|
||||
revisit_floor_quantile=0.99,
|
||||
cap_mult=6,
|
||||
)
|
||||
assert out["ttl_days_effective"] == 60
|
||||
assert out["ttl_floor_capped"] == 1
|
||||
assert out["ttl_days_floor_raw"] == 90
|
||||
|
||||
|
||||
def test_cap_mult_is_threaded_into_update_params(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Капнутый по override'нутому потолку TTL реально уходит в UPDATE, не только считается."""
|
||||
db = _FakeDB(floor_days=90.0)
|
||||
_run(
|
||||
db,
|
||||
monkeypatch,
|
||||
listing_source="avito",
|
||||
ttl_days=10,
|
||||
revisit_floor_quantile=0.99,
|
||||
cap_mult=6,
|
||||
)
|
||||
_, update_params = db.update_query
|
||||
assert update_params is not None
|
||||
assert update_params["ttl_days"] == 60
|
||||
|
||||
|
||||
# ── ttl_days <= 0 (LOW из ревью 2026-08-15) ──────────────────────────────────────
|
||||
# До потолка max(ttl_days, floor) прикрывал ttl_days<=0, если пол посчитан и
|
||||
# положителен. С потолком min(raw, ttl_days * cap_mult) при ttl_days<=0 капнутый
|
||||
# потолок тоже <= 0 и побеждает в min() -- защита пола пропадает молча. Явный guard
|
||||
# ловит это ДО любого SQL, тем же путём, что и невалидный staleness_column.
|
||||
|
||||
|
||||
def test_ttl_days_zero_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
db = _FakeDB(floor_days=75.0)
|
||||
with pytest.raises(ValueError):
|
||||
_run(db, monkeypatch, ttl_days=0, revisit_floor_quantile=0.99)
|
||||
assert db.executed == []
|
||||
|
||||
|
||||
def test_ttl_days_negative_is_rejected_before_any_sql(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
db = _FakeDB(floor_days=75.0)
|
||||
with pytest.raises(ValueError):
|
||||
_run(db, monkeypatch, ttl_days=-5, revisit_floor_quantile=0.99)
|
||||
assert db.executed == []
|
||||
|
||||
|
||||
def test_ttl_days_zero_fails_the_run_via_mark_failed(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Тот же контракт, что и невалидный staleness_column: run помечается failed,
|
||||
а не остаётся 'running'."""
|
||||
marked_failed: list[Any] = []
|
||||
monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
|
||||
monkeypatch.setattr(
|
||||
task_mod.runs_mod,
|
||||
"mark_failed",
|
||||
lambda db, run_id, err, counters: marked_failed.append((run_id, err, counters)),
|
||||
)
|
||||
db = _FakeDB(floor_days=75.0)
|
||||
with pytest.raises(ValueError):
|
||||
task_mod.deactivate_stale_listings(
|
||||
db, # type: ignore[arg-type]
|
||||
7,
|
||||
listing_source="avito",
|
||||
ttl_days=0,
|
||||
)
|
||||
assert len(marked_failed) == 1
|
||||
assert marked_failed[0][0] == 7
|
||||
|
||||
|
||||
# ── проводка cap_mult в product_handlers ─────────────────────────────────────────
|
||||
|
||||
|
||||
def test_handler_wires_cap_mult_from_schedule_params() -> None:
|
||||
"""Тот же приём, что test_handler_wires_revisit_floor_from_schedule_params:
|
||||
читаем исходник файлом (product_handlers тянет scraper_kit, которого в юнит-
|
||||
окружении может не быть) и проверяем именно проводку default_params -> вызов."""
|
||||
handlers = Path(__file__).resolve().parents[1] / "app" / "services" / "product_handlers.py"
|
||||
src = handlers.read_text("utf-8")
|
||||
job = src.split("async def _job_deactivate_stale")[1].split("\nasync def ")[0]
|
||||
flat = " ".join(job.split())
|
||||
assert 'params.get("cap_mult", CAP_MULT)' in flat
|
||||
assert "cap_mult=cap_mult" in job
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue