perf(tradein): shared EKB seed price-brackets across avito/cian/yandex (city-dict ready for multi-city)

tradein-mvp/backend/app/services/scrapers/avito.py

@@ -46,6 +46,7 @@ from app.services.scrapers.avito_exceptions import (
 )
 from app.services.scrapers.base import BaseScraper, ScrapedLot
 from app.services.scrapers.browser_fetcher import BrowserFetcher
+from app.services.scrapers.price_brackets import get_price_seed_brackets
 
 logger = logging.getLogger(__name__)
 
@@ -237,24 +238,13 @@ _AVITO_MAX_PRICE = 200_000_000
 _AVITO_MIN_BRACKET = 50_000
 
 # Seed price-brackets (руб) — data-driven по прод-распределению вторички ЕКБ
-# (пик 4-8М, bulk 3-16М; самая плотная комнатность — студии 5783, единственный
-# >2к блок был студии 4-6М → пик дроблю по 1М). Каждый (комнатность × брекет)
-# дальше дробится _walk_price_range если total > cap (=2000 → блоки ≤2к).
+# (пик 4-8М, bulk 3-16М). Теперь общий констант для всех exhaustive-скрапперов
+# (avito/cian/yandex) — app.services.scrapers.price_brackets.get_price_seed_brackets().
+# Локальный алиас сохранён для читаемости и тестов; значения идентичны shared EKB.
+# Каждый (комнатность × брекет) дальше дробится _walk_price_range если total > cap.
 # Последний брекет ОТКРЫТ (hi=None → pmax не ставится) — ловит весь хвост люкса
 # без потолка; он крошечный (avito >50М ≈ 73), пагинируется напрямую без бисекции.
-_AVITO_PRICE_SEED_BRACKETS: list[tuple[int, int | None]] = [
-    (0, 4_000_000),
-    (4_000_000, 5_000_000),
-    (5_000_000, 6_000_000),
-    (6_000_000, 7_000_000),
-    (7_000_000, 8_000_000),
-    (8_000_000, 10_000_000),
-    (10_000_000, 13_000_000),
-    (13_000_000, 17_000_000),
-    (17_000_000, 25_000_000),
-    (25_000_000, 250_000_000),
-    (250_000_000, None),  # 250М+ открытый catch-all (без потолка)
-]
+_AVITO_PRICE_SEED_BRACKETS: list[tuple[int, int | None]] = get_price_seed_brackets()
 # Avito SERP показывает ~50 карточек на страницу.
 _AVITO_OFFERS_PER_PAGE = 50
 

tradein-mvp/backend/app/services/scrapers/cian.py

@@ -38,6 +38,7 @@ from app.core.config import settings
 from app.services.scraper_settings import get_scraper_delay
 from app.services.scrapers.base import BaseScraper, ScrapedLot
 from app.services.scrapers.cian_state_parser import extract_state
+from app.services.scrapers.price_brackets import get_price_seed_brackets
 from app.services.scrapers.repair_state_normalizer import (
     infer_repair_state_from_text,
     normalize_repair_state,
@@ -325,11 +326,15 @@ class CianScraper(BaseScraper):
     ) -> list[ScrapedLot]:
         """Exhaustive-загрузка Cian ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА.
 
-        Обходит Cian SERP-cap (~54 стр/запрос ≈ 1500 результатов на запрос):
-        внутри каждой комнатности адаптивно бьёт диапазон цены на бакеты так, чтобы
-        в каждом totalOffers < price_cap_per_bucket → пагинирует бакет полностью.
+        Обходит Cian SERP-cap (~54 стр/запрос ≈ 1500 результатов на запрос). Вместо
+        одной корень-бисекции [0, _MAX_PRICE] на комнатность (холостые probe-спуски
+        через пустой верх) — стартуем с общей data-driven seed-сетки ценовых брекетов
+        (get_price_seed_brackets() — shared EKB grid, общая с avito/yandex). Каждый
+        (комнатность × seed-брекет) дальше дробится _walk_price_range если
+        totalOffers > price_cap_per_bucket. Последний seed-брекет ОТКРЫТ (hi=None →
+        maxprice не ставится) — ловит весь хвост люкса без потолка.
         Страницы внутри leaf-бакета запрашиваются параллельно (asyncio.gather + Semaphore).
-        Дедуп по source_id (dict seen).
+        Дедуп по source_id (dict seen), ОБЩИЙ по всем брекетам и комнатностям.
 
         Параметры:
           rooms_buckets: список room-кодов Cian для перебора (default: 1-6).
@@ -362,18 +367,24 @@ class CianScraper(BaseScraper):
                 secondary_only,
             )
             before = len(seen)
-            await self._walk_price_range(
-                rooms=rooms,
-                lo=0,
-                hi=_MAX_PRICE,
-                seen=seen,
-                price_cap_per_bucket=price_cap_per_bucket,
-                max_pages_per_bucket=max_pages_per_bucket,
-                concurrency=concurrency,
-                secondary_only=secondary_only,
-                on_bucket=on_bucket,
-                skip_buckets=skip_buckets,
-            )
+            # Перебираем seed-брекеты: для ЗАКРЫТЫХ передаём hi = br_hi - 1
+            # (maxprice эксклюзивный сверху → соседние брекеты не пересекаются), для
+            # ОТКРЫТОГО (br_hi is None) — hi=None (без потолка). Дедуп по source_id
+            # в общем `seen` страхует на стыках.
+            for br_lo, br_hi in get_price_seed_brackets():
+                walk_hi = br_hi - 1 if br_hi is not None else None
+                await self._walk_price_range(
+                    rooms=rooms,
+                    lo=br_lo,
+                    hi=walk_hi,
+                    seen=seen,
+                    price_cap_per_bucket=price_cap_per_bucket,
+                    max_pages_per_bucket=max_pages_per_bucket,
+                    concurrency=concurrency,
+                    secondary_only=secondary_only,
+                    on_bucket=on_bucket,
+                    skip_buckets=skip_buckets,
+                )
             room_collected = len(seen) - before
             logger.info(
                 "cian exhaustive: %s done — collected %d (total unique=%d)",
@@ -392,7 +403,7 @@ class CianScraper(BaseScraper):
         *,
         rooms: tuple[int, ...],
         lo: int,
-        hi: int,
+        hi: int | None,
         seen: dict[str, ScrapedLot],
         price_cap_per_bucket: int,
         max_pages_per_bucket: int,
@@ -404,23 +415,28 @@ class CianScraper(BaseScraper):
     ) -> None:
         """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].
 
