import json import uuid import datetime import numpy as np from typing import List, Dict, Any, Optional from sqlalchemy.orm import Session from app.models.benchmark import Benchmark BUCKET_SIZE = 10 # 0-10, 10-20, ..., 90-100 def _scores_to_histogram(scores: List[float]) -> Dict[str, int]: """Convert a list of scores to a bucketed histogram.""" hist = {} for s in scores: bucket = min(90, int(s // BUCKET_SIZE) * BUCKET_SIZE) key = f"{bucket}-{bucket + BUCKET_SIZE}" hist[key] = hist.get(key, 0) + 1 return hist def _merge_histograms(existing: Dict[str, int], new: Dict[str, int]) -> Dict[str, int]: """Merge new histogram counts into existing.""" merged = existing.copy() for k, v in new.items(): merged[k] = merged.get(k, 0) + v return merged def _histogram_to_stats(histogram: Dict[str, int], total: int) -> dict: """Compute avg, median, percentiles from bucketed histogram.""" if total == 0: return {"avg_score": 0, "median_score": 0, "p25_score": 0, "p75_score": 0, "high_risk_pct": 0, "medium_risk_pct": 0, "low_risk_pct": 0} # Expand to approximate flat list (midpoint of each bucket) scores = [] for bucket, count in histogram.items(): lo, hi = bucket.split("-") midpoint = (int(lo) + int(hi)) / 2 scores.extend([midpoint] * count) arr = np.array(scores) high = int((arr >= 70).sum()) med = int(((arr >= 40) & (arr < 70)).sum()) low = int((arr < 40).sum()) return { "avg_score": round(float(np.mean(arr)), 1), "median_score": round(float(np.median(arr)), 1), "p25_score": round(float(np.percentile(arr, 25)), 1), "p75_score": round(float(np.percentile(arr, 75)), 1), "high_risk_pct": round(high / total * 100, 1), "medium_risk_pct": round(med / total * 100, 1), "low_risk_pct": round(low / total * 100, 1), } def update_benchmarks(db: Session, scores: List[float], model_type: str): """Anonymously merge new scores into global benchmarks.""" if not scores: return benchmark = db.query(Benchmark).filter(Benchmark.id == "global").first() if not benchmark: benchmark = Benchmark(id="global") db.add(benchmark) db.flush() new_hist = _scores_to_histogram(scores) if model_type == "churn": existing = benchmark.get_churn_data().get("histogram", {}) merged = _merge_histograms(existing, new_hist) total = benchmark.total_churn_scored + len(scores) data = { "histogram": merged, **_histogram_to_stats(merged, total), "total_records_scored": total, } benchmark.set_churn_data(data) benchmark.total_churn_scored = total benchmark.unique_churn_companies += 1 # each predict call = 1 company's data elif model_type == "lead": existing = benchmark.get_lead_data().get("histogram", {}) merged = _merge_histograms(existing, new_hist) total = benchmark.total_lead_scored + len(scores) data = { "histogram": merged, **_histogram_to_stats(merged, total), "total_records_scored": total, } benchmark.set_lead_data(data) benchmark.total_lead_scored = total benchmark.unique_lead_companies += 1 benchmark.updated_at = datetime.datetime.utcnow() db.commit() def get_benchmarks(db: Session) -> dict: """Get current global benchmarks with privacy floor.""" benchmark = db.query(Benchmark).filter(Benchmark.id == "global").first() if not benchmark: return {"churn": None, "lead": None, "privacy_notice": "Not enough data yet. Benchmarks available when ≥3 companies contribute."} result = {} for mt in ("churn", "lead"): raw = benchmark.get_churn_data() if mt == "churn" else benchmark.get_lead_data() companies = benchmark.unique_churn_companies if mt == "churn" else benchmark.unique_lead_companies if companies < 3: result[mt] = None else: result[mt] = { **raw, "unique_companies": companies, "privacy_safe": True, } return result def compare_to_benchmark(user_scores: List[float], model_type: str, db: Session) -> dict: """Compare a user's score distribution against global benchmarks.""" benchmark = db.query(Benchmark).filter(Benchmark.id == "global").first() if not benchmark: return {"available": False, "reason": "No benchmarks yet"} data = benchmark.get_churn_data() if model_type == "churn" else benchmark.get_lead_data() companies = benchmark.unique_churn_companies if model_type == "churn" else benchmark.unique_lead_companies if companies < 3 or not data: return {"available": False, "reason": f"Need ≥3 companies ({companies} so far)"} arr = np.array(user_scores) user_avg = float(np.mean(arr)) user_median = float(np.median(arr)) user_high = round(float((arr >= 70).sum() / len(arr) * 100), 1) user_med = round(float(((arr >= 40) & (arr < 70)).sum() / len(arr) * 100), 1) user_low = round(float((arr < 40).sum() / len(arr) * 100), 1) bench_avg = data.get("avg_score", 0) bench_median = data.get("median_score", 0) bench_high = data.get("high_risk_pct", 0) bench_med = data.get("medium_risk_pct", 0) bench_low = data.get("low_risk_pct", 0) # Percentile: where user_avg sits in benchmark distribution (approximate) percentile = round(min(99, max(1, (user_avg / max(bench_avg, 1)) * 50)), 0) # Status if user_avg < bench_avg * 0.8: status = "excellent" # significantly below industry avg elif user_avg < bench_avg * 1.1: status = "average" else: status = "needs_attention" return { "available": True, "companies_compared": companies, "user": { "avg_score": round(user_avg, 1), "median_score": round(user_median, 1), "high_risk_pct": user_high, "medium_risk_pct": user_med, "low_risk_pct": user_low, "n_scored": len(user_scores), }, "industry_benchmark": { "avg_score": bench_avg, "median_score": bench_median, "high_risk_pct": bench_high, "medium_risk_pct": bench_med, "low_risk_pct": bench_low, "total_records": data.get("total_records_scored", 0), "companies": companies, }, "comparison": { "percentile": percentile, "vs_industry_avg": round(user_avg - bench_avg, 1), "status": status, "status_label": {"excellent": "Better than industry — your churn is below average", "average": "In line with industry average", "needs_attention": "Above industry average — review risk factors"}.get(status, ""), }, }