feat(tradein): refine same-building anchor for within-building heterogeneity (#681)

Co-authored-by: bot-backend <bot-backend@gendsgn.local>
Co-committed-by: bot-backend <bot-backend@gendsgn.local>

tradein-mvp/backend/app/core/config.py

@@ -78,6 +78,13 @@ class Settings(BaseSettings):
     estimate_sb_min_comps: int = 2  # стоп на первом тире с ≥ N активных комплов
     estimate_sb_area_sigma: float = 0.18  # σ log-нормального area-веса (Gaussian)
     estimate_sb_rooms_match_boost: float = 1.6  # ×вес если rooms компла == target
+    # #680-WB within-building heterogeneity refine: floor-similarity Gaussian по
+    # ОТНОСИТЕЛЬНОЙ вертикальной позиции (floor/total_floors). Прижимает якорь к
+    # комплам с похожим этажом — мелкокомнатный/нижний юнит во флагман-доме больше
+    # не наследует цену видового топ-этажа. 0.0 → выключено (точно старое поведение).
+    # Откалибровано на 55 golden (offline): σ_f=0.25 даёт лучший medAPE без потери
+    # покрытия; Хохрякова 3к/153 overshoot 64%→1.5%, флагман 4к 17.5%→5.4%.
+    estimate_sb_floor_sigma: float = 0.25
     estimate_sb_guardrail_tol: float = 0.05  # hard floor: est ≥ min(comp ppm²)×(1−tol)
     asking_to_sold_haircut: float = 0.05  # дефолтная asking→sold скидка (banded по ppm²)
     estimate_fsd_k: float = 1.65  # множитель FSD → полуширина диапазона

tradein-mvp/backend/app/services/estimator.py

@@ -1009,7 +1009,8 @@ def _fetch_anchor_comps(
     Excludes lots без price_per_m2. is_active=true. Best-effort: ([], None) на ошибке.
 
     Returns (comps, tier) где tier ∈ {'A','C', None}. comps — list dict с
-    ключами price_per_m2 (int>0), area_m2 (float|None), rooms (int|None).
+    ключами price_per_m2 (int>0), area_m2 (float|None), rooms (int|None),
+    floor (int|None), total_floors (int|None) — последние два для floor-веса (#680-WB).
     """
     min_comps = settings.estimate_sb_min_comps
 
@@ -1028,7 +1029,7 @@ def _fetch_anchor_comps(
                 db.execute(
                     text(
                         """
-                        SELECT price_per_m2, area_m2, rooms
+                        SELECT price_per_m2, area_m2, rooms, floor, total_floors
                         FROM listings
                         WHERE is_active = true
                           AND price_per_m2 > 0
@@ -1056,6 +1057,8 @@ def _fetch_anchor_comps(
                 "price_per_m2": int(r["price_per_m2"]),
                 "area_m2": float(r["area_m2"]) if r["area_m2"] is not None else None,
                 "rooms": int(r["rooms"]) if r["rooms"] is not None else None,
+                "floor": int(r["floor"]) if r["floor"] is not None else None,
+                "total_floors": int(r["total_floors"]) if r["total_floors"] is not None else None,
             }
             for r in rows
             if r["price_per_m2"]
@@ -1077,7 +1080,7 @@ def _fetch_anchor_comps(
                 db.execute(
                     text(
                         """
-                        SELECT price_per_m2, area_m2, rooms
+                        SELECT price_per_m2, area_m2, rooms, floor, total_floors
                         FROM listings
                         WHERE is_active = true
                           AND price_per_m2 > 0
@@ -1120,6 +1123,8 @@ def _fetch_anchor_comps(
                 "price_per_m2": int(r["price_per_m2"]),
                 "area_m2": float(r["area_m2"]) if r["area_m2"] is not None else None,
                 "rooms": int(r["rooms"]) if r["rooms"] is not None else None,
+                "floor": int(r["floor"]) if r["floor"] is not None else None,
+                "total_floors": int(r["total_floors"]) if r["total_floors"] is not None else None,
             }
             for r in rows
             if r["price_per_m2"]
@@ -1153,15 +1158,24 @@ def _compute_same_building_anchor(
     tier: str,
     sigma: float,
     rooms_boost: float,
+    floor_target: int | None = None,
+    total_floors_target: int | None = None,
+    floor_sigma: float = 0.0,
 ) -> dict[str, Any] | None:
     """Чистая (testable без БД) свёртка комплов в anchor-оценку.
 
     1. similarity-weighted mean ppm²: w_i = exp(−(ln(area_i/area_target))²/(2σ²))
-       × (rooms_boost если rooms_i==rooms_target). area_i=None → area-вес 1.0
-       (нейтрально: Yandex-листинги без площади судим только по комнатам).
-    2. PREMIUM uplift (class-free): если target — топ-юнит ДОМА (area_target ≥ p66
-       площадей комплов) И tier == 'A' → берём weighted ~p70 верхний квантиль comp
-       ppm² вместо mean (консервативно, только вверх).
+       × (rooms_boost если rooms_i==rooms_target) × FLOOR-вес. area_i=None →
+       area-вес 1.0 (нейтрально). FLOOR-вес (#680-WB) — Gaussian по ОТНОСИТЕЛЬНОЙ
+       вертикальной позиции pos=floor/total_floors: exp(−(pos_i−pos_t)²/(2σ_f²)),
+       σ_f=floor_sigma. Прижимает якорь к комплам с похожим этажом (1-й/последний
+       и видовые этажи реально отличаются по цене). floor_sigma=0 ИЛИ нет floor у
+       target/компла → вес 1.0 (выключено / нейтрально — no regress).
+    2. PREMIUM uplift (class-free): target — топ-юнит ДОМА (area ≥ p66 площадей
+       комплов) И rooms_target ≥ медианы комнат комплов И tier == 'A' → берём
+       weighted ~p70 верхний квантиль ppm² (консервативно, только вверх). Условие
+       по комнатам (#680-WB) не даёт мелкокомнатному юниту во флагман-доме унаследовать
+       headline-премию флагмана (3к/153 в доме с 4к-флагманом ≠ цена флагмана).
     3. haircut asking→sold (banded по anchor ppm²): anchor_sold = anchor×(1−haircut).
     4. FSD = 0.07 + 0.25·CV(comp ppm²) + tier_penalty + n_penalty; range полуширина
        = k·fsd. confidence-банд по fsd.
@@ -1176,7 +1190,12 @@ def _compute_same_building_anchor(
         return None
     n = len(ppm2)
 
-    # 1. similarity weights
+    # target relative vertical position (None → floor-вес отключён/нейтрален).
+    target_pos: float | None = None
+    if floor_sigma > 0 and floor_target and total_floors_target and total_floors_target > 0:
+        target_pos = floor_target / total_floors_target
+
+    # 1. similarity weights (area × rooms × floor-position)
     weights: list[float] = []
     for c in comps:
         a = c.get("area_m2")
@@ -1186,19 +1205,35 @@ def _compute_same_building_anchor(
             area_w = 1.0  # площадь неизвестна → нейтральный area-вес
         rooms_match = rooms_target is not None and c.get("rooms") == rooms_target
         rooms_w = rooms_boost if rooms_match else 1.0
-        weights.append(area_w * rooms_w)
+        floor_w = 1.0
+        if target_pos is not None:
+            cf = c.get("floor")
+            ctf = c.get("total_floors")
+            if cf and ctf and ctf > 0:
+                comp_pos = cf / ctf
+                floor_w = math.exp(
+                    -((comp_pos - target_pos) ** 2) / (2.0 * floor_sigma * floor_sigma)
+                )
+            # компл без floor/total_floors → нейтральный floor-вес 1.0
+        weights.append(area_w * rooms_w * floor_w)
     wsum = sum(weights)
     if wsum > 0:
         anchor = sum(w * p for w, p in zip(weights, ppm2, strict=True)) / wsum
     else:
         anchor = _percentile(sorted(ppm2), 0.5)
 
-    # 2. premium uplift — топ-юнит дома по площади (≥p66) И Tier A → weighted p70.
+    # 2. premium uplift — топ-юнит дома (площадь ≥ p66 И комнаты ≥ медианы) И Tier A
+    #    → weighted p70. Условие по комнатам отсекает мелкие юниты во флагман-домах.
     used_uplift = False
     areas = [c.get("area_m2") for c in comps if c.get("area_m2")]
+    comp_rooms = [c.get("rooms") for c in comps if c.get("rooms") is not None]
     if tier == "A" and areas and area_target > 0:
         p66_area = _percentile(sorted(areas), 0.66)
-        if area_target >= p66_area:
+        rooms_ok = True
+        if rooms_target is not None and comp_rooms:
+            median_rooms = _percentile(sorted(comp_rooms), 0.5)
+            rooms_ok = rooms_target >= median_rooms
+        if area_target >= p66_area and rooms_ok:
             p70 = _percentile(sorted(ppm2), 0.70)
             if p70 > anchor:
                 anchor = p70
@@ -1643,6 +1678,9 @@ async def estimate_quality(
             tier=anchor_tier or "",
             sigma=settings.estimate_sb_area_sigma,
             rooms_boost=settings.estimate_sb_rooms_match_boost,
+            floor_target=payload.floor,
+            total_floors_target=payload.total_floors,
+            floor_sigma=settings.estimate_sb_floor_sigma,
         )
         if anchor is not None:
             est_ppm2 = anchor["anchor_sold_ppm2"]

tradein-mvp/backend/tests/test_same_building_anchor.py

@@ -168,8 +168,20 @@ def test_normalize_no_house_number_returns_none_base() -> None:
 # ── Pure anchor compute ─────────────────────────────────────────────────────
 
 
-def _comp(ppm2: int, area: float | None = None, rooms: int | None = None) -> dict[str, Any]:
-    return {"price_per_m2": ppm2, "area_m2": area, "rooms": rooms}
+def _comp(
+    ppm2: int,
+    area: float | None = None,
+    rooms: int | None = None,
+    floor: int | None = None,
+    total_floors: int | None = None,
+) -> dict[str, Any]:
+    return {
+        "price_per_m2": ppm2,
+        "area_m2": area,
+        "rooms": rooms,
+        "floor": floor,
+        "total_floors": total_floors,
+    }
 
 
 def test_anchor_premium_lift_hohryakova48() -> None:
@@ -493,3 +505,114 @@ def test_anchor_exposes_comp_max() -> None:
     assert res is not None
     assert res["comp_max_ppm2"] == 500_000
     assert res["comp_min_ppm2"] == 300_000
+
+
+# ── #680-WB: within-building heterogeneity refine (uplift-gate + floor weight) ──
+
+
+def test_anchor_uplift_gated_by_rooms_median_hohryakova_3k() -> None:
+    """Зеркало Хохрякова 48 3к/153: дом с 4-комн. флагманом (комплы rooms=4 кроме
+    одного 3к). target 3-комн. 153м² (топ по площади, ≥p66) — БЕЗ refine получил бы
+    p70-uplift и унаследовал цену флагмана. С rooms-gate (3 < медианы=4) uplift НЕ
+    срабатывает → anchor падает к similar-комплам, не к headline-флагману."""
+    comps = [
+        _comp(399_477, area=153.2, rooms=4, floor=8, total_floors=8),
+        _comp(683_995, area=146.2, rooms=4, floor=5, total_floors=14),
+        _comp(502_846, area=105.4, rooms=3, floor=6, total_floors=9),
+        _comp(683_994, area=146.2, rooms=4, floor=5, total_floors=15),
+    ]
+    res = _compute_same_building_anchor(
+        comps,
+        area_target=153.2,
+        rooms_target=3,
+        tier="A",
+        sigma=0.18,
+        rooms_boost=1.6,
+        floor_target=8,
+        total_floors_target=8,
+        floor_sigma=0.25,
+    )
+    assert res is not None
+    # rooms_target 3 < median comp rooms 4 → uplift gated off.
+    assert res["used_uplift"] is False
+    # anchor НЕ должен быть на уровне флагмана (684k); прижат к похожим юнитам.
+    assert res["anchor_ppm2"] < 550_000
+
+
+def test_anchor_uplift_still_fires_for_genuine_top_unit() -> None:
+    """Топ-юнит дома (rooms ≥ медианы И площадь ≥p66) в Tier A → uplift ВСЁ ЕЩЁ
+    срабатывает, когда дорогой компл area-несимметричен и weighted mean занижает
+    (do-no-harm для настоящих флагманов). Зеркало test_anchor_premium_uplift_when_*
+    с rooms_target == медиане комнат комплов (gate проходит)."""
+    comps = [
+        _comp(900_000, area=60.0, rooms=3),  # дорогой, малая площадь → area-вес низкий
+        _comp(300_000, area=200.0, rooms=3),
+        _comp(320_000, area=210.0, rooms=3),
+    ]
+    res = _compute_same_building_anchor(
+        comps,
+        area_target=205.0,  # топ по площади (≥p66)
+        rooms_target=3,  # == median comp rooms 3 → rooms-gate проходит
+        tier="A",
+        sigma=0.18,
+        rooms_boost=1.6,
+    )
+    assert res is not None
+    # mean занижен дил. дорогим малометражным → uplift подтягивает к p70.
+    assert res["used_uplift"] is True
+    assert res["anchor_ppm2"] >= 500_000
+
+
+def test_anchor_floor_weight_tilts_toward_similar_floor() -> None:
+    """floor_sigma>0 → комплы с похожей вертикальной позицией (floor/total_floors)
+    весят больше. Дорогой видовой топ-этаж и дешёвый средний этаж той же площади;
+    target на среднем этаже → anchor тянется к среднеэтажному компла (не к видовому).
+    target — НЕ топ-юнит по площади (есть компл крупнее) → uplift не маскирует floor."""
+    comps = [
+        _comp(300_000, area=60.0, rooms=2, floor=5, total_floors=20),  # mid, как target
+        _comp(600_000, area=60.0, rooms=2, floor=20, total_floors=20),  # видовой топ
+        _comp(310_000, area=120.0, rooms=3, floor=4, total_floors=20),  # крупнее target
+    ]
+    target_kw = dict(
+        area_target=60.0,
+        rooms_target=2,
+        tier="A",
+        sigma=0.18,
+        rooms_boost=1.6,
+        floor_target=5,
+        total_floors_target=20,
+    )
+    off = _compute_same_building_anchor(comps, floor_sigma=0.0, **target_kw)
+    on = _compute_same_building_anchor(comps, floor_sigma=0.25, **target_kw)
+    assert off is not None and on is not None
+    assert off["used_uplift"] is False  # target не топ по площади
+    # floor-вес прижимает к среднеэтажному (300k), уводя от видового топа (600k).
+    assert on["anchor_ppm2"] < off["anchor_ppm2"]
+
+
+def test_anchor_floor_weight_neutral_when_data_missing() -> None:
+    """Нет floor у компла/target → floor-вес 1.0 (нейтрально), не падает,
+    результат идентичен floor_sigma=0 (no regress на Yandex-листингах без этажа)."""
+    comps = [_comp(300_000, area=60.0, rooms=2), _comp(340_000, area=62.0, rooms=2)]
+    a = _compute_same_building_anchor(
+        comps,
+        area_target=60.0,
+        rooms_target=2,
+        tier="A",
+        sigma=0.18,
+        rooms_boost=1.6,
+        floor_target=None,
+        total_floors_target=None,
+        floor_sigma=0.25,
+    )
+    b = _compute_same_building_anchor(
+        comps,
+        area_target=60.0,
+        rooms_target=2,
+        tier="A",
+        sigma=0.18,
+        rooms_boost=1.6,
+        floor_sigma=0.0,
+    )
+    assert a is not None and b is not None
+    assert abs(a["anchor_ppm2"] - b["anchor_ppm2"]) < 1.0