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

@@ -79,6 +79,7 @@ from app.services.house_metadata import get_house_metadata
 from app.services.matching.houses import match_house_readonly, match_or_create_house
 from app.services.scraper_adapters import RealScraperConfig
 from app.services.scraper_settings import get_scraper_delay
+from app.tasks.asking_to_sold_ratio import area_bucket
 
 logger = logging.getLogger(__name__)
 
@@ -380,11 +381,22 @@ _asking_sold_ratio_cache: dict[int, tuple[float | None, str | None, float]] = {}
 def _get_asking_sold_ratio(
     db: Session,
     rooms: int | None,
+    area_m2: float | None = None,
     anchor_ppm2: float | None = None,
 ) -> tuple[float | None, str | None]:
-    """Возвращает (ratio, basis) asking→sold для бакета комнат.
+    """Возвращает (ratio, basis) asking→sold для area-бакета клиентской квартиры.
 
-    bucket = min(max(rooms or 0, 0), 4).
+    bucket = area_bucket(area_m2) при area_m2 > 0, иначе min(max(rooms or 0, 0), 4)
+    (rooms-фолбэк — только когда площадь неизвестна).
+
+    #2620-2 (deep-review, вторая половина #2620): расчёт ratio (asking_to_sold_ratio.py
+    ask_side) теперь ключуется по area-бакету (rooms_bucket в asking_to_sold_ratios —
+    это на самом деле area-бакет, легаси-имя колонки), но ДО этой правки применение
+    здесь читало rooms_bucket по РЕАЛЬНЫМ комнатам клиента — тот же mismatch, что чинили
+    в расчёте, просто переехавший в применение. Прод-замер ревьюера (2026-08): 310/1038
+    (29.9%) исторических клиентских запросов легли бы в другой ratio-бакет при
+    rooms-ключе vs area-ключе. area_bucket() — тот же Python-двойник, что и в
+    asking_to_sold_ratio.py (см. комментарий там, границы 30/44/62/85 = import-rosreestr.sh).
 
     Запрос к asking_to_sold_ratios (migration 080): per-rooms строка
     (WHERE rooms_bucket = bucket AND district = '') → fallback на global -1