-        Алгоритм:
+        Алгоритм для ЗАКРЫТОГО брекета (hi is not None):
           1. Запросить page=1 с min_price=lo, max_price=hi → totalOffers.
           2. Если totalOffers <= cap → пагинировать leaf-бакет параллельно (concurrency).
           3. Если totalOffers > cap → разбить бакет пополам (рекурсия).
           Guard: hi - lo < _MIN_BRACKET → пагинировать как есть (логируем WARNING).
 
+        ОТКРЫТЫЙ брекет (hi is None) — верхний seed-брекет без потолка (maxprice не
+        ставится, Cian отдаёт всё ≥lo). Делить нельзя (mid=(lo+hi)//2 невозможен) —
+        probe + пагинируем leaf напрямую. Хвост 250М+ крошечный; за cap обычно не
+        выходит, если вышел — пагинируем до max_pages + WARNING (tail-loss accepted).
+
         После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots) если задан.
-        bucket_key = "room_label:lo:hi" (checkpoint-ключ для resume).
+        bucket_key = "room_label:lo:hi" (закрытый) либо "room_label:lo:open" (открытый) —
+        checkpoint-ключ для resume.
         on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон.
         skip_buckets: если bucket_key в skip_buckets — пагинацию и on_bucket пропускаем.
         secondary_only: если True — новостройки (listing_segment=="novostroyki") отбрасываются
           после сбора bucket_lots, до дедупа в seen и вызова on_bucket.
         """
-        effective_hi = hi
-
         room_label = f"room{'_'.join(str(r) for r in rooms)}"
         _lo_param = lo if lo > 0 else None
+        _hi_repr = "open" if hi is None else str(hi)
 
         # ── Шаг 1: probe page 1 ────────────────────────────────────────────────
         html = await self._fetch_page_html(rooms, 1, _lo_param, hi)
@@ -433,10 +449,10 @@ class CianScraper(BaseScraper):
         # Ретрай на captcha/ошибку: rotate IP + 1 retry
         if total is None:
             logger.warning(
-                "cian: totalOffers=None for %s [%d, %d] depth=%d — rotating IP + retry",
+                "cian: totalOffers=None for %s [%d, %s] depth=%d — rotating IP + retry",
                 room_label,
                 lo,
-                hi,
+                _hi_repr,
                 _depth,
             )
             rotated = await self._rotate_ip()
@@ -448,18 +464,18 @@ class CianScraper(BaseScraper):
 
         if total is None:
             logger.error(
-                "cian: skipping bucket %s [%d, %d] — totalOffers unavailable after retry",
+                "cian: skipping bucket %s [%d, %s] — totalOffers unavailable after retry",
                 room_label,
                 lo,
-                hi,
+                _hi_repr,
             )
             return
 
         logger.info(
-            "cian: %s [%d, %d] totalOffers=%d depth=%d",
+            "cian: %s [%d, %s] totalOffers=%d depth=%d",
             room_label,
             lo,
-            hi,
+            _hi_repr,
             total,
             _depth,
         )
@@ -467,8 +483,37 @@ class CianScraper(BaseScraper):
         if total == 0:
             return
 
-        # ── Шаг 2: деление или пагинация ──────────────────────────────────────
-        bracket_size = effective_hi - lo
+        # ── ОТКРЫТЫЙ брекет (hi is None): делить нельзя — пагинируем leaf напрямую ─
+        # total известен из probe → max_pages = min(ceil(total/per_page), cap).
+        # Если total > cap (не должно для 250М+, но guard) — пагинируем как есть до
+        # max_pages_per_bucket с WARNING (хвост люкса крошечный, tail-loss accepted).
+        if hi is None:
+            if total > price_cap_per_bucket:
+                logger.warning(
+                    "cian: OPEN bucket %s [%d, open] totalOffers=%d > cap=%d — paginating "
+                    "without split (tail-loss accepted, lux tail tiny)",
+                    room_label,
+                    lo,
+                    total,
+                    price_cap_per_bucket,
+                )
+            await self._paginate_open_bucket(
+                rooms=rooms,
+                room_label=room_label,
+                lo=lo,
+                html=html,
+                total=total,
+                seen=seen,
+                max_pages_per_bucket=max_pages_per_bucket,
+                concurrency=concurrency,
+                secondary_only=secondary_only,
+                on_bucket=on_bucket,
+                skip_buckets=skip_buckets,
+            )
+            return
+
+        # ── Шаг 2: деление или пагинация (только ЗАКРЫТЫЙ брекет, hi: int) ─────
+        bracket_size = hi - lo
         need_split = total > price_cap_per_bucket
         too_narrow = bracket_size < _MIN_BRACKET
 
@@ -488,7 +533,7 @@ class CianScraper(BaseScraper):
             need_split = False  # принудительно пагинируем
 
         if need_split:
-            mid = (lo + effective_hi) // 2
+            mid = (lo + hi) // 2
             # [lo, mid]
             await self._walk_price_range(
                 rooms=rooms,
@@ -519,8 +564,77 @@ class CianScraper(BaseScraper):
             )
             return
 
-        # ── Checkpoint-ключ leaf-бакета ──────────────────────────────────────
-        bucket_key = f"{room_label}:{lo}:{hi}"
+        # Закрытый leaf-бакет — общая пагинация (см. _paginate_leaf_bucket).
+        await self._paginate_leaf_bucket(
+            rooms=rooms,
+            room_label=room_label,
+            lo=lo,
+            hi=hi,
+            html=html,
+            total=total,
+            seen=seen,
+            max_pages_per_bucket=max_pages_per_bucket,
+            concurrency=concurrency,
+            secondary_only=secondary_only,
+            on_bucket=on_bucket,
+            skip_buckets=skip_buckets,
+        )
+
+    async def _paginate_open_bucket(
+        self,
+        *,
+        rooms: tuple[int, ...],
+        room_label: str,
+        lo: int,
+        html: str | None,
+        total: int,
+        seen: dict[str, ScrapedLot],
+        max_pages_per_bucket: int,
+        concurrency: int,
+        secondary_only: bool,
+        on_bucket: Callable[..., Any] | None,
+        skip_buckets: set[str] | None,
+    ) -> None:
+        """Пагинация ОТКРЫТОГО leaf-бакета (hi=None, без maxprice)."""
+        await self._paginate_leaf_bucket(
+            rooms=rooms,
+            room_label=room_label,
+            lo=lo,
+            hi=None,
+            html=html,
+            total=total,
+            seen=seen,
+            max_pages_per_bucket=max_pages_per_bucket,
+            concurrency=concurrency,
+            secondary_only=secondary_only,
+            on_bucket=on_bucket,
+            skip_buckets=skip_buckets,
+        )
+
+    async def _paginate_leaf_bucket(
+        self,
+        *,
+        rooms: tuple[int, ...],
+        room_label: str,
+        lo: int,
+        hi: int | None,
+        html: str | None,
+        total: int,
+        seen: dict[str, ScrapedLot],
+        max_pages_per_bucket: int,
+        concurrency: int,
+        secondary_only: bool,
+        on_bucket: Callable[..., Any] | None,
+        skip_buckets: set[str] | None,
+    ) -> None:
+        """Параллельная пагинация leaf-бакета [lo, hi] (закрытого или открытого).
+
+        ОТКРЫТЫЙ (hi is None): maxprice не ставится (_build_url отдаёт всё ≥lo),
+        bucket_key = "room_label:lo:open". Страница 1 берётся из probe-`html`.
+        """
+        _lo_param = lo if lo > 0 else None
+        _hi_repr = "open" if hi is None else str(hi)
+        bucket_key = f"{room_label}:{lo}:{_hi_repr}"
 
         # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket.
         if skip_buckets and bucket_key in skip_buckets:
@@ -530,7 +644,6 @@ class CianScraper(BaseScraper):
             )
             return
 
-        # ── Параллельная пагинация leaf-бакета ────────────────────────────────
         pages_needed = math.ceil(total / _CIAN_OFFERS_PER_PAGE)
         max_pages = min(pages_needed, max_pages_per_bucket)
 
@@ -539,12 +652,12 @@ class CianScraper(BaseScraper):
             # Это происходит когда bracket < _MIN_BRACKET но totalOffers всё равно > cap.
             tail_loss = total - max_pages_per_bucket * _CIAN_OFFERS_PER_PAGE
             logger.warning(
-                "cian: %s [%d, %d] totalOffers=%d exceeds page cap "
+                "cian: %s [%d, %s] totalOffers=%d exceeds page cap "
                 "(max_pages=%d × %d=%d offers) — tail loss ~%d offers, "
                 "consider narrowing price_cap_per_bucket or _MIN_BRACKET",
                 room_label,
                 lo,
-                hi,
+                _hi_repr,
                 total,
                 max_pages_per_bucket,
                 _CIAN_OFFERS_PER_PAGE,
@@ -564,10 +677,10 @@ class CianScraper(BaseScraper):
                 await asyncio.sleep(self.request_delay_sec)
                 if page_html is None:
                     logger.warning(
-                        "cian: page_html=None %s [%d, %d] page=%d — skipping page",
+                        "cian: page_html=None %s [%d, %s] page=%d — skipping page",
                         room_label,
                         lo,
-                        hi,
+                        _hi_repr,
                         p,
                     )
                     return []
@@ -582,10 +695,10 @@ class CianScraper(BaseScraper):
         for p_idx, res in enumerate(page_results, start=1):
             if isinstance(res, BaseException):
                 logger.warning(
-                    "cian: page exception %s [%d, %d] page=%d — %r",
+                    "cian: page exception %s [%d, %s] page=%d — %r",
                     room_label,
                     lo,
-                    hi,
+                    _hi_repr,
                     p_idx,
                     res,
                 )
@@ -608,10 +721,10 @@ class CianScraper(BaseScraper):
                 seen[key] = lot
 
         logger.info(
-            "cian: %s [%d, %d] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d",
+            "cian: %s [%d, %s] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d",
             room_label,
             lo,
-            hi,
+            _hi_repr,
             max_pages,
             collected_this_bucket,
             dropped_nb,

tradein-mvp/backend/app/services/scrapers/price_brackets.py

@@ -0,0 +1,32 @@
+"""Общие seed price-brackets для exhaustive-скрапперов (avito/cian/yandex).
+
+EKB-настроенные (анализ прод-распределения вторички ЕКБ). Структура city-dict —
+готова под per-city (multi-city: Москва пик 15-40М, Тюмень — иные; multi-city —
+отдельная фича, см. vault decision). Сейчас единственный город — ekaterinburg.
+"""
+
+# (lo, hi) рубли; hi=None — открытый верхний брекет (без потолка). Закрытые
+# брекеты соприкасаются — вызывающий применяет hi-1 (эксклюзивный верх), дедуп
+# по source_id страхует на стыках.
+_EKB_SECONDARY: list[tuple[int, int | None]] = [
+    (0, 4_000_000),
+    (4_000_000, 5_000_000),
+    (5_000_000, 6_000_000),
+    (6_000_000, 7_000_000),
+    (7_000_000, 8_000_000),
+    (8_000_000, 10_000_000),
+    (10_000_000, 13_000_000),
+    (13_000_000, 17_000_000),
+    (17_000_000, 25_000_000),
+    (25_000_000, 250_000_000),
+    (250_000_000, None),  # 250М+ открытый catch-all (без потолка)
+]
+
+CITY_PRICE_SEED_BRACKETS: dict[str, list[tuple[int, int | None]]] = {
+    "ekaterinburg": _EKB_SECONDARY,
+}
+
+
+def get_price_seed_brackets(city: str = "ekaterinburg") -> list[tuple[int, int | None]]:
+    """Seed price-brackets для города (fallback на ekaterinburg)."""
+    return CITY_PRICE_SEED_BRACKETS.get(city, _EKB_SECONDARY)

tradein-mvp/backend/app/services/scrapers/yandex_realty.py

@@ -41,6 +41,7 @@ from curl_cffi.requests import AsyncSession as _CurlCffiSession
 from app.services.scraper_settings import get_scraper_delay
 from app.services.scrapers.base import BaseScraper, ScrapedLot
 from app.services.scrapers.browser_fetcher import BrowserFetcher
+from app.services.scrapers.price_brackets import get_price_seed_brackets
 
 logger = logging.getLogger(__name__)
 
@@ -816,6 +817,12 @@ class YandexRealtyScraper(BaseScraper):
         """Exhaustive EKB vtorichka load via KOMNATNOST x TSENA partitioning.
 
         Uses gate-API JSON (totalItems from pager) instead of DOM state extraction.
