feat(scraper-kit): ценовой планировщик коридоров + счётчики усечения

Потолок выдачи площадки (Avito ~1500 / Cian 1512 / Yandex 500 / DomClick 2000)
по Свердловской области не мешал — вторичка влезает в сид-брекеты. По Москве и
области лотов десятки тысяч, и прогон молча забирал верхушку выдачи, рапортуя
успех. Движок бисекции в ките уже был, но усечение он нигде не фиксировал.

pricing/planner.py — синхронный ЧИСТЫЙ планировщик plan_price_corridors():
на вход count_of(lo, hi), на выход коридоры + PlanStats. Алгоритм перенесён из
обкатанного живьём scripts/local-avito-msk/collect.py: геометрическая середина
sqrt(lo*hi) (цены логнормальны — арифметическая отрезает пустую верхнюю
половину), поиск верхней границы удвоением от 8 млн, предохранители MAX_DEPTH=12
и MIN_WIDTH_RATIO=1.05. Неделимый переполненный коридор помечается truncated, а
его недобор попадает в missed_estimate.

Добавлен третий предохранитель, которого в образце нет: атомарный коридор
(hi-lo<=1) не имеет строгой внутренней точки, geometric_mid вернул бы саму
границу и половина совпала бы с родителем — рекурсия по себе, от которой
collect.py спасает только MAX_DEPTH. Покрыто тестом.

pricing/bisection.py — BisectionStats: probes/leaves/truncated_leaves/
missed_estimate/unknown_leaves/skipped_probe_fail/degraded. Параметр stats
опционален и по умолчанию None, поэтому дерево рекурсии не меняется ни на
ветку — golden-parity всех четырёх провайдеров на месте (21 существующий тест
бисекции зелёный).

Регион 66 и НЕИЗВЕСТНЫЙ регион — бисекция выключена: там она не нужна (всё
влезает в сид-брекеты), а лишний probe — лишний запрос под антибот. None
трактуется как 66, а не как «включить»: провайдеры фоллбэчат на ЕКБ-дефолт, и
«безопаснее включить» означало бы молча поменять боевое поведение прогонов,
которые просто не прокинули параметр. Выключенный режим всё равно делает
корневой probe и репортит усечение — молчания нет и там.

tradein-mvp/backend/tests/test_price_corridor_planner.py

