app.py

import os
import uvicorn
import pandas as pd
from pydantic import BaseModel
from fastapi import FastAPI
import mlflow
import os
from dotenv import load_dotenv

description = """

# Fraud Detection API

Fast API backend to predict if a payment is fraud or not from : https://huggingface.co/spaces/sdacelo/real-time-fraud-detection

## Machine-Learning 

Where you can:
* `/predict` : prediction for a single value

Check out documentation for more information on each endpoint. 
"""

tags_metadata = [
    {
        "name": "Predictions",
        "description": "Endpoints that uses our Machine Learning model",
    },
]

load_dotenv()

# Mlflow variables
MLFLOW_TRACKING_APP_URI = os.getenv("MLFLOW_TRACKING_APP_URI")
MODEL_NAME = os.getenv("MODEL_NAME", "fraud_detection_dtc")
STAGE = os.getenv("STAGE", "production")

# AWS variables
os.environ["AWS_ACCESS_KEY_ID"] = os.getenv("AWS_ACCESS_KEY_ID")
os.environ["AWS_SECRET_ACCESS_KEY"] = os.getenv("AWS_SECRET_ACCESS_KEY")

# Load model
mlflow.set_tracking_uri(MLFLOW_TRACKING_APP_URI)
model_uri = f"models:/{MODEL_NAME}@{STAGE}"
model = mlflow.sklearn.load_model(model_uri)
if model is None:
    raise ValueError("Model not found.")
else:
    print("Model loaded.") 

app = FastAPI(
    title="API for Fraud Detection Project",
    description=description,
    version="1.0",
    contact={
        "name": "Olivier-52",
        "url": "https://huggingface.co/Olivier-52",
    },
    openapi_tags=tags_metadata,)

@app.get("/")
def index():
    """Return a message to the user.

    This endpoint does not take any parameters and returns a message
    to the user. It is used to test the API.

    Returns:
        str: A message to the user.
    """
    return "Hello world! Go to /docs to try the API."

class PredictionFeatures(BaseModel):
    category : str
    amt : float
    merch_fraud_level :str
    city_fraud_level : str
    state_fraud_level : str
    zip_fraud_level : str
    job_fraud_level : str
    age_fraud_level : str

@app.post("/predict", tags=["Predictions"])
def predict(features: PredictionFeatures):

    """
    Predict if a payment is fraud or not.

    This endpoint takes a set of features as input and returns a prediction
    of whether the payment is fraud or not.

    Parameters:
        features (PredictionFeatures): A set of features describing the payment

    Returns:
        dict: A dictionary containing the prediction of whether the payment is fraud or not
    """
    data = pd.DataFrame({
        "category" : [features.category],
        "amt" : [features.amt],
        "merch_fraud_level" : [features.merch_fraud_level],
        "city_fraud_level" : [features.city_fraud_level],
        "state_fraud_level" : [features.state_fraud_level],
        "zip_fraud_level" : [features.zip_fraud_level],
        "job_fraud_level" : [features.job_fraud_level],
        "age_fraud_level" : [features.age_fraud_level]
    })
    
    try:
        prediction = model.predict(data)[0]
        return {"prediction": int(prediction)}
    
    except Exception as e:
        return {"error": str(e)}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)