| 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 |
|
|
| |
| 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, |
| } |
|
|
|
|
