Update app.py

app.py

@@ -1,25 +1,33 @@
+# app.py
 import os
 import json
-from typing import Any, Dict, List
+from pathlib import Path
+from typing import Any, Dict, List, Tuple
 
 import numpy as np
 import tensorflow as tf
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse
-import joblib
 
-# ----------------- CONFIG -----------------
-# Use your actual filenames here (from your training export)
-MODEL_PATH   = os.getenv("MODEL_PATH", "best_model.h5")      # or "best_model.h5" if that's what you have
-IMPUTER_PATH = os.getenv("IMPUTER_PATH", "imputer.joblib")
-SCALER_PATH  = os.getenv("SCALER_PATH",  "scaler.joblib")
+# ----------------- PATHS & CONFIG -----------------
+BASE_DIR = Path(__file__).resolve().parent
 
-# Class order per training: 0=Top .. 4=Low
-CLASSES = ["Top", "Mid-Top", "Mid", "Mid-Low", "Low"]
+# Prefer env vars, fall back to files next to app.py
+MODEL_PATH  = os.getenv("MODEL_PATH")  or str(BASE_DIR / "best_model.keras")
+if not Path(MODEL_PATH).exists():
+    # fallback to .h5 if .keras not present
+    alt = BASE_DIR / "best_model.h5"
+    if alt.exists():
+        MODEL_PATH = str(alt)
 
