fix(tradein/estimate): не строить цену по одному-двум аналогам, не отдавать 0 ₽
All checks were successful
CI / changes (pull_request) Successful in 7s
CI Trade-In / changes (pull_request) Successful in 8s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 2m38s

Live-аудит на проде (2026-08-02): headline-медиана строилась даже по 1-4
листингам без проверки достаточности выборки (Серов 2к/45м2, n=3 листинга ->
42 391 руб/м2 vs честный ДКП-коридор 54 126, -36%; соседняя улица того же
города на других 1-2 лотах давала разброс до 1.3x). Каменск-Уральский на
редких room-count давал буквальный median_price_rub=0 без явного отказа.

Fix:
- HEADLINE_LISTINGS_MIN_N=5 (estimator.py) - ниже порога радиусная медиана
  подавляется (listings_clean НЕ очищается - нужен ghost-anchor guard'у #1871)
  и маршрутизируется на уже существующий deals-headline-fallback (#oblast-D,
  сделки Росреестра) или честный insufficient_data, если сделок тоже мало.
- deals-headline-fallback гейт исправлен с anchor_tier is None на anchor is
  None - anchor_tier оставался stale (не сбрасывался в None), когда
  _compute_same_building_anchor отклонял кандидата целиком, блокируя fallback
  даже при большом ДКП-покрытии (найдено live regen'ом бэктест-фикстуры).
- Честные explanation-тексты: отличают "объявлений действительно нет" от
  "нашли N, но недостаточно для доверия" - не показывают противоречивые
  сообщения одновременно.
- insufficient_data (computed field, median_price_rub<=0) остаётся
  единственным источником истины - PDF/API уже не показывают 0 как число.

Regenerated tests/fixtures/backtest_baseline.json: 8 из 277 прод-сделок
теперь используют deals-fallback вместо шумной 2-4-листинговой медианы
(mape_pct 13.23%->13.18%, headline spread_pct 17.98%->16.65%).
This commit is contained in:
bot-backend 2026-08-02 15:16:08 +03:00
parent 4b3d8b4cca
commit 31f82fc5e9
16 changed files with 702 additions and 126 deletions

View file

@ -101,6 +101,24 @@ DEALS_PERIOD_MONTHS = 12 # сделки за последний год
DKP_CORRIDOR_CITY_WIDE_MIN_N = 3
DEALS_HEADLINE_FALLBACK_MIN_N = 3
# #oblast-E (money-path sufficiency gate, live-audit 2026-08-02): минимум
# LIVE-объявлений, из которых можно честно построить headline-медиану по
# рынку. Ниже порога median() по 1-4 случайным лотам не отражает рынок —
# live repro на проде: Серов 2к/45м², n=3 листинга → 42 391 ₽/м² (36% vs
# городской ДКП-коридор 54 126 ₽/м², которых сама эта улица не показывала
# из-за тонкой street-выборки), а соседняя улица того же города с другими
# 1-2 случайными лотами даёт разброс до ×1.66. Ниже порога
# _price_from_inputs ОБНУЛЯЕТ радиусную популяцию (listings_clean=[]) —
# 100%-переиспользует уже существующий и покрытый тестами путь «листингов
# нет» (same-building anchor / #oblast-D deals-headline-fallback /
# insufficient_data ниже), а не изобретает новую ветку. Значение 5 выбрано
# по live-данным (n=3 уже недостаточно; n=5 — тот же порог, что
# MIN_ANALOGS_TIER_0 использует для "хватает на строгий когортный тир" —
# согласованная планка "достаточно, чтобы не быть шумом одного-двух лотов").
# НЕ трогает подбор аналогов/тиры/радиусы — только решение, доверять ли
# ИТОГОВОЙ выборке как headline-источнику.
HEADLINE_LISTINGS_MIN_N = 5
# #794: СберИндекс time-adjustment of frozen Rosreestr ДКП deals.
# Rosreestr deals freeze ~2026-01; the sber monthly index re-bases a stale deal's ppm²
# to the latest available month. Region fixed to Свердловская обл. (tradein MVP = ЕКБ).
@ -2402,6 +2420,12 @@ class PricingResult:
# headline. Anchor-путь → CV комплов (anchor["cv"]); radius-путь → CV
# радиусной ₽/м²-выборки. None если <2 цен (недостаточно данных).
cv: float | None = None
# #oblast-E: >0 когда n листингов было найдено но ниже HEADLINE_LISTINGS_MIN_N
# (headline suppressed, listings_clean deliberately left intact — see gate
# comment above). Caller uses this to also keep the thin listings out of the
# display `analogs` cards when no anchor overrides the headline. 0 = either
# sufficient listings were used, or genuinely zero were found.
listings_headline_thin_n: int = 0
def _price_from_inputs(
@ -2480,10 +2504,45 @@ def _price_from_inputs(
n_analogs = 0
cv = None
# 4b. Repair coefficient
# 4a. #oblast-E sufficiency gate (see HEADLINE_LISTINGS_MIN_N docstring above).
# 1..HEADLINE_LISTINGS_MIN_N-1 listings are a real find but too thin to trust
# as a market median — suppress the AGGREGATE (median/range/n_analogs/cv)
# exactly like "no usable listings", so the anchor/#oblast-D-deals-fallback/
# insufficient_data chain below all take the already-honest zero-analogs
# path automatically (no new branches there). `listings_clean` itself is
# deliberately LEFT INTACT (not cleared) — the same-building anchor's own
# ghost-anchor guard (#1871, `if not listings_clean`) uses it to tell
# "genuinely zero nearby listings" from "some nearby listings, just too few
# to trust as THIS estimate's headline" — those are different confidence
# signals and clearing the list here would conflate them. The caller
# (estimate_quality) uses `listings_headline_thin_n` on the returned
# PricingResult to also keep suppressed listings out of the display
# `analogs` cards when no anchor overrides the headline (n_analogs
# invariant: cards shown ⊆ what n_analogs counts).
listings_headline_thin_n = 0
if 0 < n_analogs < HEADLINE_LISTINGS_MIN_N:
listings_headline_thin_n = n_analogs
logger.info(
"headline sufficiency gate #oblast-E: n=%d < %d listings — suppressing "
"listings-derived median (falling back to anchor/deals/insufficient_data)",
n_analogs,
HEADLINE_LISTINGS_MIN_N,
)
median_ppm2 = 0.0
q1_ppm2 = 0.0
q3_ppm2 = 0.0
median_price = 0
range_low = 0
range_high = 0
n_analogs = 0
cv = None
# 4b. Repair coefficient — skipped when the headline was thin-suppressed
# above (median_price is already 0; applying a coefficient would leave it
# 0 but still emit a misleading "adjusted for repair state" note).
repair_coef = _repair_coefficient(repair_state)
repair_note = ""
if listings_clean and repair_coef != 1.0:
if listings_clean and not listings_headline_thin_n and repair_coef != 1.0:
median_price = int(median_price * repair_coef)
range_low = int(range_low * repair_coef)
range_high = int(range_high * repair_coef)
@ -2524,6 +2583,20 @@ def _price_from_inputs(
area_widened,
listings=listings_clean,
)
# #oblast-E: honest override — _compute_confidence's generic "не найдено
# аналогов" is FALSE here (we DID find listings_headline_thin_n of them,
# just too few to trust). Stays the final explanation unless a later block
# (anchor / #oblast-D deals-fallback) overwrites it with its OWN honest
# reasoning — both of those already check truthy `explanation` and either
# replace it (anchor) or append a construction-method clause that reads
# this same thin-count (deals-fallback), so no contradiction either way.
if listings_headline_thin_n:
confidence = "low"
explanation = (
f"Рядом найдено недостаточно объявлений ({listings_headline_thin_n} шт., "
f"минимум для оценки по рынку — {HEADLINE_LISTINGS_MIN_N}) — медиана по "
"такой маленькой выборке слишком чувствительна к случайным лотам."
)
# Tier note — информируем пользователя о качестве house-match
tier_note = ""
@ -3076,9 +3149,20 @@ def _price_from_inputs(
# generic ghost-anchor). No repair_state adjustment: the deal corridor
# mixes conditions across sold units — unlike the listings comp pool,
# there is no per-unit signal to correct against.
#
# #oblast-E: guards on `anchor is None` (the actual computed anchor dict),
# NOT `anchor_tier is None`. Found via live backtest-fixture regen: when
# `_compute_same_building_anchor` rejects a candidate outright (e.g. its
# own MAD-clip drops comps below estimate_sb_min_comps), it returns None
# WITHOUT the caller resetting `anchor_tier` back to None (it stays
# whatever `anchor_tier_fetched` was, e.g. "C") — the anchor never fired,
# but the stale tier flag falsely reads as "anchor claimed the headline"
# and blocked this fallback even with a large, valid ДКП corridor
# available (observed: 677 deals for one fixture case). `anchor is None`
# is the ground truth of whether the anchor actually produced a headline.
if (
median_ppm2 <= 0
and anchor_tier is None
and anchor is None
and dkp_raw is not None
and dkp_raw.get("count", 0) >= DEALS_HEADLINE_FALLBACK_MIN_N
and dkp_raw.get("median_ppm2", 0) > 0
@ -3090,16 +3174,27 @@ def _price_from_inputs(
n_analogs = 0
confidence = "low"
cv = None
# #oblast-E: differentiate "genuinely zero listings" (unchanged wording)
# from "found some but below HEADLINE_LISTINGS_MIN_N, suppressed above" —
# the latter must NOT claim "рядом нет объявлений" (false, contradicts the
# thin-sufficiency explanation already set above this block).
no_listings_clause = (
f" Из {listings_headline_thin_n} найденных объявлений недостаточно для "
"надёжной медианы —"
if listings_headline_thin_n
else " Рядом нет актуальных объявлений —"
)
explanation = (explanation or "") + (
" Рядом нет актуальных объявлений — оценка построена по реальным "
f"{no_listings_clause} оценка построена по реальным "
f"сделкам Росреестра ({dkp_raw['count']} шт. за {dkp_raw['period_months']} мес.),"
" точность ориентировочная."
)
logger.info(
"deals_headline_fallback #oblast-D: dkp median=%d (n=%d) → headline"
" (listings=0, anchor=None)",
" (listings=0 [thin_suppressed=%d], anchor=None)",
int(median_ppm2),
dkp_raw["count"],
listings_headline_thin_n,
)
# ── #652: ДКП-коридор реальных сделок (advisory) ─────────────────────────
@ -3231,6 +3326,7 @@ def _price_from_inputs(
sources_used_pre=sources_used_pre,
listings_clean=listings_clean,
cv=cv,
listings_headline_thin_n=listings_headline_thin_n,
)
@ -3706,6 +3802,7 @@ async def estimate_quality(
ratio_basis = pr.ratio_basis
listings_clean = pr.listings_clean
cv = pr.cv
listings_headline_thin_n = pr.listings_headline_thin_n
# 5. Deals — ДКП-only sales (вторичка) из rosreestr_deals.
# Importer фильтрует doc_type='ДКП' (PR-A 2026-05-24), ДДУ застройщиков
@ -3741,6 +3838,14 @@ async def estimate_quality(
# иначе «обновлено N мин назад»/дата парсинга/срок продажи относятся к другому
# набору (или = None при пустом listings_clean, хотя у комплов данные есть).
metadata_lots = display_pool
elif listings_headline_thin_n:
# #oblast-E: headline was suppressed (thin radius sample, no anchor to
# take over) — do NOT surface those same listings as display cards
# either, else `analogs` would show N cards while n_analogs==0 (broken
# invariant, same dishonesty this gate exists to remove). Degrades to
# the exact same empty-display state as "genuinely zero listings".
analogs_lots = []
metadata_lots = []
else:
# display-consistency fix: только ЦЕНОВЫЕ листинги — та же популяция, что
# дала n_analogs = len(prices_ppm2) в radius-ветке _price_from_inputs.

View file

@ -7,10 +7,10 @@
"n_covered": 0
},
"low": {
"coverage_pct": 81.88,
"mape_pct": 13.25,
"coverage_pct": 82.09,
"mape_pct": 13.2,
"n": 276,
"n_covered": 226
"n_covered": 220
},
"medium": {
"coverage_pct": 100.0,
@ -26,22 +26,22 @@
],
"expected_sold": {
"overall": {
"mape_pct": 13.23,
"median_bias_pct": -3.44,
"n": 277,
"mape_pct": 13.18,
"median_bias_pct": -3.71,
"n": 269,
"n_no_analogs": 0,
"p25_pct": -15.49,
"p75_pct": 10.07
"p25_pct": -16.11,
"p75_pct": 8.92
},
"per_rooms": {
"0": {
"label": "студия",
"mape_pct": 18.1,
"median_bias_pct": 16.96,
"n": 37,
"mape_pct": 19.38,
"median_bias_pct": 18.1,
"n": 35,
"n_no_analogs": 0,
"p25_pct": 1.37,
"p75_pct": 33.53
"p25_pct": 1.5,
"p75_pct": 34.32
},
"1": {
"label": "1к",
@ -72,12 +72,12 @@
},
"4": {
"label": "4+",
"mape_pct": 20.27,
"median_bias_pct": 8.54,
"n": 30,
"mape_pct": 16.3,
"median_bias_pct": 3.34,
"n": 24,
"n_no_analogs": 0,
"p25_pct": -6.86,
"p75_pct": 23.54
"p25_pct": -14.91,
"p75_pct": 15.11
}
},
"per_segment": {
@ -89,11 +89,11 @@
"p75_pct": -1.31
},
"комфорт": {
"mape_pct": 11.82,
"median_bias_pct": -4.35,
"n": 104,
"p25_pct": -16.35,
"p75_pct": 6.94
"mape_pct": 11.8,
"median_bias_pct": -4.61,
"n": 101,
"p25_pct": -17.07,
"p75_pct": 6.04
},
"премиум": {
"mape_pct": 68.92,
@ -103,11 +103,11 @@
"p75_pct": -68.92
},
"эконом": {
"mape_pct": 13.96,
"median_bias_pct": 3.39,
"n": 120,
"p25_pct": -9.57,
"p75_pct": 25.7
"mape_pct": 13.71,
"median_bias_pct": 2.54,
"n": 115,
"p25_pct": -9.8,
"p75_pct": 25.5
},
"элит": {
"mape_pct": 33.2,
@ -119,15 +119,15 @@
}
},
"headline": {
"ask_median_ppm2": 147545.8502510892,
"ask_median_ppm2": 145883.6593586467,
"deal_median_ppm2": 125063.0,
"spread_pct": 17.98
"spread_pct": 16.65
},
"range_coverage": {
"overall": {
"coverage_pct": 81.95,
"n": 277,
"n_covered": 227
"coverage_pct": 82.16,
"n": 269,
"n_covered": 221
},
"per_confidence": {
"high": {
@ -136,9 +136,9 @@
"n_covered": 0
},
"low": {
"coverage_pct": 81.88,
"n": 276,
"n_covered": 226
"coverage_pct": 82.09,
"n": 268,
"n_covered": 220
},
"medium": {
"coverage_pct": 100.0,
@ -149,6 +149,6 @@
},
"sharpness": {
"median_rel_width": 0.743,
"n": 277
"n": 269
}
}

View file

@ -54,10 +54,15 @@ def _make_listing_qa(*, price_per_m2: float, area_m2: float = 60.0) -> dict[str,
}
# #oblast-E: 5 items (>= HEADLINE_LISTINGS_MIN_N) so the new headline
# sufficiency gate doesn't suppress this fixture's median (still 210_000,
# symmetric) before the anchor-vs-radius mechanic under test runs.
_RADIUS_ANALOGS_QA: list[dict[str, Any]] = [
_make_listing_qa(price_per_m2=200_000.0),
_make_listing_qa(price_per_m2=195_000.0),
_make_listing_qa(price_per_m2=205_000.0),
_make_listing_qa(price_per_m2=210_000.0),
_make_listing_qa(price_per_m2=220_000.0),
_make_listing_qa(price_per_m2=215_000.0),
_make_listing_qa(price_per_m2=225_000.0),
]
# 2 comps -- below min_comps=4 threshold introduced by #755.
@ -160,10 +165,10 @@ def test_755_anchor_n2_does_not_fire_headline_stays_radius() -> None:
f"Expected radius median 210_000, got {est.median_price_per_m2} -- "
"anchor with n=2 comps must NOT fire (min_comps=4 post-#755)"
)
# Confidence from 3 radius analogs must not be "high" (n_analogs=3 < threshold).
# Confidence from 5 radius analogs must not be "high" (unique_addr < 7 threshold).
assert (
est.confidence != "high"
), f"Confidence should not be 'high' with 3 radius analogs, got {est.confidence!r}"
), f"Confidence should not be 'high' with 5 radius analogs, got {est.confidence!r}"
def test_755_anchor_n2_pure_unit_confidence_never_high() -> None:

View file

@ -7,8 +7,11 @@ the resulting metrics dict is asserted for structure + determinism.
The deals are crafted so every call ``_price_from_inputs`` makes to the 3 injected
callables is recorded up-front, and so each headline ``median_ppm2`` is an exact,
predictable value (3 listings the middle /; no anchor / quarter-index / ДКП
mutation), which is what the recorded ``ratio_calls`` key must match.
predictable value (5 listings, symmetric around the middle / #oblast-E: the
listing count must clear HEADLINE_LISTINGS_MIN_N or the new sufficiency gate
suppresses the median before the spine ever calls these callables; no anchor /
quarter-index / ДКП mutation), which is what the recorded ``ratio_calls`` key
must match.
NOTE: importing scripts.backtest_estimator app.services.estimator
app.core.config.Settings REQUIRES DATABASE_URL. Set a dummy value BEFORE importing
@ -105,16 +108,21 @@ def _build_fixture() -> dict[str, Any]:
deal 3: carries non-empty anchor_comps (2 comps < min_comps=4 anchor never
fires, so the median stays the radius median) + a ratio call.
"""
# ── deal 1 — median of [90k, 100k, 110k] = 100k → ratio_resolver(100000.0). ──
# ── deal 1 — median of [80k,90k,100k,110k,120k] = 100k → ratio_resolver(100000.0).
# #oblast-E: 5 listings (>= HEADLINE_LISTINGS_MIN_N), symmetric around the same
# 100k median as before — below that count the new headline sufficiency gate
# would suppress the median before the spine even calls ratio_resolver. ──
deal1 = _deal_record(
deal_id=1,
sold_ppm2=100_000.0, # SOLD эконом (< 120k)
area_m2=50.0,
rooms=1,
listings=[
{"price_per_m2": 80_000.0, "source": "avito"},
{"price_per_m2": 90_000.0, "source": "avito"},
{"price_per_m2": 100_000.0, "source": "avito"},
{"price_per_m2": 110_000.0, "source": "avito"},
{"price_per_m2": 120_000.0, "source": "avito"},
],
anchor_comps=[],
anchor_tier_fetched=None,
@ -124,7 +132,8 @@ def _build_fixture() -> dict[str, Any]:
address="ул. Тестовая, 1",
)
# ── deal 2 — median 150k; first lot's cadnum → quarter "66:41:0204016". ──
# ── deal 2 — median 150k (5 listings, symmetric — #oblast-E, see deal 1
# comment); first lot's cadnum → quarter "66:41:0204016". ──
deal2 = _deal_record(
deal_id=2,
sold_ppm2=200_000.0, # SOLD бизнес (160k..220k)
@ -132,12 +141,14 @@ def _build_fixture() -> dict[str, Any]:
rooms=2,
listings=[
{
"price_per_m2": 140_000.0,
"price_per_m2": 130_000.0,
"source": "cian",
"building_cadastral_number": "66:41:0204016:350",
},
{"price_per_m2": 140_000.0, "source": "cian"},
{"price_per_m2": 150_000.0, "source": "cian"},
{"price_per_m2": 160_000.0, "source": "cian"},
{"price_per_m2": 170_000.0, "source": "cian"},
],
anchor_comps=[],
anchor_tier_fetched=None,
@ -148,16 +159,19 @@ def _build_fixture() -> dict[str, Any]:
address="ул. Тестовая, 2",
)
# ── deal 3 — median 310k; anchor_comps present but below min_comps → no fire. ──
# ── deal 3 — median 310k (5 listings, symmetric — #oblast-E, see deal 1
# comment); anchor_comps present but below min_comps → no fire. ──
deal3 = _deal_record(
deal_id=3,
sold_ppm2=290_000.0, # SOLD элит (220k..300k)
area_m2=80.0,
rooms=3,
listings=[
{"price_per_m2": 290_000.0, "source": "yandex"},
{"price_per_m2": 300_000.0, "source": "yandex"},
{"price_per_m2": 310_000.0, "source": "yandex"},
{"price_per_m2": 320_000.0, "source": "yandex"},
{"price_per_m2": 330_000.0, "source": "yandex"},
],
anchor_comps=[
{"price_per_m2": 305_000.0, "area_m2": 80.0, "rooms": 3, "floor": 5, "total_floors": 9},

View file

@ -224,14 +224,22 @@ def test_ekb_with_dense_listings_ignores_deals_fallback() -> None:
"""EKB has plenty of local listings — the radius-path headline must win,
NOT the deal corridor, even though dkp_raw is present (byte-green guard:
EKB must stay on the existing listings-median path unconditionally).
#oblast-E: 5 analogs (>= HEADLINE_LISTINGS_MIN_N) — "plenty" per this
test's own docstring means the new headline sufficiency gate must NOT
suppress them; a merely-3-analog EKB sample is exactly the thin case the
gate targets (see test_estimator_headline_sufficiency.py), so it would no
longer count as "dense" post-fix.
"""
from app.schemas.trade_in import TradeInEstimateInput
from app.services.geocoder import GeocodeResult
analogs = [
_make_listing(price_per_m2=140_000.0),
_make_listing(price_per_m2=138_000.0),
_make_listing(price_per_m2=142_000.0),
_make_listing(price_per_m2=145_000.0),
_make_listing(price_per_m2=150_000.0),
_make_listing(price_per_m2=148_000.0),
_make_listing(price_per_m2=152_000.0),
]
dkp_raw = {
"count": 20,

View file

@ -153,11 +153,14 @@ def _make_listing(*, price_per_m2: float, area_m2: float = 40.0) -> dict[str, An
}
# Three fixed analogs → deterministic median ppm2 = 150_000 (< 5 ⇒ no outlier drop).
# #oblast-E: 5 fixed analogs (>= HEADLINE_LISTINGS_MIN_N) → deterministic
# median ppm2 = 150_000 (symmetric around it, < outlier-drop threshold).
_ANALOGS: list[dict[str, Any]] = [
_make_listing(price_per_m2=140_000.0),
_make_listing(price_per_m2=135_000.0),
_make_listing(price_per_m2=145_000.0),
_make_listing(price_per_m2=150_000.0),
_make_listing(price_per_m2=160_000.0),
_make_listing(price_per_m2=155_000.0),
_make_listing(price_per_m2=165_000.0),
]

View file

@ -47,10 +47,15 @@ def _make_listing(*, price_per_m2: float, area_m2: float = 50.0) -> dict[str, An
}
# #oblast-E: 5 analogs (>= HEADLINE_LISTINGS_MIN_N) so the new headline
# sufficiency gate doesn't suppress the median before this test's target
# mechanic (expected-sold clamp) runs. Median stays 270_000.
_ANALOGS: list[dict[str, Any]] = [
_make_listing(price_per_m2=250_000.0),
_make_listing(price_per_m2=260_000.0),
_make_listing(price_per_m2=270_000.0),
_make_listing(price_per_m2=280_000.0),
_make_listing(price_per_m2=290_000.0),
]

View file

@ -0,0 +1,373 @@
"""#oblast-E — headline sufficiency gate (money-path audit, 2026-08-02).
Live-prod repro that motivated this gate: Серов 2к/45м², n=3 scraped listings
headline 42 391 /м² (36% vs the city ДКП corridor, 54 126 /м²); a neighbouring
street in the same town swung ±66% on 1-2 different random listings. Каменск-
Уральский returned a LITERAL 0 for a room/area combo with no local ДКП match
either, with no honest refusal surfaced. Первоуральск (0 listings) already fell
back to the (pre-existing) ДКП deals-headline fallback correctly this gate
routes the THIN (1..HEADLINE_LISTINGS_MIN_N-1 listings) case into that SAME,
already-tested path instead of trusting a 1-4-lot median as the headline.
Two layers:
1. `_price_from_inputs` unit tests (no DB, no estimate_quality overhead)
boundary behaviour of the gate itself.
2. `estimate_quality` integration tests proves the money-path invariants
that matter to a caller: literal 0 never leaks as a "confident" price,
display `analogs` cards never outnumber what `n_analogs` claims, and the
explanation text describes what actually happened (not a stock "аналогов
не найдено" when some WERE found, just too few).
"""
from __future__ import annotations
import os
from datetime import UTC, datetime
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import anyio
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from app.services import estimator
from app.services.estimator import HEADLINE_LISTINGS_MIN_N, _price_from_inputs
from app.services.geocoder import GeocodeResult
# ─────────────────────────────────────────────────────────────────────────────
# Layer 1 — `_price_from_inputs` direct unit tests
# ─────────────────────────────────────────────────────────────────────────────
def _geo() -> GeocodeResult:
return GeocodeResult(
lat=59.604,
lon=60.577,
full_address="Свердловская обл., Серов, ул. Ленина, 5",
provider="nominatim",
)
def _lot(ppm2: float, address: str = "ул. Ленина, 5", source: str = "avito") -> dict[str, Any]:
return {"price_per_m2": ppm2, "address": address, "source": source}
def _lots(prices: list[float]) -> list[dict[str, Any]]:
return [_lot(p, address=f"ул. Ленина, {i + 5}") for i, p in enumerate(prices)]
def _call(
*,
listings: list[dict[str, Any]],
area_m2: float = 45.0,
rooms: int | None = 2,
dkp_raw: dict[str, Any] | None = None,
anchor_comps: list[dict[str, Any]] | None = None,
anchor_tier_fetched: str | None = None,
) -> estimator.PricingResult:
def ratio_resolver(_appm2: float | None) -> tuple[float | None, str | None]:
return None, None
return _price_from_inputs(
listings=listings,
area_m2=area_m2,
rooms=rooms,
repair_state=None,
floor=5,
total_floors=9,
target_year=None,
analog_tier="W",
fallback_used=False,
area_widened=False,
anchor_comps=anchor_comps or [],
anchor_tier_fetched=anchor_tier_fetched,
dkp_raw=dkp_raw,
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_threshold_is_five_not_lower() -> None:
"""The chosen sufficiency floor — see estimator.py module docstring (#oblast-E)
for the data-driven justification (n=3 live-repro'd 36%, n=5 matches the
existing MIN_ANALOGS_TIER_0 "enough to trust" convention)."""
assert HEADLINE_LISTINGS_MIN_N == 5
def test_four_listings_below_threshold_suppressed_no_fallback() -> None:
"""n=4 (< 5), no ДКП signal → headline suppressed to the honest zero state,
NOT the naive median of 4 listings."""
pr = _call(listings=_lots([200_000.0, 210_000.0, 220_000.0, 230_000.0]))
assert pr.median_ppm2 == 0.0
assert pr.median_price == 0
assert pr.n_analogs == 0
assert pr.range_low == 0
assert pr.range_high == 0
def test_five_listings_at_threshold_not_suppressed() -> None:
"""n=5 (== threshold) → the real listings median is trusted as the headline."""
pr = _call(listings=_lots([200_000.0, 205_000.0, 210_000.0, 215_000.0, 220_000.0]))
assert pr.median_ppm2 == 210_000.0
assert pr.n_analogs == 5
assert pr.median_price == round(210_000.0 * 45.0)
def test_one_listing_below_threshold_suppressed() -> None:
"""n=1 — the sharpest form of the Серов bug (a single random lot deciding
the whole headline) must be suppressed exactly like n=4."""
pr = _call(listings=_lots([200_000.0]))
assert pr.median_ppm2 == 0.0
assert pr.n_analogs == 0
def test_thin_sample_with_sufficient_deals_uses_deals_headline() -> None:
"""n=3 listings (thin) + a usable ДКП corridor → headline comes from the
deal corridor median, NOT the 3-listing median (live Серов repro: 3
listings gave 42 391 vs the honest ДКП-based ~54 126)."""
dkp_raw = {
"count": 54,
"low_ppm2": 44_000,
"median_ppm2": 65_957,
"high_ppm2": 89_000,
"period_months": 12,
}
pr = _call(
listings=_lots([42_391.0, 26_818.0, 75_058.0]),
dkp_raw=dkp_raw,
)
assert pr.median_ppm2 == 65_957.0, (
f"headline={pr.median_ppm2} must equal the ДКП corridor median, not the "
"noisy 3-listing median (42 391 area)"
)
assert pr.n_analogs == 0, "honest: 0 scraped-listing analogs back this headline"
assert pr.confidence == "low"
def test_thin_sample_with_insufficient_deals_stays_zero() -> None:
"""n=3 listings (thin) + a ДКП corridor that is ITSELF too thin
(< DEALS_HEADLINE_FALLBACK_MIN_N) neither source is trusted; honest zero,
not a fabricated number from either side."""
dkp_raw = {
"count": 1,
"low_ppm2": 40_000,
"median_ppm2": 65_957,
"high_ppm2": 80_000,
"period_months": 12,
}
pr = _call(listings=_lots([42_391.0, 26_818.0, 75_058.0]), dkp_raw=dkp_raw)
assert pr.median_ppm2 == 0.0
assert pr.median_price == 0
assert pr.n_analogs == 0
def test_thin_sample_explanation_is_honest_about_count() -> None:
"""The explanation for a thin-but-nonzero sample must say HOW MANY listings
were found (not the generic 'ничего не найдено' text used for a genuine
zero-listing case) #4 in the task: explanation must match reality."""
pr = _call(listings=_lots([200_000.0, 210_000.0])) # n=2
assert pr.explanation is not None
assert "2" in pr.explanation
assert "недостаточно" in pr.explanation.lower()
# Must NOT reuse the "nothing found at all" copy — 2 listings WERE found.
assert "не найдено аналогов" not in pr.explanation.lower()
def test_thin_sample_deals_fallback_explanation_does_not_claim_zero_listings() -> None:
"""#4: once the ДКП fallback fires for a thin (not zero) sample, the
explanation must not falsely claim 'рядом нет объявлений' some WERE
found, just not enough to trust."""
dkp_raw = {
"count": 20,
"low_ppm2": 40_000,
"median_ppm2": 60_000,
"high_ppm2": 80_000,
"period_months": 12,
}
pr = _call(listings=_lots([200_000.0, 210_000.0, 220_000.0]), dkp_raw=dkp_raw)
assert pr.explanation is not None
assert "рядом нет актуальных объявлений" not in pr.explanation.lower()
assert "сделкам росреестра" in pr.explanation.lower()
def test_thin_sample_listings_clean_preserved_for_anchor_ghost_guard() -> None:
"""Regression guard: the gate must suppress the AGGREGATE (median/n_analogs)
without clearing `listings_clean` itself the same-building anchor's own
ghost-anchor guard (#1871) reads `listings_clean` truthiness to tell
"genuinely zero nearby listings" from "some nearby, just too few to trust
as headline", and conflating the two was caught regressing
test_estimator_split_corridor_1871.py during this change."""
pr = _call(listings=_lots([200_000.0, 210_000.0, 220_000.0]))
assert pr.n_analogs == 0
assert len(pr.listings_clean) == 3
assert pr.listings_headline_thin_n == 3
def test_sufficient_sample_listings_headline_thin_n_is_zero() -> None:
"""Sanity/control: once n reaches the threshold, the thin-marker stays 0 —
downstream (estimate_quality) must not treat a healthy sample as thin."""
pr = _call(listings=_lots([200_000.0, 205_000.0, 210_000.0, 215_000.0, 220_000.0]))
assert pr.listings_headline_thin_n == 0
# ─────────────────────────────────────────────────────────────────────────────
# Layer 2 — `estimate_quality` integration tests (full stub-patched I/O path)
# ─────────────────────────────────────────────────────────────────────────────
def _make_listing(*, price_per_m2: float, address: str, area_m2: float = 45.0) -> dict[str, Any]:
return {
"source": "avito",
"source_url": f"https://avito.ru/offer/{address}",
"address": address,
"lat": 59.604,
"lon": 60.577,
"rooms": 2,
"area_m2": area_m2,
"floor": 5,
"total_floors": 9,
"price_rub": price_per_m2 * area_m2,
"price_per_m2": price_per_m2,
"listing_date": datetime(2026, 5, 1),
"days_on_market": 10,
"photo_urls": [],
"scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
"distance_m": 150.0,
"relevance_score": 0.1,
}
def _serov_payload() -> Any:
from app.schemas.trade_in import TradeInEstimateInput
return TradeInEstimateInput(
address="Серов, ул. Ленина, 5",
area_m2=45.0,
rooms=2,
floor=5,
total_floors=9,
city_hint="Серов",
)
def _run_estimate(
*,
analogs: list[dict[str, Any]],
dkp_raw: dict[str, Any] | None,
) -> Any:
from app.services.estimator import estimate_quality
db = MagicMock()
payload = _serov_payload()
async def _run() -> Any:
with (
patch("app.services.estimator.geocode", new=AsyncMock(return_value=_geo())),
patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)),
patch("app.services.estimator.match_house_readonly", return_value=None),
patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
patch(
"app.services.estimator._fetch_analogs",
return_value=(list(analogs), False, "W"),
),
patch("app.services.estimator._fetch_anchor_comps", return_value=([], None)),
patch("app.services.estimator._fetch_deals", return_value=[]),
patch(
"app.services.estimator._get_or_fetch_imv_cached",
new=AsyncMock(return_value=None),
),
patch(
"app.services.estimator._get_or_fetch_yandex_valuation_cached",
new=AsyncMock(return_value=None),
),
patch(
"app.services.estimator.estimate_via_cian_valuation",
new=AsyncMock(return_value=None),
),
patch("app.services.estimator._fetch_dkp_corridor", return_value=dkp_raw),
patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
):
return await estimate_quality(payload, db)
return anyio.run(_run)
def test_e2e_thin_no_deals_never_leaks_literal_zero_as_confident_price() -> None:
"""Каменск-Уральский-style repro: thin listings, no usable ДКП corridor —
median_price_rub must be 0 AND insufficient_data must be True TOGETHER
(the AggregatedEstimate.insufficient_data computed_field invariant that
stops a literal 0 reaching the user as a confident number)."""
analogs = [
_make_listing(price_per_m2=200_000.0, address="ул. Ленина, 5"),
_make_listing(price_per_m2=210_000.0, address="ул. Ленина, 7"),
]
est = _run_estimate(analogs=analogs, dkp_raw=None)
assert est.median_price_rub == 0
assert est.insufficient_data is True
assert est.n_analogs == 0
assert est.confidence == "low"
def test_e2e_thin_sample_display_cards_never_outnumber_n_analogs() -> None:
"""The 2 thin listings must NOT be surfaced as `analogs` display cards while
n_analogs reports 0 that would be the same dishonesty (confident-looking
UI) this whole gate exists to remove."""
analogs = [
_make_listing(price_per_m2=200_000.0, address="ул. Ленина, 5"),
_make_listing(price_per_m2=210_000.0, address="ул. Ленина, 7"),
]
est = _run_estimate(analogs=analogs, dkp_raw=None)
assert est.n_analogs == 0
assert est.analogs == []
def test_e2e_serov_repro_thin_sample_routes_to_deals_headline() -> None:
"""Live Серов repro (n=3 scraped listings, wide ДКП corridor available):
headline must come from the deal corridor, not the noisy 3-listing median,
and the estimate must be honestly non-'insufficient' (a real number, low
confidence, deals-sourced)."""
analogs = [
_make_listing(price_per_m2=42_391.0, address="ул. Льва Толстого, 8А"),
_make_listing(price_per_m2=26_818.0, address="ул. Кирова, 4"),
_make_listing(price_per_m2=75_058.0, address="ул. Льва Толстого, 34"),
]
dkp_raw = {
"count": 54,
"low_ppm2": 44_000,
"median_ppm2": 65_957,
"high_ppm2": 89_000,
"period_months": 12,
}
est = _run_estimate(analogs=analogs, dkp_raw=dkp_raw)
assert est.median_price_per_m2 == 65_957
assert est.insufficient_data is False
assert est.n_analogs == 0
assert est.confidence == "low"
assert est.confidence_explanation is not None
assert "сделкам росреестра" in est.confidence_explanation.lower()
def test_e2e_sufficient_five_analogs_unaffected_control() -> None:
"""Control (mirrors the Екатеринбург prod check in the PR): a sample that
clears the threshold is priced exactly as before headline is the real
listings median, all 5 analogs counted."""
analogs = [
_make_listing(price_per_m2=195_000.0, address="ул. Ленина, 5"),
_make_listing(price_per_m2=205_000.0, address="ул. Ленина, 7"),
_make_listing(price_per_m2=210_000.0, address="ул. Ленина, 9"),
_make_listing(price_per_m2=215_000.0, address="ул. Ленина, 11"),
_make_listing(price_per_m2=225_000.0, address="ул. Ленина, 13"),
]
est = _run_estimate(analogs=analogs, dkp_raw=None)
assert est.median_price_per_m2 == 210_000
assert est.n_analogs == 5
assert est.insufficient_data is False

View file

@ -320,10 +320,15 @@ def _make_listing(*, price_per_m2: float, area_m2: float = 40.0) -> dict[str, An
}
# #oblast-E: 5 analogs (>= HEADLINE_LISTINGS_MIN_N) so the new headline
# sufficiency gate doesn't suppress the median before this test's target
# mechanic (IMV blend) runs. Median stays 150_000.
_BLEND_ANALOGS: list[dict[str, Any]] = [
_make_listing(price_per_m2=140_000.0),
_make_listing(price_per_m2=135_000.0),
_make_listing(price_per_m2=145_000.0),
_make_listing(price_per_m2=150_000.0),
_make_listing(price_per_m2=160_000.0),
_make_listing(price_per_m2=155_000.0),
_make_listing(price_per_m2=165_000.0),
]

View file

@ -87,15 +87,22 @@ def _listing(price_per_m2: float | None, area_m2: float = 60.0) -> dict[str, Any
def test_n_analogs_counts_only_priced_radius_analogs() -> None:
"""Radius-путь: 3 аналога с ценой + 2 без цены (price_per_m2=None) пережили
outlier-фильтр n_analogs == 3 (число ВНЁСШИХ цену), а не 5 (всего).
"""Radius-путь: 5 аналогов с ценой + 2 без цены (price_per_m2=None) пережили
outlier-фильтр n_analogs == 5 (число ВНЁСШИХ цену), а не 7 (всего).
До фикса n_analogs = len(listings_clean) = 5 «Найдено 5 аналогов» вводило в
заблуждение (медиана построена лишь по 3)."""
До фикса n_analogs = len(listings_clean) = 7 «Найдено 7 аналогов» вводило в
заблуждение (медиана построена лишь по 5).
#oblast-E: priced count is 5 (>= HEADLINE_LISTINGS_MIN_N) — below that, the
new headline sufficiency gate would suppress the median before this test's
target invariant (priced-only counting) is even reachable; see
test_estimator_headline_sufficiency.py for that gate's own tests."""
radius = [
_listing(price_per_m2=200_000.0),
_listing(price_per_m2=195_000.0),
_listing(price_per_m2=205_000.0),
_listing(price_per_m2=210_000.0),
_listing(price_per_m2=220_000.0),
_listing(price_per_m2=215_000.0),
_listing(price_per_m2=225_000.0),
_listing(price_per_m2=None),
_listing(price_per_m2=None),
]
@ -105,8 +112,8 @@ def test_n_analogs_counts_only_priced_radius_analogs() -> None:
radius_analogs=radius,
payload=_h._make_payload(area=60.0, rooms=2),
)
# n_analogs = число PRICED аналогов (3), НЕ len(listings_clean) (5).
assert est.n_analogs == 3
# n_analogs = число PRICED аналогов (5), НЕ len(listings_clean) (7).
assert est.n_analogs == 5
# Медиана построена → headline ненулевой (sanity: фикс не обнулил оценку).
assert est.median_price_rub > 0

View file

@ -305,6 +305,13 @@ def _make_listing_qi(
price_per_m2: float = _BASE_PPM2,
area_m2: float = _AREA,
building_cadastral_number: str | None = None,
# #oblast-E: distinct floor per synthetic lot (default matches the old
# hardcoded 4) — _dedup_cross_source's physical key is
# (street/cadnum, floor, area_bucket, price_bucket); with every fixture lot
# sharing address/price/area, an unvaried floor made 5+ synthetic lots
# collapse into 1 via cross-source dedup, defeating the >= HEADLINE_LISTINGS_MIN_N
# gate below floor=4 is a no-op for all pre-existing single-lot call sites.
floor: int = 4,
) -> dict[str, Any]:
from datetime import UTC, datetime
@ -317,7 +324,7 @@ def _make_listing_qi(
"lon": 60.595,
"rooms": 1,
"area_m2": area_m2,
"floor": 4,
"floor": floor,
"total_floors": 16,
"price_rub": price_rub,
"price_per_m2": price_per_m2,
@ -458,8 +465,9 @@ _ANALOGS_OTHER_QUARTER = [
_make_listing_qi(
price_per_m2=_BASE_PPM2,
building_cadastral_number=f"{_OTHER_QUARTER}:100",
floor=4 + i,
)
for _ in range(3)
for i in range(5)
]
@ -477,7 +485,8 @@ def test_quarter_index_correction_applied() -> None:
# Чтобы увидеть ненулевую коррекцию, делаем аналоги БЕЗ кадастрового номера
# → avg_analog_index = 1.0 → factor = 1.2.
analogs_no_cadnum = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None, floor=4 + i)
for i in range(5)
]
est = _run_estimate_qi(
analogs=analogs_no_cadnum,
@ -501,12 +510,16 @@ def test_guard2_skip_when_majority_analogs_in_target_quarter() -> None:
# 4 аналога в target квартале, 1 в другом → ratio = 4/5 = 0.8 > 0.6 → skip
target_cadnum = f"{_TARGET_QUARTER}:100"
other_cadnum = f"{_OTHER_QUARTER}:100"
# #oblast-E: distinct floor per lot — otherwise all 5 share the same
# (street, floor, area_bucket, price_bucket) physical key and
# _dedup_cross_source collapses them into ONE lot (defeats both this
# test's 4-vs-1 ratio AND the >= HEADLINE_LISTINGS_MIN_N sufficiency gate).
analogs = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=other_cadnum),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum, floor=1),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum, floor=2),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum, floor=3),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=target_cadnum, floor=4),
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=other_cadnum, floor=5),
]
base_median = round(_BASE_PPM2 * _AREA)
est = _run_estimate_qi(
@ -526,7 +539,8 @@ def test_guard2_skip_when_majority_analogs_in_target_quarter() -> None:
def test_sparse_fallback_no_row_noop() -> None:
"""_lookup_quarter_index вернул None → no-op, медиана не меняется."""
analogs_no_cadnum = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None, floor=4 + i)
for i in range(5)
]
base_median = round(_BASE_PPM2 * _AREA)
est = _run_estimate_qi(
@ -545,7 +559,8 @@ def test_sparse_fallback_no_row_noop() -> None:
def test_bimodal_guard_skips_high_index_small_n() -> None:
"""Bimodal guard: price_index=3.5, n_deals=20 → no-op."""
analogs_no_cadnum = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None, floor=4 + i)
for i in range(5)
]
base_median = round(_BASE_PPM2 * _AREA)
est = _run_estimate_qi(
@ -563,7 +578,8 @@ def test_bimodal_guard_allows_high_index_large_n() -> None:
Медиана меняется (guard не блокирует), но масштабируется на 1.8, не 2.5.
"""
analogs_no_cadnum = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None, floor=4 + i)
for i in range(5)
]
base_median = round(_BASE_PPM2 * _AREA)
est = _run_estimate_qi(
@ -597,7 +613,7 @@ def test_guard1_anchor_tier_prevents_correction() -> None:
fake_comps = [
_make_listing_qi(price_per_m2=_BASE_PPM2 * 1.5, building_cadastral_number=None)
for _ in range(3)
for _ in range(5)
]
fake_anchor = {
"anchor_ppm2": _BASE_PPM2 * 1.5,
@ -616,7 +632,8 @@ def test_guard1_anchor_tier_prevents_correction() -> None:
}
analogs_no_cadnum = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None, floor=4 + i)
for i in range(5)
]
async def _run():
@ -696,7 +713,8 @@ def test_guard1b_imv_blend_prevents_correction() -> None:
dadata_obj = _make_fake_dadata(f"{_TARGET_QUARTER}:350")
analogs_no_cadnum = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None, floor=4 + i)
for i in range(5)
]
# IMV anchor сильно выше медианы → blend сработает
@ -778,7 +796,8 @@ def test_guard1b_imv_anchor_below_blend_threshold_prevents_correction() -> None:
dadata_obj = _make_fake_dadata(f"{_TARGET_QUARTER}:350")
analogs_no_cadnum = [
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None) for _ in range(3)
_make_listing_qi(price_per_m2=_BASE_PPM2, building_cadastral_number=None, floor=4 + i)
for i in range(5)
]
# IMV anchor НИЖЕ blend-порога: base_median = 6_000_000, threshold=1.15 → порог 6.9М.

View file

@ -136,9 +136,18 @@ def test_floor_only_widens_never_shrinks() -> None:
def test_single_analog_asking_range_gets_nonzero_width() -> None:
"""n=1 analog: q1==q3==median → zero-width asking range → floored to ±12 %."""
pr = _call(listings=[_lot(100_000)])
assert pr.n_analogs == 1
"""Degenerate (identical-price) sample: q1==q3==median → zero-width asking
range floored to ±12 %.
#oblast-E: uses 5 identical-price lots (>= HEADLINE_LISTINGS_MIN_N) rather
than a literal single analog below the new sufficiency threshold the
headline is suppressed entirely (see test_estimator_headline_sufficiency.py),
so a genuine n=1 no longer reaches this range-floor code at all. The
zero-width-IQR degenerate case this test targets is preserved identically
with identical-price lots.
"""
pr = _call(listings=_lots(100_000, n=5))
assert pr.n_analogs == 5
point = pr.median_price # 100_000 × 50 = 5_000_000
assert point == 5_000_000
half = round(RANGE_MIN_HALFWIDTH_PCT * point)
@ -164,8 +173,10 @@ def test_wide_analog_range_not_floored() -> None:
def test_expected_sold_range_floored_around_expected_point() -> None:
"""Degenerate n=1 with a ratio: expected_sold range also gets ±12 % of its point."""
pr = _call(listings=[_lot(100_000)], ratio=0.90)
"""Degenerate identical-price sample with a ratio: expected_sold range also
gets ±12 % of its point (#oblast-E: 5 identical-price lots, see comment on
test_single_analog_asking_range_gets_nonzero_width above)."""
pr = _call(listings=_lots(100_000, n=5), ratio=0.90)
assert pr.expected_sold_price is not None
assert pr.expected_sold_range_low is not None
assert pr.expected_sold_range_high is not None
@ -187,7 +198,7 @@ def test_ppm2_point_consistent_with_floored_range() -> None:
not a ppm² range, so ценаppm² consistency means the point ppm² must equal
median_price/area even after the range floor widens the rub band.
"""
pr = _call(listings=[_lot(100_000)], area_m2=50.0)
pr = _call(listings=_lots(100_000, n=5), area_m2=50.0)
assert pr.median_ppm2 == pr.median_price / 50.0
# Floor widened the rub range but did not touch the ppm² point.
assert pr.median_ppm2 == 100_000.0

View file

@ -49,16 +49,23 @@ def _make_listing(*, price_per_m2: float, area_m2: float = 50.0) -> dict[str, An
}
# #oblast-E: 5 items each (>= HEADLINE_LISTINGS_MIN_N) so the new headline
# sufficiency gate doesn't suppress the median before this test's target
# mechanic (ratio-tier resolution) runs.
_ANALOGS_LOW: list[dict[str, Any]] = [
_make_listing(price_per_m2=95_000.0),
_make_listing(price_per_m2=100_000.0),
_make_listing(price_per_m2=110_000.0),
_make_listing(price_per_m2=105_000.0),
_make_listing(price_per_m2=110_000.0),
_make_listing(price_per_m2=115_000.0),
]
_ANALOGS_HIGH: list[dict[str, Any]] = [
_make_listing(price_per_m2=295_000.0),
_make_listing(price_per_m2=300_000.0),
_make_listing(price_per_m2=310_000.0),
_make_listing(price_per_m2=305_000.0),
_make_listing(price_per_m2=310_000.0),
_make_listing(price_per_m2=315_000.0),
]

View file

@ -52,11 +52,16 @@ def _make_listing(*, price_per_m2: float, area_m2: float = 40.0) -> dict[str, An
}
# Three fixed analogs → deterministic median ppm2 = 150_000 (< 5 ⇒ no outlier drop).
# #oblast-E: 5 fixed analogs (>= HEADLINE_LISTINGS_MIN_N so the new headline
# sufficiency gate doesn't suppress the median before this test's target
# mechanic runs) → deterministic median ppm2 = 150_000 (symmetric around it,
# < outlier-drop threshold).
_ANALOGS: list[dict[str, Any]] = [
_make_listing(price_per_m2=140_000.0),
_make_listing(price_per_m2=135_000.0),
_make_listing(price_per_m2=145_000.0),
_make_listing(price_per_m2=150_000.0),
_make_listing(price_per_m2=160_000.0),
_make_listing(price_per_m2=155_000.0),
_make_listing(price_per_m2=165_000.0),
]
@ -94,25 +99,29 @@ def _run_estimate(repair_state: str | None):
async def _run():
with (
patch("app.services.estimator.geocode", new=AsyncMock(return_value=_make_fake_geo())),
patch("app.services.estimator.dadata_clean_address",
new=AsyncMock(return_value=None)),
patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)),
patch("app.services.estimator.match_house_readonly", return_value=None),
patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
# 3-tuple: (listings, fallback_used, analog_tier). Same analogs every call so
# all fallback tiers are equivalent and the median is stable.
patch("app.services.estimator._fetch_analogs",
return_value=(list(_ANALOGS), False, "S")),
patch(
"app.services.estimator._fetch_analogs", return_value=(list(_ANALOGS), False, "S")
),
patch("app.services.estimator._fetch_deals", return_value=[]),
patch("app.services.estimator._get_or_fetch_imv_cached",
new=AsyncMock(return_value=None)),
patch("app.services.estimator._get_or_fetch_yandex_valuation_cached",
new=AsyncMock(return_value=None)),
patch("app.services.estimator.estimate_via_cian_valuation",
new=AsyncMock(return_value=None)),
patch(
"app.services.estimator._get_or_fetch_imv_cached", new=AsyncMock(return_value=None)
),
patch(
"app.services.estimator._get_or_fetch_yandex_valuation_cached",
new=AsyncMock(return_value=None),
),
patch(
"app.services.estimator.estimate_via_cian_valuation",
new=AsyncMock(return_value=None),
),
# #648 S3: stub asking→sold lookup off so this test isolates the
# repair coefficient (no sold-correction, no DB).
patch("app.services.estimator._get_asking_sold_ratio",
return_value=(None, None)),
patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
):
return await estimate_quality(payload, db)
@ -133,9 +142,9 @@ def test_excellent_to_needs_repair_ratio_matches_coef() -> None:
actual_ratio = excellent.median_price_rub / needs_repair.median_price_rub
# int() truncation on both medians ⇒ allow a small tolerance.
assert abs(actual_ratio - expected_ratio) < 0.005, (
f"excellent/needs_repair ratio {actual_ratio:.5f} != coef ratio {expected_ratio:.5f}"
)
assert (
abs(actual_ratio - expected_ratio) < 0.005
), f"excellent/needs_repair ratio {actual_ratio:.5f} != coef ratio {expected_ratio:.5f}"
def test_standard_is_baseline_noop() -> None:
@ -159,9 +168,7 @@ def test_none_repair_state_is_noop() -> None:
expected_median = int(150_000.0 * 40.0)
assert baseline.median_price_rub == expected_median
assert "скорректирована на состояние ремонта" not in (
baseline.confidence_explanation or ""
)
assert "скорректирована на состояние ремонта" not in (baseline.confidence_explanation or "")
def test_repair_note_present_when_coef_differs() -> None:
@ -181,6 +188,4 @@ def test_repair_note_present_when_coef_differs() -> None:
def test_standard_emits_no_repair_note() -> None:
"""Baseline standard (1.0) is a true no-op — no repair note appended."""
standard = _run_estimate("standard")
assert "скорректирована на состояние ремонта" not in (
standard.confidence_explanation or ""
)
assert "скорректирована на состояние ремонта" not in (standard.confidence_explanation or "")

View file

@ -294,19 +294,23 @@ def test_flag_on_business_band_multiplies_point_and_range(
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.
"""Degenerate (identical-price) 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.
Identical-price analogs collapse 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.
#oblast-E: 5 identical-price lots (>= HEADLINE_LISTINGS_MIN_N) rather than a
literal n=1 see the analogous comment in test_estimator_range_floor.py.
"""
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
pr = _call(listings=_lots(200_000.0, n=5))
assert pr.n_analogs == 5
point = pr.median_price
assert point == round(10_000_000 * _BIZ) # multiplied point
half = round(RANGE_MIN_HALFWIDTH_PCT * point)

View file

@ -383,10 +383,15 @@ def _make_listing(*, price_per_m2: float, area_m2: float = 60.0) -> dict[str, An
# Радиусные аналоги — НИЗКИЕ (массовая застройка рядом размывает премиум).
# #oblast-E: 5 items (>= HEADLINE_LISTINGS_MIN_N) so the new headline
# sufficiency gate doesn't suppress this fixture's median (still 210_000,
# symmetric) before the anchor-vs-radius mechanic under test in this file runs.
_RADIUS_ANALOGS: list[dict[str, Any]] = [
_make_listing(price_per_m2=200_000.0),
_make_listing(price_per_m2=195_000.0),
_make_listing(price_per_m2=205_000.0),
_make_listing(price_per_m2=210_000.0),
_make_listing(price_per_m2=220_000.0),
_make_listing(price_per_m2=215_000.0),
_make_listing(price_per_m2=225_000.0),
]
# Same-building комплы Хохрякова 48 (флагман 684k внутри).
@ -651,11 +656,11 @@ def test_estimate_analogs_reflect_anchor_comps_when_fired() -> None:
def test_estimate_analogs_stay_radius_when_no_anchor() -> None:
"""#694: якорь НЕ сработал (tier=None, Tier D) → est.analogs отражают радиусные
аналоги (_RADIUS_ANALOGS 200k-220k) существующее поведение сохранено."""
аналоги (_RADIUS_ANALOGS) существующее поведение сохранено."""
est = _run_estimate(anchor_comps=[], anchor_tier=None)
assert len(est.analogs) == len(_RADIUS_ANALOGS)
ppm2_shown = {a.price_per_m2 for a in est.analogs}
assert ppm2_shown == {200_000, 210_000, 220_000}
assert ppm2_shown == {195_000, 205_000, 210_000, 215_000, 225_000}
def test_estimate_analogs_pass_through_display_fields() -> None: