From 24a80ee0cf141331064cf2c5324885adaee0d9c4 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 27 Jun 2026 12:51:14 +0300 Subject: [PATCH 1/2] wip: extract pricing spine (pre-test) --- tradein-mvp/backend/app/services/estimator.py | 1492 ++++++++--------- 1 file changed, 728 insertions(+), 764 deletions(-) diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 4b0742cc..0fb4ee94 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -26,6 +26,8 @@ import logging import math import re import time +from collections.abc import Callable +from dataclasses import dataclass from datetime import UTC, date, datetime, timedelta from typing import Any from uuid import uuid4 @@ -1936,6 +1938,645 @@ async def _with_budget(coro: Any, budget_s: float, *, label: str) -> Any: return None +# ── PricingResult dataclass (pure, no I/O) ────────────────────────────────── + + +@dataclass +class PricingResult: + """Return type of _price_from_inputs — все переменные, нужные estimate_quality после блока.""" + + median_ppm2: float + median_price: int + range_low: int + range_high: int + n_analogs: int + confidence: str + explanation: str + anchor_tier: str | None + anchor_comps_used: list[dict] + avito_imv_summary: AvitoImvSummary | None + dkp_corridor: DkpCorridor | None + expected_sold_per_m2: int | None + expected_sold_price: int | None + expected_sold_range_low: int | None + expected_sold_range_high: int | None + asking_to_sold_ratio: float | None + ratio_basis: str | None + sources_used_pre: list[str] + listings_clean: list[dict] + + +def _price_from_inputs( + *, + listings: list[dict], + area_m2: float, + rooms: int | None, + repair_state: str | None, + floor: int | None, + total_floors: int | None, + target_year: int | None, + analog_tier: str, + fallback_used: bool, + area_widened: bool, + anchor_comps: list[dict], + anchor_tier_fetched: str | None, + dkp_raw: dict | None, + imv_anchor: dict | None, + imv_eval: IMVEvaluation | None, + yandex_val_present: bool, + cian_val_present: bool, + ratio_resolver: Callable[[float | None], tuple[float | None, str | None]], + quarter_index_lookup: Callable[[str], tuple[float, int] | None], + quarter_indexes_lookup: Callable[[list[str]], dict[str, float]], + target_house_cadnum: str | None, + dadata_coarse: bool, + geo: GeocodeResult, + dadata_qc_geo: int | None, +) -> PricingResult: + """Deterministic pricing orchestration — pure, synchronous, zero I/O. + + Extracted from estimate_quality (#1966) to enable offline backtesting and + direct unit testing. All DB lookups are injected via callables or pre-fetched + values; behavior is byte-identical to the original block. + """ + # 3. Outlier filter + listings_clean = _filter_outliers(listings) + + # 4. Aggregation + if listings_clean: + prices_ppm2 = sorted(lot["price_per_m2"] for lot in listings_clean if lot["price_per_m2"]) + median_ppm2 = _percentile(prices_ppm2, 0.5) + q1_ppm2 = _percentile(prices_ppm2, 0.25) + q3_ppm2 = _percentile(prices_ppm2, 0.75) + median_price = int(median_ppm2 * area_m2) + range_low = int(q1_ppm2 * area_m2) + range_high = int(q3_ppm2 * area_m2) + # #2: n_analogs считается по prices_ppm2, а не len(listings_clean). + n_analogs = len(prices_ppm2) + else: + median_ppm2 = 0.0 + q1_ppm2 = 0.0 + q3_ppm2 = 0.0 + median_price = 0 + range_low = 0 + range_high = 0 + n_analogs = 0 + + # 4b. Repair coefficient + repair_coef = _repair_coefficient(repair_state) + repair_note = "" + if listings_clean 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) + median_ppm2 = median_ppm2 * repair_coef + pct = round((repair_coef - 1.0) * 100) + repair_note = ( + f" Цена скорректирована на состояние ремонта " + f"({_REPAIR_LABEL.get(repair_state, '')} {pct:+d}%)." + ) + + # Build sources_used_pre from listings + external sources + sources_used_pre = sorted({lot.get("source") for lot in listings_clean if lot.get("source")}) + if imv_eval is not None: + sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"}) + if yandex_val_present: + sources_used_pre = sorted(set(sources_used_pre) | {"yandex_valuation"}) + if cian_val_present: + sources_used_pre = sorted(set(sources_used_pre) | {"cian_valuation"}) + + # 4c. asking→sold variables — initialized; actual computation is after all mutations. + asking_to_sold_ratio: float | None = None + ratio_basis: str | None = None + expected_sold_per_m2: int | None = None + expected_sold_price: int | None = None + expected_sold_range_low: int | None = None + expected_sold_range_high: int | None = None + + confidence, explanation = _compute_confidence( + n_analogs, + median_ppm2, + q1_ppm2 if listings_clean else 0, + q3_ppm2 if listings_clean else 0, + fallback_used, + area_widened, + listings=listings_clean, + ) + + # Tier note — информируем пользователя о качестве house-match + tier_note = "" + if analog_tier == "S": + tier_note = " (аналоги из того же дома)" + elif analog_tier == "H": + tf_str = f"{total_floors}-эт." if total_floors is not None else "" + yr_str = f"{target_year}±15 г." if target_year else "" + parts_str = ", ".join(p for p in [yr_str, tf_str] if p) + tier_note = f" (аналоги из домов того же класса: {parts_str})" if parts_str else "" + else: + tier_note = " (нет аналогов в том же доме/классе — расширили поиск)" + + explanation = (explanation or "") + tier_note + repair_note + + # ── #651/#652 v2: same-building anchor ─────────────────────────────────── + anchor_tier: str | None = anchor_tier_fetched + anchor_comps_used: list[dict] = [] + + anchor = _compute_same_building_anchor( + anchor_comps, + area_target=area_m2, + rooms_target=rooms, + tier=anchor_tier or "", + sigma=settings.estimate_sb_area_sigma, + rooms_boost=settings.estimate_sb_rooms_match_boost, + floor_target=floor, + total_floors_target=total_floors, + floor_sigma=settings.estimate_sb_floor_sigma, + min_comps=settings.estimate_sb_min_comps, + mad_k=settings.estimate_sb_mad_k, + ) + + # #1795 шаг 3: гейт Tier C. + 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 + + # #audit-1: low-confidence gate. + if anchor is not None and settings.estimate_sb_low_conf_gate_enabled: + gate_low = anchor["confidence"] == "low" + gate_thin = ( + anchor["n"] < settings.estimate_sb_gate_min_n + and anchor["fsd"] > settings.estimate_sb_gate_max_fsd + ) + if gate_low or gate_thin: + logger.info( + "sb_anchor low-conf gate #audit-1: tier=%s n=%d fsd=%.3f conf=%s" + " → suppressed (gate_low=%s gate_thin=%s) → radius fallback", + anchor_tier, + anchor["n"], + anchor["fsd"], + anchor["confidence"], + gate_low, + gate_thin, + ) + anchor = None + anchor_tier = None + + if anchor is not None: + # #694: якорь мутирует headline — UI-аналоги должны отражать ЭТИ комплы. + anchor_comps_used = anchor_comps + est_ppm2 = anchor["anchor_ppm2"] + # PREMIUM GUARDRAIL (hard). + floor_ppm2 = anchor["comp_min_ppm2"] * (1.0 - settings.estimate_sb_guardrail_tol) + if est_ppm2 < floor_ppm2: + est_ppm2 = floor_ppm2 + new_ppm2 = est_ppm2 * repair_coef + point = int(new_ppm2 * area_m2) + # FSD-диапазон. + half = settings.estimate_fsd_k * anchor["fsd"] + new_range_low = int(point * max(0.0, 1.0 - half)) + new_range_high = int(point * (1.0 + half)) + # Спред комплов. + spread_low = int(anchor["comp_min_ppm2"] * repair_coef * area_m2) + spread_high = int(anchor["comp_max_ppm2"] * repair_coef * area_m2) + new_range_low = min(new_range_low, spread_low, point) + new_range_high = max(new_range_high, spread_high, point) + + logger.info( + "sb_anchor: tier=%s n=%d radius_median_ppm2=%d → anchor_asking_ppm2=%d" + " (uplift=%s haircut=%.2f) point %d → %d", + anchor_tier, + anchor["n"], + int(median_ppm2), + int(est_ppm2), + anchor["used_uplift"], + anchor["haircut"], + median_price, + point, + ) + median_ppm2 = new_ppm2 + median_price = point + range_low = new_range_low + range_high = new_range_high + confidence = anchor["confidence"] + tier_label = "того же дома" if anchor_tier == "A" else "ближайшего окружения (≤500 м)" + # #695: explanation описывает ИМЕННО якорные комплы. + explanation = ( + f"Оценка построена по {anchor['n']} аналогам из {tier_label}" + f"{' (топ-уровень в доме)' if anchor['used_uplift'] else ''}." + ) + repair_note + # #695 (QA fixup): n_analogs по anchor-популяции. + n_analogs = anchor["n"] + # #1871 P1: ghost-anchor guard. + if not listings_clean and confidence != "low": + logger.warning( + "estimator #1871 ghost-anchor guard: confidence %s→low " + "(anchor_n=%s, radius_analogs=0)", + confidence, + anchor["n"], + ) + confidence = "low" + explanation += ( + " Оценка опирается только на аналоги из того же дома — " + "сопоставимых предложений поблизости не найдено, поэтому " + "точность ориентировочная." + ) + # #1871 P2: split-дома wide-corridor disclosure. + if ( + settings.estimate_wide_corridor_disclosure_enabled + and anchor_tier == "A" + and median_price > 0 + ): + corridor_pct = (range_high - range_low) / median_price + if corridor_pct > settings.estimate_wide_corridor_threshold: + confidence = _downgrade_confidence(confidence) + explanation = (explanation or "") + ( + " Очень широкий ценовой диапазон по дому (вероятно, дом " + "разбит на секции разной этажности или разнородный фонд) — " + "оценка ориентировочная, уточните по конкретной секции." + ) + + # ── #651: IMV / Yandex blend (Tier D only, anchor_tier is None) ────────── + imv_anchor_present: bool = False + avito_imv_summary: AvitoImvSummary | None = None + if ( + anchor_tier is None + and settings.estimate_imv_blend_enabled + and listings_clean + and median_price > 0 + ): + anchor_total: int | None = None + anchor_higher: int | None = None + anchor_label: str | None = None + if imv_anchor is not None and imv_anchor.get("recommended_price"): + anchor_total = int(imv_anchor["recommended_price"]) + anchor_higher = ( + int(imv_anchor["higher_price"]) if imv_anchor.get("higher_price") else None + ) + anchor_label = "оценке Avito IMV" + _imv_mc = int(imv_anchor["market_count"]) if imv_anchor.get("market_count") else None + avito_imv_summary = AvitoImvSummary( + recommended_price=anchor_total, + lower_price=( + int(imv_anchor["lower_price"]) if imv_anchor.get("lower_price") else None + ), + higher_price=anchor_higher, + market_count=_imv_mc, + thin_market=( + _imv_mc is not None and _imv_mc < settings.avito_imv_thin_market_threshold + ), + ) + elif imv_eval is not None and imv_eval.recommended_price: + anchor_total = int(imv_eval.recommended_price) + anchor_higher = int(imv_eval.higher_price) if imv_eval.higher_price else None + anchor_label = "оценке Avito IMV" + avito_imv_summary = AvitoImvSummary( + recommended_price=anchor_total, + lower_price=(int(imv_eval.lower_price) if imv_eval.lower_price else None), + higher_price=anchor_higher, + market_count=imv_eval.market_count, + thin_market=( + imv_eval.market_count is not None + and imv_eval.market_count < settings.avito_imv_thin_market_threshold + ), + ) + + # #audit-5b: thin-market warning. + if avito_imv_summary is not None and avito_imv_summary.thin_market: + logger.warning( + "avito_imv thin_market #audit-5b: market_count=%s" + " (< avito_imv_thin_market_threshold=%d) — IMV reliability low", + avito_imv_summary.market_count, + settings.avito_imv_thin_market_threshold, + ) + + if anchor_total is not None: + imv_anchor_present = True + new_median, new_range_high, new_ppm2, blended, anchor_used = _apply_imv_blend( + median_price=median_price, + range_high=range_high, + median_ppm2=median_ppm2, + area=area_m2, + anchor_total=anchor_total, + anchor_higher=anchor_higher, + weight=settings.estimate_imv_blend_weight, + threshold=settings.estimate_imv_blend_threshold, + ) + if blended: + logger.info( + "imv_blend: median %d → %d (anchor=%d w=%.2f) range_high %d → %d", + median_price, + new_median, + anchor_used, + settings.estimate_imv_blend_weight, + range_high, + new_range_high, + ) + median_price = new_median + median_ppm2 = new_ppm2 + explanation = (explanation or "") + ( + f" Оценка скорректирована по {anchor_label} " + f"({anchor_used / 1_000_000:.1f} млн ₽)." + ) + sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"}) + # Диапазон расширяем даже если медиану не двигали. + range_high = new_range_high + + # Display-only IMV summary when headline built by same-building anchor. + if anchor_tier is not None and avito_imv_summary is None: + if imv_anchor is not None and imv_anchor.get("recommended_price"): + _disp_mc = int(imv_anchor["market_count"]) if imv_anchor.get("market_count") else None + avito_imv_summary = AvitoImvSummary( + recommended_price=int(imv_anchor["recommended_price"]), + lower_price=( + int(imv_anchor["lower_price"]) if imv_anchor.get("lower_price") else None + ), + higher_price=( + int(imv_anchor["higher_price"]) if imv_anchor.get("higher_price") else None + ), + market_count=_disp_mc, + thin_market=( + _disp_mc is not None and _disp_mc < settings.avito_imv_thin_market_threshold + ), + ) + + # ── #764: per-cadastral-quarter price index gap-correction ─────────────── + if ( + settings.estimate_quarter_index_enabled + and anchor_tier is None # Guard-1a + and not imv_anchor_present # Guard-1b + and median_price > 0 + and area_m2 + ): + target_quarter: str | None = _quarter_from_cadastre(target_house_cadnum) + if target_quarter is None: + for lot in listings_clean: + cq = _quarter_from_cadastre(lot.get("building_cadastral_number")) + if cq is not None: + target_quarter = cq + break + + if target_quarter is not None: + qindex_result = quarter_index_lookup(target_quarter) + if qindex_result is not None: + target_qi, target_n_deals = qindex_result + + # Bimodal/nominal guard (Guard-4). + if ( + target_qi > settings.estimate_quarter_index_max_for_small_n + and target_n_deals < settings.estimate_quarter_index_small_n_threshold + ): + logger.info( + "quarter_index: bimodal guard triggered " + "(index=%.3f n=%d < %d) for %s — no-op", + target_qi, + target_n_deals, + settings.estimate_quarter_index_small_n_threshold, + target_quarter, + ) + else: + lot_quarters_for_guard2: list[str] = [] + analog_quarters: list[tuple[str, float]] = [] + for lot in listings_clean: + lq = _quarter_from_cadastre(lot.get("building_cadastral_number")) + if lq is None: + continue + lot_quarters_for_guard2.append(lq) + lp = lot.get("price_per_m2") + if lp: + analog_quarters.append((lq, float(lp))) + + # Guard-2: same-quarter ratio. + same_quarter_count = sum( + 1 for lq in lot_quarters_for_guard2 if lq == target_quarter + ) + # #1385: знаменатель — только классифицируемые аналоги. + same_quarter_ratio = ( + same_quarter_count / len(lot_quarters_for_guard2) + if lot_quarters_for_guard2 + else 0.0 + ) + if same_quarter_ratio > settings.estimate_quarter_match_skip_ratio: + logger.info( + "quarter_index: Guard-2 skip (same-quarter ratio=%.2f > %.2f)" + " for %s", + same_quarter_ratio, + settings.estimate_quarter_match_skip_ratio, + target_quarter, + ) + else: + distinct_analog_quarters = list( + dict.fromkeys(lq for lq, _lp in analog_quarters) + ) + analog_index_map = quarter_indexes_lookup(distinct_analog_quarters) + weighted_sum = 0.0 + weight_total = 0.0 + for lq, lp in analog_quarters: + lot_qi = analog_index_map.get(lq) + if lot_qi is None: + continue + weighted_sum += lp * lot_qi + weight_total += lp + + avg_analog_index = weighted_sum / weight_total if weight_total > 0 else 1.0 + + ( + median_ppm2, + median_price, + range_low, + range_high, + qi_factor, + ) = _apply_quarter_index( + base_median_ppm2=median_ppm2, + base_median_price=median_price, + base_range_low=range_low, + base_range_high=range_high, + target_index=target_qi, + avg_analog_index=avg_analog_index, + min_factor=settings.estimate_quarter_index_factor_min, + max_factor=settings.estimate_quarter_index_factor_max, + ) + analogs_with_qi = sum( + 1 for lq, _lp in analog_quarters if lq in analog_index_map + ) + logger.info( + "quarter_index: applied target=%s target_qi=%.3f" + " avg_analog_qi=%.3f factor=%.3f" + " (same_quarter_ratio=%.2f analogs_with_qi=%d)", + target_quarter, + target_qi, + avg_analog_index, + qi_factor, + same_quarter_ratio, + analogs_with_qi, + ) + explanation = (explanation or "") + ( + f" Учтена локация квартала" f" (индекс цен квартала ×{qi_factor:.2f})." + ) + sources_used_pre = sorted(set(sources_used_pre) | {"quarter_index"}) + + # ── #1795 шаг 1: soft-кламп headline к коридору ДКП-сделок ────────────── + 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 "") + ( + " Оценка ограничена коридором реальных сделок Росреестра по улице." + ) + + # ── Radius-path нижний floor от DKP-коридора ───────────────────────────── + if ( + settings.estimate_radius_floor_enabled + and anchor_tier is None + and dkp_raw is not None + and dkp_raw.get("low_ppm2", 0) > 0 + and median_ppm2 > 0 + and dkp_raw.get("count", 0) >= settings.estimate_corridor_clamp_min_n + ): + radius_floor_ppm2 = float(dkp_raw["low_ppm2"]) * settings.estimate_radius_floor_factor + if median_ppm2 < radius_floor_ppm2: + floor_factor = radius_floor_ppm2 / median_ppm2 + logger.info( + "radius_floor: median_ppm2=%d < dkp_low=%d × factor=%.2f = floor=%d" + " → lifting (factor=%.3f)", + int(median_ppm2), + dkp_raw["low_ppm2"], + settings.estimate_radius_floor_factor, + int(radius_floor_ppm2), + floor_factor, + ) + median_ppm2 = radius_floor_ppm2 + median_price = round(median_price * floor_factor) + range_low = round(range_low * floor_factor) + range_high = round(range_high * floor_factor) + explanation = (explanation or "") + ( + " Оценка поднята до нижней границы коридора реальных сделок Росреестра." + ) + + # 4c (cont.). expected_sold AFTER all headline mutations. + if median_ppm2 > 0: + asking_to_sold_ratio, ratio_basis = ratio_resolver(median_ppm2) + if asking_to_sold_ratio is None: + ratio_basis = None + + if asking_to_sold_ratio is not None and median_price > 0: + effective_ratio = asking_to_sold_ratio + if settings.estimate_expected_sold_le_asking and effective_ratio > 1.0: + logger.info( + "expected_sold ratio clamped %.3f->1.0 (rooms=%s)", + effective_ratio, + rooms, + ) + effective_ratio = 1.0 + expected_sold_per_m2 = round(median_ppm2 * effective_ratio) + expected_sold_price = round(median_price * effective_ratio) + expected_sold_range_low = round(range_low * effective_ratio) + expected_sold_range_high = round(range_high * effective_ratio) + + # ── #652: ДКП-коридор реальных сделок (advisory) ───────────────────────── + dkp_corridor: DkpCorridor | None = None + if dkp_raw is not None: + dkp_corridor = DkpCorridor(**dkp_raw) + if median_ppm2 and dkp_raw["count"] >= 3: + if median_ppm2 > dkp_raw["high_ppm2"] * (1.0 + slack): + explanation = (explanation or "") + ( + " Оценка выше коридора реальных сделок Росреестра по улице." + ) + elif median_ppm2 < dkp_raw["low_ppm2"] * (1.0 - slack): + explanation = (explanation or "") + ( + " Оценка ниже коридора реальных сделок Росреестра по улице." + ) + + # ── #693: coarse-geo downgrade ─────────────────────────────────────────── + geo_coarse = _geocode_is_coarse(geo) + if (dadata_coarse or geo_coarse) and median_price > 0: + logger.info( + "coarse-geo gate #693: dadata_coarse=%s geo_coarse=%s anchor_tier=%s " + "median=%s geo.provider=%s geo.confidence=%s geo.full_address=%r geo=(%.5f,%.5f)", + dadata_coarse, + geo_coarse, + anchor_tier, + median_price, + geo.provider, + geo.confidence, + geo.full_address, + geo.lat, + geo.lon, + ) + if (dadata_coarse or geo_coarse) and anchor_tier != "A" and median_price > 0: + if dadata_coarse: + _coarse_label = {2: "населённого пункта", 3: "города", 4: "региона"}.get( + dadata_qc_geo, "населённого пункта" + ) + else: + _coarse_label = "населённого пункта" + confidence = "low" + explanation = (explanation or "") + ( + f" Адрес определён лишь до уровня {_coarse_label} — точные координаты " + "дома найти не удалось, поэтому оценка ориентировочная (аналоги взяты " + "по широкой окрестности)." + ) + + return PricingResult( + median_ppm2=median_ppm2, + median_price=median_price, + range_low=range_low, + range_high=range_high, + n_analogs=n_analogs, + confidence=confidence, + explanation=explanation, + anchor_tier=anchor_tier, + anchor_comps_used=anchor_comps_used, + avito_imv_summary=avito_imv_summary, + dkp_corridor=dkp_corridor, + expected_sold_per_m2=expected_sold_per_m2, + expected_sold_price=expected_sold_price, + expected_sold_range_low=expected_sold_range_low, + expected_sold_range_high=expected_sold_range_high, + asking_to_sold_ratio=asking_to_sold_ratio, + ratio_basis=ratio_basis, + sources_used_pre=sources_used_pre, + listings_clean=listings_clean, + ) + + # ── Public ─────────────────────────────────────────────────────────────────── async def estimate_quality( payload: TradeInEstimateInput, db: Session, created_by: str | None = None @@ -2128,52 +2769,9 @@ async def estimate_quality( area_widened = True analog_tier = analog_tier_wa - # 3. Outlier filter - listings_clean = _filter_outliers(listings) - - # 4. Aggregation - if listings_clean: - prices_ppm2 = sorted(lot["price_per_m2"] for lot in listings_clean if lot["price_per_m2"]) - median_ppm2 = _percentile(prices_ppm2, 0.5) - q1_ppm2 = _percentile(prices_ppm2, 0.25) - q3_ppm2 = _percentile(prices_ppm2, 0.75) - median_price = int(median_ppm2 * payload.area_m2) - range_low = int(q1_ppm2 * payload.area_m2) - range_high = int(q3_ppm2 * payload.area_m2) - # #2: n_analogs должен отражать число аналогов, реально внёсших цену в - # медиану, а не len(listings_clean). _filter_outliers СОХРАНЯЕТ листинги с - # price_per_m2 IS NULL, но prices_ppm2 их отбрасывает → len(listings_clean) - # завышал счётчик ("Найдено N аналогов" вводил в заблуждение; в пределе - # все price-less → median_ppm2=0, но n_analogs>0). Считаем по prices_ppm2. - # listings_clean НЕ фильтруем — downstream (analogs_lots, sources_used, - # repair_coef) по-прежнему работает с полным выживанием outlier-фильтра. - n_analogs = len(prices_ppm2) - else: - median_ppm2 = 0 - median_price = 0 - range_low = 0 - range_high = 0 - n_analogs = 0 - - # 4b. Поправка на состояние ремонта (встреча Птицы: ремонт влияет на цену). - # Аналоги — микс состояний; коэффициент сдвигает оценку под ремонт клиента. - repair_coef = _repair_coefficient(payload.repair_state) - repair_note = "" - if listings_clean 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) - median_ppm2 = median_ppm2 * repair_coef - pct = round((repair_coef - 1.0) * 100) - repair_note = ( - f" Цена скорректирована на состояние ремонта " - 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-блоке ниже (без повторного фетча). + # ── PRE-FETCH: dkp_raw (hoisted before _price_from_inputs) ────────────── + # #1795: ДКП-коридор фетчим ДО вызова _price_from_inputs, чтобы + # corridor_high был доступен для Tier C-гейта и soft-клампа headline. dkp_raw = _fetch_dkp_corridor( db, address=geo.full_address, @@ -2181,62 +2779,10 @@ async def estimate_quality( 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 - # (asking_to_sold_ratios, migration 080). Это релевантная для выкупа цена - # сделки (backtest #648 S1: bias asking-медианы +20% → −4% на held-out ДКП). - # NOTE: actual_deals (#564) остаётся ИНФОРМАЦИОННЫМ и НЕ подмешивается в - # headline — sold-коррекция здесь единственный sold-сигнал (без double-count). - # NOTE: expected_sold_* (= asking × ratio) выводятся НЕ здесь, а ПОСЛЕ #651 - # IMV-blend / SB-anchor / #1795 corridor-clamp / radius-floor (ниже), которые - # мутируют median_price/median_ppm2/range_high. Иначе expected_sold остаётся - # pre-blend → asking 75M / sold 45M (бессмысленная скидка в HeroSummary) и - # stale-значения persist'ятся в trade_in_estimates. - # FIX(ratio-tier-mismatch): _get_asking_sold_ratio вызывается ПОСЛЕ всех - # headline-мутаций (anchor/IMV-blend/quarter-index/corridor-clamp), передавая - # ФИНАЛЬНЫЙ median_ppm2 для tier-placement. Это устраняет tier-несовпадение - # когда anchor/quarter-index двигают headline в другой tier — ratio теперь - # соответствует тому ppm², к которому он будет применён. - # Инициализируем переменные здесь; реальный _get_asking_sold_ratio вызов — после - # corridor-clamp / radius-floor (ниже, ~строка 4c-cont). - asking_to_sold_ratio: float | None = None - ratio_basis: str | None = None - expected_sold_per_m2: int | None = None - expected_sold_price: int | None = None - expected_sold_range_low: int | None = None - expected_sold_range_high: int | None = None - - confidence, explanation = _compute_confidence( - n_analogs, - median_ppm2, - q1_ppm2 if listings_clean else 0, - q3_ppm2 if listings_clean else 0, - fallback_used, - area_widened, - listings=listings_clean, - ) - - # Tier note — информируем пользователя о качестве house-match - tier_note = "" - if analog_tier == "S": - tier_note = " (аналоги из того же дома)" - elif analog_tier == "H": - tf_str = f"{payload.total_floors}-эт." if payload.total_floors is not None else "" - yr_str = f"{target_year}±15 г." if target_year else "" - parts_str = ", ".join(p for p in [yr_str, tf_str] if p) - tier_note = f" (аналоги из домов того же класса: {parts_str})" if parts_str else "" - else: - tier_note = " (нет аналогов в том же доме/классе — расширили поиск)" - - explanation = (explanation or "") + tier_note + repair_note - # ── Stage 3: Avito IMV evaluation as 5-th source (on-demand cached) ── imv_eval: IMVEvaluation | None = None imv_house_type = _IMV_HOUSE_TYPE_MAP.get(target_house_type) imv_renovation = _IMV_REPAIR_MAP.get(payload.repair_state) - # IMV требует: address, rooms, area, floor, floor_at_home, house_type, renovation_type. - # Если payload не содержит required fields — skip IMV (graceful). if ( geo is not None and geo.full_address @@ -2257,21 +2803,10 @@ async def estimate_quality( house_type=imv_house_type, renovation_type=imv_renovation, has_balcony=bool(payload.has_balcony), - has_loggia=False, # payload не разделяет балкон/лоджия → дефолт False + has_loggia=False, ) - # Include IMV в sources_used если получили - sources_used_pre = sorted({lot.get("source") for lot in listings_clean if lot.get("source")}) - if imv_eval is not None: - sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"}) - # ── Stage 8: Yandex Valuation as on-demand source (anonymous, cached 24h) ── - # #654: главный латентность-подозреваемый. Этот источник UNGATED — бежит на - # КАЖДОЙ оценке (в т.ч. без floor/total_floors), а его внутренний httpx - # timeout 30s + curl_cffi impersonation + sleep_between_requests могут одни - # превысить gateway read timeout → opaque 502. Оборачиваем в budget: при - # превышении → None (cache-hit путь быстрый, timeout бьёт только по медленному - # cache-miss fetch). Деградация идентична сетевой ошибке внутри функции. yandex_val: YandexValuationResult | None = None if geo is not None and geo.full_address: yandex_val = await _with_budget( @@ -2280,7 +2815,6 @@ async def estimate_quality( label="yandex_valuation", ) if yandex_val is not None: - sources_used_pre = sorted(set(sources_used_pre) | {"yandex_valuation"}) saved_hist = _save_yandex_history_items(db, yandex_val) logger.info( "yandex_valuation: history items processed=%d saved=%d" @@ -2289,7 +2823,7 @@ async def estimate_quality( saved_hist, ) - # ── Stage 9: Cian Valuation as 7th source (on-demand, 24h cached, graceful if no cookies) ── + # ── Stage 9: Cian Valuation as 7th source (on-demand, 24h cached) ────── cian_val: CianValuationResult | None = None if ( geo is not None @@ -2316,7 +2850,6 @@ async def estimate_quality( label="cian_valuation", ) if cian_val is not None and cian_val.sale_price_rub: - sources_used_pre = sorted(set(sources_used_pre) | {"cian_valuation"}) logger.info( "cian_valuation: price=%s accuracy=%s house_id=%s", cian_val.sale_price_rub, @@ -2326,35 +2859,17 @@ async def estimate_quality( except Exception as exc: logger.warning("cian_valuation: lookup failed (graceful): %s", exc) - # ── #651/#652 v2: SAME-BUILDING ANCHOR (PRIMARY, validated) ────────────── - # Якорь из комплов ТОГО ЖЕ ДОМА (Tier A) / micro-radius (Tier C). Когда он - # сработал — он ЗАМЕНЯЕТ радиусную медиану для median_price/ppm²/range (premium - # больше не размывается массовой застройкой). За флагом; OFF ⇒ точно старое - # поведение. Сегмент в Tier C фиксирован guard'ом (#1186): вторичка + NULL-legacy. - anchor_tier: str | None = None - # #694: комплы того же дома, на которых РЕАЛЬНО построен headline-якорь. - # Заполняется только когда anchor мутировал median (ниже) — тогда UI-аналоги - # строятся из них, а не из радиусных listings_clean (cheaper/empty). - anchor_comps_used: list[dict[str, Any]] = [] - # #691: гейт НЕ требует радиусных аналогов (listings_clean) / median_price>0. - # На проде геокод часто = None → ST_DWithin не находит радиусные комплы → - # median=0, и same-building якорь скипался, ХОТЯ комплы того же дома есть - # (_fetch_anchor_comps резолвит дом по payload.address без гео — Ткачёва 13, - # Сакко 99). Запускаем по resolved-таргету (area + raw-адрес/geo); если комплов - # реально нет — _compute_same_building_anchor вернёт None и мутации не будет - # (поведение идентично старому). Новый гейт ⊇ старого: при непустом radius - # адрес всегда есть, так что ранее-якорившиеся кейсы якорятся по-прежнему. + # ── Pre-fetch: same-building anchor comps ───────────────────────────────── + # Guard mirrors the original in-block guard exactly; when false → ([], None). + _anchor_comps_pre: list[dict] + _anchor_tier_pre: str | None if ( settings.estimate_same_building_anchor_enabled and payload.area_m2 and (payload.address or (geo is not None and geo.full_address)) ): - comps, anchor_tier = _fetch_anchor_comps( + _anchor_comps_pre, _anchor_tier_pre = _fetch_anchor_comps( db, - # payload.address (сырой ввод) ПЕРВИЧЕН: на проде геокод часто - # возвращает None (lat/lon/canonical=None), и geo.full_address пуст — - # тогда same-building Tier A не находил дом и падал в радиус. Raw-адрес - # всегда есть и именно на нём валидирован нормализатор (#677/#679). address=payload.address or geo.full_address, target_house_id=target_house_id, lat=geo.lat, @@ -2362,640 +2877,89 @@ async def estimate_quality( rooms=payload.rooms, area=payload.area_m2, ) - anchor = _compute_same_building_anchor( - comps, - area_target=payload.area_m2, - rooms_target=payload.rooms, - tier=anchor_tier or "", - sigma=settings.estimate_sb_area_sigma, - rooms_boost=settings.estimate_sb_rooms_match_boost, - floor_target=payload.floor, - total_floors_target=payload.total_floors, - floor_sigma=settings.estimate_sb_floor_sigma, - 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 - # #audit-1: low-confidence gate — якорь с низкой уверенностью или малым n - # при высоком FSD НЕ заменяет headline; fallback на radius-median. - # Tier A (n≥4, FSD<0.15) проходит без изменений при дефолтных порогах. - if anchor is not None and settings.estimate_sb_low_conf_gate_enabled: - gate_low = anchor["confidence"] == "low" - gate_thin = ( - anchor["n"] < settings.estimate_sb_gate_min_n - and anchor["fsd"] > settings.estimate_sb_gate_max_fsd - ) - if gate_low or gate_thin: - logger.info( - "sb_anchor low-conf gate #audit-1: tier=%s n=%d fsd=%.3f conf=%s" - " → suppressed (gate_low=%s gate_thin=%s) → radius fallback", - anchor_tier, - anchor["n"], - anchor["fsd"], - anchor["confidence"], - gate_low, - gate_thin, - ) - anchor = None - anchor_tier = None - if anchor is not None: - # #694: якорь мутирует headline — UI-аналоги должны отражать ЭТИ комплы. - anchor_comps_used = comps - # Headline = recommended ASKING price (комплы — активные объявления; - # golden-реалы — asking). Берём anchor_ppm2 (PRE-haircut), НЕ - # anchor_sold_ppm2. asking→sold скидка применяется единственным - # механизмом — per-rooms asking_to_sold_ratio (migration 080) ниже, - # дающим distinct expected_sold. Двойная скидка (band-haircut здесь + - # ratio там) давала median == expected_sold (#677/#681 collision). - est_ppm2 = anchor["anchor_ppm2"] - # PREMIUM GUARDRAIL (hard): не ниже минимального same-building ppm² (−tol). - # Только Tier A/C (комплы реально из дома/микрорайона). Эконом — no-op - # (est уже ≥ floor), премиум — поднимает если mean занизил. ASKING-space. - floor_ppm2 = anchor["comp_min_ppm2"] * (1.0 - settings.estimate_sb_guardrail_tol) - if est_ppm2 < floor_ppm2: - est_ppm2 = floor_ppm2 - # POINT = anchor_asking × area × repair_coef (repair уже применён к старой - # median; здесь применяем к свежему якорю — заменяем headline целиком). - new_ppm2 = est_ppm2 * repair_coef - point = int(new_ppm2 * payload.area_m2) - # FSD-диапазон (tight): симметричный вокруг point, k·fsd полуширина. - half = settings.estimate_fsd_k * anchor["fsd"] - new_range_low = int(point * max(0.0, 1.0 - half)) - new_range_high = int(point * (1.0 + half)) - # Диапазон должен ПОКРЫВАТЬ same-building спред комплов (sold-adjusted) и - # удовлетворять low ≤ point ≤ high. Внутридомовая дисперсия (этаж/вид) — - # реальный разброс цены в доме; честный диапазон обязан её включать - # (иначе видовой топ-юнит вылетает за range_high — residual miss спека). - # comp spread в ASKING-пространстве (комплы — активные объявления). range_high - # покрывает RAW comp max — честно показываем верх дома (видовой/топ-юнит), - # иначе он вылетает за диапазон. range_low — RAW comp min (asking-space): - # headline теперь asking, band-haircut больше не применяется (sold-скидка — - # единственным механизмом per-rooms ratio ниже). - # #1518: repair_coef применён к point (new_ppm2), поэтому границы спреда - # тоже нормируем на ремонт — иначе центр и края в разных масштабах - # (для needs_repair coef<1 верх несимметрично завышен, для excellent — занижен). - spread_low = int(anchor["comp_min_ppm2"] * repair_coef * payload.area_m2) - spread_high = int(anchor["comp_max_ppm2"] * repair_coef * payload.area_m2) - new_range_low = min(new_range_low, spread_low, point) - new_range_high = max(new_range_high, spread_high, point) + else: + _anchor_comps_pre, _anchor_tier_pre = [], None - logger.info( - "sb_anchor: tier=%s n=%d radius_median_ppm2=%d → anchor_asking_ppm2=%d" - " (uplift=%s haircut=%.2f) point %d → %d", - anchor_tier, - anchor["n"], - int(median_ppm2), - int(est_ppm2), - anchor["used_uplift"], - anchor["haircut"], - median_price, - point, - ) - median_ppm2 = new_ppm2 - median_price = point - range_low = new_range_low - range_high = new_range_high - confidence = anchor["confidence"] - tier_label = "того же дома" if anchor_tier == "A" else "ближайшего окружения (≤500 м)" - # #695: когда якорь построил headline, explanation описывает ИМЕННО якорные - # комплы (anchor['n'] из tier_label). Радиусный base-текст («Найдено N из M - # разных адресов») и analog_tier tier_note относятся к ДРУГОЙ выборке и - # противоречат якорю (n_analogs≠anchor['n']; «разных адресов» vs «того же - # дома») — поэтому ЗАМЕНЯЕМ, а не конкатенируем. repair_note сохраняем - # (ортогонален — поправка на ремонт). confidence уже = anchor["confidence"]. - explanation = ( - f"Оценка построена по {anchor['n']} аналогам из {tier_label}" - f"{' (топ-уровень в доме)' if anchor['used_uplift'] else ''}." - ) + repair_note - # #695 (QA fixup): когда якорь подменяет headline и список аналогов на - # комплы дома, n_analogs ДОЛЖЕН считаться по anchor-популяции, а не по - # радиусу (listings_clean). Иначе все UI-счётчики («Аналогов N», - # «Показано N из M», «по N аналогам», гистограмма) расходятся: radius=4 vs - # anchor=5 → артефакт «Показано 5 из 4». anchor['n'] = число комплов, на - # которых построена оценка (= показываемый analogs, capped 10). Для anchor- - # пути это и есть «полное число найденных» по контракту n_analogs (#698). - n_analogs = anchor["n"] - # #1871 P1 (ghost-anchor): same-building anchor заменил headline, но - # радиусных аналогов ноль (listings_clean пуст) → нельзя отдавать - # высокую/среднюю уверенность. Downgrade до 'low' + честный caveat. - # _enforce_zero_analog_low ниже это не ловит: n_analogs уже перезаписан - # на anchor["n"] (>0), а не 0. - if not listings_clean and confidence != "low": - logger.warning( - "estimator #1871 ghost-anchor guard: confidence %s→low " - "(anchor_n=%s, radius_analogs=0)", - confidence, - anchor["n"], - ) - confidence = "low" - explanation += ( - " Оценка опирается только на аналоги из того же дома — " - "сопоставимых предложений поблизости не найдено, поэтому " - "точность ориентировочная." - ) - # ── #1871 P2: split-дома wide-corridor disclosure (за флагом, OFF) ── - # Tier A матчит по address-regex (намеренно НЕ house_id — дом дробится - # на несколько house_id). На split-доме разной этажности comp_min..max - # растягивается через несколько ценовых режимов → коридор 148%/170%. - # Коридор честно широкий (НЕ сужаем); только понижаем confidence на - # ступень и дописываем disclosure. НЕ трогаем point/median/range. - if ( - settings.estimate_wide_corridor_disclosure_enabled - and anchor_tier == "A" - and median_price > 0 - ): - corridor_pct = (range_high - range_low) / median_price - if corridor_pct > settings.estimate_wide_corridor_threshold: - confidence = _downgrade_confidence(confidence) - explanation = (explanation or "") + ( - " Очень широкий ценовой диапазон по дому (вероятно, дом " - "разбит на секции разной этажности или разнородный фонд) — " - "оценка ориентировочная, уточните по конкретной секции." - ) + # ── Pre-fetch: house IMV anchor ONCE (used for both blend and display) ─── + imv_anchor_data = _fetch_house_imv_anchor( + db, + target_house_id=target_house_id, + rooms=payload.rooms, + area=payload.area_m2, + ) - # ── #651: IMV / Yandex BLEND (killer accuracy fix) — SECONDARY, Tier D only ── - # Радиусная медиана системно занижает премиум/видовые квартиры (нет class/ - # segment-коррекции). Берём РЕАЛЬНЫЙ Avito IMV target-дома из house_imv_evaluations - # (avito_imv_evaluations пуст — keyed estimate_id, on-demand), используем как - # anchor: если IMV recommended_price > median × threshold — поднимаем медиану - # blend'ом и расширяем верх диапазона. Всё за флагом + null-guard (no-op без IMV). - # ВАЖНО (v2): IMV-blend выполняется ТОЛЬКО когда same-building anchor НЕ сработал - # (anchor_tier is None) — не накладываем blend поверх уже-построенного якоря дома. - # #764: imv_anchor_present — любой IMV-anchor повлиял на estimate (median OR range). - # Guard-1b использует этот флаг чтобы пропустить квартальный индекс при любом - # IMV-влиянии, не только при blended (range_high расширяется даже без blend). - imv_anchor_present: bool = False - avito_imv_summary: AvitoImvSummary | None = None - if ( - anchor_tier is None - and settings.estimate_imv_blend_enabled - and listings_clean - and median_price > 0 - ): - imv_anchor = _fetch_house_imv_anchor( - db, - target_house_id=target_house_id, - rooms=payload.rooms, - area=payload.area_m2, - ) - # Anchor chain: prefer Avito IMV recommended; fall back to on-demand imv_eval. - anchor_total: int | None = None - anchor_higher: int | None = None - anchor_label: str | None = None - if imv_anchor is not None and imv_anchor.get("recommended_price"): - anchor_total = int(imv_anchor["recommended_price"]) - anchor_higher = ( - int(imv_anchor["higher_price"]) if imv_anchor.get("higher_price") else None - ) - anchor_label = "оценке Avito IMV" - _imv_mc = int(imv_anchor["market_count"]) if imv_anchor.get("market_count") else None - avito_imv_summary = AvitoImvSummary( - recommended_price=anchor_total, - lower_price=( - int(imv_anchor["lower_price"]) if imv_anchor.get("lower_price") else None - ), - higher_price=anchor_higher, - market_count=_imv_mc, - # #audit-5b: тонкий рынок — малая выборка для IMV. - thin_market=( - _imv_mc is not None and _imv_mc < settings.avito_imv_thin_market_threshold - ), - ) - elif imv_eval is not None and imv_eval.recommended_price: - # on-demand IMV (avito_imv_evaluations) — fallback, обычно пуст - anchor_total = int(imv_eval.recommended_price) - anchor_higher = int(imv_eval.higher_price) if imv_eval.higher_price else None - anchor_label = "оценке Avito IMV" - avito_imv_summary = AvitoImvSummary( - recommended_price=anchor_total, - lower_price=int(imv_eval.lower_price) if imv_eval.lower_price else None, - higher_price=anchor_higher, - market_count=imv_eval.market_count, - # #audit-5b: тонкий рынок — market_count < threshold. - thin_market=( - imv_eval.market_count is not None - and imv_eval.market_count < settings.avito_imv_thin_market_threshold - ), - ) - - # #audit-5b: thin-market warning — IMV на малой выборке → reliability ↓. - if avito_imv_summary is not None and avito_imv_summary.thin_market: - logger.warning( - "avito_imv thin_market #audit-5b: market_count=%s" - " (< avito_imv_thin_market_threshold=%d) — IMV reliability low", - avito_imv_summary.market_count, - settings.avito_imv_thin_market_threshold, - ) - - if anchor_total is not None: - imv_anchor_present = True - new_median, new_range_high, new_ppm2, blended, anchor_used = _apply_imv_blend( - median_price=median_price, - range_high=range_high, - median_ppm2=median_ppm2, - area=payload.area_m2, - anchor_total=anchor_total, - anchor_higher=anchor_higher, - weight=settings.estimate_imv_blend_weight, - threshold=settings.estimate_imv_blend_threshold, - ) - if blended: - logger.info( - "imv_blend: median %d → %d (anchor=%d w=%.2f) range_high %d → %d", - median_price, - new_median, - anchor_used, - settings.estimate_imv_blend_weight, - range_high, - new_range_high, - ) - median_price = new_median - median_ppm2 = new_ppm2 - explanation = (explanation or "") + ( - f" Оценка скорректирована по {anchor_label} " - f"({anchor_used / 1_000_000:.1f} млн ₽)." - ) - sources_used_pre = sorted(set(sources_used_pre) | {"avito_imv"}) - # Диапазон расширяем даже если медиану не двигали (информативность). - range_high = new_range_high - - # Display-only Avito IMV summary, когда headline построен same-building якорем - # (IMV-blend выше пропущен). Якорь дома — primary; IMV остаётся cross-check в UI. - if anchor_tier is not None and avito_imv_summary is None: - imv_anchor_disp = _fetch_house_imv_anchor( - db, - target_house_id=target_house_id, - rooms=payload.rooms, - area=payload.area_m2, - ) - if imv_anchor_disp is not None and imv_anchor_disp.get("recommended_price"): - _disp_mc = ( - int(imv_anchor_disp["market_count"]) - if imv_anchor_disp.get("market_count") - else None - ) - avito_imv_summary = AvitoImvSummary( - recommended_price=int(imv_anchor_disp["recommended_price"]), - lower_price=( - int(imv_anchor_disp["lower_price"]) - if imv_anchor_disp.get("lower_price") - else None - ), - higher_price=( - int(imv_anchor_disp["higher_price"]) - if imv_anchor_disp.get("higher_price") - else None - ), - market_count=_disp_mc, - # #audit-5b: тонкий рынок. - thin_market=( - _disp_mc is not None and _disp_mc < settings.avito_imv_thin_market_threshold - ), - ) - - # ── #764: per-cadastral-quarter price index gap-correction ────────────────── - # Применяется ТОЛЬКО в pure-radius пути (Guard-1): когда same-building anchor - # не сработал И IMV-blend не поднял медиану. Оба механизма уже учитывают - # location пространственно — наложение индекса сверху даёт double-count. - # Формула: adjusted_ppm2 = base_ppm2 × (target_index / avg_analog_index), - # где avg_analog_index = взвешенная по ppm² медиана аналогов, чьи кварталы - # известны. Если аналоги без кадастрового номера — avg_analog_index=1.0 (no-op). - if ( - settings.estimate_quarter_index_enabled - and anchor_tier is None # Guard-1a: same-building anchor не сработал - and not imv_anchor_present # Guard-1b: IMV-anchor не повлиял (median или range) - and median_price > 0 - and payload.area_m2 - ): - # Резолвим квартал target'а: Primary — DaData house_cadnum. - target_quarter: str | None = _quarter_from_cadastre( - dadata.house_cadnum if dadata is not None else None - ) - # Fallback: building_cadastral_number из самих аналогов (если все в 1 доме - # — Tier S path; тогда кадастровый номер квартала тот же). Не применяем - # PostGIS point-in-quarter: нет готовой geometry-таблицы кварталов в tradein DB. - if target_quarter is None: - for lot in listings_clean: - cq = _quarter_from_cadastre(lot.get("building_cadastral_number")) - if cq is not None: - target_quarter = cq - break - - if target_quarter is not None: - qindex_result = _lookup_quarter_index( - db, - quarter_cad_number=target_quarter, - min_n_deals=settings.estimate_quarter_index_min_n_deals, - ) - if qindex_result is not None: - target_qi, target_n_deals = qindex_result - - # Bimodal/nominal guard (Guard-4): структурно неоднородный квартал - # при малой выборке → no-op (regr. Радищева 66:41:0401017 et al.) - if ( - target_qi > settings.estimate_quarter_index_max_for_small_n - and target_n_deals < settings.estimate_quarter_index_small_n_threshold - ): - logger.info( - "quarter_index: bimodal guard triggered " - "(index=%.3f n=%d < %d) for %s — no-op", - target_qi, - target_n_deals, - settings.estimate_quarter_index_small_n_threshold, - target_quarter, - ) - else: - # Вычисляем квартал каждого аналога ОДИН РАЗ — переиспользуем - # для Guard-2 (same-quarter ratio) и avg_analog_index weighting. - # lot_quarters_for_guard2: все лоты с известным кварталом (как раньше). - # analog_quarters: только лоты с известным кварталом И ценой (для весов). - lot_quarters_for_guard2: list[str] = [] - analog_quarters: list[tuple[str, float]] = [] - for lot in listings_clean: - lq = _quarter_from_cadastre(lot.get("building_cadastral_number")) - if lq is None: - continue - lot_quarters_for_guard2.append(lq) - lp = lot.get("price_per_m2") - if lp: - analog_quarters.append((lq, float(lp))) - - # Guard-2: доля аналогов ИЗ ТОГО ЖЕ квартала > skip_ratio → - # location уже в медиане — пропускаем. - same_quarter_count = sum( - 1 for lq in lot_quarters_for_guard2 if lq == target_quarter - ) - # #1385: знаменатель — ТОЛЬКО классифицируемые аналоги (с разрешённым - # кварталом), как и числитель. Аналоги без кадастра (типичные анонимные - # Avito) не классифицируемы и не должны разбавлять ratio — иначе при - # низком покрытии кадастром ratio структурно занижен, Guard-2 не - # срабатывает, и quarter-index применяется к медиане повторно. - same_quarter_ratio = ( - same_quarter_count / len(lot_quarters_for_guard2) - if lot_quarters_for_guard2 - else 0.0 - ) - if same_quarter_ratio > settings.estimate_quarter_match_skip_ratio: - logger.info( - "quarter_index: Guard-2 skip (same-quarter ratio=%.2f > %.2f)" - " for %s", - same_quarter_ratio, - settings.estimate_quarter_match_skip_ratio, - target_quarter, - ) - else: - # Вычисляем avg_analog_index — ppm²-взвешенное среднее по - # аналогам, чьи кварталы известны И присутствуют в индексе. - # Один батч-запрос вместо N последовательных FDW roundtrips. - # Аналоги без кадастрового номера — игнорируем (не штрафуем). - distinct_analog_quarters = list( - dict.fromkeys(lq for lq, _lp in analog_quarters) - ) - analog_index_map = _lookup_quarter_indexes( - db, - quarter_cad_numbers=distinct_analog_quarters, - min_n_deals=settings.estimate_quarter_index_min_n_deals, - ) - weighted_sum = 0.0 - weight_total = 0.0 - for lq, lp in analog_quarters: - lot_qi = analog_index_map.get(lq) - if lot_qi is None: - continue - weighted_sum += lp * lot_qi - weight_total += lp - - avg_analog_index = weighted_sum / weight_total if weight_total > 0 else 1.0 - - ( - median_ppm2, - median_price, - range_low, - range_high, - qi_factor, - ) = _apply_quarter_index( - base_median_ppm2=median_ppm2, - base_median_price=median_price, - base_range_low=range_low, - base_range_high=range_high, - target_index=target_qi, - avg_analog_index=avg_analog_index, - min_factor=settings.estimate_quarter_index_factor_min, - max_factor=settings.estimate_quarter_index_factor_max, - ) - analogs_with_qi = sum( - 1 for lq, _lp in analog_quarters if lq in analog_index_map - ) - logger.info( - "quarter_index: applied target=%s target_qi=%.3f" - " avg_analog_qi=%.3f factor=%.3f" - " (same_quarter_ratio=%.2f analogs_with_qi=%d)", - target_quarter, - target_qi, - avg_analog_index, - qi_factor, - same_quarter_ratio, - analogs_with_qi, - ) - explanation = (explanation or "") + ( - f" Учтена локация квартала" f" (индекс цен квартала ×{qi_factor:.2f})." - ) - sources_used_pre = sorted(set(sources_used_pre) | {"quarter_index"}) - - # ── #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 "") + ( - " Оценка ограничена коридором реальных сделок Росреестра по улице." - ) - - # ── FIX: radius-path нижний floor от DKP-коридора (анти-undershoot) ──────── - # Anchor-путь имеет hard floor (comp_min_ppm2×(1-tol)) — radius-путь аналогичной - # защиты снизу не имел: quarter-index-вниз или corridor-clamp могли занизить - # median неоправданно (асимметрия с anchor-path). Если dkp_raw доступен и - # итоговый median_ppm2 < dkp_low_ppm2 × factor → поднимаем до floor и логируем. - # Только radius-путь (anchor_tier is None); без dkp_raw → no-op. - # За флагом estimate_radius_floor_enabled (дефолт True). - if ( - settings.estimate_radius_floor_enabled - and anchor_tier is None - and dkp_raw is not None - and dkp_raw.get("low_ppm2", 0) > 0 - and median_ppm2 > 0 - and dkp_raw.get("count", 0) >= settings.estimate_corridor_clamp_min_n - ): - radius_floor_ppm2 = float(dkp_raw["low_ppm2"]) * settings.estimate_radius_floor_factor - if median_ppm2 < radius_floor_ppm2: - floor_factor = radius_floor_ppm2 / median_ppm2 - logger.info( - "radius_floor: median_ppm2=%d < dkp_low=%d × factor=%.2f = floor=%d" - " → lifting (factor=%.3f)", - int(median_ppm2), - dkp_raw["low_ppm2"], - settings.estimate_radius_floor_factor, - int(radius_floor_ppm2), - floor_factor, - ) - median_ppm2 = radius_floor_ppm2 - median_price = round(median_price * floor_factor) - range_low = round(range_low * floor_factor) - range_high = round(range_high * floor_factor) - explanation = (explanation or "") + ( - " Оценка поднята до нижней границы коридора реальных сделок Росреестра." - ) - - # 4c (cont.). expected_sold_* выводим ЗДЕСЬ — ПОСЛЕ всех headline-мутаций: - # #651 IMV-blend / SB-anchor / quarter-index / #1795 corridor-clamp / radius-floor. - # FIX(ratio-tier-mismatch): _get_asking_sold_ratio вызывается с ФИНАЛЬНЫМ - # median_ppm2 (а не с pre-anchor radius-медианой). Это устраняет tier-несовпадение - # когда anchor/quarter-index двигают headline в другой ppm2-tier. - # Сохраняем per_rooms/global_fallback семантику ratio_basis из самой функции. - # Graceful: нет ratio (нет migration-080 строки / БД-ошибка) → expected_sold_* - # остаются None → UI не показывает «−N%» badge (не фабрикуем). - if median_ppm2 > 0: - asking_to_sold_ratio, ratio_basis = _get_asking_sold_ratio( - db, payload.rooms, anchor_ppm2=median_ppm2 - ) - # Не было ratio — не вводим в заблуждение пустым basis. - if asking_to_sold_ratio is None: - ratio_basis = None - - if asking_to_sold_ratio is not None and median_price > 0: - # FIX(expected_sold-le-asking): clamp ratio <= 1.0 чтобы expected_sold не - # превышала объявление. Smoke: 3к/27.0М -> expected_sold 31.4М > asking. - # Причина: high-price tier ratio > 1.0 (product artefact, не реальные сделки - # выше прайса). Флаг OFF -> старое поведение без clamp. - effective_ratio = asking_to_sold_ratio - if settings.estimate_expected_sold_le_asking and effective_ratio > 1.0: - logger.info( - "expected_sold ratio clamped %.3f->1.0 (rooms=%s)", - effective_ratio, - payload.rooms, - ) - effective_ratio = 1.0 - expected_sold_per_m2 = round(median_ppm2 * effective_ratio) - expected_sold_price = round(median_price * effective_ratio) - expected_sold_range_low = round(range_low * effective_ratio) - expected_sold_range_high = round(range_high * effective_ratio) - - # ── #652: ДКП-коридор реальных сделок (ADVISORY + soft sanity-bound) ───── - # dkp_raw уже зафетчен выше (#1795) — переиспользуем, не фетчим повторно. - dkp_corridor: DkpCorridor | None = None - if dkp_raw is not None: - dkp_corridor = DkpCorridor(**dkp_raw) - # Soft sanity-bound: если итоговая медиана ₽/м² заметно вне коридора — - # текстовая пометка. При сработавшем 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 "") + ( - " Оценка выше коридора реальных сделок Росреестра по улице." - ) - elif median_ppm2 < dkp_raw["low_ppm2"] * (1.0 - slack): - explanation = (explanation or "") + ( - " Оценка ниже коридора реальных сделок Росреестра по улице." - ) - - # ── #693: coarse-geo downgrade ────────────────────────────────────────── - # Когда DaData дала только грубую точность (settlement/city/region/unknown, - # qc_geo >= 2), PostGIS-радиус крутится вокруг центроида НП и собирает - # аналоги из случайных частей города → уверенная «medium» вводит в - # заблуждение. Помечаем как ориентировочную (low). КОНСЕРВАТИВНО: - # 1) есть позитивный сигнал грубости (dadata.qc_geo >= 2 ИЛИ _geocode_is_coarse); - # 2) НЕ сработал якорь «ТОГО ЖЕ ДОМА» (Tier A) — он стоит на реальных комплах - # дома (#691 path) и геоцентроид не важен. ВАЖНО (#693 QA-fail 2/3): Tier C - # (micro-radius ≤500 м) — это НЕ «тот же дом», а случайные квартиры в 500 м - # вокруг геокода. Если геокод грубый (мусор → ЕКБ-центроид), Tier C набирает - # ≥5 комплов в плотном центре и РАНЬШЕ блокировал downgrade (guard был - # `anchor_tier is None`) → канонический кейс «фывапролд 999» оставался medium. - # Теперь защищаем ТОЛЬКО Tier A; Tier C при грубом геокоде — тоже понижаем. - # 3) есть радиусное число (median_price > 0) — что квалифицировать. - # ИСТОЧНИКИ сигнала грубости (OR): dadata.qc_geo>=2 (если DaData активна) ИЛИ - # _geocode_is_coarse(geo) — прод-фолбэк, работает БЕЗ DaData. + # ── Coarse-geo signals ──────────────────────────────────────────────────── dadata_coarse = dadata is not None and dadata.qc_geo is not None and dadata.qc_geo >= 2 - geo_coarse = _geocode_is_coarse(geo) - if (dadata_coarse or geo_coarse) and median_price > 0: - # #693 QA diag (2/3): на проде QA просил залогировать сигналы геокода для - # негеокодируемых адресов — фиксируем, что увидел гейт. - logger.info( - "coarse-geo gate #693: dadata_coarse=%s geo_coarse=%s anchor_tier=%s " - "median=%s geo.provider=%s geo.confidence=%s geo.full_address=%r geo=(%.5f,%.5f)", - dadata_coarse, - geo_coarse, - anchor_tier, - median_price, - geo.provider, - geo.confidence, - geo.full_address, - geo.lat, - geo.lon, + + # ── DB-callable wrappers injected into pure pricing ─────────────────────── + def _ratio_resolver( + appm2: float | None, + ) -> tuple[float | None, str | None]: + return _get_asking_sold_ratio(db, payload.rooms, anchor_ppm2=appm2) + + def _qi_lookup(q: str) -> tuple[float, int] | None: + return _lookup_quarter_index( + db, + quarter_cad_number=q, + min_n_deals=settings.estimate_quarter_index_min_n_deals, ) - if (dadata_coarse or geo_coarse) and anchor_tier != "A" and median_price > 0: - # Уровень грубости берём из DaData (точнее), иначе generic «населённого пункта». - if dadata_coarse: - _coarse_label = {2: "населённого пункта", 3: "города", 4: "региона"}.get( - dadata.qc_geo, "населённого пункта" - ) - else: - _coarse_label = "населённого пункта" - confidence = "low" - explanation = (explanation or "") + ( - f" Адрес определён лишь до уровня {_coarse_label} — точные координаты " - "дома найти не удалось, поэтому оценка ориентировочная (аналоги взяты " - "по широкой окрестности)." + + def _qis_lookup(qs: list[str]) -> dict[str, float]: + return _lookup_quarter_indexes( + db, + quarter_cad_numbers=qs, + min_n_deals=settings.estimate_quarter_index_min_n_deals, ) + # ── Deterministic pricing orchestration ────────────────────────────────── + pr = _price_from_inputs( + listings=listings, + area_m2=payload.area_m2, + rooms=payload.rooms, + repair_state=payload.repair_state, + floor=payload.floor, + total_floors=payload.total_floors, + target_year=target_year, + analog_tier=analog_tier, + fallback_used=fallback_used, + area_widened=area_widened, + anchor_comps=_anchor_comps_pre, + anchor_tier_fetched=_anchor_tier_pre, + dkp_raw=dkp_raw, + imv_anchor=imv_anchor_data, + imv_eval=imv_eval, + yandex_val_present=yandex_val is not None, + cian_val_present=cian_val is not None and bool(cian_val.sale_price_rub), + ratio_resolver=_ratio_resolver, + quarter_index_lookup=_qi_lookup, + quarter_indexes_lookup=_qis_lookup, + target_house_cadnum=dadata.house_cadnum if dadata else None, + dadata_coarse=dadata_coarse, + geo=geo, + dadata_qc_geo=dadata.qc_geo if dadata else None, + ) + + # Unpack pricing result + median_price = pr.median_price + median_ppm2 = pr.median_ppm2 + range_low = pr.range_low + range_high = pr.range_high + n_analogs = pr.n_analogs + confidence = pr.confidence + explanation = pr.explanation + anchor_tier = pr.anchor_tier + anchor_comps_used = pr.anchor_comps_used + avito_imv_summary = pr.avito_imv_summary + dkp_corridor = pr.dkp_corridor + expected_sold_per_m2 = pr.expected_sold_per_m2 + expected_sold_price = pr.expected_sold_price + expected_sold_range_low = pr.expected_sold_range_low + expected_sold_range_high = pr.expected_sold_range_high + asking_to_sold_ratio = pr.asking_to_sold_ratio + ratio_basis = pr.ratio_basis + sources_used_pre = pr.sources_used_pre + listings_clean = pr.listings_clean + # 5. Deals — ДКП-only sales (вторичка) из rosreestr_deals. # Importer фильтрует doc_type='ДКП' (PR-A 2026-05-24), ДДУ застройщиков # исключены — больше не скёюят median вторички ~110-120 К/м². -- 2.45.3 From 2ea49b637f1ef70936242319767efe4f34405708 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 27 Jun 2026 13:00:21 +0300 Subject: [PATCH 2/2] refactor(tradein/estimator): extract deterministic pricing into pure _price_from_inputs (#1966 foundation) Behavior-preserving structural refactor: the ~870-line deterministic pricing block inside estimate_quality() is extracted into a new pure synchronous function _price_from_inputs() returning a PricingResult dataclass. Key design decisions: - All async DB fetches (imv_eval, yandex_val, cian_val) hoisted to estimate_quality BEFORE the call; passed as pre-fetched values / bool flags. - DB-dependent helpers whose arguments are computed inside the block (_get_asking_sold_ratio, _lookup_quarter_index, _lookup_quarter_indexes) injected as Callable parameters (ratio_resolver, quarter_index_lookup, quarter_indexes_lookup). - _fetch_house_imv_anchor called once in estimate_quality (was two separate conditional calls inside the block); single result passed as imv_anchor dict. - _fetch_dkp_corridor and _fetch_anchor_comps hoisted to estimate_quality with identical guards. All 2443 existing tests pass UNMODIFIED. 9 new hermetic unit tests added that call _price_from_inputs directly with stub callables. --- .../tests/test_estimator_price_spine.py | 401 ++++++++++++++++++ 1 file changed, 401 insertions(+) create mode 100644 tradein-mvp/backend/tests/test_estimator_price_spine.py diff --git a/tradein-mvp/backend/tests/test_estimator_price_spine.py b/tradein-mvp/backend/tests/test_estimator_price_spine.py new file mode 100644 index 00000000..cebee96b --- /dev/null +++ b/tradein-mvp/backend/tests/test_estimator_price_spine.py @@ -0,0 +1,401 @@ +"""Hermetic unit tests for _price_from_inputs (#1966). + +Calls the pure synchronous pricing function directly with stub callables and +hand-built inputs — no DB, no network, no mocks. Verifies that the extraction +preserved the pricing logic identically to the original block in estimate_quality. + +NOTE: importing app.services.estimator pulls app.core.config.Settings which +requires DATABASE_URL. Set it BEFORE importing app modules. +""" + +import os + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services import estimator +from app.services.geocoder import GeocodeResult + +# ── helpers ────────────────────────────────────────────────────────────────── + + +def _geo(coarse: bool = False) -> GeocodeResult: + """Minimal GeocodeResult for test injection.""" + full_address = "Екатеринбург" if coarse else "ул. Тестовая, 1" + return GeocodeResult( + lat=56.838, + lon=60.597, + full_address=full_address, + provider="nominatim", + confidence="approximate", + ) + + +def _lot(ppm2: float, address: str = "ул. Тестовая, 1", source: str = "avito") -> dict: + return {"price_per_m2": ppm2, "address": address, "source": source} + + +def _lots(ppm2: float, n: int = 7) -> list[dict]: + """n unique-address lots all at the same ppm2.""" + return [_lot(ppm2, address=f"ул. Тестовая, {i + 1}") for i in range(n)] + + +def _dkp_raw( + low: int = 80_000, + median: int = 120_000, + high: int = 150_000, + count: int = 20, + period_months: int = 12, +) -> dict: + return { + "low_ppm2": low, + "median_ppm2": median, + "high_ppm2": high, + "count": count, + "period_months": period_months, + } + + +def _anchor_comp(ppm2: float, area: float = 50.0, rooms: int = 2) -> dict: + return {"price_per_m2": ppm2, "area_m2": area, "rooms": rooms} + + +# Stub callables — returned in each test via closure. +def _ratio_stub( + ratio: float | None, + basis: str | None = "per_rooms", +) -> "tuple[float | None, str | None]": + return ratio, basis if ratio is not None else None + + +def _qi_stub_none(q: str) -> "tuple[float, int] | None": + return None + + +def _qis_stub_empty(qs: list[str]) -> dict[str, float]: + return {} + + +def _call( + *, + listings: list[dict] | None = None, + area_m2: float = 50.0, + rooms: int | None = 2, + repair_state: str | None = None, + floor: int | None = 5, + total_floors: int | None = 10, + target_year: int | None = None, + analog_tier: str = "W", + fallback_used: bool = False, + area_widened: bool = False, + anchor_comps: list[dict] | None = None, + anchor_tier_fetched: str | None = None, + dkp_raw: dict | None = None, + imv_anchor: dict | None = None, + imv_eval=None, + yandex_val_present: bool = False, + cian_val_present: bool = False, + ratio: float | None = None, + quarter_index_lookup=None, + quarter_indexes_lookup=None, + target_house_cadnum: str | None = None, + dadata_coarse: bool = False, + geo: GeocodeResult | None = None, + dadata_qc_geo: int | None = None, +) -> estimator.PricingResult: + if listings is None: + listings = _lots(100_000) + if anchor_comps is None: + anchor_comps = [] + if geo is None: + geo = _geo(coarse=dadata_coarse) + if quarter_index_lookup is None: + quarter_index_lookup = _qi_stub_none + if quarter_indexes_lookup is None: + quarter_indexes_lookup = _qis_stub_empty + + _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 if _ratio is not None else None + + return estimator._price_from_inputs( + listings=listings, + area_m2=area_m2, + rooms=rooms, + repair_state=repair_state, + floor=floor, + total_floors=total_floors, + target_year=target_year, + analog_tier=analog_tier, + fallback_used=fallback_used, + area_widened=area_widened, + anchor_comps=anchor_comps, + anchor_tier_fetched=anchor_tier_fetched, + dkp_raw=dkp_raw, + imv_anchor=imv_anchor, + imv_eval=imv_eval, + yandex_val_present=yandex_val_present, + cian_val_present=cian_val_present, + ratio_resolver=ratio_resolver, + quarter_index_lookup=quarter_index_lookup, + quarter_indexes_lookup=quarter_indexes_lookup, + target_house_cadnum=target_house_cadnum, + dadata_coarse=dadata_coarse, + geo=geo, + dadata_qc_geo=dadata_qc_geo, + ) + + +# ── Tests ──────────────────────────────────────────────────────────────────── + + +def test_radius_only_median_and_expected_sold() -> None: + """Pure radius path: 7 uniform lots → correct median, n_analogs, expected_sold.""" + pr = _call(listings=_lots(100_000, n=7), ratio=0.95) + + assert pr.median_price == int(100_000 * 50.0) # 5_000_000 + assert pr.median_ppm2 == 100_000.0 + assert pr.n_analogs == 7 + assert pr.anchor_tier is None + assert pr.dkp_corridor is None + assert pr.asking_to_sold_ratio == 0.95 + assert pr.expected_sold_price == round(5_000_000 * 0.95) # 4_750_000 + assert pr.expected_sold_per_m2 == round(100_000 * 0.95) # 95_000 + assert "avito" in pr.sources_used_pre + assert len(pr.listings_clean) == 7 + + +def test_same_building_anchor_tier_a_mutates_headline() -> None: + """Tier A same-building anchor replaces radius median with higher price. + + 5 radius lots at 100k ppm2 (5M total). 5 anchor comps at 200k ppm2 (10M total). + After anchor fires: median_price >> radius median, n_analogs == anchor count. + """ + comps = [_anchor_comp(200_000) for _ in range(5)] + pr = _call( + listings=_lots(100_000, n=5), + anchor_comps=comps, + anchor_tier_fetched="A", + ratio=None, + ) + + # Anchor must have fired (not suppressed). + assert pr.anchor_tier == "A" + # Headline is anchor-derived — must be above radius median (5_000_000). + assert pr.median_price > 5_000_000 + # n_analogs resets to anchor population. + assert pr.n_analogs == 5 + # anchor_comps_used is the injected comps list. + assert len(pr.anchor_comps_used) == 5 + # No ratio → expected_sold is None. + assert pr.expected_sold_price is None + + +def test_tier_c_corridor_gate_suppresses_anchor() -> None: + """Tier C anchor ppm2 >> corridor_high × mult → anchor suppressed. + + anchor_tier remains "C" in the result (gate sets anchor=None but doesn't + reset anchor_tier); headline stays at the radius median. + """ + # 5 comps at 300k ppm2; corridor_high=150k; gate threshold=150k×1.5=225k. + # 300k > 225k → suppressed. + comps = [_anchor_comp(300_000) for _ in range(5)] + radius_median_price = int(100_000 * 50.0) + pr = _call( + listings=_lots(100_000, n=5), + anchor_comps=comps, + anchor_tier_fetched="C", + dkp_raw=_dkp_raw(high=150_000, count=15), + ratio=None, + ) + + # Tier C gate sets anchor=None but leaves anchor_tier="C". + assert pr.anchor_tier == "C" + # Headline was NOT mutated by the suppressed anchor — stays at radius median. + assert pr.median_price == radius_median_price + # anchor_comps_used stays empty (anchor didn't fire). + assert pr.anchor_comps_used == [] + + +def test_low_conf_gate_suppresses_anchor() -> None: + """Low-confidence anchor is suppressed; anchor_tier reset to None. + + 4 comps with wide spread → high cv → fsd > 0.20 → confidence='low' → suppressed. + """ + # 4 comps at [100k, 200k, 300k, 400k] → cv≈0.45, fsd≈0.20 → "low". + comps = [_anchor_comp(p) for p in [100_000, 200_000, 300_000, 400_000]] + radius_median_price = int(100_000 * 50.0) + pr = _call( + listings=_lots(100_000, n=5), + anchor_comps=comps, + anchor_tier_fetched="A", # starts as A, gate resets to None + ratio=None, + ) + + # Gate resets anchor_tier to None on suppression. + assert pr.anchor_tier is None + # Headline stays at radius median. + assert pr.median_price == radius_median_price + assert pr.anchor_comps_used == [] + + +def test_imv_blend_raises_median_when_anchor_tier_none() -> None: + """IMV blend pushes radius median up when IMV >> median × threshold. + + radius median=5M, IMV recommended=7M, area=50, weight=0.5, threshold=1.15. + IMV/radius = 7M/5M = 1.4 > 1.15 → blend: new_median = round(5M×0.5 + 7M×0.5). + """ + imv_anchor = { + "recommended_price": 7_000_000, + "lower_price": 6_000_000, + "higher_price": 8_000_000, + "market_count": 50, + } + pr = _call( + listings=_lots(100_000, n=5), + imv_anchor=imv_anchor, + ratio=None, + ) + + expected_median = round(5_000_000 * 0.5 + 7_000_000 * 0.5) # 6_000_000 + assert pr.median_price == expected_median + assert pr.range_high == 8_000_000 # from anchor_higher + assert pr.avito_imv_summary is not None + assert pr.avito_imv_summary.recommended_price == 7_000_000 + assert "avito_imv" in pr.sources_used_pre + + +def test_quarter_index_guard2_skip_when_all_analogs_in_target_quarter() -> None: + """Guard-2: when all analogs are in the target quarter, index is NOT applied. + + same_quarter_ratio=1.0 > skip_ratio=0.6 → Guard-2 fires → median unchanged. + """ + target_cadnum = "66:41:0204016:350" + target_quarter = "66:41:0204016" + # Analogs are all in the SAME quarter as the target. + lots = [ + { + "price_per_m2": 100_000, + "address": f"ул. Тестовая, {i + 1}", + "source": "avito", + "building_cadastral_number": f"{target_quarter}:{100 + i}", + } + for i in range(5) + ] + + qi_called: list[str] = [] + + def qi_lookup(q: str) -> tuple[float, int] | None: + qi_called.append(q) + return (1.5, 100) if q == target_quarter else None # high index — would change price + + pr = _call( + listings=lots, + target_house_cadnum=target_cadnum, + quarter_index_lookup=qi_lookup, + quarter_indexes_lookup=_qis_stub_empty, + ratio=None, + ) + + # Guard-2 fired: median must remain unchanged (5M, not ×1.5). + assert pr.median_price == int(100_000 * 50.0) + # quarter_index was looked up but did NOT add "quarter_index" to sources. + assert "quarter_index" not in pr.sources_used_pre + + +def test_quarter_index_applied_when_analogs_in_different_quarter() -> None: + """Quarter-index IS applied when analogs are in a different quarter from target. + + target_qi=1.2, avg_analog_qi=1.0 → factor=1.2 → median_price×1.2. + Guard-2 skips (same_quarter_ratio=0.0 < 0.6). + """ + target_cadnum = "66:41:0204016:350" + target_quarter = "66:41:0204016" + analog_quarter = "66:41:0999999" + lots = [ + { + "price_per_m2": 100_000, + "address": f"ул. Иная, {i + 1}", + "source": "avito", + "building_cadastral_number": f"{analog_quarter}:{i + 1}", + } + for i in range(5) + ] + + def qi_lookup(q: str) -> tuple[float, int] | None: + return (1.2, 100) if q == target_quarter else None + + def qis_lookup(qs: list[str]) -> dict[str, float]: + return {q: 1.0 for q in qs if q == analog_quarter} + + pr = _call( + listings=lots, + target_house_cadnum=target_cadnum, + quarter_index_lookup=qi_lookup, + quarter_indexes_lookup=qis_lookup, + ratio=None, + ) + + # factor=1.2/1.0=1.2; original radius=5M → adjusted=6M (±rounding via _apply_quarter_index) + assert pr.median_price > int(100_000 * 50.0) # index pushed price up + assert "quarter_index" in pr.sources_used_pre + + +def test_corridor_soft_clamp_headline_above_cap() -> None: + """Headline above corridor_high × (1+slack) is clamped down. + + radius lots at 250k ppm2. corridor_high=150k, slack=0.40 → + cap=150k×1.40=210k. 250k > 210k → clamped to 210k. + """ + pr = _call( + listings=_lots(250_000, n=7), + dkp_raw=_dkp_raw(low=80_000, median=120_000, high=150_000, count=15), + ratio=None, + ) + + # cap = 150_000 × 1.40 = 210_000 + # Clamped: new ppm2 == 210_000, new_price = round(210_000 × 50) + assert pr.median_ppm2 == pytest.approx(210_000.0, rel=1e-4) + assert pr.median_price == round(210_000 * 50.0) + # DKP corridor present in result. + assert pr.dkp_corridor is not None + assert pr.dkp_corridor.high_ppm2 == 150_000 + + +def test_expected_sold_from_ratio_and_none_when_ratio_none() -> None: + """expected_sold = headline × ratio; when ratio is None, all expected_sold fields None.""" + # Case A: ratio=0.90 → expected_sold fields filled. + pr_ratio = _call(listings=_lots(100_000, n=5), ratio=0.90) + assert pr_ratio.asking_to_sold_ratio == 0.90 + assert pr_ratio.expected_sold_price is not None + assert pr_ratio.expected_sold_price == round(pr_ratio.median_price * 0.90) + assert pr_ratio.expected_sold_per_m2 is not None + assert pr_ratio.expected_sold_range_low is not None + assert pr_ratio.expected_sold_range_high is not None + + # Case B: ratio=None → all expected_sold fields None. + pr_none = _call(listings=_lots(100_000, n=5), ratio=None) + assert pr_none.asking_to_sold_ratio is None + assert pr_none.ratio_basis is None + assert pr_none.expected_sold_price is None + assert pr_none.expected_sold_per_m2 is None + assert pr_none.expected_sold_range_low is None + assert pr_none.expected_sold_range_high is None + + +def test_coarse_geo_downgrades_confidence_to_low() -> None: + """dadata_coarse=True with qc_geo=2 → confidence='low' with settlement label.""" + pr = _call( + listings=_lots(100_000, n=7), + dadata_coarse=True, + dadata_qc_geo=2, + ratio=None, + # No anchor, no IMV → radius path → anchor_tier is None (not "A" → downgrade applies) + geo=_geo(coarse=False), # geo itself not coarse; using dadata_coarse signal + ) + + assert pr.confidence == "low" + assert "населённого пункта" in pr.explanation -- 2.45.3