feat(tradein/estimator): сегментная поправка headline+выкуп по ценовому бэнду за флагом (#2255) (#2293)
All checks were successful
Deploy Trade-In / changes (push) Successful in 9s
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 1m29s
Deploy Trade-In / deploy (push) Successful in 44s
Deploy Trade-In / build-backend (push) Successful in 56s

This commit is contained in:
bot-backend 2026-07-03 18:43:38 +00:00
parent b22edc6d6f
commit d15ff3d99d
4 changed files with 713 additions and 30 deletions

View file

@ -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

View file

@ -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 askingppm²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(

View file

@ -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"])

View file

@ -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
)