Upload 4 files

handler.py

@@ -0,0 +1,302 @@
+# handler.py — Quantium insights Inference Endpoint (Residence_type canonicalized)
+import os
+import json
+import traceback
+from typing import Any, Dict, List, Tuple
+
+import joblib
+import numpy as np
+import pandas as pd
+
+
+# =========================
+# Feature schema (canonical)
+# =========================
+NUMERIC_COLS = ["age", "avg_glucose_level", "bmi", "hypertension", "heart_disease"]
+# Canonical Residence key uses capital R
+CAT_COLS     = ["gender", "ever_married", "work_type", "smoking_status", "Residence_type"]
+ALL_CANON    = NUMERIC_COLS + CAT_COLS
+
+# For explain UI ordering (match canonical names)
+EXPLAIN_ORDER = [
+    "age", "avg_glucose_level", "bmi", "hypertension", "heart_disease",
+    "gender", "ever_married", "work_type", "smoking_status", "Residence_type"
+]
+
+
+# =========================
+# Utility: dtype coercion
+# =========================
+def _to_int01(x: Any) -> int:
+    if isinstance(x, (bool, np.bool_)):
+        return int(bool(x))
+    try:
+        if isinstance(x, str):
+            s = x.strip().lower()
+            if s in {"1", "true", "t", "yes", "y"}:
+                return 1
+            if s in {"0", "false", "f", "no", "n"}:
+                return 0
+        return int(float(x))
+    except Exception:
+        return 0
+
+
+def _coerce_dataframe(rows: List[Dict[str, Any]]) -> pd.DataFrame:
+    """
+    Build a clean DataFrame:
+    - Canonical Residence key is 'Residence_type' (capital R).
+    - Accept 'residence_type' and map it to 'Residence_type' if needed.
+    - Ensure numerics are float64 and 0/1 flags are ints then float64.
+    - Ensure categoricals are plain Python strings (object), no NA.
+    - Also mirror lowercase 'residence_type' for legacy models.
+    """
+    norm_rows: List[Dict[str, Any]] = []
+    for r in rows:
+        r = dict(r or {})
+        # Normalize residence key to capitalized canonical
+        if "Residence_type" not in r and "residence_type" in r:
+            r["Residence_type"] = r["residence_type"]
+        # Keep only canonical columns
+        entry = {k: r.get(k, None) for k in ALL_CANON}
+        norm_rows.append(entry)
+
+    df = pd.DataFrame(norm_rows, columns=ALL_CANON)
+
+    # binary flags first
+    for col in ["hypertension", "heart_disease"]:
+        df[col] = df[col].map(_to_int01)
+
+    # strong numeric coercion
+    for col in ["age", "avg_glucose_level", "bmi"]:
+        df[col] = pd.to_numeric(df[col], errors="coerce")
+
+    # final cast to float64
+    df[NUMERIC_COLS] = df[NUMERIC_COLS].astype("float64")
+
+    # categoricals as plain strings, no NA
+    for col in CAT_COLS:
+        df[col] = df[col].where(df[col].notna(), "Unknown")
+        df[col] = df[col].map(lambda v: "Unknown" if v is None else str(v)).astype(object)
+
+    # Mirror lowercase 'residence_type' for backward compatibility
+    df["residence_type"] = df["Residence_type"].astype(object)
+
+    return df
+
+
+# =========================
+# Safety patches for OHE
+# =========================
+def _iter_estimators(est):
+    yield est
+    # Pipelines
+    if hasattr(est, "named_steps"):
+        for step in est.named_steps.values():
+            yield from _iter_estimators(step)
+    # ColumnTransformer
+    if hasattr(est, "transformers"):
+        for _, tr, _ in est.transformers:
+            yield from _iter_estimators(tr)
+
+
+def _numeric_like(x) -> bool:
+    if x is None:
+        return True
+    if isinstance(x, (int, np.integer, float, np.floating)):
+        return True
+    if isinstance(x, str):
+        try:
+            float(x)
+            return True
+        except Exception:
+            return False
+    return False
+
+
+def _sanitize_onehot_categories(model):
+    """Coerce OneHotEncoder.categories_ to consistent dtypes to avoid np.isnan crashes."""
+    try:
+        from sklearn.preprocessing import OneHotEncoder  # type: ignore
+    except Exception:
+        OneHotEncoder = None
+
+    if OneHotEncoder is None:
+        return
+
+    for node in _iter_estimators(model):
+        if isinstance(node, OneHotEncoder) and hasattr(node, "categories_"):
+            new_cats = []
+            for cats in node.categories_:
+                arr = np.asarray(cats, dtype=object)
+                if all(_numeric_like(v) for v in arr):
+                    vals = []
+                    for v in arr:
+                        try:
+                            vals.append(np.nan if v is None else float(v))
+                        except Exception:
+                            vals.append(np.nan)
+                    new_cats.append(np.asarray(vals, dtype=float))
+                else:
+                    strs = ["Unknown" if (v is None or (isinstance(v, float) and np.isnan(v))) else str(v) for v in arr]
+                    new_cats.append(np.asarray(strs, dtype=object))
+            node.categories_ = new_cats
+            if hasattr(node, "handle_unknown"):
+                node.handle_unknown = "ignore"
+
+
+def _patch_check_unknown():
+    """
+    Monkey-patch sklearn.utils._encode._check_unknown to avoid np.isnan on object/string arrays
+    on certain sklearn builds.
+    """
+    try:
+        from sklearn.utils import _encode  # type: ignore
+        _orig = _encode._check_unknown
+
+        def _safe_check_unknown(values, known_values, return_mask=False):
+            try:
+                return _orig(values, known_values, return_mask=return_mask)
+            except TypeError:
+                vals  = np.asarray(values, dtype=object)
+                known = np.asarray(known_values, dtype=object)
+                mask = np.isin(vals, known, assume_unique=False)
+                diff = vals[~mask]
+                if return_mask:
+                    return diff, mask
+                return diff
+
+        _encode._check_unknown = _safe_check_unknown  # type: ignore[attr-defined]
+        print("[handler] Patched sklearn.utils._encode._check_unknown", flush=True)
+    except Exception as e:
+        print(f"[handler] Patch for _check_unknown not applied: {e}", flush=True)
+
+
+# =========================
+# Model introspection (debug)
+# =========================
+def _introspect_model(model) -> Dict[str, Any]:
+    info: Dict[str, Any] = {"type": str(type(model))}
+    try:
+        if hasattr(model, "named_steps"):
+            info["pipeline_steps"] = list(model.named_steps.keys())
+            for name, step in model.named_steps.items():
+                if step.__class__.__name__ == "ColumnTransformer":
+                    info["column_transformer"] = str(step)
+                    try:
+                        info["transformers_"] = [(n, str(t.__class__), cols) for (n, t, cols) in step.transformers]
+                    except Exception:
+                        pass
+    except Exception:
+        pass
+    try:
+        info["feature_names_in_"] = list(getattr(model, "feature_names_in_", []))
+    except Exception:
+        pass
+    return info
+
+
+# =========================
+# Handler
+# =========================
+class EndpointHandler:
+    def __init__(self, path: str = "/repository") -> None:
+        _patch_check_unknown()  # apply safety patch early
+
+        model_path = os.path.join(path, "model.joblib")
+        self.model = joblib.load(model_path)
+
+        # Threshold (UI also reads this if present in response)
+        try:
+            self.threshold = float(os.getenv("THRESHOLD", "0.38"))
+        except Exception:
+            self.threshold = 0.38
+
+        # Optional explainer (for old models); XGB wrapper may provide .top_contrib instead
+        self.explainer = getattr(self.model, "explainer_", None)
+
+        # Sanitize OneHotEncoder categories (if present)
+        _sanitize_onehot_categories(self.model)
+
+        print("[handler] Model loaded", flush=True)
+        print(f"[handler] Using threshold: {self.threshold}", flush=True)
+
+    def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        debug   = bool(data.get("debug", False))
+        explain = bool(data.get("explain", False))
+
+        rows = data.get("inputs") or []
+        if isinstance(rows, dict):
+            rows = [rows]
+        if not isinstance(rows, list) or not rows:
+            return {"error": "inputs must be a non-empty list of records", "threshold": self.threshold}
+
+        df = _coerce_dataframe(rows)
+
+        debug_info = {
+            "columns": list(df.columns),
+            "dtypes": {c: str(df[c].dtype) for c in df.columns},
+            "threshold": self.threshold,
+            "model": _introspect_model(self.model),
+            "head": df.head(1).to_dict(orient="records"),
+        }
+
+        # Predict
+        try:
+            if hasattr(self.model, "predict_proba"):
+                proba = self.model.predict_proba(df)[:, 1].astype(float)
+            else:
+                # e.g., model exposes only decision_function
+                raw = self.model.predict(df).astype(float)
+                proba = 1.0 / (1.0 + np.exp(-raw))
+        except Exception as e:
+            return {
+                "error": f"model.predict failed: {e}",
+                "trace": traceback.format_exc(),
+                "debug": debug_info,
+                "threshold": self.threshold,
+            }
+
+        p = float(proba[0])
+        label = int(p >= self.threshold)
+
+        resp: Dict[str, Any] = {
+            "risk_probability": p,
+            "risk_label": label,
+            "threshold": self.threshold,  # echo for the UI
+        }
+
+        # Explanations
+        if explain:
+            # Preferred path: XGB wrapper implements top_contrib()
+            if hasattr(self.model, "top_contrib"):
+                try:
+                    names, vals = self.model.top_contrib(df, k=5)
+                    if names:
+                        resp["shap"] = {"feature_names": names, "values": vals}
+                except Exception as e:
+                    resp["shap_error"] = f"top_contrib failed: {e}"
+            # Fallback: use stored explainer_ if present
+            elif self.explainer is not None:
+                try:
+                    shap_vals = self.explainer(df)
+                    vals = shap_vals.values[0] if hasattr(shap_vals, "values") else shap_vals[0]
+                    contrib = []
+                    for feat in EXPLAIN_ORDER:
+                        if feat in df.columns:
+                            idx = list(df.columns).index(feat)
+                            contrib.append({"feature": feat, "effect": float(vals[idx])})
+                    resp["shap"] = {"contrib": contrib}
+                except Exception as e:
+                    resp["shap_error"] = f"explainer failed: {e}"
+
+        if debug:
+            resp["debug"] = debug_info
+
+        # Optional console log (visible in Endpoint Logs)
+        try:
+            print(f"[handler] prob={p:.4f} label={label}", flush=True)
+        except Exception:
+            pass
+
+        return resp

model.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:b397f797a1a6106bb2ed6e21ddfeed2738491fdddde9681916cb8c9593cc4e07
+size 1261562

requirements.txt

@@ -0,0 +1,8 @@
+scikit-learn==1.5.1
+imbalanced-learn==0.12.3
+xgboost==2.0.3
+shap==0.45.0
+numpy==1.26.4
+pandas==2.2.2
+joblib==1.3.2
+scipy==1.11.4