+        Стартует с общей data-driven seed-сетки ценовых брекетов
+        (get_price_seed_brackets() — shared EKB grid, общая с avito/cian) вместо
+        одной корень-бисекции [0, _YANDEX_MAX_PRICE]. Закрытые seed-брекеты передают
+        hi = br_hi - 1 (priceMax эксклюзивный сверху → нет пересечений), открытый
+        (br_hi is None) — hi=None (без priceMax, весь хвост люкса). Дедуп по
+        source_id в общем seen страхует на стыках.
         """
         _buckets = rooms_buckets if rooms_buckets is not None else list(ROOM_PATH.keys())
         seen: dict[str, ScrapedLot] = {}
@@ -828,17 +835,19 @@ class YandexRealtyScraper(BaseScraper):
                 concurrency,
             )
             before = len(seen)
-            await self._walk_price_range(
-                rooms=rooms_key,
-                lo=0,
-                hi=_YANDEX_MAX_PRICE,
-                seen=seen,
-                price_cap_per_bucket=price_cap_per_bucket,
-                max_pages_per_bucket=max_pages_per_bucket,
-                concurrency=concurrency,
-                on_bucket=on_bucket,
-                skip_buckets=skip_buckets,
-            )
+            for br_lo, br_hi in get_price_seed_brackets():
+                walk_hi = br_hi - 1 if br_hi is not None else None
+                await self._walk_price_range(
+                    rooms=rooms_key,
+                    lo=br_lo,
+                    hi=walk_hi,
+                    seen=seen,
+                    price_cap_per_bucket=price_cap_per_bucket,
+                    max_pages_per_bucket=max_pages_per_bucket,
+                    concurrency=concurrency,
+                    on_bucket=on_bucket,
+                    skip_buckets=skip_buckets,
+                )
             room_collected = len(seen) - before
             logger.info(
                 "yandex gate exhaustive: rooms=%s done -- collected %d (total unique=%d)",
@@ -857,7 +866,7 @@ class YandexRealtyScraper(BaseScraper):
         *,
         rooms: str | None,
         lo: int,
-        hi: int,
+        hi: int | None,
         seen: dict[str, ScrapedLot],
         price_cap_per_bucket: int,
         max_pages_per_bucket: int,
@@ -869,11 +878,15 @@ class YandexRealtyScraper(BaseScraper):
         """Recursive binary price-range partitioning.
 
         Probe page=1 -> read totalItems from pager.
