diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 3c9a84a9..4e2fc03e 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -565,6 +565,7 @@ async def estimate_quality( 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, ) explanation = (explanation or "") + repair_note @@ -1068,16 +1069,34 @@ def _compute_confidence( q3: float, fallback_radius_used: bool, area_widened: bool = False, + listings: list[dict] | None = None, ) -> tuple[str, str]: """Confidence + explanation string. - high — n≥10 AND IQR/median < 0.15 - medium — n≥5 OR IQR/median < 0.25 + Уровень определяется по количеству уникальных адресов, а не по raw n_analogs. + Это защищает от overstated confidence когда много лотов из одного здания + (например, MIN_ANALOGS_PER_SOURCE=5 + same-building bias). + + high — unique_addr ≥ 7 AND IQR/median < 0.15 + medium — unique_addr ≥ 4 OR (unique_addr ≥ 2 AND IQR/median < 0.25) low — иначе + + Downgrade на один уровень если avg_lots_per_addr > 2.5 (concentration bias). """ if median_ppm2 == 0: return "low", "Не найдено аналогов — попробуйте уточнить адрес или расширить параметры." + # Вычисляем метрики уникальных адресов + if listings: + unique_addrs = { + (lot.get("address") or "").strip().lower() for lot in listings if lot.get("address") + } + unique_addr_count = len(unique_addrs) + avg_lots_per_addr = n_analogs / max(unique_addr_count, 1) + else: + unique_addr_count = n_analogs # fallback: считаем каждый лот уникальным + avg_lots_per_addr = 1.0 + iqr = q3 - q1 iqr_pct = iqr / median_ppm2 if median_ppm2 > 0 else 1.0 notes = [] @@ -1087,22 +1106,32 @@ def _compute_confidence( notes.append("расширили допуск по площади до ±25%") fallback_note = f" ({', '.join(notes)} из-за нехватки данных)" if notes else "" - if n_analogs >= 10 and iqr_pct < 0.15: - return ( - "high", - f"Найдено {n_analogs} аналогов, разброс цены ±{int(iqr_pct * 100 / 2)}% от медианы{fallback_note}.", + # Базовый уровень по уникальным адресам + if unique_addr_count >= 7 and iqr_pct < 0.15: + base = "high" + elif unique_addr_count >= 4: + base = "medium" + elif unique_addr_count >= 2 and iqr_pct < 0.25: + base = "medium" + else: + base = "low" + + # Downgrade на один шаг если слишком много лотов сконцентрировано на малом числе адресов + if avg_lots_per_addr > 2.5 and base != "low": + downgrade_map = {"high": "medium", "medium": "low"} + downgraded = downgrade_map[base] + explanation = ( + f"Найдено {n_analogs} аналогов из {unique_addr_count} разных адресов, " + f"разброс цены ±{int(iqr_pct * 100 / 2)}% от медианы{fallback_note}. " + f"Снижена точность (≥2.5 лотов на адрес — возможен bias)." ) - # medium только если есть достаточно точек ИЛИ узкий разброс при ≥3 точках - if n_analogs >= 5 or (n_analogs >= 3 and iqr_pct < 0.25): - return ( - "medium", - f"Найдено {n_analogs} аналогов, разброс цены ±{int(iqr_pct * 100 / 2)}% от медианы{fallback_note}.", - ) - return ( - "low", - f"Только {n_analogs} аналог{'а' if 2 <= n_analogs <= 4 else 'ов' if n_analogs != 1 else ''}, " - f"разброс ±{int(iqr_pct * 100 / 2)}% — рекомендуется ручная проверка{fallback_note}.", + return downgraded, explanation + + explanation = ( + f"Найдено {n_analogs} аналогов из {unique_addr_count} разных адресов, " + f"разброс цены ±{int(iqr_pct * 100 / 2)}% от медианы{fallback_note}." ) + return base, explanation def _listing_to_analog(row: dict[str, Any]) -> AnalogLot: diff --git a/tradein-mvp/scripts/probe-confidence-distribution.py b/tradein-mvp/scripts/probe-confidence-distribution.py new file mode 100644 index 00000000..f3e1d635 --- /dev/null +++ b/tradein-mvp/scripts/probe-confidence-distribution.py @@ -0,0 +1,77 @@ +"""Probe confidence-related metrics across last N estimates. + +Usage: python scripts/probe-confidence-distribution.py [N=100] + +Prints distribution of: + - n_analogs per estimate + - unique addresses per estimate + - lots_per_address ratio (helps tune downgrade threshold) + - confidence label distribution +""" +import sys +from collections import Counter +from statistics import median, quantiles + +# Run inside tradein-mvp/backend with DATABASE_URL set +from sqlalchemy import create_engine, text +from app.core.config import settings + + +def main(limit: int = 100) -> None: + engine = create_engine(settings.database_url, future=True) + with engine.connect() as conn: + rows = conn.execute( + text(""" + SELECT id, n_analogs, confidence, analogs + FROM trade_in_estimates + WHERE n_analogs > 0 + ORDER BY created_at DESC + LIMIT :limit + """), + {"limit": limit}, + ).mappings().all() + + if not rows: + print("No estimates with analogs.") + return + + n_analogs_vals: list[float] = [] + unique_vals: list[float] = [] + ratios: list[float] = [] + conf_counter: Counter[str] = Counter() + for r in rows: + n = r["n_analogs"] + analogs = r["analogs"] or [] + unique = len( + {(a.get("address") or "").strip().lower() for a in analogs if a.get("address")} + ) + n_analogs_vals.append(n) + unique_vals.append(unique) + ratios.append(n / max(unique, 1)) + conf_counter[r["confidence"]] += 1 + + def stats(label: str, vals: list[float]) -> None: + if not vals: + print(f"{label}: empty") + return + vals_sorted = sorted(vals) + q = quantiles(vals_sorted, n=4) if len(vals_sorted) >= 4 else [vals_sorted[0]] * 3 + print( + f"{label}: n={len(vals)} min={min(vals):.2f} p25={q[0]:.2f} " + f"median={median(vals):.2f} p75={q[2]:.2f} max={max(vals):.2f}" + ) + + print(f"\n=== Distribution over last {len(rows)} estimates ===") + stats("n_analogs ", n_analogs_vals) + stats("unique_addresses", unique_vals) + stats("lots_per_address", ratios) + print(f"confidence : {dict(conf_counter)}") + p75_idx = int(len(ratios) * 0.75) + print( + f"\n-> p75 lots_per_address = {sorted(ratios)[p75_idx]:.2f} " + "(use as downgrade threshold)" + ) + + +if __name__ == "__main__": + main(int(sys.argv[1]) if len(sys.argv) > 1 else 100)