Deploy RevAI API

.dockerignore

@@ -0,0 +1,12 @@
+__pycache__
+*.pyc
+.env
+*.db
+revai.db
+*.log
+tmp/
+tests/
+test_*.py
+postman_collection.json
+sample_data/
+models/

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.12-slim
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app/ app/
+
+ENV RAPIDAPI_PROXY_SECRET=18bb8fa3b320c4e75d58116528f45f9e580a67970fbef4e390d9729842a155ca
+ENV DATABASE_URL=sqlite:///./revai.db
+ENV SECRET_KEY=revai-production-hf-secret-key-2026
+
+EXPOSE 7860
+HEALTHCHECK --interval=30s --timeout=5s CMD curl -f http://localhost:7860/v1/health || exit 1
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,34 @@
+---
+title: RevAI API
+emoji: 📊
+colorFrom: indigo
+colorTo: blue
+sdk: docker
+app_port: 7860
+pinned: false
+license: mit
+---
+
+# RevAI — Churn Prediction & Lead Scoring API
+
+AI-powered churn prediction and lead scoring API.
+
+### Quick Start
+```bash
+curl https://<username>-revai-api.hf.space/v1/health
+```
+
+### Features
+- Churn prediction with explainable heuristic rules
+- Lead scoring with priority tiers (Hot/Warm/Cold)
+- Custom XGBoost model training
+- Anonymized industry benchmarks
+- Support call audio analysis (Whisper + NLP sentiment)
+
+### Pricing
+Free tier: 100 predictions/month on RapidAPI.
+See full pricing at rapidapi.com.
+
+---
+
+Built with FastAPI + XGBoost + VADER Sentiment

app/config.py

@@ -0,0 +1,43 @@
+from pydantic_settings import BaseSettings
+from functools import lru_cache
+
+
+class Settings(BaseSettings):
+    app_name: str = "RevAI API"
+    version: str = "1.0.0"
+    debug: bool = True
+
+    database_url: str = "sqlite:///./revai.db"
+    secret_key: str = "change-me-in-production-use-a-long-random-string"
+    algorithm: str = "HS256"
+    access_token_expire_minutes: int = 1440
+
+    stripe_secret_key: str = ""
+    stripe_webhook_secret: str = ""
+
+    # RapidAPI integration
+    rapidapi_proxy_secret: str = ""
+
+    # Tier limits
+    free_predictions: int = 100
+    maker_predictions: int = 5000
+    growth_predictions: int = 50000
+    scale_predictions: int = 500000
+
+    free_models: int = 1
+    maker_models: int = 3
+    growth_models: int = 10
+    scale_models: int = 999
+
+    free_rpm: int = 60
+    maker_rpm: int = 100
+    growth_rpm: int = 500
+    scale_rpm: int = 2000
+
+    class Config:
+        env_file = ".env"
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    return Settings()

app/database.py

@@ -0,0 +1,24 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, DeclarativeBase
+from app.config import get_settings
+
+settings = get_settings()
+engine = create_engine(settings.database_url, connect_args={"check_same_thread": False})
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+def init_db():
+    from app.models import user, apikey, usage, mlmodel, benchmark  # noqa: F401
+    Base.metadata.create_all(bind=engine)

app/dependencies.py

