77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
"""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)
|