@@ -0,0 +1,261 @@
+"""Тесты чистого планировщика ценовых коридоров (`scraper_kit.pricing.planner`).
+
+Планировщик синхронный и чистый: «сколько позиций в коридоре» инжектится
+функцией, поэтому здесь НЕТ ни сети, ни моков площадок — только модели
+распределения цен.
+"""
+
+from __future__ import annotations
+
+import bisect
+
+import pytest
+from scraper_kit.pricing.planner import (
+    EKB_REGION_CODE,
+    MAX_DEPTH,
+    MIN_WIDTH_RATIO,
+    geometric_mid,
+    plan_price_corridors,
+    price_bisection_enabled,
+    width_ratio,
+)
+
+
+def _counter(prices: list[int]):
+    """`count_of` поверх отсортированного списка цен: сколько попало в [lo, hi]."""
+    ordered = sorted(prices)
+
+    def count_of(lo: int | None, hi: int | None) -> int:
+        left = 0 if lo is None else bisect.bisect_left(ordered, lo)
+        right = len(ordered) if hi is None else bisect.bisect_right(ordered, hi)
+        return max(0, right - left)
+
+    return count_of
+
+
+def _covers_every_price(plan, prices: list[int]) -> bool:
+    """Каждая цена попадает хотя бы в один коридор плана."""
+    return all(
+        any((c.lo is None or p >= c.lo) and (c.hi is None or p <= c.hi) for c in plan.corridors)
+        for p in prices
+    )
+
+
+# ── равномерное распределение ────────────────────────────────────────────────
+
+
+def test_uniform_spread_splits_until_every_corridor_fits_target() -> None:
+    """Равномерная выборка делится, пока каждый коридор не влезет в target."""
+    prices = list(range(1_000_000, 21_000_000, 2_000))  # 10 000 позиций
+    plan = plan_price_corridors(
+        count_of=_counter(prices),
+        target=1_200,
+        hard_cap=1_500,
+        lo=1_000_000,
+        hi=21_000_000,
+    )
+
+    assert plan.stats.root_count == len(prices)
+    assert plan.stats.corridors > 1, "10k позиций при target=1200 обязаны поделиться"
+    assert plan.stats.truncated_corridors == 0
+    assert plan.stats.missed_estimate == 0
+    assert plan.stats.complete
+    assert all(c.count is not None and c.count <= 1_200 for c in plan.corridors)
+    assert _covers_every_price(plan, prices)
+
+
+def test_small_sample_stays_one_corridor() -> None:
+    """Выборка ниже target не делится вовсе — один коридор, один probe."""
+    prices = list(range(2_000_000, 3_000_000, 10_000))  # 100 позиций
+    plan = plan_price_corridors(count_of=_counter(prices), target=1_200)
+
+    assert plan.stats.corridors == 1
+    assert plan.stats.probes == 1
+    assert plan.stats.complete
+
+
+# ── вырожденный случай: всё в одной цене ─────────────────────────────────────
+
+
+def test_single_price_point_hits_min_width_ratio_and_is_marked_truncated() -> None:
+    """«Всё в одной цене» упирается в MIN_WIDTH_RATIO, а не зацикливается.
+
+    Коридор вокруг единственной цены сузить до нужного размера невозможно —
+    делить бессмысленно. Планировщик обязан остановиться и ПОМЕТИТЬ коридор
+    усечённым с честным недобором, а не молча отдать план «всё влезло».
+    """
+    prices = [3_000_000] * 5_000
+    plan = plan_price_corridors(
+        count_of=_counter(prices),
+        target=1_200,
+        hard_cap=1_500,
+        lo=1_000_000,
+        hi=10_000_000,
+    )
+
+    assert plan.stats.truncated_corridors >= 1
+    assert plan.stats.missed_estimate > 0
+    assert not plan.stats.complete
+
+    overflowing = [c for c in plan.corridors if c.count and c.count > 1_500]
+    assert overflowing, "переполненный коридор обязан остаться в плане"
+    for c in overflowing:
+        assert c.truncated, f"коридор {c.label()} переполнен, но не помечен"
+        assert c.missed == c.count - 1_500
+        # Остановились по предохранителю, а не «просто так».
+        assert c.depth >= MAX_DEPTH or width_ratio(c.lo, c.hi) <= MIN_WIDTH_RATIO
+
+    assert plan.stats.missed_estimate == sum(c.missed for c in plan.truncated)
+
+
+def test_degenerate_plan_terminates_within_depth_and_probe_budget() -> None:
+    """Вырожденный вход завершается: глубина ограничена, число probe конечно."""
+    plan = plan_price_corridors(
+        count_of=_counter([3_000_000] * 5_000),
+        target=1_200,
+        lo=1_000_000,
+        hi=10_000_000,
+    )
+
+    assert plan.stats.max_depth_reached <= MAX_DEPTH
+    assert plan.stats.probes < 2 ** (MAX_DEPTH + 1)
+
+
+def test_max_depth_is_never_exceeded_on_open_ended_range() -> None:
+    """Даже на открытом диапазоне глубина не превышает MAX_DEPTH."""
+    prices = list(range(500_000, 100_000_000, 5_000))
+    plan = plan_price_corridors(count_of=_counter(prices), target=1_000)
+
+    assert plan.stats.max_depth_reached <= MAX_DEPTH
+    assert all(c.depth <= MAX_DEPTH for c in plan.corridors)
+
+
+@pytest.mark.parametrize("max_depth", [1, 2, 3])
+def test_custom_max_depth_caps_recursion(max_depth: int) -> None:
+    """Пониженный max_depth режет дерево раньше — и недобор виден в статистике."""
+    prices = list(range(1_000_000, 21_000_000, 2_000))
+    plan = plan_price_corridors(
+        count_of=_counter(prices),
+        target=1_200,
+        hard_cap=1_500,
+        lo=1_000_000,
+        hi=21_000_000,
+        max_depth=max_depth,
+    )
+
+    assert plan.stats.max_depth_reached <= max_depth
+    # Мелкий потолок глубины физически не даёт добрать всё — это обязано
+    # проявиться усечением, а не «успехом».
+    assert plan.stats.truncated_corridors > 0
+    assert plan.stats.missed_estimate > 0
+
+
+# ── регион 66: поведение прежнее ─────────────────────────────────────────────
+
+
+def test_region_66_disables_bisection() -> None:
+    """Свердловская обл. — бисекция выключена; неизвестный регион трактуем как 66."""
+    assert price_bisection_enabled(EKB_REGION_CODE) is False
+    assert price_bisection_enabled(None) is False
+    assert price_bisection_enabled(77) is True  # Москва
+    assert price_bisection_enabled(50) is True  # Московская обл.
+
+
+def test_region_66_yields_exactly_one_corridor_over_whole_range() -> None:
+    """С выключенной бисекцией план = ровно один коридор на весь диапазон.
+
+    Это и есть «поведение региона 66 осталось прежним»: ни одного лишнего
+    запроса к площадке сверх единственного корневого probe.
+    """
+    prices = list(range(1_000_000, 21_000_000, 2_000))
+    plan = plan_price_corridors(
+        count_of=_counter(prices),
+        target=1_200,
+        lo=1_000_000,
+        hi=21_000_000,
+        enabled=price_bisection_enabled(EKB_REGION_CODE),
+    )
+
+    assert plan.stats.corridors == 1
+    assert plan.stats.probes == 1
+    only = plan.corridors[0]
+    assert (only.lo, only.hi) == (1_000_000, 21_000_000)
+    assert only.depth == 0
+    assert _covers_every_price(plan, prices)
+
+
+def test_disabled_plan_still_reports_truncation() -> None:
+    """Выключенная бисекция не значит «молчать»: переполнение всё равно в счётчиках."""
+    plan = plan_price_corridors(
+        count_of=_counter(list(range(1_000_000, 21_000_000, 2_000))),
+        target=1_200,
+        hard_cap=1_500,
+        enabled=False,
+    )
+
+    assert plan.stats.corridors == 1
+    assert plan.stats.truncated_corridors == 1
+    assert plan.stats.missed_estimate == 10_000 - 1_500
+    assert not plan.stats.complete
+
+
+# ── непрочитанный счётчик ────────────────────────────────────────────────────
+
+
+def test_unreadable_counter_is_marked_not_silently_trusted() -> None:
+    """`count_of` вернул None → коридор слепой: truncated + unknown, не «пусто»."""
+    plan = plan_price_corridors(count_of=lambda lo, hi: None, target=1_200)
+
+    assert plan.stats.corridors == 1
+    assert plan.stats.unknown_corridors == 1
+    assert plan.corridors[0].unknown_count is True
+    assert plan.corridors[0].truncated is True
+    assert not plan.stats.complete
+
+
+# ── геометрическая середина ──────────────────────────────────────────────────
+
+
+def test_geometric_mid_is_geometric_not_arithmetic() -> None:
+    """sqrt(lo*hi), а не (lo+hi)/2 — иначе верхняя половина вырожденно пуста."""
+    assert geometric_mid(1_000_000, 100_000_000) == 10_000_000
+    assert geometric_mid(1_000_000, 100_000_000) != (1_000_000 + 100_000_000) // 2
+
+
+def test_geometric_mid_stays_strictly_inside_the_corridor() -> None:
+    """При hi-lo>=2 середина строго внутри — обе половины непусты."""
+    for lo, hi in ((1_000_000, 1_000_002), (1_000_000, 100_000_000), (None, 700_000)):
+        mid = geometric_mid(lo, hi)
+        assert mid < hi
+        if lo is not None:
+            assert mid > lo
+
+
+def test_atomic_corridor_is_not_split_and_is_marked_truncated() -> None:
+    """Коридор шириной в 1 ₽ делить нечем: предохранитель, а не рекурсия по себе.
+
+    `geometric_mid` при hi-lo<=1 строгой внутренней точки дать не может, и
+    половина совпала бы с родителем. Планировщик обязан остановиться сам, не
+    полагаясь на MAX_DEPTH как на единственный тормоз — поэтому
+    min_width_ratio здесь занижен до 1.0, чтобы снять штатный предохранитель.
+    """
+    plan = plan_price_corridors(
+        count_of=_counter([5_000_000] * 3_000),
+        target=100,
+        hard_cap=100,
+        lo=5_000_000,
+        hi=5_000_001,
+        min_width_ratio=1.0,
+    )
+
+    assert plan.stats.corridors == 1
+    only = plan.corridors[0]
+    assert (only.lo, only.hi) == (5_000_000, 5_000_001)
+    assert only.truncated is True
+    assert only.missed == 3_000 - 100
+    assert plan.stats.max_depth_reached == 0
+
+
+def test_width_ratio_open_top_is_infinite() -> None:
+    assert width_ratio(1_000_000, None) == float("inf")
+    assert width_ratio(1_000_000, 2_000_000) == pytest.approx(2.0)

tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/__init__.py

@@ -1,23 +1,63 @@
 """scraper_kit.pricing — общие примитивы ценового партиционирования SERP-обхода.
 
-Единый движок адаптивной бисекции «комнатность × цена» для всех провайдеров
-(avito / cian / yandex / domclick). См. `bisection.py`.
+Два слоя, намеренно разделённые:
+
+* `bisection.py` — боевой async-движок обхода `walk_price_range`: ходит в сеть
+  через provider-callback'и, пагинирует листья. `BisectionStats` — счётчики
+  усечения для уже существующих свипов.
+* `planner.py` — СИНХРОННЫЙ ЧИСТЫЙ планировщик `plan_price_corridors`: на вход
+  «сколько позиций в коридоре», на выход список коридоров + `PlanStats`.
+  Геометрическая середина `sqrt(lo*hi)` (цены логнормальны) и предохранители
+  `MAX_DEPTH` / `MIN_WIDTH_RATIO`; перенесён из обкатанного живьём
+  `scripts/local-avito-msk/collect.py`.
 """
 
 from __future__ import annotations
 
 from scraper_kit.pricing.bisection import (
     BisectionConfig,
+    BisectionStats,
     ProbeFailPolicy,
     ProbeResult,
     done_range_skipper,
     walk_price_range,
 )