@@ -0,0 +1,219 @@
+import hashlib
+import time
+import datetime
+import uuid
+from fastapi import Depends, HTTPException, Header, Security, Request
+from fastapi.security import APIKeyHeader
+from sqlalchemy.orm import Session
+from sqlalchemy import func
+from app.database import get_db
+from app.models.user import User
+from app.models.apikey import APIKey
+from app.models.usage import UsageRecord
+from app.config import get_settings
+
+settings = get_settings()
+api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
+
+RAPIDAPI_PLAN_MAP = {
+    "BASIC": "free", "FREE": "free",
+    "MAKER": "maker", "PRO": "maker",
+    "GROWTH": "growth",
+    "SCALE": "scale", "ULTRA": "scale", "MEGA": "scale",
+}
+
+
+def hash_api_key(key: str) -> str:
+    return hashlib.sha256(key.encode()).hexdigest()
+
+
+def get_current_user(
+    request: Request,
+    x_api_key: str = Security(api_key_header),
+    db: Session = Depends(get_db),
+) -> User:
+    # ── RapidAPI proxy auth ──
+    proxy_secret = request.headers.get("X-RapidAPI-Proxy-Secret", "")
+    rapidapi_user_id = request.headers.get("X-RapidAPI-User", "")
+    rapidapi_plan = request.headers.get("X-RapidAPI-Subscription", "FREE").upper()
+
+    if proxy_secret and settings.rapidapi_proxy_secret:
+        if proxy_secret != settings.rapidapi_proxy_secret:
+            raise HTTPException(status_code=401, detail="Invalid RapidAPI proxy secret")
+        if not rapidapi_user_id:
+            raise HTTPException(status_code=401, detail="Missing X-RapidAPI-User header")
+
+        user = db.query(User).filter(User.email == f"rapidapi:{rapidapi_user_id}").first()
+        if not user:
+            tier = RAPIDAPI_PLAN_MAP.get(rapidapi_plan, "free")
+            user = User(
+                email=f"rapidapi:{rapidapi_user_id}",
+                hashed_password="rapidapi-proxy-auth",
+                name=f"RapidAPI User {rapidapi_user_id[:12]}",
+                tier=tier,
+            )
+            db.add(user)
+            db.commit()
+            db.refresh(user)
+        else:
+            new_tier = RAPIDAPI_PLAN_MAP.get(rapidapi_plan, "free")
+            if user.tier != new_tier:
+                user.tier = new_tier
+                db.commit()
+        return user
+
+    # ── Direct API key auth ──
+    if not x_api_key:
+        raise HTTPException(status_code=401, detail="Missing X-API-Key header")
+
+    key_hash = hash_api_key(x_api_key)
+    api_key = db.query(APIKey).filter(
+        APIKey.key_hash == key_hash,
+        APIKey.is_active == True
+    ).first()
+
+    if not api_key:
+        raise HTTPException(status_code=401, detail="Invalid or inactive API key")
+
+    api_key.last_used_at = datetime.datetime.utcnow()
+    db.commit()
+
+    user = db.query(User).filter(User.id == api_key.user_id).first()
+    if not user or not user.is_active:
+        raise HTTPException(status_code=403, detail="Account is inactive")
+
+    return user
+
+
+def check_rate_limit(user: User, db: Session, endpoint: str, request_count: int = 1):
+    """Check if user has exceeded rate limits for their tier."""
+    tier_limits = {
+        "free": settings.free_rpm, "maker": settings.maker_rpm,
+        "growth": settings.growth_rpm, "scale": settings.scale_rpm, "enterprise": 10000
+    }
+    per_minute_limit = tier_limits.get(user.tier, 10)
+    
+    # Check recent requests in last 60 seconds
+    cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=60)
+    recent = db.query(func.sum(UsageRecord.count)).filter(
+        UsageRecord.user_id == user.id,
+        UsageRecord.created_at >= cutoff
+    ).scalar() or 0
+    
+    if recent + request_count > per_minute_limit:
+        raise HTTPException(
+            status_code=429,
+            detail=f"Rate limit exceeded — quota reached. Tier '{user.tier}' allows {per_minute_limit} requests/minute."
+        )
+
+
+def check_prediction_quota(user: User, db: Session, count: int = 1):
+    """Check if user has remaining predictions this month."""
+    tier_limits = {
+        "free": settings.free_predictions,
+        "maker": settings.maker_predictions,
+        "growth": settings.growth_predictions,
+        "scale": settings.scale_predictions,
+        "enterprise": 10_000_000,
+    }
+    limit = tier_limits.get(user.tier, 100)
+    month_str = datetime.datetime.utcnow().strftime("%Y-%m")
+
+    used = db.query(func.sum(UsageRecord.count)).filter(
+        UsageRecord.user_id == user.id,
+        UsageRecord.month == month_str,
+        UsageRecord.endpoint.in_(["predict/churn", "predict/lead", "analyze/call"])
+    ).scalar() or 0
+
+    if used + count > limit:
+        raise HTTPException(
+            status_code=429,
+            detail=f"Monthly prediction quota exceeded. Tier '{user.tier}': {used}/{limit} predictions used this month."
+        )
+
+    return used, limit
+
+
+def check_model_quota(user: User, db: Session):
+    """Check if user can create more custom models."""
+    tier_limits = {
+        "free": settings.free_models,
+        "maker": settings.maker_models,
+        "growth": settings.growth_models,
+        "scale": settings.scale_models,
+        "enterprise": 10000,
+    }
+    limit = tier_limits.get(user.tier, 0)
+    current = db.query(MLModel).filter(MLModel.user_id == user.id).count()
+
+    if current >= limit:
+        raise HTTPException(
+            status_code=429,
+            detail=f"Model quota exceeded. Tier '{user.tier}': {current}/{limit} models used."
+        )
+
+    return current, limit
+
+
+def track_usage(user: User, db: Session, endpoint: str, count: int = 1):
+    month_str = datetime.datetime.utcnow().strftime("%Y-%m")
+    record = UsageRecord(
+        user_id=user.id,
+        endpoint=endpoint,
+        count=count,
+        month=month_str,
+    )
+    db.add(record)
+    db.commit()
+
+
+def get_usage_summary(user: User, db: Session) -> dict:
+    month_str = datetime.datetime.utcnow().strftime("%Y-%m")
+    tier_limits = {
+        "free": settings.free_predictions,
+        "maker": settings.maker_predictions,
+        "growth": settings.growth_predictions,
+        "scale": settings.scale_predictions,
+        "enterprise": 10_000_000,
+    }
+    model_limits = {
+        "free": settings.free_models,
+        "maker": settings.maker_models,
+        "growth": settings.growth_models,
+        "scale": settings.scale_models,
+        "enterprise": 10000,
+    }
+
+    predictions_used = db.query(func.sum(UsageRecord.count)).filter(
+        UsageRecord.user_id == user.id,
+        UsageRecord.month == month_str,
+        UsageRecord.endpoint.in_(["predict/churn", "predict/lead", "analyze/call"])
+    ).scalar() or 0
+
+    models_used = db.query(MLModel).filter(MLModel.user_id == user.id).count()
+
+    usage_by_endpoint = {}
+    records = db.query(UsageRecord).filter(
+        UsageRecord.user_id == user.id,
+        UsageRecord.month == month_str
+    ).all()
+    for r in records:
+        usage_by_endpoint[r.endpoint] = usage_by_endpoint.get(r.endpoint, 0) + r.count
+
+    pred_limit = tier_limits.get(user.tier, 100)
+    model_limit = model_limits.get(user.tier, 0)
+
+    return {
+        "tier": user.tier,
+        "predictions_used": predictions_used,
+        "predictions_limit": pred_limit,
+        "remaining_predictions": max(0, pred_limit - predictions_used),
+        "models_used": models_used,
+        "models_limit": model_limit,
+        "remaining_models": max(0, model_limit - models_used),
+        "usage_by_endpoint": usage_by_endpoint,
+    }
+
+
+# Import MLModel at the bottom to avoid circular imports
+from app.models.mlmodel import MLModel

app/main.py

@@ -0,0 +1,80 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from contextlib import asynccontextmanager
+
+from app.config import get_settings
+from app.database import init_db
+from app.routers import auth, predict, training, audio, usage, benchmarks
+
+settings = get_settings()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    init_db()
+    yield
+
+
+app = FastAPI(
+    title="RevAI API",
+    description="""
+## Churn Prediction & Lead Scoring API
+
+### Core Features
+- **Predict churn**: Send customer data, get churn risk scores + explanations
+- **Score leads**: Prioritize sales pipeline by conversion probability
+- **Train custom models**: Upload labeled data, get a company-specific XGBoost model
+- **Analyze calls**: Transcribe support/sales calls, detect churn intent signals
+
+### Authentication
+All endpoints (except `/v1/health`) require an API key header:
+```
+X-API-Key: revai_live_...
+```
+
+Get your key by registering at `/v1/auth/register`.
+
+### Rate Limits
+| Tier    | Predictions/mo | Models | Req/min |
+|---------|---------------|--------|---------|
+| Free    | 100           | 0      | 10      |
+| Maker   | 5,000         | 3      | 100     |
+| Growth  | 50,000        | 10     | 500     |
+| Scale   | 500,000       | Unlim  | 2,000   |
+
+### Pricing
+Visit [RevAI on Payhip](https://payhip.com) to subscribe.
+    """,
+    version=settings.version,
+    lifespan=lifespan,
+    docs_url="/docs",
+    redoc_url="/redoc",
+    openapi_url="/openapi.json",
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/")
+@app.get("/_health")
+async def root_health():
+    return {"status": "ok", "api": "RevAI", "version": settings.version}
+
+
+app.include_router(auth.router)
+app.include_router(predict.router)
+app.include_router(training.router)
+app.include_router(audio.router)
+app.include_router(benchmarks.router)
+app.include_router(usage.router)
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=settings.debug)

