Merge pull request 'feat(tradein/estimator): deal_city_price_bands по ключу (region_code, city) — миграция 298, refresh per-region (#3051 sub-PR B)' (#3431) from feat/3051-bands-region-key into main

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

@@ -345,22 +345,24 @@ DEAL_MAX_FLOOR = 60  # выше реального максимума ЕКБ →
 # глобальный пол 50k ₽/м² для дешёвых городов области НЕ anti-outlier guard, а
 # cut-off легитимного рынка (Североуральск median ≈21.7k, Новоуральск ≈36k,
 # Асбест ≈41k — все ниже 50k → 46.6% не-ЕКБ сделок молча дропались).
-# deal_city_price_bands (миграция 178) — per-city [ppm2_min, ppm2_max] band из
+# deal_city_price_bands (миграция 178, ключ (region_code, city) — миграция 298
-# перцентилей p1/p99 реальных rosreestr-сделок города (hard floor/ceiling
+# #3051) — per-(region, city) [ppm2_min, ppm2_max] band из перцентилей p1/p99
-# 8000/800000 всё равно режут доли/опечатки). Екатеринбург НАМЕРЕННО исключён
+# реальных rosreestr-сделок города (hard floor/ceiling 8000/800000 всё равно
-# из таблицы — .get(city, ...) fallback на глобальные DEAL_MIN_PPM2/MAX_PPM2
+# режут доли/опечатки). Екатеринбург (region_code=66) НАМЕРЕННО исключён из
-# ниже гарантирует byte-identical ЕКБ-поведение.
+# таблицы — .get((region_code, city), ...) fallback на глобальные
+# DEAL_MIN_PPM2/MAX_PPM2 ниже гарантирует byte-identical ЕКБ-поведение.
 _CITY_PRICE_BANDS_CACHE_TTL_S = 300.0
-_city_price_bands_cache: tuple[dict[str, tuple[int, int]], float] | None = None
+_city_price_bands_cache: tuple[dict[tuple[int, str], tuple[int, int]], float] | None = None
 
 
-def _load_city_price_bands(db: Session) -> dict[str, tuple[int, int]]:
+def _load_city_price_bands(db: Session) -> dict[tuple[int, str], tuple[int, int]]:
-    """#2478: {city: (ppm2_min, ppm2_max)} из deal_city_price_bands (миграция 178).
+    """#2478 + #3051 (298): {(region_code, city): (ppm2_min, ppm2_max)} из deal_city_price_bands.
 
     Кэш в процессе (TTL _CITY_PRICE_BANDS_CACHE_TTL_S) — таблица рефрешится
-    периодическим ре-запуском миграции, не на каждый estimate. Таблицы нет /
+    периодическим ре-запуском derivation (298 / deal_city_price_bands_refresh),
-    любая ошибка → {} (graceful) — вызывающий код fallback'ит на глобальные
+    не на каждый estimate. Таблицы нет / любая ошибка → {} (graceful) —
-    DEAL_MIN_PPM2/DEAL_MAX_PPM2, т.е. поведение как до #2478.
+    вызывающий код fallback'ит на глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2, т.е.
+    поведение как до #2478.
     """
     global _city_price_bands_cache
     if _city_price_bands_cache is not None:
@@ -369,14 +371,18 @@ def _load_city_price_bands(db: Session) -> dict[str, tuple[int, int]]:
             return bands
     try:
         rows = (
-            db.execute(text("SELECT city, ppm2_min, ppm2_max FROM deal_city_price_bands"))
+            db.execute(
+                text("SELECT region_code, city, ppm2_min, ppm2_max FROM deal_city_price_bands")
+            )
             .mappings()
             .all()
         )
     except Exception as exc:
         logger.warning("deal_city_price_bands lookup failed (graceful): %s", exc)
         return {}
-    bands = {r["city"]: (int(r["ppm2_min"]), int(r["ppm2_max"])) for r in rows}
+    bands = {
+        (int(r["region_code"]), r["city"]): (int(r["ppm2_min"]), int(r["ppm2_max"])) for r in rows
+    }
     _city_price_bands_cache = (bands, time.monotonic())
     return bands
 
@@ -1885,7 +1891,8 @@ def _fetch_dkp_corridor(
                     f"""
                     SELECT d.price_per_m2, d.deal_date
                     FROM deals d
-                    LEFT JOIN deal_city_price_bands b ON b.city = d.city
+                    LEFT JOIN deal_city_price_bands b
+                        ON b.region_code = d.region_code AND b.city = d.city
                     WHERE d.source = 'rosreestr'
                       AND d.region_code = CAST(:region_code AS int)
                       AND d.address ILIKE :street_pattern
@@ -1896,11 +1903,12 @@ def _fetch_dkp_corridor(
                           - (CAST(:period_months AS integer) || ' months')::interval
                       AND d.price_per_m2 > 0
                       {city_filter_sql}
-                      -- #699 + #2478: режем нерыночные ppm²-выбросы из коридора
+                      -- #699 + #2478 + #3051: режем нерыночные ppm²-выбросы из коридора
-                      -- expected_sold. Per-city band (deal_city_price_bands, миграция
+                      -- expected_sold. Per-(region, city) band (deal_city_price_bands,
-                      -- 178) когда для города сделки есть строка (не-ЕКБ область);
+                      -- миграция 178, ключ (region_code, city) — миграция 298) когда для
-                      -- иначе (в т.ч. Екатеринбург, НАМЕРЕННО не в таблице) fallback
+                      -- (региона, города) сделки есть строка; иначе (в т.ч. Екатеринбург
-                      -- на глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2 — byte-identical.
+                      -- region_code=66, НАМЕРЕННО не в таблице) fallback на глобальные
+                      -- DEAL_MIN_PPM2/DEAL_MAX_PPM2 — byte-identical.
                       AND d.price_per_m2 BETWEEN COALESCE(b.ppm2_min, CAST(:ppm_min AS int))
                                               AND COALESCE(b.ppm2_max, CAST(:ppm_max AS int))
                     """
@@ -1968,7 +1976,8 @@ def _fetch_dkp_corridor(
                         """
                         SELECT d.price_per_m2, d.deal_date
                         FROM deals d
-                        LEFT JOIN deal_city_price_bands b ON b.city = d.city
+                        LEFT JOIN deal_city_price_bands b
+                            ON b.region_code = d.region_code AND b.city = d.city
                         WHERE d.source = 'rosreestr'
                           AND d.region_code = CAST(:region_code AS int)
                           AND d.city IS NOT NULL
@@ -6552,25 +6561,29 @@ def _is_plausible_deal(
     area_m2: float | None = None,
     price_rub: float | None = None,
     city: str | None = None,
-    bands: dict[str, tuple[int, int]] | None = None,
+    bands: dict[tuple[int, str], tuple[int, int]] | None = None,
+    region_code: int = regions_mod.DEFAULT_REGION_CODE,
 ) -> bool:
-    """#699 + Mera-audit fix-2 + #2478: True если ДКП-сделка правдоподобна (не выброс).
+    """#699 + Mera-audit fix-2 + #2478 + #3051 (298): True если ДКП-сделка правдоподобна.
 
     Абсолютные guard-bands (см. DEAL_* константы). None-поля не судим (keep —
     нечем сравнивать). Проверки:
     - price_per_m2 вне [ppm_min, ppm_max] → drop, где (ppm_min, ppm_max) =
-      bands.get(city, (DEAL_MIN_PPM2, DEAL_MAX_PPM2)) — #2478: per-city band
+      bands.get((region_code, city), (DEAL_MIN_PPM2, DEAL_MAX_PPM2)) — #2478:
-      (deal_city_price_bands, миграция 178) для не-ЕКБ городов области, где
+      per-(region, city) band (deal_city_price_bands, миграция 178, ключ
-      глобальный пол 50k ₽/м² режет легитимно дешёвый рынок. city не в bands
+      (region_code, city) — миграция 298) для не-ЕКБ городов области, где
-      (в т.ч. Екатеринбург — НАМЕРЕННО не в таблице) или bands=None →
+      глобальный пол 50k ₽/м² режет легитимно дешёвый рынок. region_code —
-      fallback на глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2 (byte-identical
+      kw-only с дефолтом regions_mod.DEFAULT_REGION_CODE (66), позиционные
-      сегодняшнему поведению).
+      вызовы не ломаются. (region_code, city) не в bands (в т.ч. Екатеринбург
+      region_code=66 — НАМЕРЕННО не в таблице) или bands=None → fallback на
+      глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2 (byte-identical сегодняшнему
+      поведению).
     - floor < 1 или floor > DEAL_MAX_FLOOR → drop (битый парсер: floor=-5/999)
     - floor > total_floors физически невозможен → drop
     - area_m2 задана и <= 0 → drop (битый парсер)
     - price_rub задана и <= 0 → drop (нерыночная/технческая сделка)
     """
-    ppm_min, ppm_max = (bands or {}).get(city, (DEAL_MIN_PPM2, DEAL_MAX_PPM2))
+    ppm_min, ppm_max = (bands or {}).get((region_code, city), (DEAL_MIN_PPM2, DEAL_MAX_PPM2))
     if price_per_m2 is not None and not (ppm_min <= price_per_m2 <= ppm_max):
         return False
     if floor is not None:
@@ -6595,7 +6608,7 @@ def _fetch_deals(
             SELECT
                 source, address, lat, lon,
                 rooms, area_m2, floor, total_floors,
-                price_rub, price_per_m2, city,
+                price_rub, price_per_m2, city, region_code,
                 deal_date, days_on_market,
                 cadastral_number,
                 ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography) AS distance_m
@@ -6625,8 +6638,11 @@ def _fetch_deals(
 
     # #699 + Mera-audit fix-2: отсекаем ДКП-выбросы (битый этаж / нерыночный ppm²
     # / нулевая площадь / нулевая цена) до выдачи в actual_deals и expected_sold.
-    # #2478: per-city ppm² band (deal_city_price_bands) вместо глобального
+    # #2478 + #3051 (298): per-(region, city) ppm² band (deal_city_price_bands)
-    # DEAL_MIN_PPM2/MAX_PPM2 — грузим один раз на вызов, graceful {} при ошибке.
+    # вместо глобального DEAL_MIN_PPM2/MAX_PPM2 — грузим один раз на вызов,
+    # graceful {} при ошибке. region_code=NULL (не должно случаться для
+    # rosreestr-строк, см. 177/288 backfill) → DEFAULT_REGION_CODE, чтобы band
+    # lookup не падал на None-ключе.
     bands = _load_city_price_bands(db)
     deals = [dict(r) for r in rows]
     clean = [
@@ -6640,6 +6656,7 @@ def _fetch_deals(
             d.get("price_rub"),
             city=d.get("city"),
             bands=bands,
+            region_code=d.get("region_code") or regions_mod.DEFAULT_REGION_CODE,
         )
     ]
     if len(clean) < len(deals):

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

@@ -15,17 +15,29 @@ asking_to_sold_ratio_refresh (06:00-07:00 UTC), чтобы бэнды счита
 свежему срезу deals, что и ratio-таблица того же дня.
 
 SQL derivation ниже — БАЙТ-В-БАЙТ та же логика, что seed в
-data/sql/194_deal_city_price_bands_tiers.sql (region_stats / city_stats / tiered:
+data/sql/298_deal_city_price_bands_region.sql (region_stats / city_stats / tiered:
 трёхуровневая схема full N>=30 / rough N 10-29 / region_fallback N 1-9, см.
-комментарий в 194 для полного обоснования тиров и hard floor/ceiling клампов).
+комментарий в 194/298 для полного обоснования тиров и hard floor/ceiling клампов).
+
+#3051 «Москва» (298): ключ (region_code, city) вместо (city) — region_stats и
+city_stats теперь группируются ПО РЕГИОНУ (region_code), а не по всей таблице
+deals целиком. Без этого пул для tier='region_fallback' одного региона
+подмешивал бы сделки другого (Москва в deals region_code=77 иначе тянула бы
+p1-floor малых городов Свердловской обл. region_code=66 вверх). Для
+region_code=66 derivation байт-в-байт прежняя (194): фильтр
+NOT (region_code = 66 AND city = 'Екатеринбург') — тот же инвариант, что
+раньше `city <> 'Екатеринбург'`; region_stats/city_stats для региона 66 видят
+ТУ ЖЕ популяцию строк, что видели до появления региона 77 в deals.
 
 Нет DELETE перед re-derive (в отличие от asking_to_sold_ratio.py true-mirror
-паттерна) — множество городов монотонно растёт (rosreestr_dkp_import только
+паттерна) — множество (region_code, city) монотонно растёт
-INSERT/ON CONFLICT DO UPDATE, никогда не удаляет сделки), поэтому merge-по-city
+(rosreestr_dkp_import только INSERT/ON CONFLICT DO UPDATE, никогда не удаляет
-(ON CONFLICT DO UPDATE) достаточен: город, перешедший в другой tier, просто
+сделки), поэтому merge-по-ключу (ON CONFLICT DO UPDATE) достаточен: город,
-перезаписывается на следующем refresh. Екатеринбург НЕ включён (WHERE city <>
+перешедший в другой tier, просто перезаписывается на следующем refresh.
-'Екатеринбург') — estimator.py fallback на глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2
+Екатеринбург НЕ включён для региона 66 (WHERE NOT (region_code = 66 AND
-для ЕКБ остаётся byte-identical (invariant из 178/194 сохранён).
+city = 'Екатеринбург')) — estimator.py fallback на глобальные
+DEAL_MIN_PPM2/DEAL_MAX_PPM2 для ЕКБ остаётся byte-identical (invariant из
+178/194/298 сохранён).
 """
 
 from __future__ import annotations
@@ -39,22 +51,28 @@ from app.services import scrape_runs as runs_mod
 
 logger = logging.getLogger(__name__)
 
-# ── Derivation + re-seed (БАЙТ-В-БАЙТ из 194) ─────────────────────────────────
+# ── Derivation + re-seed (БАЙТ-В-БАЙТ из 298, region-aware) ──────────────────
 _REDERIVE_SQL = text(
     """
     WITH region_stats AS (
-        SELECT GREATEST(
+        SELECT
-                   round(percentile_cont(0.01) WITHIN GROUP (ORDER BY price_per_m2))::int,
+            region_code,
-                   8000
+            GREATEST(
-               ) AS region_ppm2_min
+                round(percentile_cont(0.01) WITHIN GROUP (ORDER BY price_per_m2))::int,
+                8000
+            ) AS region_ppm2_min
         FROM deals
         WHERE source = 'rosreestr'
+          AND doc_type = 'ДКП'
           AND price_per_m2 IS NOT NULL
           AND city IS NOT NULL
-          AND city <> 'Екатеринбург'
+          AND region_code IS NOT NULL
+          AND NOT (region_code = 66 AND city = 'Екатеринбург')
+        GROUP BY region_code
     ),
     city_stats AS (
         SELECT
+            region_code,
             city,
             GREATEST(round(percentile_cont(0.01) WITHIN GROUP (ORDER BY price_per_m2))::int, 8000)
                 AS ppm2_p1,
@@ -63,13 +81,15 @@ _REDERIVE_SQL = text(
             count(*) AS n_deals
         FROM deals
         WHERE source = 'rosreestr'
+          AND doc_type = 'ДКП'
           AND price_per_m2 IS NOT NULL
           AND city IS NOT NULL
-          AND city <> 'Екатеринбург'
+          AND region_code IS NOT NULL
-        GROUP BY city
+          AND NOT (region_code = 66 AND city = 'Екатеринбург')
+        GROUP BY region_code, city
     ),
     tiered AS (
-        SELECT city, ppm2_p1 AS ppm2_min, ppm2_p99 AS ppm2_max, n_deals,
+        SELECT region_code, city, ppm2_p1 AS ppm2_min, ppm2_p99 AS ppm2_max, n_deals,
                'full'::text AS tier
         FROM city_stats
         WHERE n_deals >= 30
@@ -77,23 +97,24 @@ _REDERIVE_SQL = text(
 
         UNION ALL
 
-        SELECT city, LEAST(ppm2_p1, 700000) AS ppm2_min, 800000 AS ppm2_max, n_deals,
+        SELECT region_code, city, LEAST(ppm2_p1, 700000) AS ppm2_min, 800000 AS ppm2_max,
-               'rough'::text AS tier
+               n_deals, 'rough'::text AS tier
         FROM city_stats
         WHERE n_deals BETWEEN 10 AND 29
 
         UNION ALL
 
-        SELECT c.city, r.region_ppm2_min AS ppm2_min, 800000 AS ppm2_max, c.n_deals,
+        SELECT c.region_code, c.city, r.region_ppm2_min AS ppm2_min, 800000 AS ppm2_max,
-               'region_fallback'::text AS tier
+               c.n_deals, 'region_fallback'::text AS tier
         FROM city_stats c
-        CROSS JOIN region_stats r
+        JOIN region_stats r ON r.region_code = c.region_code
         WHERE c.n_deals < 10
     )
-    INSERT INTO deal_city_price_bands (city, ppm2_min, ppm2_max, n_deals, tier, refreshed_at)
+    INSERT INTO deal_city_price_bands
-    SELECT city, ppm2_min, ppm2_max, n_deals, tier, now()
+        (region_code, city, ppm2_min, ppm2_max, n_deals, tier, refreshed_at)
+    SELECT region_code, city, ppm2_min, ppm2_max, n_deals, tier, now()
     FROM tiered
-    ON CONFLICT (city) DO UPDATE
+    ON CONFLICT (region_code, city) DO UPDATE
         SET ppm2_min = EXCLUDED.ppm2_min,
             ppm2_max = EXCLUDED.ppm2_max,
             n_deals = EXCLUDED.n_deals,
@@ -103,13 +124,18 @@ _REDERIVE_SQL = text(
 )
 
 # ── Post-insert counters ──────────────────────────────────────────────────────
+# #3051 (298): regions — число различных region_code в таблице после re-derive
+# (Свердловская обл. + Москва после включения региона 77). Добавлено в конец
+# SELECT-списка, прежние 4 счётчика на тех же местах — контракт _COUNTERS_SQL
+# (rows_written/full_rows/rough_rows/region_fallback_rows) не ломается.
 _COUNTERS_SQL = text(
     """
     SELECT
         COUNT(*)                                          AS rows_written,
         COUNT(*) FILTER (WHERE tier = 'full')              AS full_rows,
         COUNT(*) FILTER (WHERE tier = 'rough')             AS rough_rows,
-        COUNT(*) FILTER (WHERE tier = 'region_fallback')   AS region_fallback_rows
+        COUNT(*) FILTER (WHERE tier = 'region_fallback')   AS region_fallback_rows,
+        COUNT(DISTINCT region_code)                        AS regions
     FROM deal_city_price_bands
     """
 )
@@ -124,13 +150,15 @@ def refresh_deal_city_price_bands(db: Session, run_id: int) -> dict[str, int]:
 
     Финализирует scrape_runs (mark_done / mark_failed) и пишет counters.
 
-    Returns {"rows_written": N, "full_rows": .., "rough_rows": .., "region_fallback_rows": ..}.
+    Returns {"rows_written": N, "full_rows": .., "rough_rows": .., "region_fallback_rows": ..,
+    "regions": ..}.
     """
     counters: dict[str, int] = {
         "rows_written": 0,
         "full_rows": 0,
         "rough_rows": 0,
         "region_fallback_rows": 0,
+        "regions": 0,
     }
     try:
         db.execute(_REDERIVE_SQL)
@@ -141,17 +169,19 @@ def refresh_deal_city_price_bands(db: Session, run_id: int) -> dict[str, int]:
             counters["full_rows"] = int(row["full_rows"] or 0)
             counters["rough_rows"] = int(row["rough_rows"] or 0)
             counters["region_fallback_rows"] = int(row["region_fallback_rows"] or 0)
+            counters["regions"] = int(row["regions"] or 0)
 
         db.commit()
         runs_mod.mark_done(db, run_id, counters)
         logger.info(
             "refresh_deal_city_price_bands run_id=%d done: "
-            "rows_written=%d full=%d rough=%d region_fallback=%d",
+            "rows_written=%d full=%d rough=%d region_fallback=%d regions=%d",
             run_id,
             counters["rows_written"],
             counters["full_rows"],
             counters["rough_rows"],
             counters["region_fallback_rows"],
+            counters["regions"],
         )
         return counters
     except Exception as exc:

tradein-mvp/backend/data/sql/298_deal_city_price_bands_region.sql

@@ -0,0 +1,170 @@
+-- 298_deal_city_price_bands_region.sql
+-- deal_city_price_bands: ключ (region_code, city) вместо (city) — #3051, sub-PR B.
+--
+-- Dependencies: 178_deal_city_price_bands.sql (таблица, PK(city)), 194_deal_city_price_bands_tiers.sql
+--               (tier-схема, derivation), 288_deals_doc_type.sql (deals.doc_type).
+-- Apply after: 289_rosreestr_fdw_msk_columns_seed77.sql
+--
+-- WHY:
+--   В deals уже 212 937 строк region_code=77 city='Москва' рядом со 108 623
+--   строками region_code=66 (Свердловская обл.) — Москва в той же таблице
+--   `deals`, что и область. Старая derivation (178/194) ключует бэнды ТОЛЬКО
+--   по city и считает region_stats (пул для tier='region_fallback') ПО ВСЕЙ
+--   таблице deals без фильтра региона. С двумя регионами в одной таблице это
+--   ломает две вещи одновременно:
+--     1. city='Москва' коллизирует c одноимёнными city в других регионах
+--        (маловероятно для этой пары, но ключ (city) в принципе не region-safe).
+--     2. region_stats-пул для 'region_fallback' смешивает 212k строк Москвы в
+--        p1-перцентиль floor для тонких городов Свердловской обл. — Москва на
+--        порядок дороже, пул сдвигается вверх, floor малых городов области
+--        задирается выше их реального рынка.
+--
+-- WHAT:
+--   - deal_city_price_bands.region_code int NOT NULL DEFAULT 66 — бэкфилл
+--     существующих 383 строк на 66 (таблица до этой миграции была исключительно
+--     Свердловская обл., см. 178/194 WHERE-фильтры без region_code).
+--   - PK (city) → PK (region_code, city) — через DO-guard по pg_get_constraintdef,
+--     идемпотентно (см. ниже).
+--   - Re-seed той же derivation, что 194/refresh-таск, НО region_stats и
+--     city_stats группируются по (region_code[, city]) — Москва больше не
+--     утекает в пул малых городов области, и наоборот.
+--
+-- БАЙТ-ИДЕНТИЧНОСТЬ ДЛЯ region_code=66:
+--   - Фильтр `NOT (region_code = 66 AND city = 'Екатеринбург')` — тот же
+--     инвариант, что `city <> 'Екатеринбург'` в 178/194, ограниченный явно на
+--     регион 66 (ЕКБ существует только там).
+--   - `doc_type = 'ДКП'` — НЕ сужает выборку region_code=66: backfill 288
+--     проставил doc_type='ДКП' на 100% строк source='rosreestr' (на момент
+--     миграции 288 других doc_type для source='rosreestr' не было и не могло
+--     быть — импорт всегда фильтровал ДКП на входе).
+--   - `region_stats` теперь GROUP BY region_code — для region_code=66 пул p1
+--     считается ИСКЛЮЧИТЕЛЬНО по строкам региона 66 (та же популяция, что была
+--     у 194 ДО появления в deals региона 77), т.е. Москва (region_code=77)
+--     физически не участвует в агрегате region_stats-строки region_code=66.
+--   - `city_stats` теперь GROUP BY region_code, city — для региона 66 разбивка
+--     по городам не меняется (раньше GROUP BY city неявно уже был per-region,
+--     т.к. в deals была только область).
+--   - Клампы (8000/800000/700000), tier-пороги (30/10) и p1/p99-перцентили —
+--     формулы дословно как в 194/refresh, только с добавленным region_code в
+--     SELECT/GROUP BY/JOIN.
+--   - `tiered` JOIN region_stats теперь по region_code (вместо CROSS JOIN на
+--     единственную строку) — для региона 66 эквивалентно прежнему CROSS JOIN,
+--     т.к. region_stats содержит ровно одну строку на регион.
+--
+-- ИДЕМПОТЕНТНОСТЬ:
+--   ADD COLUMN IF NOT EXISTS; PK-guard проверяет ТЕКУЩЕЕ определение PK через
+--   pg_get_constraintdef и меняет его только если оно ещё (city) — повторный
+--   прогон видит PK(region_code, city) и не трогает его. INSERT ... ON CONFLICT
+--   (region_code, city) DO UPDATE — повторный прогон рефрешит бэнды под свежие
+--   сделки (та же семантика 178/194).
+
+BEGIN;
+
+SET LOCAL lock_timeout = '5s';
+
+ALTER TABLE deal_city_price_bands
+    ADD COLUMN IF NOT EXISTS region_code int NOT NULL DEFAULT 66;
+
+COMMENT ON COLUMN deal_city_price_bands.region_code IS
+    'Регион ключа бэнда (deals.region_code). 66 = Свердловская обл. (Екатеринбург '
+    'намеренно исключён из таблицы для ЭТОГО региона — estimator fallback на '
+    'глобальные DEAL_MIN_PPM2/DEAL_MAX_PPM2). Часть составного PK (region_code, city).';
+
+-- PK (city) → PK (region_code, city), идемпотентно: смотрим ТЕКУЩЕЕ определение
+-- PK и меняем его только если это ещё старый PK(city).
+DO $$
+DECLARE
+    v_pk_def text;
+BEGIN
+    SELECT pg_get_constraintdef(oid) INTO v_pk_def
+    FROM pg_constraint
+    WHERE conrelid = 'deal_city_price_bands'::regclass
+      AND contype = 'p';
+
+    IF v_pk_def = 'PRIMARY KEY (city)' THEN
+        ALTER TABLE deal_city_price_bands DROP CONSTRAINT deal_city_price_bands_pkey;
+    END IF;
+
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_constraint
+        WHERE conrelid = 'deal_city_price_bands'::regclass
+          AND contype = 'p'
+    ) THEN
+        ALTER TABLE deal_city_price_bands
+            ADD CONSTRAINT deal_city_price_bands_pkey PRIMARY KEY (region_code, city);
+    END IF;
+END $$;
+
+-- Страховка: если region_code=66 город='Москва' когда-либо просочился (не
+-- должен — таблица была исключительно Свердловской обл.), убираем перед re-seed,
+-- чтобы ON CONFLICT ниже не унаследовал мусорную строку под чужим regionʼом.
+DELETE FROM deal_city_price_bands WHERE region_code = 66 AND city = 'Москва';
+
+-- Re-seed — та же derivation, что 194/refresh-таск, региональная (см. WHY/WHAT).
+WITH region_stats AS (
+    SELECT
+        region_code,
+        GREATEST(
+            round(percentile_cont(0.01) WITHIN GROUP (ORDER BY price_per_m2))::int,
+            8000
+        ) AS region_ppm2_min
+    FROM deals
+    WHERE source = 'rosreestr'
+      AND doc_type = 'ДКП'
+      AND price_per_m2 IS NOT NULL
+      AND city IS NOT NULL
+      AND region_code IS NOT NULL
+      AND NOT (region_code = 66 AND city = 'Екатеринбург')
+    GROUP BY region_code
+),
+city_stats AS (
+    SELECT
+        region_code,
+        city,
+        GREATEST(round(percentile_cont(0.01) WITHIN GROUP (ORDER BY price_per_m2))::int, 8000)
+            AS ppm2_p1,
+        LEAST(round(percentile_cont(0.99) WITHIN GROUP (ORDER BY price_per_m2))::int, 800000)
+            AS ppm2_p99,
+        count(*) AS n_deals
+    FROM deals
+    WHERE source = 'rosreestr'
+      AND doc_type = 'ДКП'
+      AND price_per_m2 IS NOT NULL
+      AND city IS NOT NULL
+      AND region_code IS NOT NULL
+      AND NOT (region_code = 66 AND city = 'Екатеринбург')
+    GROUP BY region_code, city
+),
+tiered AS (
+    SELECT region_code, city, ppm2_p1 AS ppm2_min, ppm2_p99 AS ppm2_max, n_deals,
+           'full'::text AS tier
+    FROM city_stats
+    WHERE n_deals >= 30
+      AND ppm2_p99 >= 8000
+
+    UNION ALL
+
+    SELECT region_code, city, LEAST(ppm2_p1, 700000) AS ppm2_min, 800000 AS ppm2_max, n_deals,
+           'rough'::text AS tier
+    FROM city_stats
+    WHERE n_deals BETWEEN 10 AND 29
+
+    UNION ALL
+
+    SELECT c.region_code, c.city, r.region_ppm2_min AS ppm2_min, 800000 AS ppm2_max, c.n_deals,
+           'region_fallback'::text AS tier
+    FROM city_stats c
+    JOIN region_stats r ON r.region_code = c.region_code
+    WHERE c.n_deals < 10
+)
+INSERT INTO deal_city_price_bands (region_code, city, ppm2_min, ppm2_max, n_deals, tier, refreshed_at)
+SELECT region_code, city, ppm2_min, ppm2_max, n_deals, tier, now()
+FROM tiered
+ON CONFLICT (region_code, city) DO UPDATE
+    SET ppm2_min = EXCLUDED.ppm2_min,
+        ppm2_max = EXCLUDED.ppm2_max,
+        n_deals = EXCLUDED.n_deals,
+        tier = EXCLUDED.tier,
+        refreshed_at = EXCLUDED.refreshed_at;
+
+COMMIT;

tradein-mvp/backend/scripts/backtest_estimator.py

@@ -1233,14 +1233,19 @@ _CANDIDATES_SQL = text(
 )
 
 # (oblast D) Per-city PPM2 sanity band — same table the estimator's
-# _fetch_dkp_corridor COALESCEs against (migration 178). Looked up only when
+# _fetch_dkp_corridor COALESCEs against (migration 178, key (region_code, city)
-# --city is set (see _resolve_city_ppm2_band); the default (city=None) path
+# since migration 298 — #3051 "Москва"). Looked up only when --city is set (see
-# never issues this query.
+# _resolve_city_ppm2_band); the default (city=None) path never issues this
+# query. region_code defaults to 66 (Свердловская обл.) — this script has no
+# region CLI flag yet, so every call is scoped to the oblast, matching every
+# existing --city invocation (byte-identical to the pre-298 unscoped lookup,
+# which only ever saw region-66 rows since the table was oblast-only before).
 _CITY_PPM2_BAND_SQL = text(
     """
     SELECT ppm2_min, ppm2_max
     FROM deal_city_price_bands
-    WHERE city = CAST(:city AS text)
+    WHERE region_code = CAST(:region_code AS int)
+      AND city = CAST(:city AS text)
     """
 )
 
@@ -1294,22 +1299,32 @@ def _sample_sql(city: str | None, scattered: bool = False) -> Any:
     )
 
 
-def _resolve_city_ppm2_band(db: Session, city: str | None) -> tuple[float, float]:
+def _resolve_city_ppm2_band(
-    """Per-city PPM2 sanity band (oblast D) — falls back to the module globals.
+    db: Session, city: str | None, region_code: int = 66
+) -> tuple[float, float]:
+    """Per-(region, city) PPM2 sanity band (oblast D, #3051) — falls back to module globals.
 
     ``city is None`` → ``(PPM2_MIN, PPM2_MAX)``, unchanged default behaviour,
     NO extra query. Otherwise looks up ``deal_city_price_bands`` for that
-    city's own ``[ppm2_min, ppm2_max]`` — a region-66 town can have a
+    (region_code, city)'s own ``[ppm2_min, ppm2_max]`` — a region-66 town can
-    materially different sane ₽/m² range than the EKB-tuned 30k..600k globals
+    have a materially different sane ₽/m² range than the EKB-tuned 30k..600k
-    (e.g. Нижний Тагил: 16 955..108 175). No row for the city (e.g.
+    globals (e.g. Нижний Тагил: 16 955..108 175). No row for the (region, city)
-    Екатеринбург — intentionally excluded from the table, mirrors
+    (e.g. Екатеринбург region_code=66 — intentionally excluded from the table,
-    estimator._fetch_dkp_corridor's own COALESCE fallback) or any DB error →
+    mirrors estimator._fetch_dkp_corridor's own COALESCE fallback) or any DB
-    the globals; read-only best-effort, never raises.
+    error → the globals; read-only best-effort, never raises.
+
+    ``region_code`` defaults to 66 (Свердловская обл.) — this script has no
+    region CLI flag yet (out of scope, #3051 sub-PR B); every existing
+    --city caller keeps its byte-identical lookup.
     """
     if city is None:
         return float(PPM2_MIN), float(PPM2_MAX)
     try:
-        row = db.execute(_CITY_PPM2_BAND_SQL, {"city": city}).mappings().first()
+        row = (
+            db.execute(_CITY_PPM2_BAND_SQL, {"city": city, "region_code": region_code})
+            .mappings()
+            .first()
+        )
     except Exception as exc:  # pragma: no cover — defensive, read-only best-effort
         logger.warning("city PPM2 band lookup failed for %r (fallback to global): %s", city, exc)
         return float(PPM2_MIN), float(PPM2_MAX)