+from scraper_kit.pricing.planner import (
+    EKB_REGION_CODE,
+    MAX_DEPTH,
+    MIN_WIDTH_RATIO,
+    PRICE_CEIL,
+    PRICE_FLOOR,
+    PRICE_PROBE_START,
+    Corridor,
+    CountFn,
+    PlanStats,
+    PricePlan,
+    geometric_mid,
+    plan_price_corridors,
+    price_bisection_enabled,
+    width_ratio,
+)
 
 __all__ = [
+    "EKB_REGION_CODE",
+    "MAX_DEPTH",
+    "MIN_WIDTH_RATIO",
+    "PRICE_CEIL",
+    "PRICE_FLOOR",
+    "PRICE_PROBE_START",
     "BisectionConfig",
+    "BisectionStats",
+    "Corridor",
+    "CountFn",
+    "PlanStats",
+    "PricePlan",
     "ProbeFailPolicy",
     "ProbeResult",
     "done_range_skipper",
+    "geometric_mid",
+    "plan_price_corridors",
+    "price_bisection_enabled",
     "walk_price_range",
+    "width_ratio",
 ]

tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/bisection.py

@@ -88,6 +88,63 @@ class ProbeResult:
     payload: Any = None
 
 
+@dataclass
+class BisectionStats:
+    """Счётчики одного обхода — чтобы усечение выдачи было ЧИСЛОМ, а не тишиной.
+
+    Движок делит диапазон, пока `count > cap`, но деление не бесконечно: упёрлись
+    в `min_bracket` или в `max_depth` — и переполненный диапазон всё равно уходит
+    в `on_leaf`, где пагинация штатно дорежет его до `cap`. Разница `count - cap`
+    в этот момент теряется БЕЗ СЛЕДА: по Свердловской области это никогда не
+    срабатывало (вторичка влезает в сид-брекеты), а по Москве и области, где
+    лотов десятки тысяч, прогон забирал верхушку и рапортовал успех.
+
+    Объект МУТАБЕЛЕН и передаётся в `walk_price_range` опционально: по умолчанию
+    ``None`` — ни одной новой ветки в обходе, дерево рекурсии байт-в-байт прежнее
+    (golden-parity провайдеров не двигается). Вызывающий складывает
+    `missed_estimate` в счётчики прогона.
+    """
+
+    #: Сколько раз дёрнули provider-probe (стоимость планирования в запросах).
+    probes: int = 0
+    #: Сколько диапазонов дошло до пагинации.
+    leaves: int = 0
+    #: Листья, где `count > cap` — поделить не вышло, выдача будет обрезана.
+    truncated_leaves: int = 0
+    #: Оценка недобора: сумма `count - cap` по усечённым листьям.
+    missed_estimate: int = 0
+    #: Листья с непрочитанным счётчиком (полнота недоказуема, не то же, что 0).
+    unknown_leaves: int = 0
+    #: Диапазоны, брошенные по ProbeFailPolicy.SKIP — тоже дыра в покрытии.
+    skipped_probe_fail: int = 0
+    #: Переходы в «пагинировать-до-пустоты» без знания total (yandex DEGRADE).
+    degraded: int = 0
+    max_depth_reached: int = 0
+
+    def record_probe(self, depth: int) -> None:
+        self.probes += 1
+        self.max_depth_reached = max(self.max_depth_reached, depth)
+
+    def record_leaf(self, count: int | None, cap: int) -> None:
+        self.leaves += 1
+        if count is None:
+            self.unknown_leaves += 1
+            return
+        if count > cap:
+            self.truncated_leaves += 1
+            self.missed_estimate += count - cap
+
+    @property
+    def complete(self) -> bool:
+        """Обход покрыл выборку целиком — ни усечений, ни пропусков, ни слепых листьев."""
+        return (
+            self.truncated_leaves == 0
+            and self.unknown_leaves == 0
+            and self.skipped_probe_fail == 0
+            and self.degraded == 0
+        )
+
+
 @dataclass(frozen=True)
 class BisectionConfig:
     """Per-provider константы/политики бисекции (все различия провайдеров тут).
@@ -222,6 +279,7 @@ async def walk_price_range(
     on_leaf: LeafFn,
     on_degraded: DegradedFn | None = None,
     should_skip: SkipFn | None = None,
+    stats: BisectionStats | None = None,
     depth: int = 0,
 ) -> None:
     """Рекурсивно обойти ценовой диапазон [lo, hi] адаптивной бисекцией.