app/models/__init__.py

@@ -0,0 +1,8 @@
+from app.database import Base
+from app.models.user import User
+from app.models.apikey import APIKey
+from app.models.usage import UsageRecord
+from app.models.mlmodel import MLModel
+from app.models.benchmark import Benchmark
+
+__all__ = ["User", "APIKey", "UsageRecord", "MLModel", "Benchmark", "Base"]

app/models/apikey.py

@@ -0,0 +1,20 @@
+import uuid
+import datetime
+from sqlalchemy import Column, String, DateTime, Boolean, ForeignKey
+from sqlalchemy.orm import relationship
+from app.database import Base
+
+
+class APIKey(Base):
+    __tablename__ = "api_keys"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    user_id = Column(String, ForeignKey("users.id"), nullable=False)
+    key_hash = Column(String, unique=True, nullable=False)
+    prefix = Column(String, nullable=False)  # revai_live_abc123 → store "revai_live_abc..."
+    name = Column(String, default="Default")
+    is_active = Column(Boolean, default=True)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+    last_used_at = Column(DateTime, nullable=True)
+
+    user = relationship("User", back_populates="api_keys")

app/models/benchmark.py

@@ -0,0 +1,29 @@
+import json
+import datetime
+from sqlalchemy import Column, String, DateTime, Text, Integer, Float
+from app.database import Base
+
+
+class Benchmark(Base):
+    __tablename__ = "benchmarks"
+
+    id = Column(String, primary_key=True, default="global")  # single row for global benchmarks
+    churn_data_json = Column(Text, default="{}")   # aggregated churn score distribution
+    lead_data_json = Column(Text, default="{}")    # aggregated lead score distribution
+    total_churn_scored = Column(Integer, default=0)
+    total_lead_scored = Column(Integer, default=0)
+    unique_churn_companies = Column(Integer, default=0)
+    unique_lead_companies = Column(Integer, default=0)
+    updated_at = Column(DateTime, default=datetime.datetime.utcnow)
+
+    def get_churn_data(self) -> dict:
+        return json.loads(self.churn_data_json) if self.churn_data_json else {}
+
+    def get_lead_data(self) -> dict:
+        return json.loads(self.lead_data_json) if self.lead_data_json else {}
+
+    def set_churn_data(self, data: dict):
+        self.churn_data_json = json.dumps(data)
+
+    def set_lead_data(self, data: dict):
+        self.lead_data_json = json.dumps(data)

app/models/mlmodel.py

@@ -0,0 +1,34 @@
+import uuid
+import datetime
+import json
+from sqlalchemy import Column, String, DateTime, Text, Integer, Float, ForeignKey
+from sqlalchemy.orm import relationship
+from app.database import Base
+
+
+class MLModel(Base):
+    __tablename__ = "ml_models"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    user_id = Column(String, ForeignKey("users.id"), nullable=False)
+    name = Column(String, nullable=False)
+    model_type = Column(String, nullable=False)  # churn, lead
+    feature_names_json = Column(Text, nullable=False)  # JSON list of feature names
+    encoders_json = Column(Text, nullable=False)  # JSON dict of label encoders
+    model_binary = Column(Text, nullable=False)  # base64 encoded XGBoost model
+    metrics_json = Column(Text, default="{}")  # accuracy, roc_auc, etc.
+    summary_json = Column(Text, default="{}")  # n_rows, n_features, etc.
+    n_rows = Column(Integer, default=0)
+    n_features = Column(Integer, default=0)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+
+    user = relationship("User", back_populates="ml_models")
+
+    def get_feature_names(self):
+        return json.loads(self.feature_names_json)
+
+    def get_metrics(self):
+        return json.loads(self.metrics_json)
+
+    def get_summary(self):
+        return json.loads(self.summary_json)

app/models/usage.py

@@ -0,0 +1,18 @@
+import uuid
+import datetime
+from sqlalchemy import Column, String, DateTime, Integer, ForeignKey
+from sqlalchemy.orm import relationship
+from app.database import Base
+
+
+class UsageRecord(Base):
+    __tablename__ = "usage_records"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    user_id = Column(String, ForeignKey("users.id"), nullable=False)
+    endpoint = Column(String, nullable=False)  # predict/churn, predict/lead, train, analyze/call
+    count = Column(Integer, default=1)  # number of predictions/requests
+    month = Column(String, nullable=False)  # "2026-05"
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+
+    user = relationship("User", back_populates="usage_records")

app/models/user.py

@@ -0,0 +1,23 @@
+import uuid
+import datetime
+from sqlalchemy import Column, String, DateTime, Boolean
+from sqlalchemy.orm import relationship
+from app.database import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    email = Column(String, unique=True, index=True, nullable=False)
+    hashed_password = Column(String, nullable=False)
+    name = Column(String, default="")
+    tier = Column(String, default="free")  # free, maker, growth, scale, enterprise
+    stripe_customer_id = Column(String, default="")
+    stripe_subscription_id = Column(String, default="")
+    is_active = Column(Boolean, default=True)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+
+    api_keys = relationship("APIKey", back_populates="user", cascade="all, delete-orphan")
+    usage_records = relationship("UsageRecord", back_populates="user", cascade="all, delete-orphan")
+    ml_models = relationship("MLModel", back_populates="user", cascade="all, delete-orphan")

app/routers/audio.py

