backend/app/services/site_finder/best_layouts.py

@@ -224,19 +224,21 @@ def _normalize_pct(buckets: dict[str, float]) -> dict[str, int]:
 
 
 def _cap_and_redistribute(pct_map: dict[str, int]) -> tuple[dict[str, int], bool]:
-    """Fix SF-09: iterative clamp до сходимости — гарантирует все bucket'ы ≤ MAX_BUCKET_SHARE_PCT.
+    """Fix SF-09 round 2: capacity-aware redistribute, bounded iterations.
 
-    Алгоритм (iterative, сходится за O(#buckets) итераций):
-    1. Найти clamped (> cap) и free bucket'ы.
-    2. Если ВСЕ bucket'ы clamped — pathological case: возвращаем как есть (сумма уже = 100),
-       логируем warning, cap_skipped=True.
-    3. Clamp превышающие bucket'ы к cap, surplus пропорционально добавить к free.
-    4. Повторить: после добавления surplus free-bucket'ы могут превысить cap — следующая
-       итерация их clamp'ирует. Цикл завершается когда нет clamped.
-    5. Финальный Hamilton pass на итоговый float pct_map → целые проценты с суммой ровно 100.
+    Round 1 bug: surplus распределялся пропорционально текущему `v` free bucket'а,
+    что переливало его выше cap — на 2-bucket вход цикл осциллировал бесконечно.
+
+    Round 2 fix: surplus распределяется пропорционально **available capacity**
+    `(cap - v)` каждого free bucket'а. Тогда free никогда не вылетит выше cap →
+    цикл сходится за ≤ len(pct_map) итераций. Hard guard `for _ in range(N+1)`.
+
+    Если surplus > total_capacity (геометрически невозможно поместить излишек ниже
+    cap) — забиваем все free к cap, возвращаем `cap_skipped=True` + warning log.
 
     Returns:
-        (result_map, cap_skipped) — cap_skipped=True если pathological case (cap не применён).
+        (result_map, cap_skipped) — cap_skipped=True если cap не удержан
+        (pathological: всё хочет > cap, или surplus > available capacity).
     """
     if not pct_map:
         return pct_map, False
@@ -247,39 +249,63 @@ def _cap_and_redistribute(pct_map: dict[str, int]) -> tuple[dict[str, int], bool
     if all(v <= cap for v in pct_map.values()):
         return pct_map, False
 
-    # Работаем с float-копией для точного redistribute
     work: dict[str, float] = {k: float(v) for k, v in pct_map.items()}
 
-    while True:
-        clamped = {k: v for k, v in work.items() if v > cap}
+    # Bounded iteration: после k-й итерации число clamped не убывает только если
+    # surplus > capacity (тогда — pathological). При корректном capacity-aware
+    # redistribute достаточно ≤ len(pct_map) итераций.
+    for _ in range(len(pct_map) + 1):
+        clamped = [k for k, v in work.items() if v > cap]
         if not clamped:
-            break  # сошлись — все bucket'ы ≤ cap
+            break
 
-        free = {k: v for k, v in work.items() if v <= cap}
+        free = [k for k, v in work.items() if v < cap]
         if not free:
-            # Pathological: все bucket'ы хотят > cap — невозможно перераспределить.
+            # Все bucket'ы либо >cap либо ровно =cap — некуда переливать.
             logger.warning(
-                "MAX_BUCKET_SHARE cap: все %d bucket'ов выше %d%% — пропускаем redistribute",
+                "MAX_BUCKET_SHARE cap: нет free bucket'ов (%d total) — cap_skipped",
                 len(pct_map),
-                cap,
             )
             return pct_map, True
 
-        surplus = sum(v - cap for v in clamped.values())
+        surplus = sum(work[k] - cap for k in clamped)
+        capacities = {k: cap - work[k] for k in free}
+        total_capacity = sum(capacities.values())
+
         for k in clamped:
             work[k] = float(cap)
 
-        free_total = sum(free.values()) or 1.0
-        for k in free:
-            work[k] = free[k] + free[k] / free_total * surplus
+        if surplus > total_capacity + 1e-9:
+            # Излишек не помещается ниже cap — pathological.
+            # Возвращаем оригинал (sum=100 invariant) + флаг для frontend banner.
+            logger.warning(
+                "MAX_BUCKET_SHARE cap: surplus %.2f > total_capacity %.2f — cap_skipped",
+                surplus,
+                total_capacity,
+            )
+            return pct_map, True
 
-    # Hamilton pass: work → целые проценты с суммой ровно 100
+        for k in free:
+            work[k] += capacities[k] / total_capacity * surplus
+    else:
+        # Hard guard: не сошлись за N+1 итераций — bug. Лог + cap_skipped.
+        logger.error(
+            "MAX_BUCKET_SHARE cap: не сошлись за %d итераций — algorithm bug",
+            len(pct_map) + 1,
+        )
+        return pct_map, True
+
+    return _hamilton_round(work), False
+
+
+def _hamilton_round(work: dict[str, float]) -> dict[str, int]:
+    """Hamilton apportionment: float → integer pct с суммой ровно 100."""
     floors = {k: int(v) for k, v in work.items()}
     remainder = 100 - sum(floors.values())
     fracs = sorted(work.keys(), key=lambda k: -(work[k] - floors[k]))
     for k in fracs[: max(0, remainder)]:
         floors[k] += 1
-    return floors, False
+    return floors
 
 
 # ── Главная функция ───────────────────────────────────────────────────────────

backend/tests/services/site_finder/test_best_layouts.py

@@ -348,31 +348,36 @@ def test_total_sold_in_window_matches_deals_window() -> None:
 @pytest.mark.parametrize(
     "pct_map, expect_pathological",
     [
-        # 1. normal: одиночный bucket > 35, free достаточно
+        # 1. normal: одиночный bucket > 35, free достаточно capacity
         ({"1k": 50, "studio": 30, "2k": 20}, False),
-        # 2. heavy skew из тикета: 1k=75 — главный воспроизводимый case
+        # 2. heavy skew (3-bucket): surplus=40, capacity=20+25=45 — помещается
         ({"1k": 75, "studio": 15, "2k": 10}, False),
         # 3. multiple buckets > 35
         ({"1k": 50, "studio": 40, "2k": 10}, False),
-        # 4. all > 35 — pathological, redistribution невозможен
+        # 4. all > 35 — pathological
         ({"1k": 50, "studio": 50}, True),
         # 5. граничный: один bucket ровно на cap — не clamp
         ({"1k": 35, "studio": 35, "2k": 30}, False),
         # 6. single bucket 100% — pathological (нет free)
         ({"1k": 100}, True),
-        # 7. 3-bucket heavy: 1k=90
-        ({"1k": 90, "studio": 10}, False),
+        # 7. 2-bucket heavy: surplus=55, capacity=25 — pathological (не помещается)
+        ({"1k": 90, "studio": 10}, True),
         # 8. все ≤ cap — fast-path без изменений
         ({"1k": 30, "studio": 35, "2k": 35}, False),
+        # 9. 2-bucket: 70/30 → surplus=35, capacity=5 → pathological
+        ({"1k": 70, "studio": 30}, True),
+        # 10. 2-bucket: 99/1 → surplus=64, capacity=34 → pathological
+        ({"1k": 99, "studio": 1}, True),
     ],
 )
 def test_cap_and_redistribute_invariants(
     pct_map: dict[str, int],
     expect_pathological: bool,
 ) -> None:
-    """Invariant: после redistribute max(pct) <= cap И sum(pct) == 100.
+    """Invariant: max(pct) ≤ cap И sum(pct) == 100 (или cap_skipped=True в pathological).
 
-    Pathological case (all > cap) — возвращает входной pct_map без изменений, cap_skipped=True.
+    Pathological — `cap_skipped=True`, max МОЖЕТ быть > cap (геометрически surplus
+    не вмещается в free capacity).
     """
     result, cap_skipped = _cap_and_redistribute(pct_map)
 
@@ -389,24 +394,58 @@ def test_cap_and_redistribute_invariants(
 
 
 @pytest.mark.parametrize(
-    "deals, label",
+    "deals, expect_pathological, label",
     [
-        ({"1k": 75, "studio": 15, "2k": 10}, "{1k:75, studio:15, 2k:10}"),
-        ({"1k": 80, "studio": 12, "2k": 8}, "{1k:80, studio:12, 2k:8}"),
-        ({"1k": 90, "studio": 10}, "{1k:90, studio:10}"),
-        ({"1k": 70, "studio": 30}, "{1k:70, studio:30}"),
-        ({"1k": 99, "studio": 1}, "{1k:99, studio:1}"),
-        ({"a": 50, "b": 30, "c": 20}, "{50, 30, 20}"),
+        # 3-bucket с достаточной capacity — surplus помещается, cap соблюдён
+        ({"1k": 75, "studio": 15, "2k": 10}, False, "{1k:75, studio:15, 2k:10}"),
+        ({"1k": 80, "studio": 12, "2k": 8}, False, "{1k:80, studio:12, 2k:8}"),
+        ({"1k": 60, "studio": 30, "2k": 10}, False, "{1k:60, studio:30, 2k:10}"),
+        ({"a": 50, "b": 30, "c": 20}, False, "{50, 30, 20}"),
+        # 2-bucket — surplus геометрически не помещается, cap_skipped=True
+        ({"1k": 90, "studio": 10}, True, "{1k:90, studio:10}"),
+        ({"1k": 70, "studio": 30}, True, "{1k:70, studio:30}"),
+        ({"1k": 99, "studio": 1}, True, "{1k:99, studio:1}"),
     ],
 )
-def test_cap_reproduced_failing_cases(deals: dict[str, int], label: str) -> None:
-    """Воспроизведённые из review failing cases — все должны давать max <= 35."""
+def test_cap_reproduced_failing_cases(
+    deals: dict[str, int], expect_pathological: bool, label: str
+) -> None:
+    """Review round-2 reproduced cases: 2-bucket — pathological, 3-bucket — fit cap."""
     result, cap_skipped = _cap_and_redistribute(deals)
-    assert not cap_skipped, f"cap_skipped=True неожиданно для {label}"
     assert (
-        max(result.values()) <= MAX_BUCKET_SHARE_PCT
-    ), f"max={max(result.values())} > {MAX_BUCKET_SHARE_PCT} для {label} → {result}"
-    assert sum(result.values()) == 100, f"sum={sum(result.values())} != 100 для {label} → {result}"
+        cap_skipped == expect_pathological
+    ), f"cap_skipped={cap_skipped} ожидали {expect_pathological} для {label}"
+    assert sum(result.values()) == 100, f"sum != 100 для {label} → {result}"
+    if not expect_pathological:
+        assert (
+            max(result.values()) <= MAX_BUCKET_SHARE_PCT
+        ), f"max={max(result.values())} > {MAX_BUCKET_SHARE_PCT} для {label} → {result}"
+
+
+def test_cap_iteration_count_bounded() -> None:
+    """Round 2 regression: алгоритм завершается за ≤ len(pct_map)+1 итераций.
+
+    Round 1 bag: на 2-bucket {1k:70, studio:30} цикл осциллировал бесконечно.
+    Round 2 fix: capacity-aware redistribute + hard `for _ in range(N+1)` guard.
+    Этот тест гарантирует что вызов не зависает (pytest-timeout не нужен).
+    """
+    import time
+
+    pathological_cases = [
+        {"1k": 70, "studio": 30},
+        {"1k": 99, "studio": 1},
+        {"1k": 90, "studio": 10},
+        {"1k": 50, "studio": 50},
+    ]
+    for case in pathological_cases:
+        start = time.perf_counter()
+        result, cap_skipped = _cap_and_redistribute(case)
+        elapsed_ms = (time.perf_counter() - start) * 1000
+        assert elapsed_ms < 100, f"Завис ({elapsed_ms:.0f}ms) на {case}"
+        assert sum(result.values()) == 100, f"sum != 100 для {case}"
+        # 2-bucket с одним > cap всегда pathological (surplus > free capacity)
+        if case != {"1k": 50, "studio": 50}:
+            assert cap_skipped, f"Ожидали cap_skipped=True для {case}"
 
 
 def test_cap_and_redistribute_no_dominant_unchanged() -> None: