|
|
|
import os |
|
import sys |
|
|
|
import uvicorn |
|
from fastapi import FastAPI, Request, File, UploadFile |
|
from fastapi.responses import HTMLResponse, JSONResponse |
|
from fastapi.staticfiles import StaticFiles |
|
from fastapi.templating import Jinja2Templates |
|
import pandas as pd |
|
import numpy as np |
|
from typing import List |
|
import joblib |
|
from fastapi import FastAPI |
|
from pydantic import BaseModel |
|
|
|
|
|
app = FastAPI(debug=True) |
|
|
|
|
|
num_imputer = joblib.load('numerical_imputer(1).joblib') |
|
cat_imputer = joblib.load('cat_imputer.joblib') |
|
encoder = joblib.load('encoder.joblib') |
|
scaler = joblib.load('scaler.joblib') |
|
model = joblib.load('lr_model_vif_smote.joblib') |
|
|
|
original_feature_names = ['MONTANT', 'FREQUENCE_RECH', 'REVENUE', 'ARPU_SEGMENT', 'FREQUENCE', |
|
'DATA_VOLUME', 'ON_NET', 'ORANGE', 'TIGO', 'ZONE1', 'ZONE2', 'REGULARITY', 'FREQ_TOP_PACK', |
|
'REGION_DAKAR', 'REGION_DIOURBEL', 'REGION_FATICK', 'REGION_KAFFRINE', 'REGION_KAOLACK', |
|
'REGION_KEDOUGOU', 'REGION_KOLDA', 'REGION_LOUGA', 'REGION_MATAM', 'REGION_SAINT-LOUIS', |
|
'REGION_SEDHIOU', 'REGION_TAMBACOUNDA', 'REGION_THIES', 'REGION_ZIGUINCHOR', |
|
'TENURE_Long-term', 'TENURE_Medium-term', 'TENURE_Mid-term', 'TENURE_Short-term', |
|
'TENURE_Very short-term', 'TOP_PACK_data', 'TOP_PACK_international', 'TOP_PACK_messaging', |
|
'TOP_PACK_other_services', 'TOP_PACK_social_media', 'TOP_PACK_value_added_services', |
|
'TOP_PACK_voice'] |
|
|
|
|
|
class InputData(BaseModel): |
|
MONTANT: float |
|
FREQUENCE_RECH: float |
|
REVENUE: float |
|
ARPU_SEGMENT: float |
|
FREQUENCE: float |
|
DATA_VOLUME: float |
|
ON_NET: float |
|
ORANGE: float |
|
TIGO: float |
|
ZONE1: float |
|
ZONE2: float |
|
REGULARITY: float |
|
FREQ_TOP_PACK: float |
|
REGION: str |
|
TENURE: str |
|
TOP_PACK: str |
|
|
|
|
|
def preprocess_input(input_data): |
|
input_df = pd.DataFrame(input_data, index=[0]) |
|
|
|
cat_columns = ['REGION', 'TENURE', 'TOP_PACK'] |
|
num_columns = [col for col in input_df.columns if col not in cat_columns] |
|
|
|
input_df_imputed_cat = cat_imputer.transform(input_df[cat_columns]) |
|
input_df_imputed_num = num_imputer.transform(input_df[num_columns]) |
|
|
|
input_encoded_df = pd.DataFrame(encoder.transform(input_df_imputed_cat).toarray(), |
|
columns=encoder.get_feature_names_out(cat_columns)) |
|
|
|
input_df_scaled = scaler.transform(input_df_imputed_num) |
|
input_scaled_df = pd.DataFrame(input_df_scaled, columns=num_columns) |
|
final_df = pd.concat([input_encoded_df, input_scaled_df], axis=1) |
|
final_df = final_df.reindex(columns=original_feature_names, fill_value=0) |
|
|
|
return final_df |
|
|
|
|
|
def make_prediction(data, model): |
|
probabilities = model.predict_proba(data) |
|
churn_labels = ["No Churn" if class_idx == 0 else "Churn" for class_idx in range(len(probabilities[0]))] |
|
churn_probabilities = probabilities[0] |
|
|
|
|
|
predicted_class_index = np.argmax(churn_probabilities) |
|
predicted_churn_label = churn_labels[predicted_class_index] |
|
predicted_probability = churn_probabilities[predicted_class_index] |
|
|
|
|
|
if predicted_churn_label == "Churn": |
|
output_message = f"⚠️ Customer is likely to churn with a probability of {predicted_probability:.2f}. This indicates a high risk of losing the customer. ⚠️" |
|
else: |
|
output_message = f"✅ Customer is not likely to churn with a probability of {predicted_probability:.2f}. This indicates a lower risk of losing the customer. ✅" |
|
|
|
return output_message |
|
|
|
@app.get("/") |
|
def read_root(): |
|
|
|
info = """ |
|
Welcome to the Expressor Churn Prediction API!. This API provides advanced machine learning predictions for churn. ⚡📊 For more information and to explore the API's capabilities, please visit the documentation: https://abubakari-expresso-churn-prediction-fastapi.hf.space/docs |
|
""" |
|
return info.strip() |
|
|
|
|
|
@app.post('/model-info') |
|
async def model_info(): |
|
model_name = model.__class__.__name__ |
|
model_params = model.get_params() |
|
model_information = { |
|
'model info': { |
|
'model name': model_name, |
|
'model parameters': model_params |
|
} |
|
} |
|
return model_information |
|
|
|
@app.post('/predict') |
|
async def predict(input_data: InputData): |
|
input_features = input_data.dict() |
|
preprocessed_data = preprocess_input(input_features) |
|
prediction = make_prediction(preprocessed_data, model) |
|
|
|
return {"prediction": prediction} |
|
|
|
|
|
@app.post('/batch_predict') |
|
async def predict(input_data: List[InputData]): |
|
preprocessed_data = [] |
|
|
|
for data in input_data: |
|
input_features = data.dict() |
|
preprocessed = preprocess_input(input_features) |
|
preprocessed_data.append(preprocessed) |
|
|
|
predictions = [make_prediction(data, model) for data in preprocessed_data] |
|
|
|
return {"predictions": predictions} |
|
|
|
|