@@ -0,0 +1,114 @@
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.dependencies import get_current_user, check_rate_limit, check_prediction_quota, track_usage, get_usage_summary
+from app.models.user import User
+from app.schemas import CallAnalysisResponse, CallAnalysisItem
+from app.services.audio import analyze_call
+
+router = APIRouter(prefix="/v1/analyze", tags=["audio"])
+
+
+@router.post("/call", response_model=CallAnalysisResponse)
+async def analyze_call_endpoint(
+    file: UploadFile = File(...),
+    openai_api_key: str = Form(...),
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    if not file.filename:
+        raise HTTPException(status_code=400, detail="No file provided")
+
+    allowed_ext = (".mp3", ".wav", ".m4a", ".mp4", ".ogg", ".webm")
+    if not file.filename.lower().endswith(allowed_ext):
+        raise HTTPException(status_code=400,
+                            detail=f"Unsupported format. Allowed: {', '.join(allowed_ext)}")
+
+    check_rate_limit(user, db, "analyze/call", 1)
+    used, limit = check_prediction_quota(user, db, 1)
+
+    try:
+        audio_bytes = await file.read()
+        result = analyze_call(audio_bytes, file.filename, openai_api_key)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Analysis failed: {str(e)}")
+
+    track_usage(user, db, "analyze/call", 1)
+
+    item = CallAnalysisItem(
+        filename=result["filename"],
+        transcript_snippet=result["transcript_snippet"],
+        sentiment_score=result["sentiment_score"],
+        sentiment_label=result["sentiment_label"],
+        churn_intent_score=float(result["churn_intent_score"]),
+        flagged_keywords=result["flagged_keywords"],
+    )
+
+    summary = {
+        "total_calls": 1,
+        "average_sentiment": result["sentiment_score"],
+        "high_churn_intent": result["churn_intent_score"] >= 50,
+    }
+
+    return CallAnalysisResponse(
+        results=[item],
+        summary=summary,
+        usage=get_usage_summary(user, db),
+    )
+
+
+@router.post("/calls/batch", response_model=CallAnalysisResponse)
+async def analyze_calls_batch(
+    files: list[UploadFile] = File(...),
+    openai_api_key: str = Form(...),
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    if not files:
+        raise HTTPException(status_code=400, detail="No files provided")
+
+    n_files = len(files)
+    check_rate_limit(user, db, "analyze/call", n_files)
+    used, limit = check_prediction_quota(user, db, n_files)
+
+    results = []
+    errors = []
+
+    for f in files:
+        if not f.filename.lower().endswith((".mp3", ".wav", ".m4a", ".mp4", ".ogg", ".webm")):
+            errors.append(f"Skipped {f.filename}: unsupported format")
+            continue
+        try:
+            audio_bytes = await f.read()
+            result = analyze_call(audio_bytes, f.filename, openai_api_key)
+            results.append(CallAnalysisItem(
+                filename=result["filename"],
+                transcript_snippet=result["transcript_snippet"],
+                sentiment_score=result["sentiment_score"],
+                sentiment_label=result["sentiment_label"],
+                churn_intent_score=float(result["churn_intent_score"]),
+                flagged_keywords=result["flagged_keywords"],
+            ))
+        except Exception as e:
+            errors.append(f"Failed {f.filename}: {str(e)}")
+
+    processed = len(results)
+    track_usage(user, db, "analyze/call", processed)
+
+    avg_sentiment = sum(r.sentiment_score for r in results) / max(processed, 1)
+    high_intent = sum(1 for r in results if r.churn_intent_score >= 50)
+
+    summary = {
+        "total_calls": n_files,
+        "processed": processed,
+        "errors": errors,
+        "average_sentiment": round(avg_sentiment, 4),
+        "high_churn_intent_count": high_intent,
+    }
+
+    return CallAnalysisResponse(
+        results=results,
+        summary=summary,
+        usage=get_usage_summary(user, db),
+    )

app/routers/auth.py

@@ -0,0 +1,116 @@
+import uuid
+import datetime
+import hashlib
+import os
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from jose import jwt
+
+from app.database import get_db
+from app.models.user import User
+from app.models.apikey import APIKey
+from app.dependencies import hash_api_key, get_current_user
+from app.schemas import UserRegister, UserLogin, TokenResponse, APIKeyResponse
+from app.config import get_settings
+
+router = APIRouter(prefix="/v1/auth", tags=["auth"])
+settings = get_settings()
+
+
+def hash_password(password: str) -> str:
+    salt = os.urandom(16).hex()
+    h = hashlib.sha256((password + salt).encode()).hexdigest()
+    return f"{salt}${h}"
+
+
+def verify_password(password: str, hashed: str) -> bool:
+    salt, h = hashed.split("$", 1)
+    return hashlib.sha256((password + salt).encode()).hexdigest() == h
+
+
+def generate_api_key() -> tuple:
+    raw = f"revai_live_{uuid.uuid4().hex}{uuid.uuid4().hex[:8]}"
+    return raw, hash_api_key(raw)
+
+
+def create_token(user: User) -> str:
+    payload = {
+        "sub": user.id,
+        "email": user.email,
+        "tier": user.tier,
+        "exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=settings.access_token_expire_minutes),
+    }
+    return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)
+
+
+@router.post("/register", response_model=TokenResponse)
+def register(body: UserRegister, db: Session = Depends(get_db)):
+    existing = db.query(User).filter(User.email == body.email).first()
+    if existing:
+        raise HTTPException(status_code=409, detail="Email already registered")
+
+    user = User(
+        email=body.email,
+        hashed_password=hash_password(body.password),
+        name=body.name,
+    )
+    db.add(user)
+    db.flush()
+
+    raw_key, key_hash = generate_api_key()
+    api_key = APIKey(
+        user_id=user.id,
+        key_hash=key_hash,
+        prefix=raw_key[:20] + "...",
+    )
+    db.add(api_key)
+    db.commit()
+
+    token = create_token(user)
+    return TokenResponse(
+        access_token=token,
+        user_id=user.id,
+        email=user.email,
+        tier=user.tier,
+        api_key=raw_key,
+    )
+
+
+@router.post("/login", response_model=TokenResponse)
+def login(body: UserLogin, db: Session = Depends(get_db)):
+    user = db.query(User).filter(User.email == body.email).first()
+    if not user or not verify_password(body.password, user.hashed_password):
+        raise HTTPException(status_code=401, detail="Invalid email or password")
+
+    # Return existing active key or create new one
+    api_key = db.query(APIKey).filter(
+        APIKey.user_id == user.id, APIKey.is_active == True
+    ).first()
+
+    raw_key = None
+    if not api_key:
+        raw_key, key_hash = generate_api_key()
+        api_key = APIKey(user_id=user.id, key_hash=key_hash, prefix=raw_key[:20] + "...")
+        db.add(api_key)
+        db.commit()
+        raw_key = raw_key
+
+    token = create_token(user)
+    return TokenResponse(
+        access_token=token,
+        user_id=user.id,
+        email=user.email,
+        tier=user.tier,
+        api_key=raw_key,
+    )
+
+
+@router.get("/keys", response_model=list[APIKeyResponse])
+def list_api_keys(user: User = Depends(get_current_user),
+                  db: Session = Depends(get_db)):
+    keys = db.query(APIKey).filter(APIKey.user_id == user.id).all()
+    return [APIKeyResponse(
+        id=k.id, prefix=k.prefix, name=k.name,
+        is_active=k.is_active, created_at=k.created_at,
+        last_used_at=k.last_used_at
+    ) for k in keys]

app/routers/benchmarks.py

@@ -0,0 +1,18 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.dependencies import get_current_user
+from app.models.user import User
+from app.services.benchmarking import get_benchmarks
+
+router = APIRouter(prefix="/v1/benchmarks", tags=["benchmarks"])
+
+
+@router.get("")
+def get_global_benchmarks(
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    """Get global anonymized benchmarks. Minimum 3 companies required for privacy."""
+    return get_benchmarks(db)

app/routers/predict.py

@@ -0,0 +1,186 @@
+import json
+import base64
+import pickle
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.dependencies import (
+    get_current_user, check_rate_limit, check_prediction_quota,
+    track_usage, get_usage_summary,
+)
+from app.models.user import User
+from app.models.mlmodel import MLModel
+from app.schemas import PredictionInput, PredictionResponse, SinglePrediction
+from app.services.scoring import _churn_factor, _lead_factor
+from app.services.training import predict_with_model
+from app.services.benchmarking import update_benchmarks, compare_to_benchmark
+
+router = APIRouter(prefix="/v1/predict", tags=["prediction"])
+
+
+def _apply_heuristics(data: list, model_type: str) -> list:
+    """Apply heuristic rules to a list of data dicts."""
+    results = []
+    rule_func = _churn_factor if model_type == "churn" else _lead_factor
+
+    for i, row in enumerate(data):
+        score, factors = rule_func(row)
+
+        if score >= 70:
+            risk = "High Risk" if model_type == "churn" else "Hot"
+            action = ("Immediate outreach + retention offer" if model_type == "churn"
+                      else "Call immediately — high intent signals")
+        elif score >= 40:
+            risk = "Medium Risk" if model_type == "churn" else "Warm"
+            action = ("Monitor + engagement campaign" if model_type == "churn"
+                      else "Send case study + schedule demo")
+        else:
+            risk = "Low Risk" if model_type == "churn" else "Cold"
+            action = ("No action needed" if model_type == "churn"
+                      else "Add to email drip sequence")
+
+        cid = row.get("customer_id") or row.get("lead_id") or row.get("id") or None
+
+        results.append(SinglePrediction(
+            index=i,
+            score=float(score),
+            risk_level=risk,
+            recommended_action=action,
+            risk_factors=factors,
+            scoring_mode="heuristic",
+            customer_id=str(cid) if cid else None,
+            input_fields=row,
+        ))
+
+    return results
+
+
+@router.post("/churn", response_model=PredictionResponse)
+def predict_churn(
+    body: PredictionInput,
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    n_predictions = len(body.data)
+    check_rate_limit(user, db, "predict/churn", n_predictions)
+    used, limit = check_prediction_quota(user, db, n_predictions)
+
+    if body.model_id:
+        # Use custom trained model
+        ml_model = db.query(MLModel).filter(
+            MLModel.id == body.model_id, MLModel.user_id == user.id
+        ).first()
+        if not ml_model:
+            raise HTTPException(status_code=404, detail="Model not found")
+
+        scores = predict_with_model(
+            ml_model.model_binary,
+            ml_model.get_feature_names(),
+            ml_model.encoders_json,
+            body.data,
+        )
+
+        results = []
+        for i, (row, score) in enumerate(zip(body.data, scores)):
+            cid = row.get("customer_id") or row.get("id")
+            if score >= 70:
+                risk = "High Risk"
+                action = "Immediate outreach + retention offer"
+            elif score >= 40:
+                risk = "Medium Risk"
+                action = "Monitor + engagement campaign"
+            else:
+                risk = "Low Risk"
+                action = "No action needed"
+
+            results.append(SinglePrediction(
+                index=i, score=score, risk_level=risk,
+                recommended_action=action, risk_factors=[],
+                scoring_mode="custom_ml",
+                customer_id=str(cid) if cid else None,
+                input_fields=row,
+            ))
+
+        model_label = f"custom_ml_{body.model_id[:8]}"
+    else:
+        results = _apply_heuristics(body.data, "churn")
+        model_label = "heuristic"
+
+    track_usage(user, db, "predict/churn", n_predictions)
+
+    # ── Anonymous benchmark update + comparison ──
+    all_scores = [p.score for p in results]
+    update_benchmarks(db, all_scores, "churn")
+    benchmark = compare_to_benchmark(all_scores, "churn", db)
+
+    return PredictionResponse(
+        predictions=results,
+        model_used=model_label,
+        usage=get_usage_summary(user, db),
+        benchmark=benchmark,
+    )
+
+
+@router.post("/lead", response_model=PredictionResponse)
+def predict_lead(
+    body: PredictionInput,
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    n_predictions = len(body.data)
+    check_rate_limit(user, db, "predict/lead", n_predictions)
+    used, limit = check_prediction_quota(user, db, n_predictions)
+
+    if body.model_id:
+        ml_model = db.query(MLModel).filter(
+            MLModel.id == body.model_id, MLModel.user_id == user.id
+        ).first()
+        if not ml_model:
+            raise HTTPException(status_code=404, detail="Model not found")
+
+        scores = predict_with_model(
+            ml_model.model_binary,
+            ml_model.get_feature_names(),
+            ml_model.encoders_json,
+            body.data,
+        )
+
+        results = []
+        for i, (row, score) in enumerate(zip(body.data, scores)):
+            cid = row.get("lead_id") or row.get("id")
+            if score >= 70:
+                risk = "Hot"
+                action = "Call immediately — high intent signals"
+            elif score >= 40:
+                risk = "Warm"
+                action = "Send case study + schedule demo"
+            else:
+                risk = "Cold"
+                action = "Add to email drip sequence"
+
+            results.append(SinglePrediction(
+                index=i, score=score, risk_level=risk,
+                recommended_action=action, risk_factors=[],
+                scoring_mode="custom_ml",
+                customer_id=str(cid) if cid else None,
+                input_fields=row,
+            ))
+
+        model_label = f"custom_ml_{body.model_id[:8]}"
+    else:
+        results = _apply_heuristics(body.data, "lead")
+        model_label = "heuristic"
+
+    track_usage(user, db, "predict/lead", n_predictions)
+
+    all_scores = [p.score for p in results]
+    update_benchmarks(db, all_scores, "lead")
+    benchmark = compare_to_benchmark(all_scores, "lead", db)
+
+    return PredictionResponse(
+        predictions=results,
+        model_used=model_label,
+        usage=get_usage_summary(user, db),
+        benchmark=benchmark,
+    )

app/routers/training.py

@@ -0,0 +1,120 @@
+import uuid
+import json
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.dependencies import get_current_user, check_rate_limit, check_model_quota, track_usage
+from app.models.user import User
+from app.models.mlmodel import MLModel
+from app.schemas import TrainingInput, TrainingResponse, MLModelResponse, MLModelListResponse
+from app.services.training import train_from_data
+
+router = APIRouter(prefix="/v1", tags=["training"])
+
+
+@router.post("/train", response_model=TrainingResponse)
+def train_model(
+    body: TrainingInput,
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    check_rate_limit(user, db, "train", 1)
+    check_model_quota(user, db)
+
+    if body.model_type not in ("churn", "lead"):
+        raise HTTPException(status_code=400, detail="model_type must be 'churn' or 'lead'")
+
+    if len(body.data) < 50:
+        raise HTTPException(status_code=400, detail="Need at least 50 rows to train")
+
+    try:
+        result = train_from_data(body.data, body.target_column, body.model_type)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+    model_name = body.model_name or f"{body.model_type}-model-{uuid.uuid4().hex[:6]}"
+
+    ml_model = MLModel(
+        user_id=user.id,
+        name=model_name,
+        model_type=body.model_type,
+        feature_names_json=json.dumps(result["feature_names"]),
+        encoders_json=result["encoders_json"],
+        model_binary=result["model_binary"],
+        metrics_json=result["metrics_json"],
+        summary_json=result["summary_json"],
+        n_rows=result["n_rows"],
+        n_features=result["n_features"],
+    )
+    db.add(ml_model)
+    db.commit()
+    db.refresh(ml_model)
+
+    track_usage(user, db, "train", 1)
+
+    return TrainingResponse(
+        model_id=ml_model.id,
+        model_name=ml_model.name,
+        model_type=ml_model.model_type,
+        metrics=result["metrics"],
+        summary=result["summary"],
+    )
+
+
+@router.get("/models", response_model=MLModelListResponse)
+def list_models(
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    models = db.query(MLModel).filter(MLModel.user_id == user.id).order_by(
+        MLModel.created_at.desc()
+    ).all()
+
+    items = []
+    for m in models:
+        items.append(MLModelResponse(
+            id=m.id, name=m.name, model_type=m.model_type,
+            n_features=m.n_features, n_rows=m.n_rows,
+            metrics=m.get_metrics(), summary=m.get_summary(),
+            created_at=m.created_at,
+        ))
+
+    return MLModelListResponse(models=items, total=len(items))
+
+
+@router.get("/models/{model_id}", response_model=MLModelResponse)
+def get_model(
+    model_id: str,
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    m = db.query(MLModel).filter(
+        MLModel.id == model_id, MLModel.user_id == user.id
+    ).first()
+    if not m:
+        raise HTTPException(status_code=404, detail="Model not found")
+
+    return MLModelResponse(
+        id=m.id, name=m.name, model_type=m.model_type,
+        n_features=m.n_features, n_rows=m.n_rows,
+        metrics=m.get_metrics(), summary=m.get_summary(),
+        created_at=m.created_at,
+    )
+
+
+@router.delete("/models/{model_id}")
+def delete_model(
+    model_id: str,
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    m = db.query(MLModel).filter(
+        MLModel.id == model_id, MLModel.user_id == user.id
+    ).first()
+    if not m:
+        raise HTTPException(status_code=404, detail="Model not found")
+
+    db.delete(m)
+    db.commit()
+    return {"detail": "Model deleted", "model_id": model_id}

app/routers/usage.py

@@ -0,0 +1,29 @@
+import time
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.dependencies import get_current_user, get_usage_summary
+from app.models.user import User
+from app.schemas import UsageResponse, HealthResponse
+from app.config import get_settings
+
+router = APIRouter(tags=["usage"])
+settings = get_settings()
+_start_time = time.time()
+
+
+@router.get("/v1/usage", response_model=UsageResponse)
+def get_usage(
+    user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    return get_usage_summary(user, db)
+
+
+@router.get("/v1/health", response_model=HealthResponse)
+def health_check():
+    return HealthResponse(
+        version=settings.version,
+        uptime=round(time.time() - _start_time, 1),
+    )

app/schemas/__init__.py

@@ -0,0 +1,124 @@
+from pydantic import BaseModel, Field, EmailStr, field_validator
+from typing import Optional, List, Dict, Any
+from datetime import datetime
+
+
+# ── Auth ──
+class UserRegister(BaseModel):
+    email: str
+    password: str = Field(min_length=8)
+    name: str = ""
+
+class UserLogin(BaseModel):
+    email: str
+    password: str
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    user_id: str
+    email: str
+    tier: str
+    api_key: Optional[str] = None
+
+class APIKeyResponse(BaseModel):
+    id: str
+    prefix: str
+    name: str
+    is_active: bool
+    created_at: datetime
+    last_used_at: Optional[datetime] = None
+
+
+# ── Prediction ──
+class PredictionInput(BaseModel):
+    data: List[Dict[str, Any]] = Field(..., min_length=1, max_length=1000)
+    model_id: Optional[str] = None  # use custom trained model if provided
+
+class RiskFactor(BaseModel):
+    rule: str
+    points: int
+    detail: str
+
+class SinglePrediction(BaseModel):
+    index: int
+    score: float
+    risk_level: str
+    recommended_action: Optional[str] = None
+    risk_factors: List[RiskFactor] = []
+    scoring_mode: str  # "heuristic" or "custom_ml"
+    customer_id: Optional[str] = None
+    # Re-echo input row merged with output
+    input_fields: Dict[str, Any] = {}
+
+class PredictionResponse(BaseModel):
+    predictions: List[SinglePrediction]
+    model_used: str  # "heuristic" or "custom_ml_<id>"
+    usage: Dict[str, Any]
+    benchmark: Optional[Dict[str, Any]] = None  # comparison vs industry
+
+
+# ── Training ──
+class TrainingInput(BaseModel):
+    data: List[Dict[str, Any]] = Field(..., min_length=50)
+    target_column: str
+    model_type: str = "churn"  # churn or lead
+    model_name: Optional[str] = None
+
+class TrainingResponse(BaseModel):
+    model_id: str
+    model_name: str
+    model_type: str
+    metrics: Dict[str, Any]
+    summary: Dict[str, Any]
+
+
+# ── Call Analysis ──
+class CallAnalysisItem(BaseModel):
+    filename: str
+    transcript_snippet: str
+    sentiment_score: float  # -1 to 1
+    sentiment_label: str  # positive, neutral, negative
+    churn_intent_score: float  # 0-100
+    flagged_keywords: List[str]
+    duration_seconds: Optional[float] = None
+
+class CallAnalysisResponse(BaseModel):
+    results: List[CallAnalysisItem]
+    summary: Dict[str, Any]
+    usage: Dict[str, Any]
+
+
+# ── Usage ──
+class UsageResponse(BaseModel):
+    tier: str
+    predictions_used: int
+    predictions_limit: int
+    models_used: int
+    models_limit: int
+    remaining_predictions: int
+    remaining_models: int
+    usage_by_endpoint: Dict[str, int]
+
+
+# ── Models ──
+class MLModelResponse(BaseModel):
+    id: str
+    name: str
+    model_type: str
+    n_features: int
+    n_rows: int
+    metrics: Dict[str, Any]
+    summary: Dict[str, Any]
+    created_at: datetime
+
+class MLModelListResponse(BaseModel):
+    models: List[MLModelResponse]
+    total: int
+
+
+# ── Health ──
+class HealthResponse(BaseModel):
+    status: str = "ok"
+    version: str
+    uptime: float

app/services/audio.py

@@ -0,0 +1,105 @@
+import io
+import tempfile
+import os
+from typing import List, Dict, Any, Optional
+from pathlib import Path
+
+try:
+    from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
+    _vader = SentimentIntensityAnalyzer()
+except ImportError:
+    _vader = None
+
+from app.services.scoring import detect_churn_keywords, CHURN_KEYWORDS
+
+
+def analyze_sentiment(text: str) -> dict:
+    """Run VADER sentiment analysis on transcript text."""
+    if not text.strip():
+        return {"compound": 0.0, "pos": 0.0, "neg": 0.0, "neu": 1.0, "label": "neutral"}
+
+    if _vader is None:
+        # fallback: simple heuristic
+        positive = sum(1 for w in ["good", "great", "excellent", "happy", "love", "thanks", "amazing", "helpful"] if w in text.lower())
+        negative = sum(1 for w in ["bad", "terrible", "awful", "hate", "frustrated", "angry", "broken", "useless"] if w in text.lower())
+        compound = (positive - negative) / max(positive + negative + 1, 1)
+        return {
+            "compound": round(compound, 4),
+            "pos": 0.0, "neg": 0.0, "neu": 0.0,
+            "label": "positive" if compound > 0.1 else ("negative" if compound < -0.1 else "neutral")
+        }
+
+    scores = _vader.polarity_scores(text)
+    compound = scores["compound"]
+    label = "positive" if compound >= 0.05 else ("negative" if compound <= -0.05 else "neutral")
+
+    return {
+        "compound": round(compound, 4),
+        "pos": round(scores["pos"], 4),
+        "neg": round(scores["neg"], 4),
+        "neu": round(scores["neu"], 4),
+        "label": label,
+    }
+
+
+def transcribe_with_whisper(audio_bytes: bytes, filename: str, api_key: str) -> str:
+    """Transcribe audio using OpenAI Whisper API."""
+    import httpx
+
+    # Determine content type from extension
+    ext = Path(filename).suffix.lower()
+    content_type_map = {
+        ".mp3": "audio/mpeg",
+        ".wav": "audio/wav",
+        ".m4a": "audio/mp4",
+        ".mp4": "video/mp4",
+        ".ogg": "audio/ogg",
+        ".webm": "audio/webm",
+    }
+    content_type = content_type_map.get(ext, "audio/mpeg")
+
+    # Write bytes to temp file
+    with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as tmp:
+        tmp.write(audio_bytes)
+        tmp_path = tmp.name
+
+    try:
+        with open(tmp_path, "rb") as f:
+            response = httpx.post(
+                "https://api.openai.com/v1/audio/transcriptions",
+                headers={"Authorization": f"Bearer {api_key}"},
+                data={"model": "whisper-1", "response_format": "text"},
+                files={"file": (filename, f, content_type)},
+                timeout=60.0,
+            )
+        if response.status_code != 200:
+            raise Exception(f"Whisper API error ({response.status_code}): {response.text[:200]}")
+
+        return response.text
+    finally:
+        os.unlink(tmp_path)
+
+
+def analyze_call(audio_bytes: bytes, filename: str, api_key: str) -> dict:
+    """Full call analysis: transcript → sentiment → churn keywords."""
+    transcript = transcribe_with_whisper(audio_bytes, filename, api_key)
+    sentiment = analyze_sentiment(transcript)
+    keywords = detect_churn_keywords(transcript)
+
+    # Combined churn intent from transcript
+    churn_score = keywords["churn_intent_score"]
+    if sentiment["label"] == "negative":
+        churn_score = min(churn_score + 20, 100)
+    elif sentiment["label"] == "positive":
+        churn_score = max(churn_score - 15, 0)
+
+    return {
+        "filename": filename,
+        "transcript_snippet": transcript[:300] + ("..." if len(transcript) > 300 else ""),
+        "transcript_full": transcript,
+        "sentiment_score": sentiment["compound"],
+        "sentiment_label": sentiment["label"],
+        "churn_intent_score": churn_score,
+        "flagged_keywords": keywords["flagged_keywords"],
+        "flagged_keyword_count": keywords["flagged_keyword_count"],
+    }

app/services/benchmarking.py

@@ -0,0 +1,192 @@
+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, ""),
+        },
+    }

app/services/scoring.py

@@ -0,0 +1,196 @@
+import pandas as pd
+import numpy as np
+from typing import List, Dict, Any, Tuple, Optional
+
+
+def _safe_float(val: Any, default: float = 0.0) -> float:
+    try:
+        return float(val) if val is not None else default
+    except (ValueError, TypeError):
+        return default
+
+
+def _safe_int(val: Any, default: int = 0) -> int:
+    try:
+        return int(float(val)) if val is not None else default
+    except (ValueError, TypeError):
+        return default
+
+
+# ═══════════════════════════════════════════════════
+#  CHURN HEURISTIC RULES
+# ═══════════════════════════════════════════════════
+
+def _churn_factor(row: Dict[str, Any]) -> tuple:
+    rules = []
+
+    days_off = _safe_float(row.get("days_since_last_login"), 0)
+    if days_off >= 14:
+        rules.append(("Login Recency", 25, f"No login in {int(days_off)} days"))
+    elif days_off >= 7:
+        rules.append(("Login Recency", 15, f"No login in {int(days_off)} days"))
+
+    logins = _safe_float(row.get("login_frequency_7d"), 10)
+    if logins <= 1:
+        rules.append(("Low Engagement", 20, f"Only {int(logins)} logins this week"))
+    elif logins <= 3:
+        rules.append(("Low Engagement", 10, f"Only {int(logins)} logins this week"))
+
+    tickets = _safe_float(row.get("support_tickets_last_30d"), 0)
+    if tickets >= 5:
+        rules.append(("Support Friction", 15, f"{int(tickets)} tickets in 30 days"))
+    elif tickets >= 3:
+        rules.append(("Support Friction", 8, f"{int(tickets)} tickets in 30 days"))
+
+    delays = _safe_float(row.get("payment_delays_90d"), 0)
+    if delays >= 3:
+        rules.append(("Payment Failure", 25, f"{int(delays)} payment delays"))
+    elif delays >= 1:
+        rules.append(("Payment Failure", 12, f"{int(delays)} payment delays in 90 days"))
+
+    adoption = _safe_float(row.get("feature_adoption_score"), 100)
+    if adoption <= 30:
+        rules.append(("Low Adoption", 10, f"Only {adoption:.0f}% feature adoption"))
+
+    nps = _safe_float(row.get("nps_score"), 10)
+    if nps <= 4:
+        rules.append(("Low NPS", 10, f"NPS score of {int(nps)}/10"))
+
+    tenure = _safe_float(row.get("tenure_days"), 365)
+    if tenure <= 60:
+        rules.append(("Short Tenure", 10, f"Only {int(tenure)} days as customer"))
+    elif tenure <= 90:
+        rules.append(("Short Tenure", 5, f"Only {int(tenure)} days as customer"))
+
+    ct = str(row.get("contract_type", "")).strip().lower()
+    if ct in ("month-to-month", "month to month", "monthly"):
+        rules.append(("Contract Risk", 10, "Month-to-month contract"))
+
+    session = _safe_float(row.get("avg_session_minutes"), 60)
+    if session <= 5:
+        rules.append(("Low Sessions", 5, f"Avg session {session:.1f} min"))
+
+    call_score = _safe_float(row.get("call_sentiment_churn_risk"), None)
+    if call_score is not None and call_score >= 70:
+        rules.append(("Cancellation Intent (Audio)", 30, f"Call churn score: {int(call_score)}%"))
+    elif call_score is not None and call_score >= 40:
+        rules.append(("Call Concern (Audio)", 15, f"Call churn score: {int(call_score)}%"))
+
+    call_sentiment = _safe_float(row.get("call_sentiment"), None)
+    if call_sentiment is not None and call_sentiment < -0.5:
+        rules.append(("Negative Sentiment (Audio)", 15, f"Sentiment: {call_sentiment:.2f}"))
+
+    keyword_count = _safe_int(row.get("flagged_keyword_count"), None)
+    if keyword_count is not None and keyword_count >= 2:
+        rules.append(("Churn Keywords (Audio)", 10, f"{int(keyword_count)} flagged keywords"))
+
+    score = min(sum(r[1] for r in rules), 100)
+    factors = [{"rule": r[0], "points": r[1], "detail": r[2]} for r in rules]
+    return score, factors
+
+
+# ═══════════════════════════════════════════════════
+#  LEAD HEURISTIC RULES
+# ═══════════════════════════════════════════════════
+
+def _lead_factor(row: Dict[str, Any]) -> tuple:
+    rules = []
+
+    demo = _safe_int(row.get("demo_requested"), 0)
+    if demo == 1:
+        rules.append(("Demo Requested", 25, "Demo has been requested"))
+
+    budget = _safe_int(row.get("budget_confirmed"), 0)
+    if budget == 1:
+        rules.append(("Budget Confirmed", 20, "Budget is confirmed"))
+
+    dm = _safe_int(row.get("decision_maker_contacted"), 0)
+    if dm == 1:
+        rules.append(("DM Access", 20, "Decision maker contacted"))
+
+    eng = _safe_float(row.get("engagement_score"), 0)
+    if eng >= 60:
+        rules.append(("High Engagement", 15, f"Engagement score {eng:.0f}/100"))
+
+    src = str(row.get("source", "")).strip().lower()
+    if src in ("referral", "organic", "paid ads"):
+        rules.append(("Quality Source", 10, f"Source: {src.title()}"))
+
+    dip = _safe_float(row.get("days_in_pipeline"), 60)
+    if dip <= 14:
+        rules.append(("Fresh Lead", 10, f"Only {int(dip)} days in pipeline"))
+
+    convs = _safe_float(row.get("previous_conversations"), 0)
+    if convs >= 3:
+        rules.append(("Active Relationship", 10, f"{int(convs)} conversations"))
+
+    downloads = _safe_float(row.get("content_downloads"), 0)
+    if downloads >= 3:
+        rules.append(("Content Interest", 5, f"{int(downloads)} downloads"))
+
+    opens = _safe_float(row.get("email_opens"), 0)
+    if opens >= 5:
+        rules.append(("Email Engagement", 5, f"{int(opens)} email opens"))
+
+    visitors = _safe_float(row.get("website_visits"), 0)
+    if visitors >= 5:
+        rules.append(("Website Activity", 5, f"{int(visitors)} visits"))
+
+    score = min(sum(r[1] for r in rules), 100)
+    factors = [{"rule": r[0], "points": r[1], "detail": r[2]} for r in rules]
+    return score, factors
+
+
+# ═══════════════════════════════════════════════════
+#  CHURN KEYWORDS FOR AUDIO ANALYSIS
+# ═══════════════════════════════════════════════════
+
+CHURN_KEYWORDS = [
+    ("cancel", 30),
+    ("cancel my account", 30),
+    ("not renewing", 30),
+    ("refund", 25),
+    ("chargeback", 25),
+    ("too expensive", 20),
+    ("cheaper", 20),
+    ("overpriced", 20),
+    ("can't afford", 20),
+    ("competitor", 15),
+    ("switching to", 15),
+    ("better option", 15),
+    ("not working", 20),
+    ("broken", 20),
+    ("bug", 15),
+    ("glitch", 15),
+    ("unusable", 20),
+    ("frustrated", 15),
+    ("fed up", 20),
+    ("done with this", 25),
+    ("never works", 20),
+    ("waste of money", 25),
+    ("leaving", 15),
+    ("close account", 25),
+    ("unsubscribe", 15),
+    ("stop service", 20),
+    ("downgrade", 10),
+    ("not happy", 15),
+    ("disappointed", 15),
+]
+
+
+def detect_churn_keywords(transcript: str) -> dict:
+    transcript_lower = transcript.lower()
+    flagged = []
+    score = 0
+
+    for keyword, points in CHURN_KEYWORDS:
+        if keyword in transcript_lower:
+            flagged.append(keyword)
+            score = max(score, points)
+
+    # Count total flagged keywords for aggregate signal
+    return {
+        "flagged_keywords": flagged,
+        "churn_intent_score": min(score, 100),
+        "flagged_keyword_count": len(flagged),
+    }