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