|
|
import os |
|
|
import json |
|
|
import joblib |
|
|
import numpy as np |
|
|
import onnxruntime as ort |
|
|
from fastapi import FastAPI |
|
|
|
|
|
MODEL_PATH = "models/fraud_model.onnx" |
|
|
SCALER_PATH = "models/scaler.joblib" |
|
|
META_PATH = "models/metadata.json" |
|
|
|
|
|
|
|
|
for p in [MODEL_PATH, SCALER_PATH, META_PATH]: |
|
|
if not os.path.exists(p): |
|
|
raise FileNotFoundError(f"Missing file: {p}") |
|
|
|
|
|
scaler = joblib.load(SCALER_PATH) |
|
|
|
|
|
with open(META_PATH) as f: |
|
|
meta = json.load(f) |
|
|
|
|
|
num_features = meta["num_features"] |
|
|
|
|
|
session = ort.InferenceSession( |
|
|
MODEL_PATH, |
|
|
providers=["CPUExecutionProvider"] |
|
|
) |
|
|
|
|
|
app = FastAPI(title="Fraud Detection API") |
|
|
|
|
|
@app.get("/") |
|
|
def health(): |
|
|
return {"status": "running"} |
|
|
|
|
|
@app.post("/predict") |
|
|
def predict(payload: dict): |
|
|
x = np.array([[payload[f] for f in num_features]], dtype=np.float32) |
|
|
x = scaler.transform(x) |
|
|
|
|
|
input_name = session.get_inputs()[0].name |
|
|
prob = session.run(None, {input_name: x})[0][0][0] |
|
|
|
|
|
return {"fraud_probability": float(prob)} |
|
|
|