| """SENTINEL scoring API (create_app factory). |
| |
| Run: API_KEYS=dev-key-change-me uvicorn api.main:app --port 8000 |
| """ |
|
|
| import json |
| import logging |
| import threading |
| import time |
| import uuid |
| from contextlib import asynccontextmanager |
| from datetime import datetime, timezone |
|
|
| import numpy as np |
| import pandas as pd |
| from fastapi import FastAPI, HTTPException, Query, Request |
| from fastapi.exceptions import RequestValidationError |
| from fastapi.middleware.cors import CORSMiddleware |
| from fastapi.responses import FileResponse, JSONResponse |
|
|
| from src import __version__ |
| from src.decision import risk_level |
| from src.features import RAW_FEATURES, make_features |
| from src.model_bundle import ArtifactError, ModelBundle |
| from src.reasons import DISCLAIMER, build_reasons |
|
|
| from .monitoring import Monitor |
| from .schemas import BatchRequest, BatchResponse, Prediction, Transaction |
| from .security import AccessDep, SlidingWindowLimiter, configure_logging, new_request_id, request_id_var |
| from .settings import Settings |
|
|
| logger = logging.getLogger("sentinel.api") |
|
|
|
|
| def _now() -> str: |
| return datetime.now(timezone.utc).isoformat() |
|
|
|
|
| class State: |
| """Everything loaded at startup lives here (no module-level globals).""" |
|
|
| def __init__(self): |
| self.bundle = None |
| self.load_error = None |
| self.metrics = {} |
| self.reference = {} |
| self.replay = None |
| self.replay_proba = np.array([]) |
| self.replay_importance = [] |
| self.replay_offset = 0 |
| self.explain_offset = 0 |
| self.replay_lock = threading.Lock() |
| self.monitor = None |
|
|
|
|
| def load_state(settings: Settings) -> State: |
| st = State() |
| d = settings.model_dir |
| try: |
| st.bundle = ModelBundle.load(d, strict_versions=settings.strict_versions) |
| except (ArtifactError, OSError, ValueError) as exc: |
| st.load_error = str(exc) |
| logger.error("Model not loaded: %s", exc) |
| st.monitor = Monitor({}, settings.drift_window, settings.drift_min_samples, settings.log_every) |
| return st |
|
|
| for name, attr in (("metrics.json", "metrics"), ("reference_stats.json", "reference")): |
| try: |
| setattr(st, attr, json.loads((d / name).read_text(encoding="utf-8"))) |
| except (OSError, json.JSONDecodeError): |
| logger.warning("%s unavailable", name) |
| st.monitor = Monitor(st.reference.get("psi_features", {}), settings.drift_window, |
| settings.drift_min_samples, settings.log_every) |
|
|
| replay_path = d / "replay_sample.csv" |
| if replay_path.exists(): |
| try: |
| df = pd.read_csv(replay_path) |
| st.replay = df.reset_index(drop=True) |
| st.replay_proba = st.bundle.predict_proba(df[RAW_FEATURES]) |
| _, contrib = st.bundle.contributions(df[RAW_FEATURES]) |
| mean_abs = np.abs(contrib).mean(axis=0) |
| ranked = sorted(zip(st.bundle.feature_names, mean_abs), key=lambda kv: kv[1], reverse=True)[:15] |
| st.replay_importance = [{"feature": n, "importance": round(float(v), 6)} for n, v in ranked] |
| except Exception: |
| logger.exception("Could not score replay sample") |
| st.replay = None |
| logger.info("Loaded model %s (threshold %.4f)", st.bundle.version, st.bundle.threshold) |
| return st |
|
|
|
|
| def create_app(settings: Settings = None) -> FastAPI: |
| settings = settings or Settings() |
| configure_logging(settings.log_level) |
|
|
| @asynccontextmanager |
| async def lifespan(app: FastAPI): |
| app.state.sentinel = load_state(settings) |
| yield |
|
|
| app = FastAPI(title="SENTINEL Fraud Detection API", version=__version__, lifespan=lifespan) |
| app.state.settings = settings |
| app.state.limiter = SlidingWindowLimiter(settings.rate_limit_per_min) |
| app.state.sentinel = State() |
| app.state.sentinel.monitor = Monitor({}, 1, 1, 1) |
|
|
| app.add_middleware(CORSMiddleware, allow_origins=list(settings.allowed_origins), |
| allow_methods=["GET", "POST"], allow_headers=["*"]) |
|
|
| |
| @app.middleware("http") |
| async def request_context(request: Request, call_next): |
| rid = new_request_id(request.headers.get("x-request-id")) |
| token = request_id_var.set(rid) |
| request.state.request_id = rid |
| started = time.perf_counter() |
| try: |
| response = await call_next(request) |
| response.headers["X-Request-ID"] = rid |
| logger.info("%s %s -> %s (%.1f ms)", request.method, request.url.path, response.status_code, |
| (time.perf_counter() - started) * 1000) |
| return response |
| finally: |
| request_id_var.reset(token) |
|
|
| @app.exception_handler(RequestValidationError) |
| async def validation_error(request: Request, exc: RequestValidationError): |
| |
| errors = [{"loc": list(e["loc"]), "msg": e["msg"], "type": e["type"]} for e in exc.errors()] |
| return JSONResponse({"detail": errors}, status_code=422) |
|
|
| @app.exception_handler(Exception) |
| async def unhandled(request: Request, exc: Exception): |
| rid = getattr(request.state, "request_id", "-") |
| logger.error("Unhandled error on %s %s [%s]", request.method, request.url.path, rid, exc_info=exc) |
| return JSONResponse({"detail": "Internal server error", "request_id": rid}, status_code=500, |
| headers={"X-Request-ID": rid}) |
|
|
| |
| def state() -> State: |
| return app.state.sentinel |
|
|
| def require_model() -> State: |
| st = state() |
| if st.bundle is None: |
| raise HTTPException(503, "Model is not loaded; service is not ready") |
| return st |
|
|
| def explain_rows(st: State, sub: pd.DataFrame): |
| """TreeSHAP contributions + reason codes for raw rows. Returns (contrib[n, f], [reasons per row]).""" |
| bundle = st.bundle |
| _, contrib = bundle.contributions(sub) |
| unscaled = make_features(sub) |
| stats = st.reference.get("feature_stats", {}) |
| names = bundle.feature_names |
| hours = ((sub["Time"].values % 86400) // 3600).astype(int) |
| out = [build_reasons(names, contrib[i], unscaled.iloc[i][names].values, stats, hour=int(hours[i])) |
| for i in range(len(sub))] |
| return contrib, out |
|
|
| def score(st: State, txns, explain: bool): |
| started = time.perf_counter() |
| raw = pd.DataFrame([t.model_dump() for t in txns])[RAW_FEATURES] |
| bundle = st.bundle |
| proba = bundle.predict_proba(raw) |
| flagged = proba >= bundle.threshold |
|
|
| reasons = {} |
| if explain and flagged.any(): |
| idx = np.where(flagged)[0] |
| _, per_row = explain_rows(st, raw.iloc[idx]) |
| reasons = {int(orig): per_row[k] for k, orig in enumerate(idx)} |
| results = [] |
| for i, t in enumerate(txns): |
| p = float(proba[i]) |
| results.append({ |
| "transaction_id": str(uuid.uuid4()), |
| "is_fraud": bool(flagged[i]), |
| "fraud_probability": round(p, 6), |
| "risk_level": risk_level(p, bundle.threshold), |
| "threshold": round(bundle.threshold, 6), |
| "amount": t.Amount, |
| "timestamp": _now(), |
| "model_version": bundle.version, |
| "reasons": reasons.get(i), |
| }) |
| drift_keys = list(st.reference.get("psi_features", {}).keys()) |
| drift_rows = [{k: float(raw.iloc[i][k]) for k in drift_keys if k in raw.columns} for i in range(len(raw))] |
| st.monitor.record(len(raw), int(flagged.sum()), (time.perf_counter() - started) * 1000, drift_rows) |
| return results |
|
|
| |
| @app.get("/live") |
| def live(): |
| return {"status": "alive"} |
|
|
| @app.get("/ready") |
| def ready(): |
| st = state() |
| if st.bundle is None: |
| return JSONResponse({"status": "not_ready", "reason": "model not loaded"}, status_code=503) |
| return {"status": "ready", "model_version": st.bundle.version} |
|
|
| @app.get("/health") |
| def health(): |
| st = state() |
| ok = st.bundle is not None |
| body = {"status": "ready" if ok else "not_ready", "model_loaded": ok, |
| "model_version": st.bundle.version if ok else None, "timestamp": _now()} |
| return JSONResponse(body, status_code=200 if ok else 503) |
|
|
| @app.get("/", include_in_schema=False) |
| def index(): |
| if not settings.frontend_path.exists(): |
| raise HTTPException(404, "Dashboard not found") |
| return FileResponse(settings.frontend_path) |
|
|
| |
| @app.post("/predict", response_model=Prediction, dependencies=[AccessDep]) |
| def predict(t: Transaction, explain: bool = Query(False, description="Add reason codes (flagged transactions only)")): |
| st = require_model() |
| return score(st, [t], explain)[0] |
|
|
| @app.post("/predict/batch", response_model=BatchResponse, dependencies=[AccessDep]) |
| def predict_batch(req: BatchRequest, explain: bool = Query(False)): |
| st = require_model() |
| results = score(st, req.transactions, explain) |
| fraud = sum(r["is_fraud"] for r in results) |
| return {"total": len(results), "fraud_count": fraud, "fraud_rate": round(fraud / len(results), 6), |
| "model_version": st.bundle.version, "reasons_note": DISCLAIMER if explain else None, |
| "results": results} |
|
|
| |
| @app.get("/model/info", dependencies=[AccessDep]) |
| def model_info(): |
| st = require_model() |
| m = st.bundle.metadata |
| return {"model_type": "XGBoost", "model_version": m["model_version"], "strategy": m["strategy"], |
| "calibration": m["calibration_method"], "threshold": m["threshold"], |
| "risk_bands": m["risk_bands"], "costs": m["costs"], "features": len(m["feature_names"]), |
| "feature_names": m["feature_names"], "input_schema": m["raw_input_features"], |
| "data": {k: m["data"][k] for k in ("source", "synthetic", "rows", "frauds", "split_kind")}, |
| "library_versions": m["library_versions"], "trained_at": m["trained_at"]} |
|
|
| @app.get("/stats/summary", dependencies=[AccessDep]) |
| def stats(): |
| st = require_model() |
| t = st.metrics.get("test", {}) |
| val = lambda k: t.get(k, {}).get("value", 0.0) |
| total = flagged = caught = 0 |
| saved = 0.0 |
| if st.replay is not None and len(st.replay_proba): |
| f = st.replay_proba >= st.bundle.threshold |
| total, flagged = len(f), int(f.sum()) |
| label = st.replay["Class"].values == 1 |
| caught = int((f & label).sum()) |
| saved = round(float(st.replay.loc[f & label, "Amount"].sum()), 2) |
| snap = st.monitor.snapshot() |
| return { |
| "source": "model/metrics.json (held-out test split)", |
| "synthetic_data": st.metrics.get("synthetic_data", None), |
| "model_version": st.bundle.version, |
| "threshold": st.bundle.threshold, |
| "replay_total": total, "replay_flagged": flagged, "replay_true_frauds_caught": caught, |
| "replay_flag_rate": round(flagged / total, 4) if total else 0.0, |
| "replay_fraud_amount_caught": saved, |
| "avg_response_ms": round(snap["latency_ms"]["mean"], 3), |
| "pr_auc": val("pr_auc"), "roc_auc": val("roc_auc"), "precision": val("precision"), |
| "recall": val("recall"), "f1": val("f1"), "accuracy": val("accuracy"), |
| "ci95": {k: t[k]["ci95"] for k in ("pr_auc", "precision", "recall", "f1") if k in t}, |
| "confusion": st.metrics.get("test_confusion"), |
| } |
|
|
| @app.get("/shap/global", dependencies=[AccessDep]) |
| def shap_global(): |
| st = require_model() |
| return {"method": "mean |TreeSHAP| on replay sample (log-odds)", "feature_importance": st.replay_importance} |
|
|
| @app.get("/replay/feed", dependencies=[AccessDep]) |
| def replay_feed(): |
| """Rolling 20-row window over held-out test rows, scored by the loaded model. This is a replay, not live traffic.""" |
| st = require_model() |
| if st.replay is None or not len(st.replay_proba): |
| return {"transactions": [], "count": 0, "source": "replay_of_held_out_test_data"} |
| n = len(st.replay_proba) |
| size = min(20, n) |
| with st.replay_lock: |
| start = st.replay_offset |
| st.replay_offset = (start + min(5, n)) % n |
| rows = [] |
| for pos, i in enumerate([(start + k) % n for k in range(size)]): |
| p = float(st.replay_proba[i]) |
| r = st.replay.iloc[i] |
| rows.append({ |
| "id": f"REPLAY-{i:05d}", "row_index": i, "amount": round(float(r["Amount"]), 2), |
| "time_seconds": float(r["Time"]), "fraud_probability": round(p, 6), |
| "risk_level": risk_level(p, st.bundle.threshold), |
| "is_fraud": p >= st.bundle.threshold, "true_label": int(r["Class"]), |
| }) |
| return {"transactions": rows, "count": len(rows), "source": "replay_of_held_out_test_data"} |
|
|
| @app.get("/replay/threshold", dependencies=[AccessDep]) |
| def replay_threshold(value: float = Query(..., ge=0.0, le=1.0)): |
| """Real precision/recall/cost at an arbitrary threshold, measured on the labelled replay sample.""" |
| st = require_model() |
| if st.replay is None: |
| raise HTTPException(404, "No replay sample available") |
| y = st.replay["Class"].values.astype(int) |
| flagged = st.replay_proba >= value |
| data = st.bundle.metadata["data"] |
| test_frauds = data["split_frauds"]["test"] |
| test_legit = data["split_rows"]["test"] - test_frauds |
| legit = y == 0 |
|
|
| if "weight" in st.replay.columns: |
| wts = st.replay["weight"].values.astype(float) |
| else: |
| n_legit = int(legit.sum()) |
| wts = np.where(legit, test_legit / n_legit if n_legit else 1.0, 1.0) |
| weighted = int(y.sum()) == test_frauds |
| if not weighted: |
| wts = np.ones(len(y)) |
|
|
| tp, fn = int((flagged & (y == 1)).sum()), int((~flagged & (y == 1)).sum()) |
| fp_s = int((flagged & legit).sum()) |
| fp = float(wts[flagged & legit].sum()) |
| tn = float(wts[~flagged & legit].sum()) |
| hard = legit & (wts == 1.0) |
| exact_from = float(st.replay_proba[hard].min()) if weighted and hard.any() else None |
| exact = bool(weighted and exact_from is not None and value >= exact_from) |
|
|
| precision = tp / (tp + fp) if tp + fp else 0.0 |
| recall = tp / (tp + fn) if tp + fn else 0.0 |
| costs = st.bundle.metadata["costs"] |
| if exact: |
| note = (f"Exact for the full held-out test split at this threshold (all {int(y.sum())} frauds and every " |
| f"legitimate row scoring >= {exact_from:.3f} are in the replay sample).") |
| elif weighted: |
| note = (f"Estimated for the full test split: below {exact_from if exact_from is not None else 0:.3f} the " |
| f"false-alarm count is scaled up from a random sample of easy rows, so it is noisy. " |
| f"Only {int(y.sum())} frauds: a guide, not a guarantee.") |
| else: |
| note = ("Measured on the raw replay sample (not prevalence-corrected); precision and cost are not " |
| "comparable to metrics.json.") |
| return {"threshold": value, "model_threshold": st.bundle.threshold, "rows": len(y), "frauds": int(y.sum()), |
| "weighted": weighted, "exact": exact, "exact_from": exact_from, |
| "tp": tp, "fp": int(round(fp)), "fn": fn, "tn": int(round(tn)), "fp_sampled": fp_s, |
| "precision": round(precision, 4), "recall": round(recall, 4), |
| "f1": round(2 * precision * recall / (precision + recall), 4) if precision + recall else 0.0, |
| "expected_cost": round(fn * costs["false_negative"] + fp * costs["false_positive"], 2), |
| "note": note} |
|
|
| @app.get("/replay/explain", dependencies=[AccessDep]) |
| def replay_explain(flagged: bool = Query(True)): |
| """Real TreeSHAP explanation for the next flagged (or unflagged) replay row.""" |
| st = require_model() |
| if st.replay is None: |
| raise HTTPException(404, "No replay sample available") |
| mask = st.replay_proba >= st.bundle.threshold |
| pool = np.where(mask if flagged else ~mask)[0] |
| if not len(pool): |
| raise HTTPException(404, "No matching rows in the replay sample") |
| with st.replay_lock: |
| i = int(pool[st.explain_offset % len(pool)]) |
| st.explain_offset += 1 |
| sub = st.replay.iloc[[i]][RAW_FEATURES] |
| contrib, reasons = explain_rows(st, sub) |
| names = st.bundle.feature_names |
| order = np.argsort(-np.abs(contrib[0]))[:10] |
| p = float(st.replay_proba[i]) |
| return {"row_index": i, "amount": float(sub["Amount"].iloc[0]), "fraud_probability": round(p, 6), |
| "is_fraud": p >= st.bundle.threshold, "true_label": int(st.replay["Class"].iloc[i]), |
| "risk_level": risk_level(p, st.bundle.threshold), |
| "contributions": [{"feature": names[j], "contribution": round(float(contrib[0][j]), 4)} for j in order], |
| "reasons": reasons[0], "note": DISCLAIMER} |
|
|
| |
| @app.get("/monitoring/metrics", dependencies=[AccessDep]) |
| def monitoring_metrics(): |
| require_model() |
| return state().monitor.snapshot() |
|
|
| @app.get("/monitoring/drift", dependencies=[AccessDep]) |
| def monitoring_drift(): |
| require_model() |
| return state().monitor.drift_report() |
|
|
| return app |
|
|
|
|
| app = create_app() |
|
|
