File size: 2,477 Bytes
ae7c16d | 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 | from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, Optional
from fastapi import FastAPI
from fastapi import HTTPException
from pydantic import BaseModel, Field
from app.model import FraudModel
ARTIFACTS_PATH = Path(__file__).resolve().parents[1] / "artifacts" / "model.joblib"
app = FastAPI(title="Fraud Pattern Detector", version="1.0.0")
engine: Optional[FraudModel] = None
class Transaction(BaseModel):
TransactionAmt: float = Field(..., ge=0)
ProductCD: Optional[str] = None
card1: Optional[str] = None
card2: Optional[str] = None
card3: Optional[str] = None
card5: Optional[str] = None
card6: Optional[str] = None
addr1: Optional[str] = None
addr2: Optional[str] = None
P_emaildomain: Optional[str] = None
R_emaildomain: Optional[str] = None
DeviceType: Optional[str] = None
dist1: Optional[float] = None
dist2: Optional[float] = None
# C and D style numeric features (optional)
C1: Optional[float] = None
C2: Optional[float] = None
C3: Optional[float] = None
C4: Optional[float] = None
C5: Optional[float] = None
C6: Optional[float] = None
C7: Optional[float] = None
C8: Optional[float] = None
C9: Optional[float] = None
C10: Optional[float] = None
C11: Optional[float] = None
C12: Optional[float] = None
C13: Optional[float] = None
C14: Optional[float] = None
D1: Optional[float] = None
D2: Optional[float] = None
D3: Optional[float] = None
D4: Optional[float] = None
D5: Optional[float] = None
D10: Optional[float] = None
D11: Optional[float] = None
D15: Optional[float] = None
@app.on_event("startup")
def _load() -> None:
global engine
if ARTIFACTS_PATH.exists():
engine = FraudModel(ARTIFACTS_PATH)
else:
engine = None
@app.get("/health")
def health() -> Dict[str, Any]:
ok = ARTIFACTS_PATH.exists()
return {"ok": ok, "artifacts_path": str(ARTIFACTS_PATH)}
@app.post("/score")
def score(tx: Transaction) -> Dict[str, Any]:
if engine is None:
raise HTTPException(
status_code=503,
detail="Model artifacts not found. Train first (python -m scripts.train_xgb) to create artifacts/model.joblib.",
)
res = engine.score(tx.model_dump(), top_k=8)
return {
"probability_fraud": res.probability,
"decision": res.decision,
"top_factors": res.top_factors,
}
|