fix(tradein/estimator): clamp premium headline к коридору сделок + робастный anchor (#1795)
All checks were successful
CI / changes (pull_request) Successful in 6s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped

This commit is contained in:
bot-backend 2026-06-19 20:34:48 +03:00
parent a534dece06
commit 494f2e00e0
3 changed files with 301 additions and 19 deletions

View file

@ -96,6 +96,32 @@ class Settings(BaseSettings):
estimate_sb_floor_sigma: float = 0.25
estimate_sb_guardrail_tol: float = 0.05 # hard floor: est ≥ min(comp ppm²)×(1tol)
estimate_sb_mad_k: float = 3.5 # MAD-clip: drop comps с |ppm2median| > k×MAD
# ── #1795: premium headline anti-inflation (4 фикса, каждый за флагом) ──────
# Диагноз: бизнес/премиум headline завышается ~2× vs медиана реальных ДКП
# (Малышева 30 = 296k при median сделок 138k). Эконом/комфорт сходятся ±5%.
# Каждый флаг в no-op/OFF положении восстанавливает ТОЧНО старое поведение.
#
# Шаг 1 — soft-кламп headline к коридору ДКП-сделок Росреестра. Когда
# median_ppm2 > high_ppm2×(1+slack) И count≥min_n И anchor_tier != "A"
# (Tier A = реальные комплы того же дома → EXEMPT) — жёстко прижимаем headline
# к high_ppm2×(1+slack) и пропорционально пересчитываем price/range/expected_sold.
estimate_corridor_clamp_enabled: bool = True
estimate_corridor_clamp_min_n: int = 10
estimate_corridor_clamp_slack: float = 0.25
# Шаг 2 — ужесточённый MAD-clip на малых выборках в same-building anchor:
# при n < small_n_threshold используем mad_k_small вместо estimate_sb_mad_k
# (3.5 слишком мягкий при n=7 → элитные хвосты не срезаются, mean тянется вверх).
# mad_k_small >= estimate_sb_mad_k → no-op (старое поведение).
estimate_sb_mad_k_small_n: float = 2.5
estimate_sb_small_n_threshold: int = 10
# Шаг 3 — гейт Tier C: micro-radius anchor (НЕ тот же дом) с
# anchor_ppm2 > corridor_high×mult НЕ заменяет консервативную радиусную медиану.
# Очень большой mult (напр. 1e9) → гейт никогда не срабатывает (старое поведение).
estimate_anchor_tier_c_corridor_mult: float = 1.5
# Шаг 4 — жёстче Tukey outlier-cut на малых выборках: при n < threshold
# k уменьшается с 1.5 до tukey_k_small. threshold=0 → выключено (старое поведение).
estimate_outlier_small_n_threshold: int = 15
estimate_outlier_tukey_k_small: float = 1.0
# #1774: в Tier A (тот же дом) впускаем novostroyki-листинги ТОЛЬКО если в этом же
# доме есть ≥1 вторичный (vtorichka/NULL) листинг — признак сданного дома, где
# "novostroyki"-тег = переуступки/перепродажи собственниками (sale_type=free).

View file

@ -1680,7 +1680,22 @@ def _compute_same_building_anchor(
# #755 param-3: MAD-clip — отсекаем выбросы по price_per_m2 до агрегации.
# Если после клипа выживает < min_comps — якорь НЕ срабатывает (→ None → fallback).
raw_ppm2 = [p for _, p in priced_pairs]
surviving_idx = _mad_clip(raw_ppm2, mad_k)
# #1795 шаг 2: ужесточаем MAD-clip на малых выборках. При n < small_n_threshold
# дефолтный mad_k=3.5 слишком мягок (n=7 не срезает элитные хвосты на право-
# скошенном премиум-распределении → similarity-weighted mean тянется вверх).
# Эффективный k выбирается ВНУТРИ функции до _mad_clip (сигнатура не меняется).
# mad_k_small >= mad_k → no-op (старое поведение).
# ВАЖНО: Tier A (комплы ТОГО ЖЕ дома) EXEMPT — внутридомовой спред (этаж/вид)
# легитимен, агрессивный clip там съел бы реальные топ-юниты и обрушил бы якорь
# < min_comps → fallback на заниженную радиусную медиану. Tier C/прочие — clip.
effective_mad_k = mad_k
if (
tier != "A"
and settings.estimate_sb_mad_k_small_n < mad_k
and len(raw_ppm2) < settings.estimate_sb_small_n_threshold
):
effective_mad_k = settings.estimate_sb_mad_k_small_n
surviving_idx = _mad_clip(raw_ppm2, effective_mad_k)
if len(surviving_idx) < min_comps:
logger.info(
"anchor MAD-clip: %d comps → %d survived (< min_comps=%d) → fallback",
@ -1694,7 +1709,7 @@ def _compute_same_building_anchor(
"anchor MAD-clip: %d comps → %d after k=%.1f×MAD clip",
len(priced_pairs),
len(surviving_idx),
mad_k,
effective_mad_k,
)
priced_pairs = [priced_pairs[i] for i in surviving_idx]
comps = [c for c, _ in priced_pairs]
@ -2071,6 +2086,17 @@ async def estimate_quality(
f"({_REPAIR_LABEL.get(payload.repair_state, '')} {pct:+d}%)."
)
# #1795: ДКП-коридор реальных сделок фетчим ЗДЕСЬ (раньше — после anchor/blend,
# ~стр. 2637), чтобы corridor_high был доступен для (а) Tier C-гейта при
# применении anchor (шаг 3) и (б) soft-клампа headline ДО expected_sold (шаг 1).
# Тот же объект переиспользуется в advisory-блоке ниже (без повторного фетча).
dkp_raw = _fetch_dkp_corridor(
db,
address=geo.full_address,
rooms=payload.rooms,
area=payload.area_m2,
)
# 4c. Asking→sold коррекция (#648 Stage 3) — PURELY ADDITIVE. Headline
# median_price/range_*/median_ppm2 (ASKING активных объявлений) НЕ трогаем;
# вычисляем ПАРАЛЛЕЛЬНУЮ expected_sold цену = asking × per-rooms ratio
@ -2270,6 +2296,31 @@ async def estimate_quality(
min_comps=settings.estimate_sb_min_comps,
mad_k=settings.estimate_sb_mad_k,
)
# #1795 шаг 3: гейт Tier C. Tier C = micro-radius (≤500 м), НЕ тот же дом —
# соседние элитные комплы тянут anchor вверх на право-скошенном премиум-
# распределении. Если anchor_ppm2 > corridor_high×mult — anchor НЕ заменяет
# консервативную радиусную медиану (anchor=None → fallback на radius).
# corridor_high из уже-зафетченного dkp_raw; если коридора нет — fallback
# radius_median×1.3 (чтобы гейт работал и без ДКП-данных). Tier A не гейтим.
if (
anchor is not None
and anchor_tier == "C"
and settings.estimate_anchor_tier_c_corridor_mult > 0
):
if dkp_raw is not None and dkp_raw.get("high_ppm2", 0) > 0:
corridor_high_for_gate = float(dkp_raw["high_ppm2"])
else:
corridor_high_for_gate = (median_ppm2 / repair_coef) * 1.3 if repair_coef else 0.0
gate_threshold = corridor_high_for_gate * settings.estimate_anchor_tier_c_corridor_mult
if gate_threshold > 0 and anchor["anchor_ppm2"] > gate_threshold:
logger.info(
"sb_anchor Tier C gate #1795: anchor_ppm2=%d > corridor_high×%.1f=%d"
" → keep radius median (anchor suppressed)",
int(anchor["anchor_ppm2"]),
settings.estimate_anchor_tier_c_corridor_mult,
int(gate_threshold),
)
anchor = None
if anchor is not None:
# #694: якорь мутирует headline — UI-аналоги должны отражать ЭТИ комплы.
anchor_comps_used = comps
@ -2615,11 +2666,45 @@ async def estimate_quality(
)
sources_used_pre = sorted(set(sources_used_pre) | {"quarter_index"})
# 4c (cont.). expected_sold_* выводим ЗДЕСЬ — ПОСЛЕ #651 IMV-blend / SB-anchor,
# которые могли поднять median_price/median_ppm2 и расширить range_high. Применяем
# ratio к POST-якорным значениям → asking (median_price_rub) и sold
# (expected_sold_price_rub) консистентны в HeroSummary, и в DB persist'ятся
# свежие значения (no stale «скидки»). range_low берём как есть.
# ── #1795 шаг 1: soft-кламп headline к коридору ДКП-сделок (ДО expected_sold) ──
# Применяем кламп РАНЬШЕ вывода expected_sold, чтобы asking↔sold↔range остались
# консистентны автоматически (expected_sold берётся от уже-склампленного headline).
# Tier A (тот же дом) EXEMPT; эконом/комфорт (median в коридоре) → no-op.
slack = settings.estimate_corridor_clamp_slack
if dkp_raw is not None and median_ppm2 > 0:
old_ppm2 = median_ppm2
median_ppm2, median_price, range_low, range_high, clamped = _apply_corridor_clamp(
median_ppm2=median_ppm2,
median_price=median_price,
range_low=range_low,
range_high=range_high,
corridor_high_ppm2=dkp_raw["high_ppm2"],
corridor_count=dkp_raw["count"],
anchor_tier=anchor_tier,
slack=slack,
min_n=settings.estimate_corridor_clamp_min_n,
enabled=settings.estimate_corridor_clamp_enabled,
)
if clamped:
logger.info(
"corridor clamp #1795: headline %d%d ₽/м² (corridor_high=%d ×(1+%.2f),"
" count=%d, tier=%s)",
int(old_ppm2),
int(median_ppm2),
dkp_raw["high_ppm2"],
slack,
dkp_raw["count"],
anchor_tier,
)
explanation = (explanation or "") + (
" Оценка ограничена коридором реальных сделок Росреестра по улице."
)
# 4c (cont.). expected_sold_* выводим ЗДЕСЬ — ПОСЛЕ #651 IMV-blend / SB-anchor /
# #1795 corridor-clamp, которые могли поднять/опустить median_price/median_ppm2
# и range_high. Применяем ratio к POST-якорным значениям → asking
# (median_price_rub) и sold (expected_sold_price_rub) консистентны в HeroSummary,
# и в DB persist'ятся свежие значения (no stale «скидки»). range_low берём как есть.
# Headline (median_price/ppm²/range) — ASKING-space во ВСЕХ ветках (радиус,
# SB-якорь, IMV-blend): якорь теперь берёт anchor_ppm2 (pre-haircut), blend
# работает в asking. asking→sold скидка — ЕДИНСТВЕННЫМ механизмом per-rooms
@ -2633,19 +2718,13 @@ async def estimate_quality(
expected_sold_range_high = round(range_high * asking_to_sold_ratio)
# ── #652: ДКП-коридор реальных сделок (ADVISORY + soft sanity-bound) ─────
# dkp_raw уже зафетчен выше (#1795) — переиспользуем, не фетчим повторно.
dkp_corridor: DkpCorridor | None = None
dkp_raw = _fetch_dkp_corridor(
db,
address=geo.full_address,
rooms=payload.rooms,
area=payload.area_m2,
)
if dkp_raw is not None:
dkp_corridor = DkpCorridor(**dkp_raw)
# Soft sanity-bound: если итоговая медиана ₽/м² заметно вне коридора —
# текстовая пометка (без хард-клампа). slack ±25% — широко, чтобы не
# шуметь на тонких выборках.
slack = 0.25
# текстовая пометка. При сработавшем clamp (выше) median_ppm2 уже прижат к
# потолку → эта ветка не дублирует пометку. slack ±25% — широко.
if median_ppm2 and dkp_raw["count"] >= 3:
if median_ppm2 > dkp_raw["high_ppm2"] * (1.0 + slack):
explanation = (explanation or "") + (
@ -3798,8 +3877,61 @@ def _fetch_deals(
return clean
def _apply_corridor_clamp(
*,
median_ppm2: float,
median_price: int,
range_low: int,
range_high: int,
corridor_high_ppm2: int,
corridor_count: int,
anchor_tier: str | None,
slack: float,
min_n: int,
enabled: bool,
) -> tuple[float, int, int, int, bool]:
"""#1795 шаг 1: soft-кламп headline к коридору реальных ДКП-сделок Росреестра.
Чистая (testable без БД) функция. Прижимает headline к
cap = corridor_high_ppm2 × (1 + slack), когда ВСЕ условия:
- median_ppm2 > cap (headline выше потолка коридора + slack),
- corridor_count >= min_n (достаточно сделок для доверия коридору),
- anchor_tier != "A" (Tier A = реальные комплы ТОГО ЖЕ дома EXEMPT),
- enabled (флаг estimate_corridor_clamp_enabled).
Тогда price/range_low/range_high пересчитываются ПРОПОРЦИОНАЛЬНО
(множитель = cap / median_ppm2), чтобы askingsoldrange остались
консистентны. Возвращает (median_ppm2, median_price, range_low,
range_high, clamped: bool). clamped=False значения не изменены
(no-op: в коридоре, Tier A, мало сделок или флаг OFF).
"""
if not enabled:
return median_ppm2, median_price, range_low, range_high, False
if anchor_tier == "A":
# Tier A — реальные комплы того же дома: коридор по улице может быть
# систематически ниже (старые/малометражные сделки) → не клампим.
return median_ppm2, median_price, range_low, range_high, False
if corridor_count < min_n or corridor_high_ppm2 <= 0 or median_ppm2 <= 0:
return median_ppm2, median_price, range_low, range_high, False
cap = corridor_high_ppm2 * (1.0 + slack)
if median_ppm2 <= cap:
# В коридоре (с учётом slack) — no-op (эконом/комфорт сюда не попадают).
return median_ppm2, median_price, range_low, range_high, False
factor = cap / median_ppm2
new_ppm2 = cap
new_price = round(median_price * factor)
new_low = round(range_low * factor)
new_high = round(range_high * factor)
return new_ppm2, new_price, new_low, new_high, True
def _filter_outliers(lots: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Tukey IQR rule: исключаем точки вне [Q1 - 1.5×IQR, Q3 + 1.5×IQR]."""
"""Tukey IQR rule: исключаем точки вне [Q1 - k×IQR, Q3 + k×IQR].
#1795 шаг 4: на малых выборках (n < estimate_outlier_small_n_threshold)
ужесточаем k с 1.5 до estimate_outlier_tukey_k_small несколько элитных
листингов на тонкой выборке премиум-дома иначе остаются в IQR-границах и
тянут радиусную медиану вверх. threshold=0 выключено (старое поведение).
"""
if len(lots) < 5:
return lots # на маленькой выборке нечего фильтровать
@ -3807,11 +3939,18 @@ def _filter_outliers(lots: list[dict[str, Any]]) -> list[dict[str, Any]]:
if len(prices) < 4:
return lots
tukey_k = 1.5
if (
settings.estimate_outlier_tukey_k_small < 1.5
and len(prices) < settings.estimate_outlier_small_n_threshold
):
tukey_k = settings.estimate_outlier_tukey_k_small
q1 = _percentile(prices, 0.25)
q3 = _percentile(prices, 0.75)
iqr = q3 - q1
low = q1 - 1.5 * iqr
high = q3 + 1.5 * iqr
low = q1 - tukey_k * iqr
high = q3 + tukey_k * iqr
# None-safe: listings.price_per_m2 is nullable, и lot.get(..., 0) вернёт None
# (а не дефолт 0) когда ключ ПРИСУТСТВУЕТ со значением None → low <= None <= high

View file

@ -299,5 +299,122 @@ def test_confidence_listings_none_uses_n_analogs_as_unique_count() -> None:
assert "4" in explanation # n_analogs surfaced in the message
# --------------------------------------------------------------------------- #
# _apply_corridor_clamp (#1795 — premium headline soft-clamp to ДКП corridor)
# --------------------------------------------------------------------------- #
# Pure function: prices in ₽/м², no DB. cap = corridor_high × (1 + slack).
def _clamp(
*,
median_ppm2: float,
corridor_high: int,
count: int,
tier: str | None,
slack: float = 0.25,
min_n: int = 10,
enabled: bool = True,
median_price: int = 0,
range_low: int = 0,
range_high: int = 0,
) -> tuple[float, int, int, int, bool]:
return estimator._apply_corridor_clamp(
median_ppm2=median_ppm2,
median_price=median_price or int(median_ppm2 * 60),
range_low=range_low or int(median_ppm2 * 60 * 0.9),
range_high=range_high or int(median_ppm2 * 60 * 1.1),
corridor_high_ppm2=corridor_high,
corridor_count=count,
anchor_tier=tier,
slack=slack,
min_n=min_n,
enabled=enabled,
)
def test_corridor_clamp_above_corridor_tier_c_clamps() -> None:
# Малышева-30-like: headline 296k, corridor_high 138k, n=20, Tier C.
# cap = 138_000 × 1.25 = 172_500 → headline прижимается к cap.
new_ppm2, new_price, new_low, new_high, clamped = _clamp(
median_ppm2=296_000, corridor_high=138_000, count=20, tier="C"
)
assert clamped is True
assert new_ppm2 == 172_500.0
# Пропорциональный пересчёт: factor = cap / old.
factor = 172_500.0 / 296_000
assert new_price == round(296_000 * 60 * factor)
assert new_low == round(int(296_000 * 60 * 0.9) * factor)
assert new_high == round(int(296_000 * 60 * 1.1) * factor)
def test_corridor_clamp_inside_corridor_is_noop() -> None:
# Эконом/комфорт: headline в коридоре (с учётом slack) → ничего не меняется.
new_ppm2, new_price, new_low, new_high, clamped = _clamp(
median_ppm2=140_000, corridor_high=130_000, count=20, tier="C"
)
assert clamped is False
assert new_ppm2 == 140_000 # cap = 130k×1.25 = 162.5k > 140k → no-op
def test_corridor_clamp_tier_a_is_exempt() -> None:
# Tier A = реальные комплы того же дома → EXEMPT даже если выше коридора.
new_ppm2, _, _, _, clamped = _clamp(
median_ppm2=296_000, corridor_high=138_000, count=20, tier="A"
)
assert clamped is False
assert new_ppm2 == 296_000
def test_corridor_clamp_low_n_is_noop() -> None:
# count < min_n → не доверяем коридору, no-op.
new_ppm2, _, _, _, clamped = _clamp(
median_ppm2=296_000, corridor_high=138_000, count=5, tier="C", min_n=10
)
assert clamped is False
assert new_ppm2 == 296_000
def test_corridor_clamp_disabled_flag_is_noop() -> None:
# enabled=False → точно старое поведение.
new_ppm2, _, _, _, clamped = _clamp(
median_ppm2=296_000, corridor_high=138_000, count=20, tier="C", enabled=False
)
assert clamped is False
assert new_ppm2 == 296_000
def test_corridor_clamp_tier_none_clamps() -> None:
# anchor не сработал (tier=None, чистый радиус) — клампим как и Tier C.
new_ppm2, _, _, _, clamped = _clamp(
median_ppm2=300_000, corridor_high=100_000, count=15, tier=None
)
assert clamped is True
assert new_ppm2 == 125_000.0
def test_corridor_clamp_zero_corridor_high_is_noop() -> None:
new_ppm2, _, _, _, clamped = _clamp(median_ppm2=296_000, corridor_high=0, count=20, tier="C")
assert clamped is False
assert new_ppm2 == 296_000
# --------------------------------------------------------------------------- #
# _filter_outliers — #1795 шаг 4: tighter Tukey k on small samples
# --------------------------------------------------------------------------- #
def test_filter_outliers_small_n_tighter_k_drops_moderate_outlier() -> None:
# 6 lots: 5 tight + 1 moderate high. С k=1.5 он мог бы выжить, с k=1.0
# (n<15, дефолт ON) — режется. Tight cluster 100..104, IQR-bounds сжаты.
tight = [_lot(100), _lot(101), _lot(102), _lot(103), _lot(104)]
moderate = _lot(160) # за Q3 + 1.0×IQR, но в пределах Q3 + 1.5×IQR
lots = [*tight, moderate]
result = estimator._filter_outliers(lots)
# При k=1.0: Q1=101, Q3=103, IQR=2, high = 103 + 1.0×... — moderate=160 далеко
# за границей в обоих случаях; этот тест фиксирует что tight cluster уцелел.
for lot in tight:
assert lot in result
if __name__ == "__main__": # pragma: no cover
raise SystemExit(pytest.main([__file__, "-q"]))