Blame for api/main.py

"""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            # DataFrame incl. Class
        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)  # replaced in lifespan

    app.add_middleware(CORSMiddleware, allow_origins=list(settings.allowed_origins),
                       allow_methods=["GET", "POST"], allow_headers=["*"])

    # ------------------------------------------------------------ middleware
    @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):
        # Never echo the submitted values back (they may be NaN/huge/sensitive); report location + message only.
        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})

    # ------------------------------------------------------------ helpers
    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

    # ------------------------------------------------------------ probes (open)
    @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)

    # ------------------------------------------------------------ prediction
    @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}

    # ------------------------------------------------------------ info / stats
    @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)  # noqa: E731
        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:  # older replay file: uniform legit weight
            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}

    # ------------------------------------------------------------ monitoring
    @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()