-# EXACT feature order used during training (from your Section 3.1)
-FEATURES: List[str] = [
+STATS_PATH    = os.getenv("STATS_PATH")    or str(BASE_DIR / "means_std.json")
+IMPUTER_PATH  = os.getenv("IMPUTER_PATH")  or str(BASE_DIR / "imputer.joblib")
+SCALER_PATH   = os.getenv("SCALER_PATH")   or str(BASE_DIR / "scaler.joblib")
+
+CLASSES = ["Top", "Mid-Top", "Mid", "Mid-Low", "Low"]  # ordinal: 0..4
+
+# IMPORTANT — exact feature order used during training
+FEATURE_ORDER: List[str] = [
     "autosuf_oper",
     "improductiva",
     "gastos_fin_over_avg_cart",
@@ -42,67 +50,127 @@ FEATURES: List[str] = [
     "rend_cart_over_avg_cart",
     "roa_pre_tax",
 ]
-# ------------------------------------------
 
+print("Resolved paths:")
+print("  MODEL_PATH :", MODEL_PATH)
+print("  STATS_PATH :", STATS_PATH)
+print("  IMPUTER_PATH:", IMPUTER_PATH)
+print("  SCALER_PATH :", SCALER_PATH)
+
+# ----------------- LOAD ARTIFACTS -----------------
 print("Loading model / imputer / scaler...")
-# Loss is not needed for inference; compile=False avoids needing custom loss objects.
+# If the model used custom losses/metrics you’d pass custom_objects here.
 model = tf.keras.models.load_model(MODEL_PATH, compile=False)
-imputer = joblib.load(IMPUTER_PATH)   # median imputation from training
-scaler  = joblib.load(SCALER_PATH)    # StandardScaler from training
-
-print("Model loaded. Feature order:", FEATURES)
 
+# Optional: imputer & scaler from training pipeline
+imputer = None
+scaler = None
+try:
+    import joblib  # in requirements
+    if Path(IMPUTER_PATH).exists():
+        imputer = joblib.load(IMPUTER_PATH)
+        print("Loaded imputer:", IMPUTER_PATH)
+    if Path(SCALER_PATH).exists():
+        scaler = joblib.load(SCALER_PATH)
+        print("Loaded scaler :", SCALER_PATH)
+except Exception as e:
+    print("Imputer/scaler not loaded:", e)
+
+# Optional: stats fallback for manual z-scoring
+stats: Dict[str, Dict[str, float]] = {}
+if Path(STATS_PATH).exists():
+    with open(STATS_PATH, "r") as f:
+        stats = json.load(f)
+        print("Loaded means/std from:", STATS_PATH)
+
+# ----------------- HELPERS -----------------
 def coerce_float(val: Any) -> float:
     """
-    Robust numeric parse:
-      "49.709,14"  -> 49709.14
-      "49,709.14"  -> 49709.14
-      "0,005"      -> 0.005
-      1.23         -> 1.23
-    Raises ValueError on failure.
+    Accepts numeric or strings like:
+      '49.709,14' -> 49709.14 ; '49,709.14' -> 49709.14 ; '0,005' -> 0.005
     """
-    if isinstance(val, (int, float, np.integer, np.floating)):
+    if isinstance(val, (int, float, np.number)):
         return float(val)
-
     s = str(val).strip()
     if s == "":
         raise ValueError("empty")
-
     s = s.replace(" ", "")
-    has_dot = "." in s
-    has_comma = "," in s
-
+    has_dot, has_comma = "." in s, "," in s
     if has_dot and has_comma:
-        last_dot = s.rfind(".")
-        last_comma = s.rfind(",")
-        if last_comma > last_dot:
-            # decimal is comma, thousands is dot
+        if s.rfind(",") > s.rfind("."):
             s = s.replace(".", "")
             s = s.replace(",", ".")
         else:
-            # decimal is dot, thousands is comma
             s = s.replace(",", "")
     elif has_comma and not has_dot:
         s = s.replace(",", ".")
-    # else: dots only or pure digits
-
     return float(s)
 
+def build_matrix_from_payload(payload: Dict[str, Any]) -> Tuple[np.ndarray, Dict[str, float], List[str]]:
+    """
+    Returns:
+      X (1, 21) ready for model (imputed+scaled if artifacts exist; else z-scored via stats),
+      z_detail (dict feature -> standardized value used),
+      missing list (features not present in payload)
+    """
+    raw = []
+    missing: List[str] = []
+    for f in FEATURE_ORDER:
+        if f in payload:
+            try:
+                raw.append(coerce_float(payload[f]))
+            except Exception:
+                raw.append(np.nan)
+        else:
+            raw.append(np.nan)
+            missing.append(f)
+
+    arr = np.array([raw], dtype=np.float32)  # shape (1, 21)
+
+    # primary path: use imputer + scaler if both available
+    if imputer is not None and scaler is not None:
+        arr_imp = imputer.transform(arr)   # median impute
+        arr_std = scaler.transform(arr_imp)  # z-score to training distribution
+        z_row = arr_std[0].tolist()
+        z_detail = {f: float(z_row[i]) for i, f in enumerate(FEATURE_ORDER)}
+        return arr_std.astype(np.float32), z_detail, missing
+
+    # fallback path: manual z-score using means_std.json
+    z_vals = []
+    z_detail = {}
+    for i, f in enumerate(FEATURE_ORDER):
+        v = arr[0, i]
+        if f in stats and "mean" in stats[f] and "std" in stats[f] and stats[f]["std"]:
+            mean = float(stats[f]["mean"])
+            std = float(stats[f]["std"])
+            vv = 0.0 if np.isnan(v) else float(v)
+            z = (vv - mean) / std
+        else:
+            z = 0.0  # safest fallback
+        z_vals.append(z)
+        z_detail[f] = float(z)
+    return np.array([z_vals], dtype=np.float32), z_detail, missing
+
 def coral_probs_from_logits(logits_np: np.ndarray) -> np.ndarray:
     """
-    Decode CORAL logits (N, K-1) -> probabilities (N, K)
+    CORAL decode: (N, K-1) logits -> (N, K) probs.
+    Adds a small monotonicity fix (non-increasing thresholds).
     """
-    logits = tf.convert_to_tensor(logits_np, dtype=tf.float32)
-    sig = tf.math.sigmoid(logits)  # (N, K-1)
-    left  = tf.concat([tf.ones_like(sig[:, :1]), sig], axis=1)
-    right = tf.concat([sig, tf.zeros_like(sig[:, :1])], axis=1)
+    logits = tf.convert_to_tensor(logits_np, dtype=tf.float32)  # (N, K-1)
+    sig = tf.math.sigmoid(logits)  # p(y>k)
+    # Enforce non-increasing along thresholds (numerical guard)
+    sig = tf.clip_by_value(sig, 1e-12, 1.0 - 1e-12)
+    sig_sorted = tf.minimum(sig, tf.math.cummin(sig, axis=1, exclusive=False))
+    left  = tf.concat([tf.ones_like(sig_sorted[:, :1]), sig_sorted], axis=1)
+    right = tf.concat([sig_sorted, tf.zeros_like(sig_sorted[:, :1])], axis=1)
     probs = tf.clip_by_value(left - right, 1e-12, 1.0)
+    # Normalize row just in case
+    probs = probs / tf.reduce_sum(probs, axis=1, keepdims=True)
     return probs.numpy()
 
-# ------------- FastAPI app ----------------
+# ----------------- FASTAPI -----------------
 app = FastAPI(title="Static Fingerprint API", version="1.0.0")
 
-# Allow Excel / local tools to call the API
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -113,17 +181,26 @@ app.add_middleware(
 
 @app.get("/")
 def root():
-    return {"message": "Static Fingerprint API is running.", "try": ["GET /health", "POST /predict"]}
+    return {
+        "message": "Static Fingerprint API is running.",
+        "try": ["GET /health", "POST /predict"],
+    }
 
 @app.get("/health")
 def health():
     return {
         "status": "ok",
-        "features": FEATURES,
         "classes": CLASSES,
-        "model_file": MODEL_PATH,
-        "imputer_file": IMPUTER_PATH,
-        "scaler_file": SCALER_PATH,
+        "feature_order": FEATURE_ORDER,
+        "paths": {
+            "model": MODEL_PATH,
+            "stats": STATS_PATH if Path(STATS_PATH).exists() else None,
+            "imputer": IMPUTER_PATH if Path(IMPUTER_PATH).exists() else None,
+            "scaler": SCALER_PATH if Path(SCALER_PATH).exists() else None,
+            "base_dir_files": [p.name for p in BASE_DIR.iterdir()],
+        },
+        "has_imputer": imputer is not None,
+        "has_scaler": scaler is not None,
     }
 
 @app.post("/echo")
@@ -134,67 +211,43 @@ async def echo(req: Request):
 @app.post("/predict")
 async def predict(req: Request):
     """
-    Body: JSON object mapping feature -> value (raw numbers; median+z will be applied here)
-    Missing features are imputed by the training imputer (median).
+    Body: JSON dict mapping feature -> value (raw numbers). Example:
+    {
+      "autosuf_oper": 1.0,
+      "cov_improductiva": 0.9,
+      ...
+    }
     """
-    try:
-        payload = await req.json()
-    except Exception as e:
-        return JSONResponse(status_code=400, content={"error": f"Invalid JSON: {e}"})
-
+    payload = await req.json()
     if not isinstance(payload, dict):
-        return JSONResponse(status_code=400, content={"error": "Expected a JSON object mapping feature -> value."})
-
-    # Build raw vector in EXACT training order; use np.nan for missing so imputer handles it
-    x_raw = []
-    missing = []
-    for f in FEATURES:
-        if f in payload:
-            try:
-                x_raw.append(coerce_float(payload[f]))
-            except Exception:
-                # treat unparsable as missing -> np.nan (imputer will fill)
-                x_raw.append(np.nan)
-                missing.append(f)
-        else:
-            x_raw.append(np.nan)
-            missing.append(f)
+        return {"error": "Expected a JSON object mapping feature -> value."}
 
-    X_raw = np.array([x_raw], dtype=np.float64)          # (1, 21)
-    X_imp = imputer.transform(X_raw)                      # median imputation
-    X_std = scaler.transform(X_imp).astype(np.float32)    # z-scores as per training
+    X, z_detail, missing = build_matrix_from_payload(payload)   # shape (1, 21)
 
-    # Predict
-    raw = model.predict(X_std, verbose=0)
+    raw = model.predict(X, verbose=0)
 
-    # CORAL vs softmax detection (your model is CORAL with 4 logits)
+    # Auto-detect output head: CORAL (K-1) or softmax (K)
+    decode_mode = "auto_coral"
     if raw.ndim == 2 and raw.shape[1] == (len(CLASSES) - 1):
         probs = coral_probs_from_logits(raw)[0]
-        decode_mode = "coral"
-    elif raw.ndim == 2 and raw.shape[1] == len(CLASSES):
-        p = raw[0]
-        s = float(np.sum(p))
-        probs = (p / s) if s > 0 else p
-        decode_mode = "softmax"
+        decode_mode = "auto_coral_monotone"
     else:
-        # Fallback: try CORAL first
-        probs = coral_probs_from_logits(raw)[0]
-        decode_mode = "auto"
+        # assume logits for K classes
+        logits = tf.convert_to_tensor(raw, dtype=tf.float32)
+        probs = tf.nn.softmax(logits, axis=1).numpy()[0]
+        decode_mode = "softmax"
 
     pred_idx = int(np.argmax(probs))
-
-    # Build z-score dict for transparency
-    z_detail = {FEATURES[i]: float(X_std[0, i]) for i in range(len(FEATURES))}
-
-    return {
+    out = {
         "input_ok": (len(missing) == 0),
-        "missing": missing,                    # features that were np.nan (imputed)
-        "z_scores": z_detail,                  # exactly what the model saw
+        "missing": missing,
+        "z_scores": z_detail,
         "probabilities": {CLASSES[i]: float(probs[i]) for i in range(len(CLASSES))},
         "predicted_state": CLASSES[pred_idx],
         "debug": {
             "raw_shape": list(raw.shape),
             "decode_mode": decode_mode,
-            "first_row_logits": [float(v) for v in (raw[0].tolist() if raw.ndim == 2 else np.atleast_1d(raw).tolist())],
+            "raw_first_row": [float(x) for x in raw[0].tolist()],
         },
-    }
+    }
+    return out