|
from fastapi import FastAPI, HTTPException |
|
from pydantic import BaseModel |
|
from typing import Dict, Union |
|
from fraud_detection_pipeline import load_pipeline |
|
|
|
app = FastAPI() |
|
pipeline = load_pipeline() |
|
|
|
class PredictionInput(BaseModel): |
|
account_age: int |
|
cred_changes_freq: float |
|
return_order_ratio: float |
|
vpn_usage: int |
|
credit_score: int |
|
|
|
@app.post("/predict") |
|
async def predict(input_data: PredictionInput) -> Dict[str, Union[str, float]]: |
|
try: |
|
features = input_data.dict() |
|
prediction = pipeline._forward(features) |
|
return prediction |
|
except Exception as e: |
|
raise HTTPException(status_code=500, detail=str(e)) |
|
|
|
@app.get("/") |
|
async def root(): |
|
return {"message": "Fraud Detection API is running"} |
