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:

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)