fix(sf-09): iterative cap algorithm + cap_skipped flag + unit tests

Review verdict MINOR (PR #282) — single-pass cap_and_redistribute
не гарантировал invariant max ≤ MAX_BUCKET_SHARE_PCT. Surplus от clamped
поднимал free выше cap (например {1k:75, studio:15, 2k:10} → studio=39).

- backend/app/services/site_finder/best_layouts.py
  - _cap_and_redistribute → iterative while-loop до сходимости
  - Returns (result_map, cap_skipped) — флаг для pathological all-clamped
  - Float work map, Hamilton финальный pass для invariant sum=100
- backend/app/schemas/parcel.py
  - LayoutTzRecommendation.cap_skipped: bool = False
- backend/tests/services/site_finder/test_best_layouts.py
  - 6 parametrized tests: invariants, 7 failing cases from review,
    no-dominant unchanged, empty, cap_skipped propagation, normal-case
- frontend/src/types/best-layouts.ts — cap_skipped: boolean
- frontend/src/components/site-finder/BestLayoutsBlock.tsx
  - banner условие сменилось с maxPct > 60 на rec.cap_skipped (honest signal)

backend/app/schemas/parcel.py

@@ -277,6 +277,9 @@ class LayoutTzRecommendation(BaseModel):
     based_on_total_deals: int
     data_window_start: dt.date
     data_window_end: dt.date
+    # Fix SF-09 review: True если pathological case — все bucket'ы выше cap,
+    # redistribute невозможен. Frontend использует для отображения warning banner.
+    cap_skipped: bool = False
 
 
 class LayoutDataQuality(BaseModel):

backend/app/services/site_finder/best_layouts.py

@@ -223,55 +223,63 @@ def _normalize_pct(buckets: dict[str, float]) -> dict[str, int]:
     return floors
 
 
-def _cap_and_redistribute(pct_map: dict[str, int]) -> dict[str, int]:
+def _cap_and_redistribute(pct_map: dict[str, int]) -> tuple[dict[str, int], bool]:
-    """Fix SF-09: clamp каждый bucket к MAX_BUCKET_SHARE_PCT, перераспределить излишек.
+    """Fix SF-09: iterative clamp до сходимости — гарантирует все bucket'ы ≤ MAX_BUCKET_SHARE_PCT.
 
-    Алгоритм:
+    Алгоритм (iterative, сходится за O(#buckets) итераций):
     1. Найти clamped (> cap) и free bucket'ы.
-    2. Если ВСЕ bucket'ы clamped — pathological case (редко, напр. 2 bucket'а по 50%):
+    2. Если ВСЕ bucket'ы clamped — pathological case: возвращаем как есть (сумма уже = 100),
-       вернуть как есть с warning (сумма уже = 100).
+       логируем warning, cap_skipped=True.
-    3. Surplus пропорционально добавить к free bucket'ам (взвешено по их original share).
+    3. Clamp превышающие bucket'ы к cap, surplus пропорционально добавить к free.
-    4. Финальный Hamilton pass чтобы сумма осталась ровно 100.
+    4. Повторить: после добавления surplus free-bucket'ы могут превысить cap — следующая
+       итерация их clamp'ирует. Цикл завершается когда нет clamped.
+    5. Финальный Hamilton pass на итоговый float pct_map → целые проценты с суммой ровно 100.
+
+    Returns:
+        (result_map, cap_skipped) — cap_skipped=True если pathological case (cap не применён).
     """
     if not pct_map:
-        return pct_map
+        return pct_map, False
 
     cap = MAX_BUCKET_SHARE_PCT
-    clamped = {k: v for k, v in pct_map.items() if v > cap}
-    if not clamped:
-        return pct_map  # нет доминирующих — ничего не делаем
 
-    free = {k: v for k, v in pct_map.items() if v <= cap}
+    # Быстрый path: нет доминирующих
-    if not free:
+    if all(v <= cap for v in pct_map.values()):
-        # Все bucket'ы выше cap — pathological (напр. 2 bucket'а по 50%).
+        return pct_map, False
-        # Нельзя перераспределить — возвращаем как есть.
-        logger.warning(
-            "MAX_BUCKET_SHARE cap: все %d bucket'ов выше %d%% — пропускаем redistribute",
-            len(pct_map),
-            cap,
-        )
-        return pct_map
 
-    surplus = sum(v - cap for v in clamped.values())
+    # Работаем с float-копией для точного redistribute
-    # Новые значения для clamped bucket'ов
+    work: dict[str, float] = {k: float(v) for k, v in pct_map.items()}
-    result: dict[str, int] = {k: cap for k in clamped}
 
-    # Пропорциональный redistribute в free bucket'ы
+    while True:
-    free_total = sum(free.values())
+        clamped = {k: v for k, v in work.items() if v > cap}
-    # Дробные проценты после redistribution
+        if not clamped:
-    free_raw: dict[str, float] = {}
+            break  # сошлись — все bucket'ы ≤ cap
-    for k, v in free.items():
-        free_raw[k] = v + (v / free_total * surplus if free_total > 0 else surplus / len(free))
 
-    # Hamilton pass чтобы сумма clamped + free = 100
+        free = {k: v for k, v in work.items() if v <= cap}
-    target_sum = 100 - sum(result.values())  # сколько должно остаться у free
+        if not free:
-    free_floors = {k: int(v) for k, v in free_raw.items()}
+            # Pathological: все bucket'ы хотят > cap — невозможно перераспределить.
-    free_remainder = target_sum - sum(free_floors.values())
+            logger.warning(
-    free_fracs = sorted(free_raw.keys(), key=lambda k: -(free_raw[k] - free_floors[k]))
+                "MAX_BUCKET_SHARE cap: все %d bucket'ов выше %d%% — пропускаем redistribute",
-    for k in free_fracs[: max(0, free_remainder)]:
+                len(pct_map),
-        free_floors[k] += 1
+                cap,
-    result.update(free_floors)
+            )
-    return result
+            return pct_map, True
+
+        surplus = sum(v - cap for v in clamped.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
+
+    # Hamilton pass: work → целые проценты с суммой ровно 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
 
 
 # ── Главная функция ───────────────────────────────────────────────────────────
@@ -607,7 +615,7 @@ def _build_recommendation(
 
     total_deals = sum(rb_deals.values())
     pct_map = _normalize_pct(rb_deals)
-    pct_map = _cap_and_redistribute(pct_map)
+    pct_map, cap_skipped = _cap_and_redistribute(pct_map)
 
     mix: list[LayoutTzMixRow] = []
     for rb, pct in sorted(pct_map.items(), key=lambda x: -x[1]):
@@ -656,6 +664,7 @@ def _build_recommendation(
         based_on_total_deals=int(total_deals),
         data_window_start=window_start,
         data_window_end=window_end,
+        cap_skipped=cap_skipped,
     )
 
 

backend/tests/services/site_finder/test_best_layouts.py

@@ -25,7 +25,12 @@ from unittest.mock import MagicMock
 import pytest
 
 from app.schemas.parcel import BestLayoutsRequest
-from app.services.site_finder.best_layouts import _TIME_WINDOW_PARAMS, get_best_layouts
+from app.services.site_finder.best_layouts import (
+    _TIME_WINDOW_PARAMS,
+    MAX_BUCKET_SHARE_PCT,
+    _cap_and_redistribute,
+    get_best_layouts,
+)
 
 _TODAY = dt.date.today()
 CAD_NUM = "66:41:0303161:123"
@@ -335,3 +340,119 @@ def test_total_sold_in_window_matches_deals_window() -> None:
 
     assert len(resp.top_layouts) == 1
     assert resp.top_layouts[0].total_sold_in_window == int(deals)
+
+
+# ── Тесты _cap_and_redistribute (Fix SF-09 review) ───────────────────────────
+
+
+@pytest.mark.parametrize(
+    "pct_map, expect_pathological",
+    [
+        # 1. normal: одиночный bucket > 35, free достаточно
+        ({"1k": 50, "studio": 30, "2k": 20}, False),
+        # 2. heavy skew из тикета: 1k=75 — главный воспроизводимый case
+        ({"1k": 75, "studio": 15, "2k": 10}, False),
+        # 3. multiple buckets > 35
+        ({"1k": 50, "studio": 40, "2k": 10}, False),
+        # 4. all > 35 — pathological, redistribution невозможен
+        ({"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),
+        # 8. все ≤ cap — fast-path без изменений
+        ({"1k": 30, "studio": 35, "2k": 35}, False),
+    ],
+)
+def test_cap_and_redistribute_invariants(
+    pct_map: dict[str, int],
+    expect_pathological: bool,
+) -> None:
+    """Invariant: после redistribute max(pct) <= cap И sum(pct) == 100.
+
+    Pathological case (all > cap) — возвращает входной pct_map без изменений, cap_skipped=True.
+    """
+    result, cap_skipped = _cap_and_redistribute(pct_map)
+
+    assert (
+        cap_skipped == expect_pathological
+    ), f"cap_skipped={cap_skipped} но ожидали {expect_pathological} для {pct_map}"
+    assert (
+        sum(result.values()) == 100
+    ), f"sum={sum(result.values())} != 100 для {pct_map} → {result}"
+    if not expect_pathological:
+        assert (
+            max(result.values()) <= MAX_BUCKET_SHARE_PCT
+        ), f"max={max(result.values())} > cap={MAX_BUCKET_SHARE_PCT} для {pct_map} → {result}"
+
+
+@pytest.mark.parametrize(
+    "deals, 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}"),
+    ],
+)
+def test_cap_reproduced_failing_cases(deals: dict[str, int], label: str) -> None:
+    """Воспроизведённые из review failing cases — все должны давать max <= 35."""
+    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}"
+
+
+def test_cap_and_redistribute_no_dominant_unchanged() -> None:
+    """Если все bucket'ы ≤ cap — результат идентичен входу (fast-path)."""
+    pct_map = {"studio": 20, "1": 35, "2": 30, "3": 15}
+    result, cap_skipped = _cap_and_redistribute(pct_map)
+    assert not cap_skipped
+    assert result == pct_map
+
+
+def test_cap_and_redistribute_empty() -> None:
+    """Пустой dict → возвращается как есть."""
+    result, cap_skipped = _cap_and_redistribute({})
+    assert result == {}
+    assert not cap_skipped
+
+
+def test_cap_skipped_flag_propagates_to_recommendation() -> None:
+    """Pathological case → cap_skipped=True в recommendation_for_tz ответа."""
+    # 2 bucket'а по 50% — pathological
+    id_rows = [_obj_id_row(1), _obj_id_row(2)]
+    vel_rows = [
+        _vel_row("studio", deals_window=50.0, obj_ids=[1]),
+        _vel_row("1", deals_window=50.0, obj_ids=[2]),
+    ]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
+    req = _request(time_window="last_quarter")
+    resp = get_best_layouts(db, CAD_NUM, req)
+
+    # С deals 50/50 → normalize_pct даёт {studio:50, 1:50} — оба выше cap
+    assert resp.recommendation_for_tz.cap_skipped is True
+
+
+def test_cap_skipped_false_for_normal_case() -> None:
+    """Normal case с capping → cap_skipped=False в recommendation_for_tz."""
+    id_rows = [_obj_id_row(1), _obj_id_row(2), _obj_id_row(3)]
+    vel_rows = [
+        _vel_row("1k", deals_window=75.0, obj_ids=[1]),
+        _vel_row("studio", deals_window=15.0, obj_ids=[2]),
+        _vel_row("2k", deals_window=10.0, obj_ids=[3]),
+    ]
+    db = _make_db(coord=_coord_row(), id_rows=id_rows, vel_rows=vel_rows)
+    req = _request(time_window="last_quarter")
+    resp = get_best_layouts(db, CAD_NUM, req)
+
+    assert resp.recommendation_for_tz.cap_skipped is False
+    mix = resp.recommendation_for_tz.mix
+    assert all(row.pct <= MAX_BUCKET_SHARE_PCT for row in mix)
+    assert sum(row.pct for row in mix) == 100

frontend/src/components/site-finder/BestLayoutsBlock.tsx

@@ -368,11 +368,12 @@ function RecommendationCard({
           gap: 16,
         }}
       >
-        {/* SF-09: предупреждение о сильном перекосе unit-mix */}
+        {/* SF-09: предупреждение о перекосе рынка — показываем только если cap не смог применить */}
-        {maxPct > 60 && (
+        {rec.cap_skipped && (
           <div className="bg-amber-50 border border-amber-200 text-amber-800 rounded p-3 text-sm">
             Рекомендация имеет сильный перекос ({maxPct}% в одном формате) —
-            проверьте competitive density и district pipeline.
+            рынок настолько асимметричен, что cap не применён. Проверьте
+            competitive density и district pipeline.
           </div>
         )}
 

frontend/src/types/best-layouts.ts

@@ -48,6 +48,8 @@ export interface LayoutTzRecommendation {
   based_on_total_deals: number;
   data_window_start: string;
   data_window_end: string;
+  /** Fix SF-09: true если pathological case — все bucket'ы > cap, redistribute невозможен. */
+  cap_skipped: boolean;
 }
 
 export interface LayoutDataQuality {