Spaces:
Sleeping
Sleeping
File size: 3,158 Bytes
9cdea9f 9718e97 9cdea9f c3530e3 9cdea9f 5e295a6 9cdea9f c3530e3 9cdea9f c3530e3 9cdea9f 2741c49 9cdea9f c3530e3 9cdea9f b828e05 ac50d1a 2c009f0 b828e05 9cdea9f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | 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)
|