-        totalItems > cap and bracket >= MIN_BRACKET -> split; else paginate leaf.
+        Closed bracket (hi is not None): totalItems > cap and bracket >= MIN_BRACKET
+        -> split; else paginate leaf.
+        Open bracket (hi is None): top seed-bracket without priceMax ceiling. Cannot
+        split (no mid) -> paginate leaf directly (lux tail tiny, tail-loss accepted).
         Fallback: totalItems=None -> paginate-until-empty (safe degradation).
         skip_buckets: resume checkpoint -- skip leaf buckets already done.
         """
         _lo_param = lo if lo > 0 else None
+        _hi_repr = "open" if hi is None else str(hi)
 
         probe_payload = await self._fetch_page_json(rooms, 1, _lo_param, hi)
         await asyncio.sleep(self.request_delay_sec)
@@ -883,10 +896,10 @@ class YandexRealtyScraper(BaseScraper):
 
         if probe_payload is None:
             logger.warning(
-                "yandex gate: probe failed rooms=%s [%d, %d] depth=%d -- rotating + retry",
+                "yandex gate: probe failed rooms=%s [%d, %s] depth=%d -- rotating + retry",
                 rooms,
                 lo,
-                hi,
+                _hi_repr,
                 _depth,
             )
             rotated = await self._rotate_ip()
@@ -902,16 +915,16 @@ class YandexRealtyScraper(BaseScraper):
                 probe_lots = _parse_gate_json(probe_payload, page_param=1)
             else:
                 logger.warning(
-                    "yandex gate: probe data extract failed rooms=%s [%d, %d]", rooms, lo, hi
+                    "yandex gate: probe data extract failed rooms=%s [%d, %s]", rooms, lo, _hi_repr
                 )
 
         if total is None:
             # Degraded path: paginate-until-empty without knowing total
             logger.warning(
-                "yandex gate: totalItems unknown rooms=%s [%d, %d] -- paginate until empty",
+                "yandex gate: totalItems unknown rooms=%s [%d, %s] -- paginate until empty",
                 rooms,
                 lo,
-                hi,
+                _hi_repr,
             )
             page = 1
             bucket_key = _combo_label(rooms, lo, hi)
@@ -935,12 +948,12 @@ class YandexRealtyScraper(BaseScraper):
                 on_bucket(bucket_key, len(seen))
             return
 
-        # Split or paginate
+        # Split or paginate. ОТКРЫТЫЙ брекет (hi is None) делить нельзя (нет mid) —
+        # сразу пагинируем leaf напрямую (хвост люкса крошечный, tail-loss accepted).
         cap = price_cap_per_bucket
         _min_bracket = 500_000
-        bracket = hi - lo
 
-        if total > cap and bracket >= _min_bracket and _depth < 8:
+        if hi is not None and total > cap and (hi - lo) >= _min_bracket and _depth < 8:
             mid = (lo + hi) // 2
             await self._walk_price_range(
                 rooms=rooms,

tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py

@@ -23,6 +23,9 @@ from app.services.scrapers.cian import (
     _MIN_BRACKET,
     CianScraper,
 )
+from app.services.scrapers.price_brackets import get_price_seed_brackets
+
+_SEED = get_price_seed_brackets()
 
 
 def _make_lot(source_id: str, price: int = 3_000_000) -> ScrapedLot:
@@ -258,8 +261,10 @@ async def test_fetch_all_secondary_on_bucket_called_per_leaf(scraper: CianScrape
             on_bucket=fake_on_bucket,
         )
 
-    # on_bucket вызван ровно 1 раз (один leaf-бакет для одной комнатности)
-    assert len(bucket_calls) == 1, f"Ожидался 1 вызов on_bucket, получено {len(bucket_calls)}"
+    # on_bucket вызван по разу на каждый seed-брекет (все непустые leaf-бакеты).
+    assert len(bucket_calls) == len(
+        _SEED
+    ), f"Ожидалось {len(_SEED)} вызовов on_bucket, получено {len(bucket_calls)}"
     key, lots = bucket_calls[0]
     # Ключ бакета содержит room_label:lo:hi
     assert "room1" in key, f"bucket_key должен содержать room1, got {key!r}"
@@ -366,9 +371,11 @@ async def test_skip_buckets_skips_pagination_and_on_bucket(scraper: CianScraper)
     def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
         on_bucket_calls.append(bucket_key)
 
-    # Сформируем ключ того же бакета, который получится при walk (room1:0:200000000)
-    bucket_key_to_skip = f"room1:0:{_MAX_PRICE}"
-    skip_set = {bucket_key_to_skip}
+    # Скипаем ВСЕ seed-бакеты комнатности room1. Закрытые → ':<hi-1>', открытый → ':open'.
+    skip_set: set[str] = set()
+    for br_lo, br_hi in _SEED:
+        hi_repr = "open" if br_hi is None else str(br_hi - 1)
+        skip_set.add(f"room1:{br_lo}:{hi_repr}")
 
     with (
         patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html),
@@ -384,19 +391,19 @@ async def test_skip_buckets_skips_pagination_and_on_bucket(scraper: CianScraper)
             skip_buckets=skip_set,
         )
 
-    # Probe (page=1) ВСЁ РАВНО вызывается (нужен для split-решения),
-    # но НЕ должно быть страниц пагинации (т.к. бакет в skip_buckets).
-    # Поскольку probe = page=1 и мы пропускаем ДО пагинации — fetch_page_html не должен вызываться.
-    assert (
-        len(fetch_calls) == 1
-    ), f"Ожидался ровно 1 вызов (probe), получено {len(fetch_calls)}: {fetch_calls}"
-    assert fetch_calls[0][1] == 1, "Probe должен быть page=1"
-    # on_bucket не вызывается для skip-бакета
+    # Probe (page=1) ВСЁ РАВНО вызывается на каждый seed-брекет (нужен для split-решения),
+    # но НЕ должно быть страниц пагинации (все бакеты в skip_buckets).
+    # Probe = page=1; пагинация пропущена ДО fetch остальных страниц → ровно len(_SEED) probe.
+    assert len(fetch_calls) == len(
+        _SEED
+    ), f"Ожидалось {len(_SEED)} probe-вызовов, получено {len(fetch_calls)}: {fetch_calls}"
+    assert all(c[1] == 1 for c in fetch_calls), "Все вызовы должны быть probe page=1"
+    # on_bucket не вызывается для skip-бакетов
     assert (
         len(on_bucket_calls) == 0
-    ), f"on_bucket не должен вызываться для skip-бакета, got: {on_bucket_calls}"
+    ), f"on_bucket не должен вызываться для skip-бакетов, got: {on_bucket_calls}"
     # lots пустой — ничего не собрали (пагинация пропущена)
-    assert len(lots) == 0, f"Ожидалось 0 лотов для skip-бакета, got {len(lots)}"
+    assert len(lots) == 0, f"Ожидалось 0 лотов для skip-бакетов, got {len(lots)}"
 
 
 @pytest.mark.asyncio
@@ -435,11 +442,18 @@ async def test_on_bucket_receives_key_and_lots(scraper: CianScraper) -> None:
             on_bucket=fake_on_bucket,
         )
 
-    assert len(received) == 1, f"Ожидался 1 вызов on_bucket, получено {len(received)}"
-    key, count = received[0]
-    # Ключ должен быть «room2:0:<MAX_PRICE>»
-    assert key == f"room2:0:{_MAX_PRICE}", f"Неверный bucket_key: {key!r}"
-    assert count == 5, f"Ожидалось 5 лотов, получено {count}"
+    assert len(received) == len(
+        _SEED
+    ), f"Ожидалось {len(_SEED)} вызовов on_bucket, получено {len(received)}"
+    keys = [k for (k, _c) in received]
+    # Закрытые брекеты: hi = br_hi - 1. Открытый (последний): ':open'.
+    first_lo, first_hi = _SEED[0]
+    assert first_hi is not None
+    assert keys[0] == f"room2:{first_lo}:{first_hi - 1}", f"Неверный ключ: {keys[0]!r}"
+    last_lo, last_hi = _SEED[-1]
+    assert last_hi is None
+    assert keys[-1] == f"room2:{last_lo}:open", f"Открытый ключ неверен: {keys[-1]!r}"
+    assert all(count == 5 for (_k, count) in received)
 
 
 # ── secondary_only фильтр ─────────────────────────────────────────────────────
@@ -542,12 +556,16 @@ async def test_secondary_only_drops_novostroyki(scraper: CianScraper) -> None:
     assert "vt_2" in lot_ids, "vt_2 (vtorichka) должен присутствовать"
     assert "none_seg" in lot_ids, "none_seg (listing_segment=None) должен присутствовать"
 
-    # on_bucket тоже не должен получить новостройки
-    assert len(on_bucket_calls) == 1, f"Ожидался 1 вызов on_bucket, получено {len(on_bucket_calls)}"
-    bucket_ids = {lot.source_id for lot in on_bucket_calls[0]}
-    assert "nb_1" not in bucket_ids
-    assert "nb_2" not in bucket_ids
-    assert len(on_bucket_calls[0]) == 3
+    # on_bucket вызывается на каждый seed-брекет (тот же микс на page=1 каждого),
+    # но новостройки отфильтрованы в каждом вызове.
+    assert len(on_bucket_calls) == len(
+        _SEED
+    ), f"Ожидалось {len(_SEED)} вызовов on_bucket, получено {len(on_bucket_calls)}"
+    for call_lots in on_bucket_calls:
+        bucket_ids = {lot.source_id for lot in call_lots}
+        assert "nb_1" not in bucket_ids
+        assert "nb_2" not in bucket_ids
+        assert len(call_lots) == 3
 
 
 @pytest.mark.asyncio
@@ -721,3 +739,54 @@ async def test_walk_price_range_no_warning_when_within_page_cap(
         f"Не ожидалось WARNING о tail loss при total={total_offers} <= page cap, "
         f"но получено: {[r.message for r in tail_warnings]}"
     )
+
+
+@pytest.mark.asyncio
+async def test_walk_price_range_open_bracket_no_split(scraper: CianScraper) -> None:
+    """Открытый брекет (hi=None) не делится даже при total > cap — leaf без maxprice.
+
+    bucket_key оканчивается на ':open'; maxprice в запросах всегда None (без потолка).
+    """
+    on_bucket_keys: list[str] = []
+    seen_max: list[int | None] = []
+
+    async def fake_fetch_page_html(
+        rooms: tuple, page: int, min_price: int | None, max_price: int | None
+    ) -> str:
+        seen_max.append(max_price)
+        return f"<html>page={page}</html>"
+
+    def fake_extract_total_offers(html: str) -> int | None:
+        return 5000  # > cap, но открытый брекет делить нельзя → leaf
+
+    def fake_parse_serp_html(html: str) -> list[ScrapedLot]:
+        import re
+
+        page_m = re.search(r"page=(\d+)", html)
+        page = int(page_m.group(1)) if page_m else 1
+        return [_make_lot(f"lot_p{page}")] if page == 1 else []
+
+    with (
+        patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers),
+        patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html),
+        patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock),
+        patch.object(scraper, "_rotate_ip", return_value=False),
+    ):
+        await scraper._walk_price_range(
+            rooms=(1,),
+            lo=250_000_000,
+            hi=None,
+            seen={},
+            price_cap_per_bucket=1400,
+            max_pages_per_bucket=54,
+            on_bucket=lambda k, lots: on_bucket_keys.append(k),
+        )
+
+    # Открытый брекет: ровно один leaf-бакет с ключом ':open', maxprice всегда None.
+    assert on_bucket_keys == [
+        "room1:250000000:open"
+    ], f"Открытый брекет должен дать один ':open'-бакет, got {on_bucket_keys}"
+    assert all(
+        m is None for m in seen_max
+    ), f"Открытый брекет: maxprice всегда None (без потолка), got {seen_max}"

tradein-mvp/backend/tests/scrapers/test_price_brackets.py

@@ -0,0 +1,48 @@
+"""Тесты общего seed price-brackets констант (avito/cian/yandex)."""
+
+from itertools import pairwise
+
+from app.services.scrapers.price_brackets import (
+    _EKB_SECONDARY,
+    CITY_PRICE_SEED_BRACKETS,
+    get_price_seed_brackets,
+)
+
+
+def test_default_returns_ekb():
+    """get_price_seed_brackets() без аргумента возвращает EKB-сетку."""
+    assert get_price_seed_brackets() == _EKB_SECONDARY
+    assert get_price_seed_brackets("ekaterinburg") == _EKB_SECONDARY
+
+
+def test_unknown_city_falls_back_to_ekb():
+    """Неизвестный город → fallback на EKB (не KeyError)."""
+    assert get_price_seed_brackets("atlantis") == _EKB_SECONDARY
+
+
+def test_last_bracket_is_open():
+    """Последний брекет ОТКРЫТ (hi=None) — catch-all хвоста люкса без потолка."""
+    brackets = get_price_seed_brackets()
+    assert brackets[-1][1] is None
+    # Только последний брекет открыт — остальные закрыты.
+    for lo, hi in brackets[:-1]:
+        assert hi is not None
+        assert hi > lo
+
+
+def test_brackets_are_contiguous():
+    """Закрытые брекеты соприкасаются: hi[i] == lo[i+1] (без дыр/перекрытий)."""
+    brackets = get_price_seed_brackets()
+    for (lo, hi), (next_lo, _next_hi) in pairwise(brackets):
+        assert hi == next_lo, f"разрыв между [{lo},{hi}] и [{next_lo},...]"
+
+
+def test_first_bracket_starts_at_zero():
+    """Первый брекет стартует с 0 (вся вторичка с нижнего края)."""
+    assert get_price_seed_brackets()[0][0] == 0
+
+
+def test_city_dict_contains_ekaterinburg():
+    """City-dict готов под multi-city — сейчас единственный ключ ekaterinburg."""
+    assert "ekaterinburg" in CITY_PRICE_SEED_BRACKETS
+    assert CITY_PRICE_SEED_BRACKETS["ekaterinburg"] is _EKB_SECONDARY

tradein-mvp/backend/tests/test_yandex_realty_serp.py

@@ -1108,3 +1108,71 @@ async def test_fetch_all_secondary_deduplicates_across_rooms():
     assert "unique_001" in ids
     assert "unique_002" in ids
     assert len(ids) == 3
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_walks_shared_seed_grid():
+    """fetch_all_secondary итерирует общую seed-сетку: len(_SEED) walk-вызовов/комнатность.
+
+    Закрытые брекеты передают hi = br_hi - 1; открытый (последний) — hi=None.
+    """
+    from app.services.scrapers.price_brackets import get_price_seed_brackets
+
+    seed = get_price_seed_brackets()
+    s = YandexRealtyScraper()
+    walk_args: list[tuple[int, int | None]] = []
+
+    async def fake_walk(*, rooms, lo, hi, seen, **kwargs):
+        walk_args.append((lo, hi))
+
+    with patch.object(s, "_walk_price_range", side_effect=fake_walk):
+        await s.fetch_all_secondary(rooms_buckets=["1"])
+
+    assert len(walk_args) == len(
+        seed
+    ), f"Ожидалось {len(seed)} walk-вызовов (seed-брекеты), получено {len(walk_args)}"
+    # Первый брекет закрыт: hi = seed[0].hi - 1.
+    first_lo, first_hi = seed[0]
+    assert first_hi is not None
+    assert walk_args[0] == (first_lo, first_hi - 1)
+    # Последний брекет открыт: hi=None (без priceMax-потолка).
+    last_lo, last_hi = seed[-1]
+    assert last_hi is None
+    assert walk_args[-1] == (last_lo, None)
+
+
+@pytest.mark.asyncio
+async def test_walk_price_range_open_bracket_no_split():
+    """Открытый брекет (hi=None) не делится — пагинируется leaf напрямую (one bucket).
+
+    totalItems > cap, но hi=None → split невозможен → ровно один leaf-бакет.
+    """
+    s = YandexRealtyScraper()
+    bucket_calls: list = []
+    seen_prices: list[int | None] = []
+
+    async def fake_fetch_page_json(rooms, page, price_min, price_max):
+        seen_prices.append(price_max)
+        # total > cap, но 1 страница лотов → leaf пагинируется без split
+        pager = {"page": 0, "pageSize": 20, "totalItems": 9999, "totalPages": 1}
+        return _make_gate_payload([_ENTITY_FULL], pager)
+
+    seen: dict = {}
+    with patch.object(s, "_fetch_page_json", side_effect=fake_fetch_page_json):
+        with patch("asyncio.sleep", new_callable=AsyncMock):
+            await s._walk_price_range(
+                rooms="1",
+                lo=250_000_000,
+                hi=None,
+                seen=seen,
+                price_cap_per_bucket=500,
+                max_pages_per_bucket=5,
+                concurrency=2,
+                on_bucket=lambda k, n: bucket_calls.append(k),
+            )
+
+    # Открытый брекет: НЕ делится (один on_bucket), priceMax всегда None (без потолка).
+    assert len(bucket_calls) == 1, f"Открытый брекет не должен делиться, got {bucket_calls}"
+    assert all(
+        pm is None for pm in seen_prices
+    ), f"Открытый брекет: priceMax всегда None, got {seen_prices}"