@@ -249,6 +307,10 @@ async def walk_price_range(
             (:func:`done_range_skipper`, #3315). Проверяется на КАЖДОМ узле
             рекурсии: у частично покрытого диапазона probe остаётся, а его уже
             готовые половины отсекаются на входе, а не в листе после probe.
+        stats: опциональный :class:`BisectionStats` — мутабельный аккумулятор
+            probe/leaf/усечений. ``None`` (умолчание) = ни одной новой ветки в
+            обходе. Передан — вызывающий получает `missed_estimate` и обязан
+            донести его до счётчиков прогона, иначе усечение снова станет тишиной.
         depth: текущая глубина рекурсии (для ``max_depth``).
     """
     if should_skip is not None and should_skip(lo, hi):
@@ -256,22 +318,34 @@ async def walk_price_range(
 
     result = await probe(lo, hi, depth)
     total = result.count
+    if stats is not None:
+        stats.record_probe(depth)
 
     # ── Probe-fail: счётчик недоступен ────────────────────────────────────────
     if total is None:
         policy = config.probe_fail_policy
         if policy is ProbeFailPolicy.SKIP:
+            if stats is not None:
+                stats.skipped_probe_fail += 1
             return
         if policy is ProbeFailPolicy.DEGRADE:
+            if stats is not None:
+                stats.degraded += 1
             if on_degraded is not None:
                 await on_degraded(lo, hi)
             return
         # SPLIT_OR_SKIP (avito): открытый брекет пагинируем best-effort; закрытый
         # ещё делимый — принудительно делим; иначе пропускаем (хвост теряем).
         if hi is None:
+            if stats is not None:
+                stats.record_leaf(None, config.cap)
             await on_leaf(lo, hi, result)
             return
         _lo = lo if lo is not None else config.open_split_floor
+        if not _can_split_span(hi - _lo, config) and stats is not None:
+            # Закрытый неделимый диапазон без счётчика просто бросается — это
+            # такая же дыра в покрытии, как SKIP, и она обязана быть видна.
+            stats.skipped_probe_fail += 1
         if _can_split_span(hi - _lo, config):
             mid = (_lo + hi) // 2
             await walk_price_range(
@@ -282,6 +356,7 @@ async def walk_price_range(
                 on_leaf=on_leaf,
                 on_degraded=on_degraded,
                 should_skip=should_skip,
+                stats=stats,
                 depth=depth + 1,
             )
             await walk_price_range(
@@ -292,6 +367,7 @@ async def walk_price_range(
                 on_leaf=on_leaf,
                 on_degraded=on_degraded,
                 should_skip=should_skip,
+                stats=stats,
                 depth=depth + 1,
             )
         return
@@ -303,6 +379,8 @@ async def walk_price_range(
     # Открытый брекет без split-потолка (avito/cian/yandex): делить нельзя —
     # пагинируем leaf напрямую (хвост люкса крошечный, tail-loss accepted).
     if hi is None and config.open_split_ceiling is None:
+        if stats is not None:
+            stats.record_leaf(total, config.cap)
         await on_leaf(lo, hi, result)
         return
 
@@ -329,6 +407,7 @@ async def walk_price_range(
             on_leaf=on_leaf,
             on_degraded=on_degraded,
             should_skip=should_skip,
+            stats=stats,
             depth=depth + 1,
         )
         await walk_price_range(
@@ -339,9 +418,15 @@ async def walk_price_range(
             on_leaf=on_leaf,
             on_degraded=on_degraded,
             should_skip=should_skip,
+            stats=stats,
             depth=depth + 1,
         )
         return
 
     # Leaf-бакет: пагинация + дедуп + on_bucket — целиком на стороне провайдера.
+    # Сюда же приезжает переполненный диапазон, который НЕ дали поделить
+    # min_bracket / max_depth: пагинация дорежет его до cap, и без stats разница
+    # count-cap пропала бы бесследно.
+    if stats is not None:
+        stats.record_leaf(total, config.cap)
     await on_leaf(lo, hi, result)

tradein-mvp/packages/scraper-kit/src/scraper_kit/pricing/planner.py

@@ -0,0 +1,284 @@
+"""Чистый планировщик ценовых коридоров под жёсткий потолок выдачи площадки.
+
+ЗАЧЕМ. У каждой площадки жёсткий cap на набор фильтров: Avito ~1500 позиций,
+Cian 1512, Yandex 500, DomClick 2000. По Свердловской области это не мешало —
+вторичка целиком влезает в сид-брекеты (`scraper_kit.price_brackets`). По Москве
+и области лотов десятки тысяч (Cian 153k на объединённом скоупе, Yandex 31k,
+DomClick 17.5k по области), и без дробления по цене прогон ТИХО забирает верхушку
+выдачи и рапортует успех. Этот модуль превращает молчаливое усечение в число.
+
+ЧЕМ ОТЛИЧАЕТСЯ ОТ `pricing.bisection`. `walk_price_range` — боевой async-движок
+обхода: он ходит в сеть, пагинирует листья и завязан на provider-callback'и, а
+его арифметика деления — `mid = (lo + hi) // 2`. Здесь СИНХРОННАЯ ЧИСТАЯ функция:
+на вход «сколько позиций в коридоре», на выход список коридоров + честная
+статистика. Ни одного сетевого вызова внутри, поэтому она тестируется как
+обычная функция и её можно прогнать на модели распределения до прогона.
+
+Алгоритм перенесён из обкатанного живьём `scripts/local-avito-msk/collect.py`
+(:245-249, :303-336, :1408-1434) и сохраняет три его решения:
+
+1. **Геометрическая середина** `sqrt(lo*hi)`, а не арифметическая. Цены
+   логнормальны: арифметическая середина коридора 1М..100М (≈50М) отрезает
+   вырожденно-пустую верхнюю половину, и дерево вырождается в лестницу —
+   деление идёт, а переполненный левый край так и не разъезжается.
+2. **Верхняя граница ищется удвоением** от `probe_start` (8 млн ₽), а не берётся
+   константой: потолок рынка по региону заранее неизвестен.
+3. **Два предохранителя** — `MAX_DEPTH` и `MIN_WIDTH_RATIO`. Неделимый
+   переполненный коридор НЕ молчит: он помечается `truncated` и его недобор
+   попадает в `PlanStats.missed_estimate`.
+"""
+
+from __future__ import annotations
+
+import math
+from collections.abc import Callable
+from dataclasses import dataclass, field
+
+# Нижняя граница осмысленного коридора, ₽ (подстановка вместо None-lo в арифметике).
+PRICE_FLOOR = 500_000
+# Старт удвоения при поиске верхней границы открытого коридора, ₽.
+PRICE_PROBE_START = 8_000_000
+# Абсолютный потолок поиска верхней границы, ₽.
+PRICE_CEIL = 2_000_000_000
+# Уже этого (по ГЕОМЕТРИЧЕСКОЙ ширине hi/lo) коридор не делим.
+MIN_WIDTH_RATIO = 1.05
+# Жёсткий предел глубины рекурсии: гарантия остановки, не более 2^12 листьев.
+MAX_DEPTH = 12
+
+# Свердловская область: текущий боевой скоуп, где бисекция не нужна (см.
+# `price_bisection_enabled`).
+EKB_REGION_CODE = 66
+
+#: «Сколько всего позиций в коридоре [lo, hi]». ``None`` — счётчик не прочитался
+#: (captcha / битая страница), это НЕ ноль. Границы включительные, ``None`` = без
+#: границы.
+CountFn = Callable[[int | None, int | None], int | None]
+
+
+@dataclass(frozen=True)
+class Corridor:
+    """Один ценовой коридор плана."""
+
+    lo: int | None
+    hi: int | None
+    count: int | None = None
+    #: Коридор заведомо не влезает в потолок площадки и поделить его уже нельзя.
+    truncated: bool = False
+    #: Оценка недобора: ``count - hard_cap`` (0, если count неизвестен).
+    missed: int = 0
+    depth: int = 0
+    #: Счётчик не прочитался — коридор уходит в прогон, но за полноту не ручаемся.
+    unknown_count: bool = False
+
+    def label(self) -> str:
+        lo = "-" if self.lo is None else f"{self.lo:_}"
+        hi = "-" if self.hi is None else f"{self.hi:_}"
+        return f"[{lo} .. {hi}]"
+
+
+@dataclass(frozen=True)
+class PlanStats:
+    """Честная сводка по плану. Усечение обязано быть видно здесь, а не в логе."""
+
+    corridors: int = 0
+    truncated_corridors: int = 0
+    #: Сколько позиций заведомо не будет собрано — сумма `missed` по усечённым.
+    missed_estimate: int = 0
+    #: Коридоры с непрочитанным счётчиком (полнота недоказуема).
+    unknown_corridors: int = 0
+    #: Сколько раз дёрнули `count_of` (стоимость планирования в запросах).
+    probes: int = 0
+    max_depth_reached: int = 0
+    #: Счётчик по корневому запросу — сколько всего в скоупе.
+    root_count: int | None = None
+
+    @property
+    def complete(self) -> bool:
+        """План покрывает выборку целиком (ни усечений, ни слепых коридоров)."""
+        return self.truncated_corridors == 0 and self.unknown_corridors == 0
+
+
+@dataclass(frozen=True)
+class PricePlan:
+    corridors: list[Corridor] = field(default_factory=list)
+    stats: PlanStats = field(default_factory=PlanStats)
+
+    @property
+    def truncated(self) -> list[Corridor]:
+        return [c for c in self.corridors if c.truncated]
+
+
+def geometric_mid(lo: int | None, hi: int, *, price_floor: int = PRICE_FLOOR) -> int:
+    """Геометрическая середина коридора: ``sqrt(lo*hi)``.
+
+    Цены логнормальны, поэтому арифметическая середина делит ВЫБОРКУ крайне
+    неровно. Результат зажат в ``[lo+1, hi-1]``, чтобы деление всегда давало две
+    непустые половины и рекурсия не могла зациклиться на границе.
+
+    ПРЕДУСЛОВИЕ: ``hi - lo >= 2``. У атомарного коридора (``hi - lo <= 1``)
+    строгой внутренней точки не существует вовсе, и зажим вернул бы саму границу
+    — половина совпала бы с родителем, то есть рекурсия по себе. Делимость
+    проверяет вызывающий (`plan_price_corridors`), см. там ветку `upper - eff_lo
+    <= 1`; в исходнике `collect.py` этой проверки нет, и от бесконечного спуска
+    его спасает только MAX_DEPTH.
+    """
+    low = max(int(lo or price_floor), 1)
+    mid = int(math.sqrt(low * float(hi)))
+    return max(low + 1, min(hi - 1, mid))
+
+
+def width_ratio(lo: int | None, hi: int | None, *, price_floor: int = PRICE_FLOOR) -> float:
+    """Геометрическая ширина коридора ``hi/lo`` (``inf`` для открытого верха)."""
+    if hi is None:
+        return float("inf")
+    return float(hi) / max(float(lo or price_floor), 1.0)
+
+
+def price_bisection_enabled(region_code: int | None) -> bool:
+    """Нужна ли ценовая бисекция для региона.
+
+    УМОЛЧАНИЕ — ВЫКЛЮЧЕНО ДЛЯ 66 (Свердловская обл.) И ДЛЯ НЕИЗВЕСТНОГО РЕГИОНА.
+    Причина не в том, что бисекция там вредна, а в том, что она там НЕ НУЖНА и
+    поэтому её включение было бы чистым регрессом: вторичка области целиком
+    влезает в сид-брекеты `price_brackets`, счётчик ни в одном брекете не бьётся
+    в cap, и каждый лишний probe — это лишний запрос к площадке под антибот. То
+    есть для 66 план обязан быть РОВНО ТЕМ ЖЕ обходом, что и сегодня.
+
+    ``None`` (регион не передали) трактуем как 66, а не как «включить»: провайдеры
+    фоллбэчат на ЕКБ-дефолт, когда регион не задан, и «безопаснее включить»
+    здесь означало бы молча поменять боевое поведение прогонов, которые просто
+    не прокинули параметр.
+
+    Для всех остальных регионов — включено: именно там (77/50 — Москва и область)
+    выдача кратно превышает cap площадки.
+    """
+    if region_code is None:
+        return False
+    return int(region_code) != EKB_REGION_CODE
+
+
+def plan_price_corridors(
+    *,
+    count_of: CountFn,
+    target: int,
+    hard_cap: int | None = None,
+    lo: int | None = None,
+    hi: int | None = None,
+    enabled: bool = True,
+    max_depth: int = MAX_DEPTH,
+    min_width_ratio: float = MIN_WIDTH_RATIO,
+    price_floor: int = PRICE_FLOOR,
+    probe_start: int = PRICE_PROBE_START,
+    price_ceil: int = PRICE_CEIL,
+) -> PricePlan:
+    """Разбить ценовой диапазон на коридоры, влезающие в потолок выдачи.
+
+    Args:
+        count_of: «сколько позиций в [lo, hi]»; ``None`` — счётчик не прочитался.
+        target: целевой размер коридора. Берётся С ЗАПАСОМ ниже ``hard_cap``:
+            счётчик площадки врёт в обе стороны и выдача плывёт между probe и
+            пагинацией, поэтому коридор ровно в cap приезжает усечённым.
+        hard_cap: жёсткий потолок выдачи площадки; по умолчанию ``target``.
+            Недобор считается от него, а не от ``target``.
+        lo / hi: границы корневого диапазона (``None`` = без границы).
+        enabled: ``False`` -> ровно ОДИН коридор на весь диапазон, ноль делений
+            (режим региона 66, см. :func:`price_bisection_enabled`). Корневой
+            probe всё равно делается — чтобы усечение, если оно там есть, попало
+            в статистику, а не осталось незамеченным.
+        max_depth / min_width_ratio: предохранители, см. модульный docstring.
+
+    Returns:
+        :class:`PricePlan` — коридоры в порядке обхода + :class:`PlanStats`.
+    """
+    cap = hard_cap if hard_cap is not None else target
+    corridors: list[Corridor] = []
+    probes = 0
+    deepest = 0
+
+    def probe(p_lo: int | None, p_hi: int | None) -> int | None:
+        nonlocal probes
+        probes += 1
+        return count_of(p_lo, p_hi)
+
+    def emit(c_lo: int | None, c_hi: int | None, count: int | None, depth: int) -> None:
+        nonlocal deepest
+        deepest = max(deepest, depth)
+        if count is None:
+            # Слепой коридор: собираем, но полноту не декларируем.
+            corridors.append(
+                Corridor(
+                    lo=c_lo,
+                    hi=c_hi,
+                    count=None,
+                    truncated=True,
+                    missed=0,
+                    depth=depth,
+                    unknown_count=True,
+                )
+            )
+            return
+        over = count > cap
+        corridors.append(
+            Corridor(
+                lo=c_lo,
+                hi=c_hi,
+                count=count,
+                truncated=over,
+                missed=max(0, count - cap) if over else 0,
+                depth=depth,
+            )
+        )
+
+    def find_upper(f_lo: int | None) -> int:
+        """Верхнюю границу открытого коридора ищем удвоением от probe_start."""
+        cand = max(int(f_lo or price_floor) * 2, probe_start)
+        while cand < price_ceil:
+            cnt = probe(cand, None)
+            if cnt is not None and cnt <= target:
+                return cand
+            cand *= 2
+        return cand
+
+    def split(s_lo: int | None, s_hi: int | None, depth: int, count: int | None) -> None:
+        if count is None or count <= target:
+            emit(s_lo, s_hi, count, depth)
+            return
+        too_narrow = width_ratio(s_lo, s_hi, price_floor=price_floor) <= min_width_ratio
+        if depth >= max_depth or too_narrow:
+            # Предохранитель сработал: делить больше нельзя. НЕ молчим — коридор
+            # уезжает в план помеченным, недобор попадает в missed_estimate.
+            emit(s_lo, s_hi, count, depth)
+            return
+        upper = s_hi if s_hi is not None else find_upper(s_lo)
+        if upper - (s_lo if s_lo is not None else price_floor) <= 1:
+            # Атомарный коридор: строгой внутренней точки нет, делить нечем.
+            # Третий предохранитель рядом с MAX_DEPTH / MIN_WIDTH_RATIO — при
+            # дефолтном min_width_ratio=1.05 недостижим, но min_width_ratio
+            # параметризуем, и без этой ветки деление ушло бы в рекурсию по себе.
+            emit(s_lo, s_hi, count, depth)
+            return
+        if s_hi is None:
+            # Хвост выше найденной границы закрывается отдельным коридором:
+            # иначе верх рынка просто выпал бы из плана.
+            emit(upper, None, probe(upper, None), depth + 1)
+        mid = geometric_mid(s_lo, upper, price_floor=price_floor)
+        for sub_lo, sub_hi in ((s_lo, mid), (mid, upper)):
+            split(sub_lo, sub_hi, depth + 1, probe(sub_lo, sub_hi))
+
+    root_count = probe(lo, hi)
+    if enabled:
+        split(lo, hi, 0, root_count)
+    else:
+        emit(lo, hi, root_count, 0)
+
+    truncated = [c for c in corridors if c.truncated]
+    stats = PlanStats(
+        corridors=len(corridors),
+        truncated_corridors=len(truncated),
+        missed_estimate=sum(c.missed for c in truncated),
+        unknown_corridors=sum(1 for c in corridors if c.unknown_count),
+        probes=probes,
+        max_depth_reached=deepest,
+        root_count=root_count,
+    )
+    return PricePlan(corridors=corridors, stats=stats)