feat(tradein/estimator): сегментная поправка headline+выкуп по ценовому бэнду за флагом (#2255)

_apply_segment_multiplier: множитель по бэнду PRICE_SEGMENTS_PPM2 (единый
источник в estimator, backtest импортирует лениво через PEP 562 —
бэндинг доказуемо идентичен скореру) к median_price/ppm2/range И к
expected_sold (выкуп) той же пропорцией — бэктест скорит expected_sold,
поправка обязана доехать до оффера. Вставка после IMV-blend/
corridor-clamp/quarter-index, перед range_floor; honest_ratio инвариантен
(числитель и знаменатель ×один factor), «выкуп ≤ headline» сохраняется.

Калибровка OFF/ON (детерминированный sample 300, туннель): только
бизнес ×1.08 / элит ×1.06 (V2). v4 (1.12/1.10) отвергнут: band-by-predicted
утечка эконом-sold→predicted-бизнес даёт +3.08pp overall MAPE; V2 =
лучший трейд-офф (+1.87pp, бизнес bias −10.4→−3.2, элит −30→−24.6,
эконом intact). Эконом/комфорт намеренно 1.00. Дальнейшее сжатие bias —
только с confidence-гейтом (follow-up).

Флаг estimate_segment_multiplier_enabled default OFF — путь байт-в-байт,
hermetic regression gate зелёный БЕЗ регенерации baseline. Битый/пустой
multipliers-dict → no-op с warning. + режим --calibrate-segments в
backtest (print-only: bias→raw→shrinkage λ→m).

Refs #2255

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

@@ -370,6 +370,37 @@ class Settings(BaseSettings):
     estimate_quarter_index_factor_min: float = 0.6
     estimate_quarter_index_factor_max: float = 1.8
 
+    # ── Сегментная поправка эстиматора по ценовому бэнду (#2255) ──────────────
+    # Эстиматор систематически занижает верхние сегменты (live-бэктест n=561,
+    # 2026-07-03): эконом +3.2%, комфорт −4.5%, бизнес −16.3% (n=76),
+    # элит −25.7% (n=16). Множитель применяется к median_price/median_ppm2 и
+    # пропорционально к range_low/range_high СРАЗУ ПЕРЕД min-width floor, после
+    # IMV-blend / corridor-clamp / quarter-index / hedonic / PI. Бэнд — по
+    # median_ppm2 границами PRICE_SEGMENTS_PPM2 (единый источник в estimator.py).
+    # Флаг default OFF → путь байт-в-байт идентичен (frozen gate не двигается).
+    estimate_segment_multiplier_enabled: bool = False
+    # Множители: калибровка live-бэктест OFF/ON sample=300 2026-07-03 (#2255).
+    # Поправка применяется ТОЛЬКО к бизнес/элит — где занижение крупное и
+    # однонаправленное. эконом/комфорт = 1.00 (no-op).
+    #
+    # Бэнд считается по PREDICTED ppm² (при оценке истинный сегмент неизвестен),
+    # а точность меряется по SOLD ppm² → band-mismatch: эконом-sold сделки,
+    # PREDICTED в бизнес (и так завышаемые +7.7%), получают ×множитель и
+    # завышаются дальше. Поэтому агрессивный v4 (бизнес 1.12 / элит 1.10)
+    # ОТВЕРГНУТ: overall MAPE +3.08pp, эконом MAPE 17.9→20.2.
+    #
+    # V2 (бизнес 1.08 / элит 1.06) — лучший трейд-офф: overall MAPE +1.87pp,
+    # бизнес bias −10.4→−3.2, элит −30→−24.6, эконом почти intact (MAPE 18.0).
+    # Дальнейшее сжатие bias без роста MAPE — только с confidence-гейтом
+    # (не применять множитель на low-confidence предсказаниях) → follow-up issue.
+    # Пересчёт биасов — см. scripts/backtest_estimator.py --calibrate-segments.
+    estimate_segment_multipliers: dict[str, float] = {
+        "эконом": 1.00,
+        "комфорт": 1.00,
+        "бизнес": 1.08,
+        "элит": 1.06,
+    }
+
     # ── Estimate enrichment time-budgets (#654) ──────────────────────────────
     # POST /estimate делает несколько ПОСЛЕДОВАТЕЛЬНЫХ блокирующих сетевых
     # вызовов (geocode → Overpass → Yandex valuation → IMV → Cian). Yandex

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

@@ -105,6 +105,19 @@ _RATIO_DESCRIPTOR_EPS = 1e-4
 # быть у́же типичной ошибки: floor только РАСШИРЯЕТ, точку не двигает.
 RANGE_MIN_HALFWIDTH_PCT = 0.12
 
+# #2255: границы ценовых сегментов по ₽/м² (ЕКБ-вторичка) — ЕДИНЫЙ источник для
+# эстиматора (_apply_segment_multiplier) и бэктеста (scripts/backtest_estimator.py
+# импортирует ЭТУ константу, числа НЕ дублирует). Значение лежит в первом бэнде,
+# чей upper_bound (exclusive) оно НЕ превышает; последний бэнд ловит хвост (+inf).
+# Верифицировано против config-заметок: p99.9 сделок ≈ 500k, премиум ~680k.
+PRICE_SEGMENTS_PPM2: tuple[tuple[str, float], ...] = (
+    ("эконом", 120_000.0),
+    ("комфорт", 160_000.0),
+    ("бизнес", 220_000.0),
+    ("элит", 300_000.0),
+    ("премиум", float("inf")),
+)
+
 # #699: санитизация ДКП-выбросов (Росреестр `deals`). В сырых сделках встречаются
 # нерыночные/битые записи — доли, сделки с обременением, опечатки этажа/площади —
 # которые шумят actual_deals (display) и dkp_corridor/expected_sold. Абсолютные
@@ -1041,6 +1054,87 @@ def _apply_quarter_index(
     return adjusted_ppm2, adjusted_median_price, adjusted_range_low, adjusted_range_high, factor
 
 
+def _segment_for_ppm2(ppm2: float) -> str:
+    """Ценовой сегмент (label) для значения ₽/м² по границам PRICE_SEGMENTS_PPM2.
+
+    Значение попадает в первый бэнд, чей upper_bound (exclusive) оно НЕ превышает.
+    Последний бэнд (+inf) ловит хвост. Чистая функция — общий бэндинг для
+    эстиматора и бэктеста.
+    """
+    for label, upper in PRICE_SEGMENTS_PPM2:
+        if ppm2 < upper:
+            return label
+    return PRICE_SEGMENTS_PPM2[-1][0]  # +inf-хвост — недостижимо, defensive
+
+
+def _apply_segment_multiplier(
+    *,
+    median_price: int,
+    median_ppm2: float,
+    range_low: int,
+    range_high: int,
+    enabled: bool,
+    multipliers: dict[str, float],
+) -> tuple[float, int, int, int, str | None, float]:
+    """#2255: сегментная поправка эстиматора по ценовому бэнду (₽/м²).
+
+    Чистая (testable без БД) функция по образцу _apply_quarter_index. Бэнд
+    определяется по median_ppm2 границами PRICE_SEGMENTS_PPM2; множитель для
+    этого бэнда берётся из multipliers. Умножается median_price, median_ppm2 и
+    ПРОПОРЦИОНАЛЬНО range_low/range_high (тот же factor → asking↔ppm²↔range
+    остаются геометрически консистентны, как в quarter-index/corridor-clamp).
+
+    No-op (factor=1.0, band=None) когда:
+      - enabled=False (флаг OFF → путь байт-в-байт идентичен),
+      - multipliers пуст/битый (нет ключа бэнда, значение не приводится к float,
+        либо ≤ 0) → warning + no-op (не роняем оценку из-за кривого конфига),
+      - median_ppm2 <= 0 (нет headline),
+      - множитель бэнда == 1.0 (премиум и любой явно-нейтральный сегмент).
+
+    Returns (median_ppm2, median_price, range_low, range_high, band, factor).
+    band=None при no-op; иначе label сегмента, к которому применён множитель.
+    """
+    if not enabled or median_ppm2 <= 0:
+        return median_ppm2, median_price, range_low, range_high, None, 1.0
+    if not multipliers:
+        logger.warning("segment_mult: пустой multipliers-dict (флаг ON) — no-op")
+        return median_ppm2, median_price, range_low, range_high, None, 1.0
+
+    band = _segment_for_ppm2(median_ppm2)
+    raw = multipliers.get(band)
+    if raw is None:
+        # Бэнд без записи (напр. премиум намеренно отсутствует) — no-op без шума.
+        return median_ppm2, median_price, range_low, range_high, None, 1.0
+    try:
+        factor = float(raw)
+    except (TypeError, ValueError):
+        logger.warning("segment_mult: битый множитель band=%s raw=%r — no-op", band, raw)
+        return median_ppm2, median_price, range_low, range_high, None, 1.0
+    if factor <= 0:
+        logger.warning("segment_mult: неположительный множитель band=%s ×%r — no-op", band, factor)
+        return median_ppm2, median_price, range_low, range_high, None, 1.0
+    if factor == 1.0:
+        # Нейтральный сегмент (эконом=1.00 и т.п.) — no-op без мутации/лога.
+        return median_ppm2, median_price, range_low, range_high, None, 1.0
+
+    new_price = round(median_price * factor)
+    logger.info(
+        "segment_mult: band=%s ×%.3f median %d→%d",
+        band,
+        factor,
+        median_price,
+        new_price,
+    )
+    return (
+        median_ppm2 * factor,
+        new_price,
+        round(range_low * factor),
+        round(range_high * factor),
+        band,
+        factor,
+    )
+
+
 def _load_sber_index_series(db: Session, *, region: str) -> dict[date, float]:
     """#794: monthly {period_month: index_value} for region from sber_price_index.
 
@@ -2615,11 +2709,62 @@ def _price_from_inputs(
             "по широкой окрестности)."
         )
 
+    # ── #2255: сегментная поправка по ценовому бэнду — СРАЗУ ПЕРЕД min-width
+    # floor, ПОСЛЕ IMV-blend / corridor-clamp / quarter-index / hedonic / PI
+    # (все ценовые мутации точки уже отработали → бэнд считается по финальному
+    # median_ppm2). Множитель к median_price/median_ppm2 и пропорционально к
+    # range_low/range_high. Флаг OFF ⇒ no-op (путь байт-в-байт идентичен).
+    #
+    # Тот же factor применяется ДВУМЯ точками — к asking-headline (здесь) И к
+    # expected_sold (выкуп) ниже. Причина: expected_sold дерайвится выше по
+    # цепочке (median×ratio + hedonic/le_asking/PI, ~стр. 2595-2651), т.е. ДО
+    # этой поправки → он уже заморожен и не «подхватит» умноженный median сам.
+    # Acceptance #2255 требует сжатия per-segment bias именно в бэктесте, а
+    # бэктест скорит expected_sold — значит поправка ОБЯЗАНА доехать до выкупа,
+    # иначе оффер клиенту в бизнес/элит остаётся заниженным. Один общий factor
+    # держит asking↔выкуп геометрически консистентными: honest_ratio
+    # (expected/median) инвариантен к общему множителю (бейдж «−N%» не врёт), и
+    # инвариант «выкуп ≤ headline» сохраняется (обе стороны ×один factor).
+    if settings.estimate_segment_multiplier_enabled:
+        (
+            median_ppm2,
+            median_price,
+            range_low,
+            range_high,
+            _seg_band,
+            _seg_factor,
+        ) = _apply_segment_multiplier(
+            median_price=median_price,
+            median_ppm2=median_ppm2,
+            range_low=range_low,
+            range_high=range_high,
+            enabled=settings.estimate_segment_multiplier_enabled,
+            multipliers=settings.estimate_segment_multipliers,
+        )
+        if _seg_band is not None:
+            sources_used_pre = sorted(set(sources_used_pre) | {"segment_multiplier"})
+            # Догоняем той же пропорцией уже-дерайвнутый expected_sold (выкуп).
+            if expected_sold_per_m2 is not None:
+                expected_sold_per_m2 = round(expected_sold_per_m2 * _seg_factor)
+            if expected_sold_price is not None:
+                expected_sold_price = round(expected_sold_price * _seg_factor)
+            if expected_sold_range_low is not None:
+                expected_sold_range_low = round(expected_sold_range_low * _seg_factor)
+            if expected_sold_range_high is not None:
+                expected_sold_range_high = round(expected_sold_range_high * _seg_factor)
+            logger.info(
+                "segment_mult: expected_sold band=%s ×%.3f price→%s per_m2→%s",
+                _seg_band,
+                _seg_factor,
+                expected_sold_price,
+                expected_sold_per_m2,
+            )
+
     # ── #2209: min-width floor — ПОСЛЕДНИМ, после всех мутаций (anchor / IMV blend
-    # / quarter-index / corridor-clamp / radius-floor / hedonic PI), чтобы никакая
-    # последующая мутация не сузила диапазон обратно. Floor только расширяет и не
-    # трогает точку (median_price / expected_sold_price). Вырожденный n=1 asking-
-    # диапазон (Q1==Q3) перестаёт быть точкой с ложной точностью.
+    # / quarter-index / corridor-clamp / radius-floor / hedonic PI / segment-mult),
+    # чтобы никакая последующая мутация не сузила диапазон обратно. Floor только
+    # расширяет и не трогает точку (median_price / expected_sold_price). Вырожденный
+    # n=1 asking-диапазон (Q1==Q3) перестаёт быть точкой с ложной точностью.
     range_low, range_high = _apply_range_floor(range_low, range_high, median_price)
     if expected_sold_range_low is not None and expected_sold_range_high is not None:
         expected_sold_range_low, expected_sold_range_high = _apply_range_floor(

tradein-mvp/backend/scripts/backtest_estimator.py

@@ -202,20 +202,34 @@ MIN_BUCKET = 20
 # range-coverage + calibration breakdowns. Any unexpected value → "other".
 CONFIDENCE_BUCKETS: tuple[str, ...] = ("high", "medium", "low")
 
-# Price-segment bands by ₽/m² (EKB вторичка). The estimator has NO reusable
-# price-tier constant — `listing_segment` is categorical (vtorichka/novostroyki)
-# and DEAL_MIN/MAX_PPM2 are sanity bounds, not class bands — so these are
-# defined here. TUNABLE: rough EKB market tiers (эконом < комфорт < бизнес <
-# элит < премиум). Each entry is (label, upper_bound_exclusive); a value lands
-# in the first band whose upper bound it is below; the last band catches the
-# tail (+inf). Verified against config notes: p99.9 deals ≈ 500k, премиум ~680k.
-PRICE_SEGMENTS_PPM2: tuple[tuple[str, float], ...] = (
-    ("эконом", 120_000.0),
-    ("комфорт", 160_000.0),
-    ("бизнес", 220_000.0),
-    ("элит", 300_000.0),
-    ("премиум", float("inf")),
-)
+# Price-segment bands by ₽/m² (EKB вторичка). SINGLE SOURCE OF TRUTH now lives in
+# ``estimator.PRICE_SEGMENTS_PPM2`` (#2255: the estimator's segment multiplier and
+# this backtest must band identically) — we no longer duplicate the numbers here.
+# Exposed lazily via module ``__getattr__``/``_price_segments()`` so importing the
+# estimator (→ app.core.config.Settings, which fail-fasts without DATABASE_URL) is
+# deferred out of `--help` / the pure-metric unit tests, mirroring _import_estimator.
+
+
+def _price_segments() -> tuple[tuple[str, float], ...]:
+    """Lazy accessor for the shared PRICE_SEGMENTS_PPM2 band table (from estimator).
+
+    Deferred import (same reason as _import_estimator): pulling the estimator
+    module eagerly would import Settings, which fail-fasts when DATABASE_URL is
+    unset. Every caller here runs at RUNTIME, never at import time.
+    """
+    ns = _import_estimator_full()
+    return ns.m.PRICE_SEGMENTS_PPM2  # type: ignore[no-any-return]
+
+
+def __getattr__(name: str) -> Any:
+    """PEP 562: expose ``PRICE_SEGMENTS_PPM2`` as a module attribute lazily.
+
+    Keeps ``backtest_estimator.PRICE_SEGMENTS_PPM2`` working (tests read it) while
+    avoiding an eager estimator/Settings import at module load.
+    """
+    if name == "PRICE_SEGMENTS_PPM2":
+        return _price_segments()
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 
 # --------------------------------------------------------------------------- #
@@ -497,11 +511,14 @@ def _bucketize_confidence(confidence: str) -> str:
 
 
 def _segment_label(ppm2: float) -> str:
-    """Price-segment label for a ₽/m² value (see PRICE_SEGMENTS_PPM2 bands)."""
-    for label, upper in PRICE_SEGMENTS_PPM2:
-        if ppm2 < upper:
-            return label
-    return PRICE_SEGMENTS_PPM2[-1][0]  # +inf tail — unreachable, defensive
+    """Price-segment label for a ₽/m² value — delegates to the estimator's helper.
+
+    Single source of truth: estimator._segment_for_ppm2 uses the shared
+    PRICE_SEGMENTS_PPM2 band table (#2255), so backtest bucketing and the
+    estimator's segment multiplier band identically.
+    """
+    ns = _import_estimator_full()
+    return ns.m._segment_for_ppm2(ppm2)  # type: ignore[no-any-return]
 
 
 def _segment_metrics(rows: list[tuple[float, float]]) -> dict[str, dict[str, Any]]:
@@ -511,15 +528,18 @@ def _segment_metrics(rows: list[tuple[float, float]]) -> dict[str, dict[str, Any
     SOLD price (ground truth), compute signed_error_pct = 100*(pred-sold)/sold,
     and run `_errors_summary` per band. Rows with sold<=0 are dropped (can't
     divide). Every band in PRICE_SEGMENTS_PPM2 is present (n=0 when empty) so the
-    report renders a stable table. Pure: no DB.
+    report renders a stable table. No DB (band table via one lazy estimator import).
     """
-    by_seg: dict[str, list[float]] = {label: [] for label, _ in PRICE_SEGMENTS_PPM2}
+    ns = _import_estimator_full()
+    segments = ns.m.PRICE_SEGMENTS_PPM2
+    seg_for = ns.m._segment_for_ppm2
+    by_seg: dict[str, list[float]] = {label: [] for label, _ in segments}
     for pred, sold in rows:
         if sold <= 0:
             continue
-        by_seg[_segment_label(sold)].append(100.0 * (pred - sold) / sold)
+        by_seg[seg_for(sold)].append(100.0 * (pred - sold) / sold)
     out: dict[str, dict[str, Any]] = {}
-    for label, _ in PRICE_SEGMENTS_PPM2:
+    for label, _ in segments:
         out[label] = _errors_summary(by_seg[label])
     return out
 
@@ -875,8 +895,9 @@ def _render_segment_block(per_segment: dict[str, Any]) -> list[str]:
         header,
         "  " + "-" * (len(header) - 2),
     ]
-    for label, _ in PRICE_SEGMENTS_PPM2:
-        m = per_segment[label]
+    # per_segment is built in band order by _segment_metrics → dict insertion
+    # order preserves it (no need to re-import the constant in a pure renderer).
+    for label, m in per_segment.items():
         out.append(
             f"  {label:<10} {m.get('n', 0):>5} "
             f"{_fmt_pct(m.get('median_bias_pct')):>8} {_fmt_pct(m.get('mape_pct')):>8} "
@@ -885,6 +906,52 @@ def _render_segment_block(per_segment: dict[str, Any]) -> list[str]:
     return out
 
 
+# --------------------------------------------------------------------------- #
+# #2255 --calibrate-segments — per-segment multiplier proposal (PRINT-ONLY)
+# --------------------------------------------------------------------------- #
+
+# Shrinkage denominator for λ = n/(n+SHRINK_N): pulls thin-sample multipliers back
+# toward 1.0 (no correction) so a 16-deal segment isn't over-trusted. Larger → more
+# conservative. 40 chosen so элит (n≈16) lands ~0.29 weight, бизнес (n≈76) ~0.66.
+_SEGMENT_SHRINK_N = 40
+
+
+def _render_calibrate_segments_block(per_segment: dict[str, Any]) -> list[str]:
+    """Propose per-segment multipliers from expected_sold bias (#2255). PRINT-ONLY.
+
+    For each band: raw = 1/(1+bias) undoes the median signed bias; λ = n/(n+40)
+    shrinks it toward 1.0 by sample size; suggested m = 1 + λ·(raw − 1). Segments
+    with no bias/n render "—". This proposes numbers for
+    ``estimate_segment_multipliers``; it applies nothing.
+    """
+    header = f"  {'segment':<10} {'n':>5} {'bias%':>8} {'raw m':>8} {'λ':>6} {'suggest m':>10}"
+    out: list[str] = [
+        "[#2255 CALIBRATE] segment multiplier proposal (PRINT-ONLY, applies nothing):",
+        f"  m undoes expected_sold bias, shrunk toward 1.0 by λ=n/(n+{_SEGMENT_SHRINK_N}).",
+        header,
+        "  " + "-" * (len(header) - 2),
+    ]
+    for label, m in per_segment.items():
+        n = int(m.get("n", 0) or 0)
+        bias = m.get("median_bias_pct")
+        if bias is None or n == 0:
+            out.append(f"  {label:<10} {n:>5} {'—':>8} {'—':>8} {'—':>6} {'—':>10}")
+            continue
+        bias_frac = float(bias) / 100.0
+        # raw multiplier that would drive median bias to 0 (guard div-by-~0).
+        raw = 1.0 / (1.0 + bias_frac) if abs(1.0 + bias_frac) > 1e-9 else 1.0
+        lam = n / (n + _SEGMENT_SHRINK_N)
+        suggested = 1.0 + lam * (raw - 1.0)
+        out.append(
+            f"  {label:<10} {n:>5} {_fmt_pct(bias):>8} {raw:>8.3f} {lam:>6.2f} {suggested:>10.3f}"
+        )
+    out.append("")
+    out.append(
+        "  NB: элит capped in config (thin n) — do NOT paste raw suggestions for n<20 verbatim."
+    )
+    return out
+
+
 def _render_coverage_block(range_coverage: dict[str, Any], conf_order: list[str]) -> list[str]:
     """Render range-coverage: overall + per-confidence (sold_total ∈ range)."""
     ov = range_coverage["overall"]
@@ -1854,7 +1921,7 @@ def run_backtest_full(
         "since": since,
         "n_matched": len(predictions),
         "n_no_prediction": n_no_prediction,
-        "price_segments_ppm2": [list(seg) for seg in PRICE_SEGMENTS_PPM2],
+        "price_segments_ppm2": [list(seg) for seg in _price_segments()],
     }
 
     # #2002: house_id resolution coverage — the key Tier-S + IMV reach number.
@@ -1981,6 +2048,15 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         "the JSON output. Default OFF → byte-identical to the prior behaviour, so "
         "the frozen regression gate is untouched.",
     )
+    p.add_argument(
+        "--calibrate-segments",
+        action="store_true",
+        help="FULL engine only (#2255): after the run, print a per-price-segment "
+        "calibration table — segment → n / bias%% / raw multiplier 1/(1+bias) / "
+        "shrinkage λ=n/(n+40) / SUGGESTED m (shrunk toward 1.0). PRINT-ONLY: it "
+        "proposes estimate_segment_multipliers, it does NOT apply them or touch "
+        "the baseline. Run with --resolve-house-id for prod-parity biases.",
+    )
     # #1966 PR 3/3 — fixture capture + hermetic replay. --dump-fixture (DB run,
     # full engine) and --from-fixture (NO DB) are mutually exclusive modes.
     fixture_mode = p.add_mutually_exclusive_group()
@@ -2034,6 +2110,8 @@ def main(argv: list[str] | None = None) -> int:
         raise SystemExit("--dump-fixture is only supported with --engine full")
     if args.resolve_house_id and args.engine != "full":
         raise SystemExit("--resolve-house-id is only supported with --engine full")
+    if args.calibrate_segments and args.engine != "full":
+        raise SystemExit("--calibrate-segments is only supported with --engine full")
 
     logger.info(
         "backtest start: engine=%s sample=%d since=%s radius=%dm "
@@ -2077,6 +2155,13 @@ def main(argv: list[str] | None = None) -> int:
     else:
         print(_render_table(metrics, metrics["headline"]))
 
+    # #2255 print-only: after the run, propose per-segment multipliers from the
+    # expected_sold per-segment bias. Renders even in --json mode (to stderr-free
+    # stdout tail) so the operator sees the proposal alongside machine output.
+    if args.calibrate_segments:
+        per_segment = metrics["expected_sold"]["per_segment"]
+        print("\n" + "\n".join(_render_calibrate_segments_block(per_segment)))
+
     return int(metrics["params"]["n_matched"])
 
 

tradein-mvp/backend/tests/test_estimator_segment_multiplier.py

@@ -0,0 +1,422 @@
+"""Unit tests for #2255 — segment multiplier by price band (default OFF).
+
+Two layers:
+  1. The pure ``_apply_segment_multiplier`` helper (no DB): flag-off no-op, per-band
+     proportional scaling of point + range, out-of-band / премиум → 1.0, and the
+     empty / broken multipliers-dict no-op (with a warning).
+  2. Integration through ``_price_from_inputs`` with the flag toggled ON via
+     monkeypatch — proves the multiplier fires BEFORE ``_apply_range_floor`` (the
+     floor widens the ALREADY-multiplied range, never the other way round) and that
+     flag-OFF leaves the priced result byte-identical.
+
+NOTE: importing app.services.estimator pulls app.core.config.Settings which requires
+DATABASE_URL. Set it BEFORE importing app modules (mirrors test_estimator_range_floor).
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import pytest
+
+from app.services import estimator
+from app.services.estimator import (
+    RANGE_MIN_HALFWIDTH_PCT,
+    _apply_segment_multiplier,
+    _segment_for_ppm2,
+)
+from app.services.geocoder import GeocodeResult
+
+# Prod multipliers as calibrated in config (#2255, V2). комфорт=1.00 deliberately
+# (predicted-band leakage into эконом — see config comment + test_flag_on_comfort_
+# band_is_noop_end_to_end); бизнес/элит softened to 1.08/1.06 (v4's 1.12/1.10 gave
+# +3.08pp overall MAPE, V2 gives +1.87pp). Kept local so integration tests reflect prod.
+_BIZ = 1.08  # prod бизнес multiplier — one symbol so a future retune touches one place
+_MULTS = {"эконом": 1.00, "комфорт": 1.00, "бизнес": _BIZ, "элит": 1.06}
+
+# Synthetic multipliers for the PURE-helper math: proves _apply_segment_multiplier
+# scales ANY band by its factor (independent of the prod config's комфорт=1.00 choice).
+_MULTS_ALL_ACTIVE = {"эконом": 1.05, "комфорт": 1.03, "бизнес": 1.12, "элит": 1.10}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 1a. _segment_for_ppm2 — band boundaries (shared source of truth)
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_segment_for_ppm2_bands_and_boundaries() -> None:
+    """Boundaries are upper-exclusive; +inf tail catches the top (premium)."""
+    assert _segment_for_ppm2(100_000) == "эконом"
+    assert _segment_for_ppm2(119_999) == "эконом"
+    assert _segment_for_ppm2(120_000) == "комфорт"
+    assert _segment_for_ppm2(159_999) == "комфорт"
+    assert _segment_for_ppm2(160_000) == "бизнес"
+    assert _segment_for_ppm2(219_999) == "бизнес"
+    assert _segment_for_ppm2(220_000) == "элит"
+    assert _segment_for_ppm2(299_999) == "элит"
+    assert _segment_for_ppm2(300_000) == "премиум"
+    assert _segment_for_ppm2(2_000_000) == "премиум"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 1b. _apply_segment_multiplier — pure helper
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_disabled_is_noop() -> None:
+    """enabled=False → values returned untouched, band=None, factor=1.0."""
+    out = _apply_segment_multiplier(
+        median_price=10_000_000,
+        median_ppm2=200_000.0,  # бизнес band
+        range_low=9_000_000,
+        range_high=11_000_000,
+        enabled=False,
+        multipliers=_MULTS,
+    )
+    assert out == (200_000.0, 10_000_000, 9_000_000, 11_000_000, None, 1.0)
+
+
+@pytest.mark.parametrize(
+    ("ppm2", "band", "mult"),
+    [
+        (200_000.0, "бизнес", 1.12),
+        (250_000.0, "элит", 1.10),
+        (140_000.0, "комфорт", 1.03),
+        (100_000.0, "эконом", 1.05),
+    ],
+)
+def test_each_band_scales_point_and_range_proportionally(
+    ppm2: float, band: str, mult: float
+) -> None:
+    """The pure helper multiplies point + range by the SAME factor for ANY band.
+
+    Uses synthetic all-active multipliers (комфорт≠1.0, эконом≠1.0) to prove the
+    helper math independent of the prod config's комфорт=1.00 leakage choice.
+    """
+    price, low, high = 10_000_000, 9_000_000, 11_000_000
+    new_ppm2, new_price, new_low, new_high, out_band, factor = _apply_segment_multiplier(
+        median_price=price,
+        median_ppm2=ppm2,
+        range_low=low,
+        range_high=high,
+        enabled=True,
+        multipliers=_MULTS_ALL_ACTIVE,
+    )
+    assert out_band == band
+    assert factor == mult
+    assert new_ppm2 == pytest.approx(ppm2 * mult)
+    assert new_price == round(price * mult)
+    assert new_low == round(low * mult)
+    assert new_high == round(high * mult)
+    # Geometric consistency: the range scales by exactly the same factor as the point.
+    assert new_low / low == pytest.approx(new_high / high)
+    assert new_price / price == pytest.approx(new_ppm2 / ppm2)
+
+
+def test_econom_neutral_multiplier_is_noop() -> None:
+    """эконом=1.00 → no mutation, band=None (nothing to attribute)."""
+    out = _apply_segment_multiplier(
+        median_price=5_000_000,
+        median_ppm2=100_000.0,  # эконом → 1.00
+        range_low=4_500_000,
+        range_high=5_500_000,
+        enabled=True,
+        multipliers=_MULTS,
+    )
+    assert out == (100_000.0, 5_000_000, 4_500_000, 5_500_000, None, 1.0)
+
+
+def test_premium_band_has_no_multiplier_and_is_noop() -> None:
+    """премиум is intentionally absent from the dict → no-op (missing key)."""
+    out = _apply_segment_multiplier(
+        median_price=40_000_000,
+        median_ppm2=400_000.0,  # премиум band, no entry
+        range_low=36_000_000,
+        range_high=44_000_000,
+        enabled=True,
+        multipliers=_MULTS,
+    )
+    assert out == (400_000.0, 40_000_000, 36_000_000, 44_000_000, None, 1.0)
+
+
+def test_zero_ppm2_is_noop() -> None:
+    """median_ppm2<=0 (no headline) → no-op regardless of flag."""
+    out = _apply_segment_multiplier(
+        median_price=0,
+        median_ppm2=0.0,
+        range_low=0,
+        range_high=0,
+        enabled=True,
+        multipliers=_MULTS,
+    )
+    assert out == (0.0, 0, 0, 0, None, 1.0)
+
+
+def test_empty_multipliers_dict_warns_and_noops(caplog: pytest.LogCaptureFixture) -> None:
+    """Empty multipliers with flag ON → no-op + a warning (bad config, don't crash)."""
+    with caplog.at_level(logging.WARNING, logger="app.services.estimator"):
+        out = _apply_segment_multiplier(
+            median_price=10_000_000,
+            median_ppm2=200_000.0,
+            range_low=9_000_000,
+            range_high=11_000_000,
+            enabled=True,
+            multipliers={},
+        )
+    assert out == (200_000.0, 10_000_000, 9_000_000, 11_000_000, None, 1.0)
+    assert any("segment_mult" in r.message and "пустой" in r.message for r in caplog.records)
+
+
+def test_broken_multiplier_value_warns_and_noops(caplog: pytest.LogCaptureFixture) -> None:
+    """Non-numeric band value → no-op + warning (don't blow up a real estimate)."""
+    with caplog.at_level(logging.WARNING, logger="app.services.estimator"):
+        out = _apply_segment_multiplier(
+            median_price=10_000_000,
+            median_ppm2=200_000.0,  # бизнес
+            range_low=9_000_000,
+            range_high=11_000_000,
+            enabled=True,
+            multipliers={"бизнес": "oops"},  # type: ignore[dict-item]
+        )
+    assert out == (200_000.0, 10_000_000, 9_000_000, 11_000_000, None, 1.0)
+    assert any("битый множитель" in r.message for r in caplog.records)
+
+
+def test_nonpositive_multiplier_warns_and_noops(caplog: pytest.LogCaptureFixture) -> None:
+    """A ≤0 multiplier is nonsensical → no-op + warning (never zero out a price)."""
+    with caplog.at_level(logging.WARNING, logger="app.services.estimator"):
+        out = _apply_segment_multiplier(
+            median_price=10_000_000,
+            median_ppm2=200_000.0,
+            range_low=9_000_000,
+            range_high=11_000_000,
+            enabled=True,
+            multipliers={"бизнес": 0.0},
+        )
+    assert out == (200_000.0, 10_000_000, 9_000_000, 11_000_000, None, 1.0)
+    assert any("неположительный множитель" in r.message for r in caplog.records)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 2. Integration through _price_from_inputs — order vs _apply_range_floor + flag OFF
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def _geo() -> GeocodeResult:
+    return GeocodeResult(
+        lat=56.838,
+        lon=60.597,
+        full_address="ул. Тестовая, 1",
+        provider="nominatim",
+        confidence="approximate",
+    )
+
+
+def _lots(ppm2: float, n: int = 7) -> list[dict]:
+    return [
+        {"price_per_m2": ppm2, "address": f"ул. Тестовая, {i + 1}", "source": "avito"}
+        for i in range(n)
+    ]
+
+
+def _call(
+    *, listings: list[dict], area_m2: float = 50.0, ratio: float | None = None
+) -> estimator.PricingResult:
+    _ratio = ratio
+    _basis = "per_rooms" if ratio is not None else None
+
+    def ratio_resolver(appm2: float | None) -> tuple[float | None, str | None]:
+        return _ratio, _basis
+
+    return estimator._price_from_inputs(
+        listings=listings,
+        area_m2=area_m2,
+        rooms=2,
+        repair_state=None,
+        floor=5,
+        total_floors=10,
+        target_year=None,
+        analog_tier="W",
+        fallback_used=False,
+        area_widened=False,
+        anchor_comps=[],
+        anchor_tier_fetched=None,
+        dkp_raw=None,
+        imv_anchor=None,
+        imv_eval=None,
+        yandex_val_present=False,
+        cian_val_present=False,
+        ratio_resolver=ratio_resolver,
+        quarter_index_lookup=lambda q: None,
+        quarter_indexes_lookup=lambda qs: {},
+        target_house_cadnum=None,
+        dadata_coarse=False,
+        geo=_geo(),
+        dadata_qc_geo=None,
+    )
+
+
+def test_flag_off_priced_result_is_unchanged() -> None:
+    """Default (flag OFF): a бизнес-band estimate is NOT multiplied — byte-identical."""
+    # 7 uniform lots at 200k ₽/m² (бизнес band), wide enough spread avoided → point stays.
+    pr = _call(listings=_lots(200_000.0, n=7))
+    assert pr.median_ppm2 == 200_000.0
+    assert pr.median_price == 200_000 * 50  # 10_000_000, untouched
+    assert "segment_multiplier" not in (pr.sources_used_pre or [])
+
+
+def test_flag_on_business_band_multiplies_point_and_range(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Flag ON: бизнес-band point + range scaled ×_BIZ; range still brackets point.
+
+    Also proves the multiplier ran BEFORE _apply_range_floor: the floor may only
+    WIDEN, so range_high/point must be at least the multiplied point ± floor. A
+    naive "floor then multiply" order would instead leave the point unmultiplied.
+    """
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", True)
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multipliers", _MULTS)
+
+    pr = _call(listings=_lots(200_000.0, n=7))
+    # Point multiplied: 200k → 200k×_BIZ ₽/m²; total 10M → 10M×_BIZ.
+    assert pr.median_ppm2 == pytest.approx(200_000.0 * _BIZ)
+    assert pr.median_price == round(10_000_000 * _BIZ)
+    assert "segment_multiplier" in pr.sources_used_pre
+    # Range brackets the MULTIPLIED point (floor only widens around the lifted point).
+    assert pr.range_low <= pr.median_price <= pr.range_high
+    # ppm² point stays consistent with the multiplied total.
+    assert pr.median_ppm2 == pytest.approx(pr.median_price / 50.0)
+
+
+def test_flag_on_order_multiplier_before_range_floor(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """n=1 degenerate range: multiplier moves the point FIRST, floor then widens it.
+
+    A single analog collapses Q1==Q3==median → zero-width asking range. With the
+    flag ON the бизнес point is multiplied to 10M×_BIZ, THEN the ±12 % floor widens
+    the (still zero-width) range symmetrically around that lifted point. If the floor
+    ran first, it would bracket the pre-multiply 10M point and the edges would not be
+    point±12 % of the multiplied point — this asserts they are.
+    """
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", True)
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multipliers", _MULTS)
+
+    pr = _call(listings=_lots(200_000.0, n=1))
+    assert pr.n_analogs == 1
+    point = pr.median_price
+    assert point == round(10_000_000 * _BIZ)  # multiplied point
+    half = round(RANGE_MIN_HALFWIDTH_PCT * point)
+    # Floor widened AROUND the multiplied point (proves multiplier-before-floor).
+    assert pr.range_low == point - half
+    assert pr.range_high == point + half
+
+
+def test_flag_on_econom_band_is_noop_end_to_end(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Flag ON but эконом band (mult 1.00): priced result stays unmultiplied."""
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", True)
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multipliers", _MULTS)
+
+    pr = _call(listings=_lots(100_000.0, n=7))  # эконом
+    assert pr.median_ppm2 == 100_000.0
+    assert pr.median_price == 100_000 * 50
+    assert "segment_multiplier" not in (pr.sources_used_pre or [])
+
+
+def test_flag_on_comfort_band_is_noop_end_to_end(monkeypatch: pytest.MonkeyPatch) -> None:
+    """DELIBERATE: комфорт=1.00 in prod config → no-op even with the flag ON.
+
+    комфорт was dropped to 1.00 (from a раннего 1.03) because the band is keyed on
+    PREDICTED ppm² while accuracy is measured by SOLD ppm²: a ×1.03 on комфорт-
+    predicted deals leaks onto эконом-sold deals (already over-predicted), inflating
+    overall MAPE for negligible комфорт benefit (live OFF/ON sample=300, 2026-07-03).
+    This test guards that decision — a future retune to комфорт≠1.0 must be conscious.
+    """
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multipliers", _MULTS)
+
+    # Baseline OFF vs flag ON on the SAME комфорт deal — both must be identical.
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", False)
+    off = _call(listings=_lots(140_000.0, n=7), ratio=0.90)  # комфорт band
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", True)
+    on = _call(listings=_lots(140_000.0, n=7), ratio=0.90)
+
+    assert estimator.settings.estimate_segment_multipliers["комфорт"] == 1.00
+    assert on.median_ppm2 == 140_000.0  # headline untouched
+    assert on.median_price == 140_000 * 50
+    # Both headline and выкуп identical OFF vs ON (комфорт=1.00 → true no-op).
+    assert on.median_price == off.median_price
+    assert on.expected_sold_price == off.expected_sold_price
+    assert "segment_multiplier" not in (on.sources_used_pre or [])
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 3. expected_sold (выкуп) must ALSO be multiplied — the acceptance-critical path
+#    (backtest scores expected_sold; #2255 requires per-segment bias to shrink).
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+def test_flag_off_expected_sold_unchanged() -> None:
+    """OFF: expected_sold is the plain ratio-derived выкуп (hedonic/PI, no multiplier).
+
+    The exact value is median×ratio×hedonic (≈0.889 of median here, not 0.90 —
+    hedonic's ln(area) term shifts it), but the point is: NO segment multiplier is
+    attributed and выкуп ≤ headline.
+    """
+    pr = _call(listings=_lots(200_000.0, n=7), ratio=0.90)
+    assert pr.expected_sold_price is not None
+    assert pr.expected_sold_price <= pr.median_price  # выкуп ≤ headline
+    assert pr.expected_sold_price < pr.median_price  # ratio<1 → strictly below
+    assert "segment_multiplier" not in (pr.sources_used_pre or [])
+
+
+def test_flag_on_expected_sold_multiplied_by_same_factor(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """ON: expected_sold (point + per_m2 + range) scaled by the SAME бизнес factor.
+
+    This is the fix for #2255: the multiplier reaches the выкуп (client offer),
+    not only the asking headline. expected_sold is derived earlier in the pipeline
+    (median×ratio + hedonic/PI) and frozen before the multiplier, so the call site
+    re-applies the identical factor to it explicitly.
+    """
+    # Baseline with the flag OFF (multipliers set but flag disabled → no-op).
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", False)
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multipliers", _MULTS)
+    off = _call(listings=_lots(200_000.0, n=7), ratio=0.90)
+
+    # Now flip the flag ON for the same deal.
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", True)
+    on = _call(listings=_lots(200_000.0, n=7), ratio=0.90)
+
+    assert "segment_multiplier" not in (off.sources_used_pre or [])
+    assert "segment_multiplier" in on.sources_used_pre
+    # выкуп point ×_BIZ vs the flag-OFF derivation of the SAME deal.
+    assert off.expected_sold_price is not None and on.expected_sold_price is not None
+    assert on.expected_sold_price == round(off.expected_sold_price * _BIZ)
+    assert on.expected_sold_per_m2 == round(off.expected_sold_per_m2 * _BIZ)
+    # Range scaled too (floor may only widen; ×_BIZ keeps it ≥ scaled edges).
+    assert on.expected_sold_range_low >= round(off.expected_sold_range_low * _BIZ) - 1
+    assert on.expected_sold_range_high >= round(off.expected_sold_range_high * _BIZ) - 1
+
+
+def test_flag_on_expected_sold_le_headline_invariant_preserved(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """ON: выкуп ≤ headline holds after the multiplier (both ×same factor)."""
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multipliers", _MULTS)
+
+    # ratio 0.90 < 1 → expected_sold < median before AND after the ×_BIZ lift.
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", False)
+    off = _call(listings=_lots(200_000.0, n=7), ratio=0.90)
+    monkeypatch.setattr(estimator.settings, "estimate_segment_multiplier_enabled", True)
+    on = _call(listings=_lots(200_000.0, n=7), ratio=0.90)
+    assert on.expected_sold_price is not None
+    # выкуп ≤ headline holds after the lift (both ×_BIZ → monotonic).
+    assert on.expected_sold_price <= on.median_price
+    # honest ratio (expected/median) is invariant to the common factor: same as OFF.
+    assert on.expected_sold_price / on.median_price == pytest.approx(
+        off.expected_sold_price / off.median_price, abs=1e-3
+    )