feat(tradein/backtest): --city per-city валидация + corridor city-parity

D validation harness для oblast-точности:
- --city '<точное deals.city>' → scope выборки по deals.city; default None = байт-идентично (тот же _SAMPLE_SQL object, literal identity).
- per-city PPM2 band из deal_city_price_bands (fallback global, 0 доп-запросов при None).
- corridor city-parity: backtest _fetch_dkp_corridor теперь передаёт city=_resolve_target_city(deal.address) — зеркалит прод C2 (был cross-city contaminated). NB: меняет и default-run live-числа (не fixture-replay) — intended catch-up к проду.
- _compute_confidence подтверждён comp-count-based (без EKB-хардкода).
- +10 unit-тестов. Регресс-гейт байт-зелёный (replay не гоняет corridor).

code-review ✅ APPROVE. Frozen fixture/baseline не тронуты. Live per-city backtest прогоню после деплоя.

tradein-mvp/backend/scripts/backtest_estimator.py

@@ -86,6 +86,9 @@ USAGE
 
     # machine-readable:
     python -m scripts.backtest_estimator --json
+
+    # oblast D: per-city validation (exact deals.city name, not a slug):
+    python -m scripts.backtest_estimator --city "Нижний Тагил" --sample 300
 """
 
 from __future__ import annotations
@@ -181,6 +184,14 @@ logger = logging.getLogger("backtest_estimator")
 # Price-per-m² sanity band — shared by the deal sample and the listings
 # subquery. Mirrors the estimator's working range for EKB вторичка and drops
 # obvious data-entry garbage / commercial outliers.
+# (oblast D) When --city is set, _load_sample sources a PER-CITY band from
+# deal_city_price_bands instead (see _resolve_city_ppm2_band) — these globals
+# remain the fallback (default/unscoped sample, and any city with no band row,
+# e.g. Екатеринбург — intentionally excluded from that table).
+# NB: the listings subquery (_CANDIDATES_SQL, asking-core engine only) still
+# always uses these globals — left global deliberately (not city-scoped) to
+# keep the change minimal; the asking-core engine is the legacy comparison
+# path, not the primary oblast-D target (the full engine, default).
 PPM2_MIN = 30_000
 PPM2_MAX = 600_000
 
@@ -1043,22 +1054,115 @@ _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
+# --city is set (see _resolve_city_ppm2_band); the default (city=None) path
+# never issues this query.
+_CITY_PPM2_BAND_SQL = text(
+    """
+    SELECT ppm2_min, ppm2_max
+    FROM deal_city_price_bands
+    WHERE city = CAST(:city AS text)
+    """
+)
 
-def _load_sample(db: Session, *, sample: int, since: str) -> list[DealSample]:
-    """Run the held-out ДКП deal sampling SELECT → list[DealSample]."""
-    rows = (
-        db.execute(
-            _SAMPLE_SQL,
-            {
-                "ppm2_min": PPM2_MIN,
-                "ppm2_max": PPM2_MAX,
-                "since": since,
-                "sample": sample,
-            },
-        )
-        .mappings()
-        .all()
+
+def _sample_sql(city: str | None) -> Any:
+    """ДКП deal-sample SELECT — optionally scoped to one ``deals.city`` (oblast D).
+
+    ``city is None`` (default) returns the SAME ``_SAMPLE_SQL`` object used
+    before this change — literal identity, not just equal text — so the
+    default CLI invocation (and the frozen EKB regression gate, which never
+    calls this path at all) see byte-identical SQL.
+
+    When ``city`` is set, an extra ``AND city = CAST(:city AS text)`` predicate
+    scopes the sample to that ``deals.city`` value — populated oblast-wide by
+    the #C1 migration (368 cities, 96,974 deals). Pass the EXACT canonical
+    Russian name as stored in ``deals.city`` (e.g. ``'Нижний Тагил'``), not a
+    slug — see the ``--city`` CLI help for the naming decision.
+    """
+    if city is None:
+        return _SAMPLE_SQL
+    return text(
+        """
+        SELECT
+            id,
+            ST_X(geom::geometry) AS lon,
+            ST_Y(geom::geometry) AS lat,
+            rooms,
+            price_per_m2 AS sold_ppm2,
+            deal_date,
+            area_m2,
+            address,
+            floor,
+            total_floors,
+            year_built,
+            house_type
+        FROM deals
+        WHERE source = 'rosreestr'
+          AND geom IS NOT NULL
+          AND price_per_m2 BETWEEN CAST(:ppm2_min AS numeric) AND CAST(:ppm2_max AS numeric)
+          AND rooms IS NOT NULL
+          AND area_m2 IS NOT NULL
+          AND area_m2 > 0
+          AND deal_date >= CAST(:since AS date)
+          AND city = CAST(:city AS text)
+        ORDER BY id DESC
+        LIMIT CAST(:sample AS integer)
+        """
     )
+
+
+def _resolve_city_ppm2_band(db: Session, city: str | None) -> tuple[float, float]:
+    """Per-city PPM2 sanity band (oblast D) — falls back to the 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
+    materially different sane ₽/m² range than the EKB-tuned 30k..600k globals
+    (e.g. Нижний Тагил: 16 955..108 175). No row for the city (e.g.
+    Екатеринбург — intentionally excluded from the table, mirrors
+    estimator._fetch_dkp_corridor's own COALESCE fallback) or any DB error →
+    the globals; read-only best-effort, never raises.
+    """
+    if city is None:
+        return float(PPM2_MIN), float(PPM2_MAX)
+    try:
+        row = db.execute(_CITY_PPM2_BAND_SQL, {"city": city}).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)
+    if row is None or row["ppm2_min"] is None or row["ppm2_max"] is None:
+        return float(PPM2_MIN), float(PPM2_MAX)
+    return float(row["ppm2_min"]), float(row["ppm2_max"])
+
+
+def _load_sample(
+    db: Session, *, sample: int, since: str, city: str | None = None
+) -> list[DealSample]:
+    """Run the held-out ДКП deal sampling SELECT → list[DealSample].
+
+    ``city`` (oblast D, default None) scopes the sample to one ``deals.city``
+    value via ``_sample_sql`` and sources the PPM2 sanity band from
+    ``deal_city_price_bands`` for that city (``_resolve_city_ppm2_band``,
+    falls back to the module globals). Default None is byte-identical to the
+    pre-oblast-D behaviour: same SQL object, same PPM2_MIN/PPM2_MAX globals,
+    no extra query.
+    """
+    if city is None:
+        ppm2_min: float = PPM2_MIN
+        ppm2_max: float = PPM2_MAX
+    else:
+        ppm2_min, ppm2_max = _resolve_city_ppm2_band(db, city)
+    params: dict[str, Any] = {
+        "ppm2_min": ppm2_min,
+        "ppm2_max": ppm2_max,
+        "since": since,
+        "sample": sample,
+    }
+    if city is not None:
+        params["city"] = city
+    rows = db.execute(_sample_sql(city), params).mappings().all()
     out: list[DealSample] = []
     for r in rows:
         if r["lon"] is None or r["lat"] is None or r["sold_ppm2"] is None:
@@ -1516,7 +1620,18 @@ def _predict_full_spine(
     )
 
     # ── Pre-fetch the spine inputs (same calls estimate_quality hoists) ───────
-    dkp_raw = m._fetch_dkp_corridor(db, address=deal.address, rooms=deal.rooms, area=deal.area_m2)
+    # (oblast C2 parity): estimate_quality now resolves a target city and passes
+    # it to _fetch_dkp_corridor (estimator.py:3210-3220) — deals.city is
+    # populated oblast-wide (368 cities), and an unscoped corridor lets
+    # same-named streets in OTHER region-66 towns (e.g. "Ленина" exists in many)
+    # contaminate the corridor. Mirror that resolve+pass here UNCONDITIONALLY
+    # (independent of --city, which only scopes WHICH deals get sampled) so this
+    # harness measures the SAME corridor prod actually computes today, not the
+    # pre-C2 unscoped behaviour — else the backtest validates stale semantics.
+    target_city = m._resolve_target_city(deal.address)
+    dkp_raw = m._fetch_dkp_corridor(
+        db, address=deal.address, rooms=deal.rooms, area=deal.area_m2, city=target_city
+    )
     # #1966 prod parity: same-building anchor pre-fetch is GATED exactly like
     # estimate_quality — no-area / no-address → ([], None) instead of an
     # unconditional fetch.
@@ -1733,6 +1848,7 @@ def run_backtest(
     radius: int,
     rooms_tolerance: int,
     holdout_split: bool = False,
+    city: str | None = None,
 ) -> dict[str, Any]:
     """Drive the full read-only backtest and return a metrics dict.
 
@@ -1747,9 +1863,13 @@ def run_backtest(
     so its bias is near-zero by construction — it proves the MECHANISM, not
     out-of-sample accuracy (see _derive_room_ratios). Pass ``holdout_split=True``
     to fit on even-id deals and evaluate on the odd-id half for an honest number.
+
+    ``city`` (oblast D, default None) scopes the deal sample to one
+    ``deals.city`` value — see ``_load_sample``. Default None is unscoped
+    (byte-identical to the pre-oblast-D behaviour).
     """
-    deals = _load_sample(db, sample=sample, since=since)
-    logger.info("loaded sample: %d ДКП deals (since=%s)", len(deals), since)
+    deals = _load_sample(db, sample=sample, since=since, city=city)
+    logger.info("loaded sample: %d ДКП deals (since=%s, city=%s)", len(deals), since, city)
 
     matched_rows: list[tuple[float, float, int]] = []
     matched_ids: list[int] = []
@@ -1809,6 +1929,7 @@ def run_backtest(
         "n_matched": len(matched_rows),
         "n_no_analogs": n_no_analogs,
         "holdout_split": holdout_split,
+        "city": city,
     }
     return metrics
 
@@ -1820,6 +1941,7 @@ def run_backtest_full(
     since: str,
     dump_fixture: str | None = None,
     resolve_house_id: bool = False,
+    city: str | None = None,
 ) -> dict[str, Any]:
     """Drive the FULL-spine read-only backtest and return a metrics dict (#1966).
 
@@ -1844,10 +1966,18 @@ def run_backtest_full(
     the Tier-S analog ladder + the house Avito-IMV anchor, mirroring prod, and a
     ``house_id_resolution`` coverage block (resolved / total / imv_reachable) is
     attached to the returned metrics. Default False → byte-identical prior output.
+
+    ``city`` (oblast D, default None) scopes the deal sample to one
+    ``deals.city`` value — see ``_load_sample``. Default None is unscoped
+    (byte-identical to the pre-oblast-D behaviour). Independently of this flag,
+    ``_predict_full_spine`` ALWAYS resolves + passes a per-deal target city to
+    ``_fetch_dkp_corridor`` (oblast C2 parity fix) — see its docstring.
     """
     est = _import_estimator_full()
-    deals = _load_sample(db, sample=sample, since=since)
-    logger.info("loaded sample: %d ДКП deals (since=%s) [full spine]", len(deals), since)
+    deals = _load_sample(db, sample=sample, since=since, city=city)
+    logger.info(
+        "loaded sample: %d ДКП deals (since=%s, city=%s) [full spine]", len(deals), since, city
+    )
 
     predictions: list[Prediction] = []
     n_no_prediction = 0
@@ -1922,6 +2052,7 @@ def run_backtest_full(
         "n_matched": len(predictions),
         "n_no_prediction": n_no_prediction,
         "price_segments_ppm2": [list(seg) for seg in _price_segments()],
+        "city": city,
     }
 
     # #2002: house_id resolution coverage — the key Tier-S + IMV reach number.
@@ -2011,6 +2142,20 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         default="2025-06-01",
         help="Only deals with deal_date >= this ISO date (default 2025-06-01).",
     )
+    p.add_argument(
+        "--city",
+        default=None,
+        metavar="NAME",
+        help="Oblast D: scope the held-out ДКП deal sample to ONE deals.city value "
+        "(region 66 is oblast-wide — deals.city covers 368 cities, 96,974 deals). "
+        "Pass the EXACT canonical Russian name as stored in deals.city, e.g. "
+        "'Нижний Тагил' or 'Каменск-Уральский' — NOT a slug/transliteration (run "
+        "`SELECT DISTINCT city FROM deals` to check the exact spelling). Also "
+        "sources the deal-sample PPM2 sanity band from deal_city_price_bands for "
+        "that city (falls back to the global PPM2_MIN/PPM2_MAX when the city has "
+        "no band row, e.g. Екатеринбург — intentionally excluded from that table). "
+        "Default None → unscoped sample, byte-identical to the pre-oblast-D SQL.",
+    )
     p.add_argument(
         "--radius",
         type=int,
@@ -2115,7 +2260,7 @@ def main(argv: list[str] | None = None) -> int:
 
     logger.info(
         "backtest start: engine=%s sample=%d since=%s radius=%dm "
-        "rooms_tolerance=%d holdout_split=%s dump_fixture=%s resolve_house_id=%s",
+        "rooms_tolerance=%d holdout_split=%s dump_fixture=%s resolve_house_id=%s city=%s",
         args.engine,
         args.sample,
         args.since,
@@ -2124,6 +2269,7 @@ def main(argv: list[str] | None = None) -> int:
         args.holdout_split,
         args.dump_fixture,
         args.resolve_house_id,
+        args.city,
     )
 
     db = _session()
@@ -2135,6 +2281,7 @@ def main(argv: list[str] | None = None) -> int:
                 since=args.since,
                 dump_fixture=args.dump_fixture,
                 resolve_house_id=args.resolve_house_id,
+                city=args.city,
             )
         else:
             metrics = run_backtest(
@@ -2144,6 +2291,7 @@ def main(argv: list[str] | None = None) -> int:
                 radius=args.radius,
                 rooms_tolerance=args.rooms_tolerance,
                 holdout_split=args.holdout_split,
+                city=args.city,
             )
     finally:
         db.close()

tradein-mvp/backend/tests/test_backtest_estimator.py

@@ -832,3 +832,177 @@ def test_render_full_table_handles_empty_sample() -> None:
     out = bt._render_full_table(m)
     assert "n/a" in out  # None metrics render as n/a, no crash
     assert "BACKTEST" in out
+
+
+# --------------------------------------------------------------------------- #
+# Oblast D — `--city` deal-sample scoping (parse + SQL predicate + PPM2 band).
+# --------------------------------------------------------------------------- #
+
+
+def test_argparse_city_defaults_none() -> None:
+    assert bt._parse_args([]).city is None
+
+
+def test_argparse_city_override() -> None:
+    ns = bt._parse_args(["--city", "Нижний Тагил"])
+    assert ns.city == "Нижний Тагил"
+
+
+def test_sample_sql_default_is_unscoped_and_same_object() -> None:
+    # city=None must return the SAME _SAMPLE_SQL object used before oblast D —
+    # literal identity, not just equal text — so the default CLI invocation (and
+    # the frozen EKB regression gate, which never calls this path at all) see a
+    # byte-identical query.
+    assert bt._sample_sql(None) is bt._SAMPLE_SQL
+
+
+def test_sample_sql_default_has_no_city_predicate() -> None:
+    assert ":city" not in bt._SAMPLE_SQL.text
+
+
+def test_sample_sql_city_scoped_adds_city_predicate() -> None:
+    sql = bt._sample_sql("Нижний Тагил")
+    assert sql is not bt._SAMPLE_SQL
+    built = sql.text
+    assert "AND city = CAST(:city AS text)" in built
+    # City-scoping is additive — the base predicates are still present.
+    assert "source = 'rosreestr'" in built
+    assert "deal_date >= CAST(:since AS date)" in built
+    assert "ORDER BY id DESC" in built
+
+
+# --------------------------------------------------------------------------- #
+# Oblast D — _resolve_city_ppm2_band (per-city PPM2 sanity band, no live DB).
+# --------------------------------------------------------------------------- #
+
+
+class _FakeBandResult:
+    """Minimal stand-in for a SQLAlchemy Result exposing .mappings().first()."""
+
+    def __init__(self, row: dict[str, object] | None) -> None:
+        self._row = row
+
+    def mappings(self) -> "_FakeBandResult":
+        return self
+
+    def first(self) -> dict[str, object] | None:
+        return self._row
+
+
+class _FakeBandSession:
+    """Minimal stand-in for a Session — only .execute() is exercised here."""
+
+    def __init__(self, row: dict[str, object] | None = None, *, raise_exc: bool = False) -> None:
+        self._row = row
+        self._raise = raise_exc
+
+    def execute(self, *_args: object, **_kwargs: object) -> _FakeBandResult:
+        if self._raise:
+            raise RuntimeError("boom")
+        return _FakeBandResult(self._row)
+
+
+def test_resolve_city_ppm2_band_none_city_returns_globals_no_query() -> None:
+    # city=None must not even touch the DB (no _FakeBandSession.execute call).
+    class _NoExecuteSession:
+        def execute(self, *_a: object, **_kw: object) -> None:
+            raise AssertionError("must not query DB when city is None")
+
+    assert bt._resolve_city_ppm2_band(_NoExecuteSession(), None) == (
+        float(bt.PPM2_MIN),
+        float(bt.PPM2_MAX),
+    )
+
+
+def test_resolve_city_ppm2_band_found_row() -> None:
+    db = _FakeBandSession({"ppm2_min": 16955, "ppm2_max": 108175})
+    assert bt._resolve_city_ppm2_band(db, "Нижний Тагил") == (16955.0, 108175.0)
+
+
+def test_resolve_city_ppm2_band_no_row_falls_back_to_globals() -> None:
+    db = _FakeBandSession(None)
+    assert bt._resolve_city_ppm2_band(db, "Екатеринбург") == (
+        float(bt.PPM2_MIN),
+        float(bt.PPM2_MAX),
+    )
+
+
+def test_resolve_city_ppm2_band_db_error_falls_back_to_globals() -> None:
+    db = _FakeBandSession(raise_exc=True)
+    assert bt._resolve_city_ppm2_band(db, "Нижний Тагил") == (
+        float(bt.PPM2_MIN),
+        float(bt.PPM2_MAX),
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Oblast C2 parity — _predict_full_spine must resolve + pass a target city to
+# _fetch_dkp_corridor, mirroring estimate_quality (estimator.py:3210-3220).
+# --------------------------------------------------------------------------- #
+
+
+def test_predict_full_spine_passes_resolved_city_to_corridor(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from types import SimpleNamespace
+
+    from app.services import estimator as est_mod
+
+    captured: dict[str, object] = {}
+
+    def _fake_resolve_target_city(address: str | None) -> str | None:
+        return "нижний тагил"
+
+    def _fake_fetch_dkp_corridor(
+        _db: object,
+        *,
+        address: object,
+        rooms: object,
+        area: object,
+        city: object = None,
+        **_kw: object,
+    ) -> None:
+        captured["called"] = True
+        captured["city"] = city
+        return None
+
+    def _fake_price_from_inputs(**_kwargs: object) -> SimpleNamespace:
+        return SimpleNamespace(
+            median_price=100_000.0,
+            median_ppm2=100_000.0,
+            confidence="low",
+            anchor_tier=None,
+            expected_sold_per_m2=95_000.0,
+            expected_sold_price=4_750_000.0,
+            expected_sold_range_low=4_000_000.0,
+            expected_sold_range_high=5_500_000.0,
+        )
+
+    monkeypatch.setattr(est_mod, "_resolve_target_city", _fake_resolve_target_city)
+    monkeypatch.setattr(est_mod, "_fetch_dkp_corridor", _fake_fetch_dkp_corridor)
+    monkeypatch.setattr(est_mod, "_fetch_anchor_comps", lambda *a, **kw: ([], None))
+    monkeypatch.setattr(est_mod, "_fetch_house_imv_anchor", lambda *a, **kw: None)
+    monkeypatch.setattr(est_mod, "_price_from_inputs", _fake_price_from_inputs)
+    monkeypatch.setattr(bt, "_select_analogs_full", lambda *a, **kw: ([], "W", False, False))
+
+    est = bt._import_estimator_full()
+    deal = bt.DealSample(
+        id=1,
+        lon=60.6,
+        lat=58.05,
+        rooms=2,
+        sold_ppm2=90_000.0,
+        deal_date=None,
+        area_m2=50.0,
+        address="Нижний Тагил, ул. Ленина, 5",
+        floor=3,
+        total_floors=9,
+        year_built=2000,
+        house_type="панель",
+    )
+
+    pred = bt._predict_full_spine(None, deal, est)
+
+    assert captured.get("called") is True
+    assert captured.get("city") == "нижний тагил"
+    assert pred is not None