From Research to Production: How I Built a Customer Churn Prediction API That Actually Works

    from sklearn.base import BaseEstimator, TransformerMixin
    from sklearn.compose import ColumnTransformer
    from sklearn.pipeline import Pipeline
    from sklearn.preprocessing import OneHotEncoder

    class durationTransform(BaseEstimator, TransformerMixin):
        def fit(self, x, y=None):
            return self

        def transform(self, x):
             #Handle both DataFrame and numpy array inputs
            if isinstance(x, pd.DataFrame):
                db = x.copy()
            else:
                db = pd.DataFrame(x, columns=["transactiondate", "membershipexpiredate"])

             #Calculate subscription duration in days
            db["transactiondate"] = pd.todatetime(db["transactiondate"])
            db["membershipexpiredate"] = pd.todatetime(db["membershipexpiredate"])

            result = (db["membershipexpiredate"] - db["transactiondate"]).dt.days
            return result.values.reshape(-1, 1)

     #Build the complete pipeline
    genencoding = ColumnTransformer([
        ("gender", OneHotEncoder(), [1])
    ], remainder='passthrough')

    substime = ColumnTransformer([
        ("durationindays", durationTransform(), [8, 9])
    ], remainder='passthrough')

    pipe = Pipeline([
        ('genencoding', genencoding),
        ('substime', substime)
    ])

    from sklearn.utils import resample

     Original data distribution
    zeros = dbtrain[dbtrain['ischurn'] == 0]   9,354 non-churners
    ones = dbtrain[dbtrain['ischurn'] == 1]    646 churners

     #Undersampling to balance the dataset
    zerosundersampled = resample(zeros, 
                                 replace=False, 
                                 nsamples=len(ones), 
                                 randomstate=42)

     #Combine and shuffle
    dbtrain = pd.concat([zerosundersampled, ones])
    dbtrain = dbtrain.sample(frac=1, randomstate=42).resetindex(drop=True)

from fastapi import FastAPI, HTTPException
    from pydantic import BaseModel
    from typing import Optional
    from datetime import date

    app = FastAPI(title="Customer Churn Prediction", 
                  description="Production-ready ML API for customer churn prediction",
                  version='1.0.0')

     #Pydantic model for data validation
    class dataval(BaseModel):
        userid: Optional[int] = None
        city: int
        gender: str
        registeredvia: int
        paymentmethodid: int
        paymentplandays: int
        actualamountpaid: int
        isautorenew: int
        transactiondate: date
        membershipexpiredate: date

    @app.post("/predict")
    def predict(
        city: int,
        gender: str,
        registeredvia: int,
        paymentmethodid: int,
        paymentplandays: int,
        actualamountpaid: int,
        isautorenew: int,
        transactiondate: date,
        membershipexpiredate: date,
        userid: Optional[int] = None
    ):
        # Validate input data
        data = dataval(
            userid=userid,
            city=city,
            gender=gender,
            registeredvia=registeredvia,
            paymentmethodid=paymentmethodid,
            paymentplandays=paymentplandays,
            actualamountpaid=actualamountpaid,
            isautorenew=isautorenew,
            transactiondate=transactiondate,
            membershipexpiredate=membershipexpiredate
        )

         # Generate user ID if not provided
        user = validuser(data.userid)

         Transform data through pipeline
        pipedata = [[
            data.city, data.gender, data.registeredvia,
            data.paymentmethodid, data.paymentplandays,
            data.actualamountpaid, data.isautorenew,
            data.transactiondate, data.membershipexpiredate
        ]]

        try:
            transformed = pipe.transform(pipedata)
            dftransformed = pd.DataFrame(transformed, 
                columns=["durationofsubscription", "female", "male", "city", 
                        "registeredvia", "paymentmethodid", "paymentplandays", 
                        "actualamountpaid", "isautorenew"])

            # Make prediction
            prediction = model.predict(dftransformed)
            result = {user: dftransformed.iloc[0].todict()}
            result[user]["prediction"] = int(prediction[0])

            # Store result
            saveprediction(result)

            return result

        except Exception as e:
            raise HTTPException(statuscode=500, 
                              detail=f"Prediction failed: {str(e)}")

import json
    import os

    def validuser(user: int):
        """Generate or validate user IDs with persistent storage"""
        if pd.isna(user):
            with open("data/users.json", "r") as f:
                data = json.load(f)
                maxuser = max(data)
                user = maxuser + 1
                data.append(int(user))
                with open("data/users.json", "w") as f:
                    json.dump(data, f, indent=2)
            return user
        else:
            with open("data/users.json", "r") as f:
                data = json.load(f)
                if user not in data:
                    data.append(int(user))
                    with open("data/users.json", "w") as f:
                        json.dump(data, f, indent=2)
            return user

    def saveprediction(result: dict):
        """Persist prediction results with user data"""
        jsonpath = "data/userdata.json"

        if os.path.exists(jsonpath):
            with open(jsonpath, "r") as f:
                jsonfile = json.load(f)
        else:
            jsonfile = {}

        jsonfile.update(result)

        with open(jsonpath, "w") as f:
            json.dump(jsonfile, f, indent=2)

import cloudpickle

     Save the trained model and pipeline
    with open("model/model.pickle", "wb") as f:
        cloudpickle.dump(adaboostmodel, f)

    with open("model/pipe.pickle", "wb") as f:
        cloudpickle.dump(pipe, f)

     Load in production
    with open("model/model.pickle", "rb") as f:
        model = cloudpickle.load(f)

    with open("model/pipe.pickle", "rb") as f:
        pipe = cloudpickle.load(f)

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Jupyter       │    │   FastAPI       │    │   Production    │
│   Notebook      │───▶│   Service       │───▶│   Deployment    │
│                 │    │                 │    │                 │
│ • Data EDA      │    │ • REST API      │    │ • Load Balancer │
│ • Model Training│    │ • Validation    │    │ • Auto-scaling  │
│ • Pipeline Dev  │    │ • Error Handling│    │ • Monitoring    │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │                       │
         ▼                       ▼                       ▼
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Custom        │    │   Pydantic      │    │   JSON Storage  │
│   Transformers  │    │   Models        │    │   & User Mgmt   │
└─────────────────┘    └─────────────────┘    └─────────────────┘