boladeE
97 lines
2.6 KiB
Python
97 lines
2.6 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
import pandas as pd
|
|
import numpy as np
|
|
import joblib
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from config import MODELS_DIR
|
|
from data_preprocessing import prepare_data
|
|
|
|
app = FastAPI(title="Fraud Detection API",
|
|
description="API for detecting fraudulent transactions",
|
|
version="1.0.0")
|
|
|
|
class Transaction(BaseModel):
|
|
trans_date_trans_time: str
|
|
cc_num: str
|
|
merchant: str
|
|
category: str
|
|
amt: float
|
|
first: str
|
|
last: str
|
|
gender: str
|
|
street: str
|
|
city: str
|
|
state: str
|
|
zip: str
|
|
lat: float
|
|
long: float
|
|
city_pop: int
|
|
job: str
|
|
dob: str
|
|
trans_num: str
|
|
unix_time: int
|
|
merch_lat: float
|
|
merch_long: float
|
|
|
|
class PredictionResponse(BaseModel):
|
|
is_fraud: bool
|
|
fraud_probability: float
|
|
confidence: str
|
|
|
|
def load_model():
|
|
"""Load the trained model and preprocessor."""
|
|
try:
|
|
model = joblib.load(MODELS_DIR / "fraud_model.joblib")
|
|
preprocessor = joblib.load(MODELS_DIR / "preprocessor.joblib")
|
|
return model, preprocessor
|
|
except FileNotFoundError:
|
|
raise HTTPException(status_code=500, detail="Model not found. Please train the model first.")
|
|
|
|
def get_confidence_level(probability: float) -> str:
|
|
"""Convert probability to confidence level."""
|
|
if probability >= 0.9:
|
|
return "Very High"
|
|
elif probability >= 0.7:
|
|
return "High"
|
|
elif probability >= 0.5:
|
|
return "Medium"
|
|
else:
|
|
return "Low"
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Welcome to the Fraud Detection API"}
|
|
|
|
@app.post("/predict", response_model=PredictionResponse)
|
|
async def predict(transaction: Transaction):
|
|
"""Predict whether a transaction is fraudulent."""
|
|
try:
|
|
# Load model and preprocessor
|
|
model, preprocessor = load_model()
|
|
|
|
# Convert transaction to DataFrame
|
|
transaction_dict = transaction.dict()
|
|
df = pd.DataFrame([transaction_dict])
|
|
|
|
# Prepare data for prediction
|
|
X, _, _ = prepare_data(df, preprocessor=preprocessor)
|
|
|
|
# Make prediction
|
|
probability = model.predict_proba(X)[0, 1]
|
|
is_fraud = probability >= 0.5
|
|
|
|
return PredictionResponse(
|
|
is_fraud=bool(is_fraud),
|
|
fraud_probability=float(probability),
|
|
confidence=get_confidence_level(probability)
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|