@@ -398,7 +410,7 @@ def _get_asking_sold_ratio(
     Таблицы нет / любая ошибка → (None, None), НЕ raise (graceful).
     Кэшируется на ключ bucket с TTL _ASKING_SOLD_RATIO_CACHE_TTL_S.
     """
-    bucket = min(max(rooms or 0, 0), 4)
+    bucket = area_bucket(area_m2) if area_m2 else min(max(rooms or 0, 0), 4)
 
     cached = _asking_sold_ratio_cache.get(bucket)
     if cached is not None:
@@ -3746,7 +3758,8 @@ async def estimate_quality(
     def _ratio_resolver(
         appm2: float | None,
     ) -> tuple[float | None, str | None]:
-        return _get_asking_sold_ratio(db, payload.rooms, anchor_ppm2=appm2)
+        # #2620-2: area-bucket key (payload.rooms — фолбэк только без площади).
+        return _get_asking_sold_ratio(db, payload.rooms, payload.area_m2, anchor_ppm2=appm2)
 
     def _qi_lookup(q: str) -> tuple[float, int] | None:
         return _lookup_quarter_index(

tradein-mvp/backend/app/tasks/asking_to_sold_ratio.py

@@ -20,12 +20,20 @@ snapshot_listing_sources / import_rosreestr_dkp.
 Окно расписания 06:00-07:00 UTC — ПОСЛЕ rosreestr_dkp_import (04:00-06:00 UTC), чтобы
 refresh потреблял свежие ДКП-сделки того же дня.
 
-SQL derivation ниже — БАЙТ-В-БАЙТ та же логика, что seed в data/sql/080_asking_to_sold_ratios.sql
+SQL derivation ниже повторяет seed в data/sql/080_asking_to_sold_ratios.sql (deal_side /
-(deal_side / ask_side / per_bucket + deal_global / ask_global / global_row: трейлинг-12мес
+ask_side / per_bucket + deal_global / ask_global / global_row: трейлинг-12мес окно, ppm²-полоса
-окно, ppm²-полоса [_PPM2_MIN, settings.asking_ratio_ppm2_max] (default [30000,1200000]),
+[_PPM2_MIN, settings.asking_ratio_ppm2_max] (default [30000,1200000]), порог n_deals>=30 AND
-бакет LEAST(GREATEST(rooms,0),4), порог n_deals>=30 AND n_listings>=30 для per_rooms,
+n_listings>=30 для per_rooms, global -1 строка всегда). ON CONFLICT убран — DELETE идёт первым,
-global -1 строка всегда). ON CONFLICT убран — DELETE идёт первым,
 конфликтов нет (повторный прогон в одной tx невозможен, refresh = re-seed по семантике).
+
+#2620 — ОДНО ПРЕДНАМЕРЕННОЕ РАСХОЖДЕНИЕ с 080: deal_side бакетится по
+LEAST(GREATEST(rooms,0),4), а ask_side — по _AREA_ROOMS_BUCKET_SQL (площадь, та же формула,
+что deals.rooms получает при импорте). Причина — deals.rooms НЕ настоящая комнатность
+(Росреестр её не отдаёт), это синтетика из площади; сравнивать её с РЕАЛЬНЫМИ комнатами
+listings значило сравнивать разные классификации. Замер на проде (2026-08, #2620) показал
+миграцию 23-55% объявлений между бакетами при таком сравнении — не только в бакете «4+»
+(который к тому же обрезан обрезкой ELSE 4, тогда как listings.rooms доходит до 10) — и
+это и была причина ratio>1 в бакете 4+ (см. _AREA_ROOMS_BUCKET_SQL ниже).
 """
 
 from __future__ import annotations
@@ -60,6 +68,37 @@ _ASKING_CITY_PATTERN: str = "%Екатеринбург%"
 # QA-note: точное значение сверить с `SELECT max(price_per_m2) FROM deals
 # WHERE source='rosreestr'` на проде — ceiling должен быть > max(ppm²) premium-сделок.
 
+# #2620 — синтетический "бакет комнат по площади", ИСТОЧНИК ИСТИНЫ:
+# tradein-mvp/deploy/import-rosreestr.sh (Росреестр не отдаёт комнатность — deals.rooms
+# синтезируется из area_m2 при импорте ровно этим CASE). Три представления ОДНОЙ формулы —
+# держи границы (30/44/62/85) в синхроне при правке: shell (import-rosreestr.sh) → SQL
+# (эта константа, ask_side ниже) → Python (area_bucket() ниже, estimator.py rekey #2620-2).
+_AREA_ROOMS_BUCKET_SQL = (
+    "CASE WHEN area_m2 < 30 THEN 0 WHEN area_m2 < 44 THEN 1 "
+    "WHEN area_m2 < 62 THEN 2 WHEN area_m2 < 85 THEN 3 ELSE 4 END"
+)
+
+
+def area_bucket(area_m2: float) -> int:
+    """Python-двойник _AREA_ROOMS_BUCKET_SQL (границы ИДЕНТИЧНЫ, #2620).
+
+    Используется estimator.py при ПРИМЕНЕНИИ ratio (не только при расчёте здесь) —
+    ratio_resolver должен ключевать по ТОМУ ЖЕ area-бакету, что и ask_side при
+    деривации, иначе mismatch просто переезжает из расчёта в применение (прод-замер
+    ревьюера #2620: 310/1038 = 29.9% исторических запросов легли бы в другой бакет
+    при rooms-ключе vs area-ключе).
+    """
+    if area_m2 < 30:
+        return 0
+    if area_m2 < 44:
+        return 1
+    if area_m2 < 62:
+        return 2
+    if area_m2 < 85:
+        return 3
+    return 4
+
+
 logger = logging.getLogger(__name__)
 
 # ── True-mirror cleanup: drop all #648 rows before re-derivation ──────────────
@@ -77,16 +116,18 @@ _DELETE_SQL = text(
 # deal_side / ask_side / per_bucket + deal_global / ask_global / global_row:
 #   sold_median = percentile_cont(0.5) по deals.price_per_m2 (source='rosreestr',
 #     ppm² ∈ [_PPM2_MIN, settings.asking_ratio_ppm2_max], deal_date >= CURRENT_DATE − 12 months),
-#     бакет LEAST(GREATEST(rooms,0),4).
+#     бакет LEAST(GREATEST(rooms,0),4) (rooms уже синтетика-из-площади при импорте, см. #2620
+#     комментарий у _AREA_ROOMS_BUCKET_SQL выше).
 #   ask_median  = percentile_cont(0.5) по listings.price_per_m2
 #     (is_active, та же ppm²-полоса [_PPM2_MIN, asking_ratio_ppm2_max], тот же город что
-#     SOLD-сторона — city IS NULL OR city ILIKE :asking_city, #2583 H2).
+#     SOLD-сторона — city IS NULL OR city ILIKE :asking_city, #2583 H2). Бакет —
+#     _AREA_ROOMS_BUCKET_SQL (площадь, #2620), НЕ listings.rooms — см. комментарий там.
 #   per_rooms строки — только при n_deals>=30 AND n_listings>=30 AND ask>0 AND sold>0.
 #   global -1 строка (basis='global_fallback') — всегда (если ask>0 AND sold>0). window_months=12.
 # Порог/окно — литералы; ppm²-полоса передаётся bind-параметрами :ppm2_min/:ppm2_max
 # (безопасно от SQL-инъекций; CAST не нужен — psycopg v3 передаёт int напрямую).
 _REDERIVE_SQL = text(
-    """
+    f"""
     WITH
     -- SOLD медианы по бакетам комнат за трейлинг-12мес (ДКП Росреестра).
     deal_side AS (
@@ -102,15 +143,24 @@ _REDERIVE_SQL = text(
           AND deal_date >= CURRENT_DATE - INTERVAL '12 months'
         GROUP BY LEAST(GREATEST(rooms, 0), 4)
     ),
-    -- ASKING медианы по бакетам комнат среди ТЕКУЩИХ активных объявлений.
+    -- ASKING медианы по ТОМУ ЖЕ area-бакету, что deal_side (#2620) — НЕ по listings.rooms.
+    -- deals.rooms — синтетика из площади (Росреестр её не отдаёт), listings.rooms — реальная
+    -- комнатность; сравнение area-бакета с area-бакетом (не area-бакета с real-rooms-бакетом)
+    -- убирает миграцию объявлений между бакетами (23-55% строк на проде, 2026-08, #2620) —
+    -- включая инверсию ratio>1 в бакете «4+» (deals.rooms обрезан ELSE 4, а listings.rooms
+    -- нет: 110/782 пяти- и более комнатных объявлений раньше схлопывались в бакет 4).
     ask_side AS (
         SELECT
-            LEAST(GREATEST(rooms, 0), 4)                                  AS rooms_bucket,
+            {_AREA_ROOMS_BUCKET_SQL}                                      AS rooms_bucket,
             percentile_cont(0.5) WITHIN GROUP (ORDER BY price_per_m2)     AS ask_median,
             COUNT(*)                                                      AS n_listings
         FROM listings
         WHERE is_active
           AND rooms IS NOT NULL
+          -- #2620 hardening: area_m2 IS NULL falls into the CASE ELSE branch (bucket 4)
+          -- of _AREA_ROOMS_BUCKET_SQL — a latent "everything unmeasured looks like a big
+          -- flat" trap. Excluded explicitly instead of relying on ELSE-as-junk-drawer.
+          AND area_m2 IS NOT NULL
           AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
           -- novostroyki guard (#1186): NULL = legacy вторичка до м.011
           AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
@@ -123,7 +173,7 @@ _REDERIVE_SQL = text(
           -- роста покрытия колонки этот предикат сам ужесточается без правок кода; когда
           -- покрытие станет полным — заменить на строго симметричный `city ILIKE :asking_city`.
           AND (city IS NULL OR city ILIKE :asking_city)
-        GROUP BY LEAST(GREATEST(rooms, 0), 4)
+        GROUP BY {_AREA_ROOMS_BUCKET_SQL}
     ),
     -- Per-rooms строки: только бакеты с обеими сторонами, прошедшие порог 30/30 и ask>0.
     -- Тонкие бакеты (n<30) сюда НЕ попадают → estimator делает fallback на -1.
@@ -167,6 +217,9 @@ _REDERIVE_SQL = text(
         FROM listings
         WHERE is_active
           AND rooms IS NOT NULL
+          -- #2620 hardening: same area_m2 IS NOT NULL as ask_side — keeps the global-row
+          -- population consistent with the per-bucket rows it's a fallback for.
+          AND area_m2 IS NOT NULL
           AND price_per_m2 BETWEEN :ppm2_min AND :ppm2_max
           -- novostroyki guard (#1186): NULL = legacy вторичка до м.011
           AND (listing_segment IS NULL OR listing_segment = 'vtorichka')
@@ -231,6 +284,18 @@ def recompute_asking_to_sold_ratios(db: Session, run_id: int) -> dict[str, int]:
 
     Финализирует scrape_runs (mark_done / mark_failed) и пишет counters.
 
+    LIMITATION (#2620, честно задокументировано — не гард, а факт данных): sold-сторона
+    (deals) НЕ имеет маркера новостройка/вторичка — Росреестр таким свойством ДКП не
+    делится, а listing_segment (гард #1186) существует только у listings. ask_side/ask_global
+    отфильтрованы на вторичку, deal_side/deal_global — нет. Замер на проде (2026-08, #2620):
+    доля сделок с year_built >= 2020 (грубый прокси новостройки) — 44.1% в бакете «4+» против
+    23.1% в бакетах 1-3 — заметный перекос, но year_built НЕ идентифицирует первичку/вторичку
+    (продажа квартиры 2021 года постройки в 2026м — легитимная вторичка), поэтому фильтр по
+    году НЕ добавлен (создал бы новую, столь же спекулятивную асимметрию). Area-бакет-фикс
+    ниже (см. _AREA_ROOMS_BUCKET_SQL) сам по себе убрал инверсию ratio>1 в бакете «4+»
+    (0.8315 на замере прод-данных 2026-08, было 1.0257) — снятие миграции между бакетами было
+    root cause, а не новостройки.
+
     Returns {"rows_written": N, "per_rooms_rows": M, "used_global_fallback": 0|1}.
     """
     counters: dict[str, int] = {

tradein-mvp/backend/data/sql/080_asking_to_sold_ratios.sql

@@ -12,7 +12,14 @@
 --
 -- СОЗДАЁТ:
 --   asking_to_sold_ratios — таблица коэффициентов (rooms_bucket, district) → ratio.
---     rooms_bucket: 0=студия,1,2,3,4(=4+); СПЕЦ-строка rooms_bucket=-1 = global fallback.
+--     rooms_bucket: ИМЯ ЛЕГАСИ — с #2620 (2026-08) семантика AREA-BASED, не «комнаты»:
+--       0=area<30, 1=area<44, 2=area<62, 3=area<85, 4=area>=85 м² (границы = ровно та же
+--       формула, что синтезирует deals.rooms из площади при импорте, см. deploy/import-
+--       rosreestr.sh и app/tasks/asking_to_sold_ratio.py: _AREA_ROOMS_BUCKET_SQL/area_bucket()).
+--       Причина: Росреестр не отдаёт реальную комнатность, поэтому обе стороны (расчёт ask_side
+--       И применение в estimator.py) ключуются по площади — сравнение «area-бакет vs реальные
+--       комнаты» давало систематический mismatch (до 55% строк не в своём бакете, #2620).
+--       СПЕЦ-строка rooms_bucket=-1 = global fallback.
 --     district: ЗАРЕЗЕРВИРОВАНО для #647 (geo-разбивка); в #648 ВСЕГДА '' (часть PK,
 --       поэтому NOT NULL DEFAULT '' — '' можно положить в PK, NULL нельзя).
 --
@@ -61,7 +68,8 @@ BEGIN;
 -- district NOT NULL DEFAULT '' — часть PK; #647 заполнит район, #648 всегда ''.
 -- sold_median/ask_median nullable — диагностика; ratio NOT NULL (строку без ratio не пишем).
 CREATE TABLE IF NOT EXISTS asking_to_sold_ratios (
-    rooms_bucket  int          NOT NULL,             -- 0=студия..4=4+; -1 = global fallback row
+    rooms_bucket  int          NOT NULL,             -- legacy name, area-based since #2620:
+                                                       -- 0=area<30..4=area>=85; -1=global fallback
     district      text         NOT NULL DEFAULT '',  -- RESERVED for #647 (always '' in #648)
     ratio         numeric      NOT NULL,             -- sold_median_ppm2 / ask_median_ppm2
     sold_median   bigint,                            -- median(deals.price_per_m2), диагностика
@@ -75,18 +83,25 @@ CREATE TABLE IF NOT EXISTS asking_to_sold_ratios (
 );
 
 COMMENT ON TABLE asking_to_sold_ratios IS
-    'Per-rooms asking→sold коэффициент (#648): ratio = median(SOLD ppm²)/median(ASKING ppm²). '
+    'Asking→sold коэффициент (#648): ratio = median(SOLD ppm²)/median(ASKING ppm²). '
-    'rooms_bucket 0=студия..4=4+; -1 = global fallback (basis=global_fallback, пишется всегда). '
+    'rooms_bucket — LEGACY NAME, area-based since #2620: 0=area<30..4=area>=85 m2 (same '
+    'formula deals.rooms is synthesized from, see import-rosreestr.sh); -1 = global fallback '
+    '(basis=global_fallback, пишется всегда). '
     'Per-rooms строки только при n_deals>=30 AND n_listings>=30 (иначе estimator читает -1). '
     'district зарезервирован под #647 (geo), в #648 всегда ''''. '
     'Caveat: ask=ТЕКУЩИЕ listings vs sold=сделки за 12 мес (не point-in-time); ДКП=registered. '
-    'Refresh — Stage 4 asking_to_sold_ratio_refresh переиспользует derivation ниже.';
+    'Refresh — Stage 4 asking_to_sold_ratio_refresh переиспользует derivation ниже (area-bucket '
+    'ask-side since #2620 — see app/tasks/asking_to_sold_ratio.py, this seed predates it).';
 
 -- ── Derivation + seed ─────────────────────────────────────────────────────────
 -- Вынесено как один INSERT...SELECT с CTE-«сторонами» (deal_side / ask_side), чтобы
 -- Stage 4 (asking_to_sold_ratio_refresh) переиспользовал ровно эту логику. Окно сделок
 -- = трейлинг 12 мес; listings — текущие активные. ppm²-полоса [30000,600000] и бакет
 -- LEAST(GREATEST(rooms,0),4) — байт-в-байт как в харнесе (PPM2_MIN/PPM2_MAX, _bucketize_rooms).
+-- #2620 (2026-08): live-рефреш (app/tasks/asking_to_sold_ratio.py) ушёл от этого fresh-install
+-- seed — ask_side там бакетируется по площади (_AREA_ROOMS_BUCKET_SQL), а не rooms; см. комментарий
+-- в СОЗДАЁТ выше и модуль asking_to_sold_ratio.py. Этот CTE-блок оставлен как есть (fresh-install
+-- seed, применяется один раз через _schema_migrations) — не источник истины для прод-derivation.
 WITH
 -- SOLD медианы по бакетам комнат за трейлинг-12мес (ДКП Росреестра).
 deal_side AS (

tradein-mvp/backend/tests/test_asking_to_sold_ratio.py

@@ -197,6 +197,17 @@ def test_migration_080_derivation_is_subset_of_refresh_sql() -> None:
 
     #2583 H2: the refresh now also adds the NULL-tolerant city-scope predicate to each
     ask_* CTE (symmetric to the #C2 SOLD-side guard) — normalised away the same way.
+
+    #2620: the refresh now buckets ask_side by AREA (_AREA_ROOMS_BUCKET_SQL) instead of
+    listings.rooms — the 080 seed still uses the rooms-based LEAST/GREATEST formula for
+    ask_side (pre-#2620, the bug this fixes). Both bucket formulas normalise to the same
+    placeholder token so this test keeps proving everything ELSE unchanged (CTE shape,
+    threshold, window, deal_side bucketing) — the #2620 divergence itself is asserted by
+    test_ask_side_buckets_by_area_not_rooms below.
+
+    #2620 hardening: the refresh also adds `AND area_m2 IS NOT NULL` to ask_side/ask_global
+    (NULL area_m2 would fall into the CASE ELSE branch = bucket 4 — a latent trap). Absent
+    in the 080 seed; dropped here the same way as the other guards above.
     """
     seed_sql = _MIGRATION_080.read_text("utf-8")
     # Extract the WITH … (up to the ON CONFLICT) from the seed.
@@ -231,12 +242,76 @@ def test_migration_080_derivation_is_subset_of_refresh_sql() -> None:
         s = re.sub(r"AND\s*\(\s*city\s+IS\s+NULL\s+OR\s+city\s+ILIKE\s+:asking_city\s*\)", "", s)
         return s
 
+    def _normalise_bucket_expr(s: str) -> str:
+        """Collapse the rooms-based and area-based bucket formulas to one placeholder.
+
+        #2620: ask_side buckets by area now (_AREA_ROOMS_BUCKET_SQL), not rooms. Both
+        forms appear an equal number of times (2x each: SELECT expr + GROUP BY) once
+        deal_side's untouched rooms-formula occurrences are also normalised, so this
+        keeps the containment check valid for the parts of the derivation #2620 did NOT
+        touch (deal_side stays rooms-bucketed — it has no other choice, deals.rooms IS
+        the synthetic area bucket already).
+        """
+        s = re.sub(r"LEAST\(GREATEST\(rooms,\s*0\),\s*4\)", "BUCKET_PLACEHOLDER", s)
+        s = re.sub(
+            r"CASE WHEN area_m2 < 30 THEN 0 WHEN area_m2 < 44 THEN 1 "
+            r"WHEN area_m2 < 62 THEN 2 WHEN area_m2 < 85 THEN 3 ELSE 4 END",
+            "BUCKET_PLACEHOLDER",
+            s,
+        )
+        return s
+
+    def _drop_area_not_null_guard(s: str) -> str:
+        """Remove the #2620 hardening guard (absent in the 080 seed).
+
+        NULL area_m2 falls into the CASE ELSE branch (bucket 4) of the area formula — a
+        latent trap. ask_side/ask_global added `AND area_m2 IS NOT NULL` explicitly.
+        """
+        return re.sub(r"AND\s+area_m2\s+IS\s+NOT\s+NULL", "", s)
+
     def _norm(s: str) -> str:
-        return _strip_sql(_normalise_ppm2(_drop_city_guard(_drop_segment_guard(s))))
+        return _strip_sql(
+            _normalise_ppm2(
+                _normalise_bucket_expr(
+                    _drop_area_not_null_guard(_drop_city_guard(_drop_segment_guard(s)))
+                )
+            )
+        )
 
     assert _norm(seed_derivation) in _norm(_REDERIVE_SQL)
 
 
+def test_ask_side_buckets_by_area_not_rooms() -> None:
+    """#2620: ask_side buckets listings by AREA (same CASE as deals), not listings.rooms.
+
+    Root cause of the ratio>1 bug in bucket "4+": deals.rooms is synthetic (derived from
+    area_m2 at import time, deploy/import-rosreestr.sh — Rosreestr doesn't report room
+    counts), while listings.rooms is a REAL room count. Comparing a synthetic area-bucket
+    to a real-rooms-bucket mixed two different classifications — measured on prod
+    (2026-08, #2620): 23-55% of listings migrate to a different bucket depending on which
+    classification is used, not just in the "4+" bucket (which was ALSO truncated: deals
+    caps rooms at 4 via ELSE 4, listings.rooms does not).
+
+    Falsifiable: this assert is FALSE on the pre-#2620 code (ask_side bucketed by
+    LEAST(GREATEST(rooms, 0), 4), same as deal_side) and TRUE once ask_side switches to
+    _AREA_ROOMS_BUCKET_SQL.
+    """
+    assert ratio_mod._AREA_ROOMS_BUCKET_SQL.startswith("CASE WHEN area_m2 < 30 THEN 0")
+    assert ratio_mod._AREA_ROOMS_BUCKET_SQL in _REDERIVE_SQL
+
+    _a = _REDERIVE_SQL.index("ask_side AS")
+    _b = _REDERIVE_SQL.index("per_bucket AS")
+    ask_side_block = _REDERIVE_SQL[_a:_b]
+    assert ratio_mod._AREA_ROOMS_BUCKET_SQL in ask_side_block
+    assert "LEAST(GREATEST(rooms" not in ask_side_block
+
+    # deal_side is UNCHANGED — deals.rooms is already the synthetic area bucket, there is
+    # no separate "real rooms" column to prefer instead (see docstring above).
+    _c = _REDERIVE_SQL.index("deal_side AS")
+    deal_side_block = _REDERIVE_SQL[_c:_a]
+    assert "LEAST(GREATEST(rooms, 0), 4)" in deal_side_block
+
+
 # ── Counters query ────────────────────────────────────────────────────────────
 
 
@@ -426,3 +501,42 @@ def test_migration_098_band_matches_settings_default() -> None:
     assert (
         Settings().asking_ratio_ppm2_max == 1_200_000
     ), "asking_ratio_ppm2_max default changed; update migration 098 seed literals to match"
+
+
+# ── area_bucket() Python twin matches _AREA_ROOMS_BUCKET_SQL (#2620) ─────────
+
+
+def test_area_bucket_matches_sql_boundaries() -> None:
+    """area_bucket() has IDENTICAL boundaries to _AREA_ROOMS_BUCKET_SQL (#2620).
+
+    Two representations of one truth (see module comment: shell/import-rosreestr.sh → SQL
+    _AREA_ROOMS_BUCKET_SQL → Python area_bucket()). If SQL (computes the ratio in ask_side)
+    and Python (applies it in estimator.py) drift apart, the #2620 mismatch is silently
+    reintroduced. Boundaries are parsed straight out of the SQL string, not hardcoded
+    independently, so this fails the moment either one is edited without the other.
+    """
+    sql_bounds = [int(n) for n in re.findall(r"area_m2 < (\d+)", ratio_mod._AREA_ROOMS_BUCKET_SQL)]
+    assert sql_bounds == [30, 44, 62, 85]
+
+    cases = {
+        0: 0,
+        29: 0,
+        29.99: 0,
+        30: 1,
+        43: 1,
+        43.99: 1,
+        44: 2,
+        61: 2,
+        61.99: 2,
+        62: 3,
+        84: 3,
+        84.99: 3,
+        85: 4,
+        200: 4,
+    }
+    for area, expected in cases.items():
+        assert ratio_mod.area_bucket(area) == expected, f"area={area} -> expected {expected}"
+
+    for i, bound in enumerate(sql_bounds):
+        assert ratio_mod.area_bucket(bound - 0.01) == i
+        assert ratio_mod.area_bucket(bound) == i + 1

tradein-mvp/backend/tests/test_estimator_expected_sold.py

@@ -3,7 +3,8 @@
 Two layers:
 
 1. `_get_asking_sold_ratio` lookup helper (DB mocked):
-   - bucket = min(max(rooms or 0, 0), 4) — clamping + None handling
+   - bucket = area_bucket(area_m2) when area_m2 known (#2620-2), else
+     min(max(rooms or 0, 0), 4) — rooms-fallback + clamping + None handling
    - per-rooms row hit returns (ratio, basis)
    - per-rooms miss falls back to the global rooms_bucket=-1 row
    - empty / missing table → (None, None), never raises (graceful)
@@ -72,6 +73,36 @@ def test_bucket_clamping() -> None:
         ), f"rooms={rooms} → bucket {bind['b']} != {expected_bucket}"
 
 
+def test_bucket_keyed_by_area_when_area_known() -> None:
+    """#2620-2: bucket = area_bucket(area_m2), NOT rooms, when area_m2 is given.
+
+    Root cause moved from calc (ask_side, fixed by #2620) to apply (this function) if the
+    lookup still keyed by listings/payload.rooms: расчёт бакетирует по площади, применение
+    должно совпадать. A "4-room" 70m² flat area-buckets to 3 (62<=70<85) — deep-review prod
+    measurement (2026-08): 310/1038 (29.9%) historical requests would land in a different
+    ratio bucket under the old rooms-key vs the area-key.
+    """
+    from app.services.estimator import _get_asking_sold_ratio
+
+    _clear_ratio_cache()
+    db = _db_returning([_FakeRow(0.83, "per_rooms")])
+    _get_asking_sold_ratio(db, rooms=4, area_m2=70.0)
+    bind = db.execute.call_args_list[0].args[1]
+    assert bind["b"] == 3, f"70m2 (rooms=4 claimed) should bucket to 3 by area, got {bind['b']}"
+
+
+def test_bucket_falls_back_to_rooms_when_area_missing() -> None:
+    """area_m2 unknown (None/0) → rooms-based bucket fallback, min(max(rooms,0),4)."""
+    from app.services.estimator import _get_asking_sold_ratio
+
+    for area_m2 in (None, 0, 0.0):
+        _clear_ratio_cache()
+        db = _db_returning([_FakeRow(0.83, "per_rooms")])
+        _get_asking_sold_ratio(db, rooms=4, area_m2=area_m2)
+        bind = db.execute.call_args_list[0].args[1]
+        assert bind["b"] == 4, f"area_m2={area_m2!r} should fall back to rooms bucket 4"
+
+
 def test_per_rooms_hit_returns_ratio_basis() -> None:
     """Per-rooms row present → returned directly, no fallback query."""
     from app.services.estimator import _get_asking_sold_ratio

tradein-mvp/backend/tests/test_estimator_ratio_tier_fix.py

@@ -107,8 +107,9 @@ def _run_estimate_with_ratio_spy(
 
     ratio_calls: list[Any] = []
 
-    def _spy_ratio(db_inner: Any, rooms: Any, anchor_ppm2: Any = None) -> Any:
+    def _spy_ratio(db_inner: Any, rooms: Any, area_m2: Any = None, anchor_ppm2: Any = None) -> Any:
-        ratio_calls.append({"rooms": rooms, "anchor_ppm2": anchor_ppm2})
+        # #2620-2: signature now carries area_m2 (rekey to area-bucket) before anchor_ppm2.
+        ratio_calls.append({"rooms": rooms, "area_m2": area_m2, "anchor_ppm2": anchor_ppm2})
         return ratio_return
 
     async def _run() -> Any:
@@ -168,7 +169,8 @@ def test_ratio_tier_uses_final_headline_after_anchor() -> None:
     # _get_asking_sold_ratio должен получить anchor_ppm2 ≈ 300k (не 105k).
     captured_anchor_ppm2: list[float | None] = []
 
-    def _spy(db_inner: Any, rooms: Any, anchor_ppm2: Any = None) -> tuple:
+    def _spy(db_inner: Any, rooms: Any, area_m2: Any = None, anchor_ppm2: Any = None) -> tuple:
+        # #2620-2: signature now carries area_m2 (rekey to area-bucket) before anchor_ppm2.
         captured_anchor_ppm2.append(anchor_ppm2)
         return (0.78, "per_rooms")