Predictive Customer Churn and Demand Forecasting: Deploying Production MLOps with MLflow and FastAPI

import mlflow.pyfunc
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import pandas as pd

app = FastAPI(title="XIYOR Customer Churn Prediction Service")

# Load production model from MLflow registry at startup
MODEL_URI = "models:/CustomerChurnXGBoost/Production"
model = None

@app.on_event("startup")
def load_model():
    global model
    model = mlflow.pyfunc.load_model(MODEL_URI)

class CustomerFeatures(BaseModel):
    tenure_months: int
    monthly_charges: float
    total_support_tickets: int
    contract_type_annual: int  # 1 for Yes, 0 for No
    avg_session_duration_min: float

@app.post("/predict-churn")
async def predict_churn(features: CustomerFeatures):
    if not model:
        raise HTTPException(status_code=500, detail="Model engine not loaded")
    
    # Convert Pydantic model to DataFrame for inference
    input_df = pd.DataFrame([features.dict()])
    
    # Execute model prediction probability
    churn_probability = float(model.predict(input_df)[0])
    risk_level = "HIGH" if churn_probability > 0.7 else ("MEDIUM" if churn_probability > 0.3 else "LOW")
    
    return {
        "churn_probability": round(churn_probability, 4),
        "risk_level": risk_level,
        "recommended_action": "TRIGGER_RETENTION_OFFER" if risk_level == "HIGH" else "NO_ACTION"
    }