Create app.py

app.py

@@ -0,0 +1,99 @@
+import json
+from typing import Any, Dict
+
+import numpy as np
+import tensorflow as tf
+from fastapi import FastAPI, HTTPException
+
+# -----------------
+# Config
+# -----------------
+MODEL_PATH = "best_model.h5"      # upload alongside app.py
+STATS_PATH = "means_std.json"     # {"feature":{"mean":..., "std":...}, ...}
+CLASSES = ["Top", "Mid-Top", "Mid", "Mid-Low", "Low"]
+
+# -----------------
+# Load artifacts
+# -----------------
+print("Loading model and stats...")
+model = tf.keras.models.load_model(MODEL_PATH, compile=False)
+
+with open(STATS_PATH, "r") as f:
+    stats: Dict[str, Dict[str, float]] = json.load(f)
+
+FEATURES = list(stats.keys())
+print("Feature order:", FEATURES)
+
+# -----------------
+# Helpers
+# -----------------
+def _z(v: Any, mean: float, sd: float) -> float:
+    try:
+        x = float(v)
+    except Exception:
+        return 0.0
+    if not sd or sd == 0.0:
+        return 0.0
+    return (x - mean) / sd
+
+def _coral_probs_from_logits(logits_np: np.ndarray) -> np.ndarray:
+    """
+    For ordinal (K-1 logits): p_k = σ(z_{k-1}) - σ(z_k) with σ(z_0)=1, σ(z_K)=0
+    """
+    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)
+    probs = tf.clip_by_value(left - right, 1e-12, 1.0)
+    return probs.numpy()
+
+def _predict_core(ratios: Dict[str, Any]) -> Dict[str, Any]:
+    missing = [f for f in FEATURES if f not in ratios]
+
+    z = [ _z(ratios.get(f, 0.0), stats[f]["mean"], stats[f]["std"]) for f in FEATURES ]
+    X = np.array([z], dtype=np.float32)  # (1, D)
+
+    raw = model.predict(X, verbose=0)    # (1, K) or (1, K-1)
+    if raw.ndim != 2:
+        raise ValueError(f"Unexpected model output shape: {raw.shape}")
+
+    if raw.shape[1] == len(CLASSES) - 1:
+        probs = _coral_probs_from_logits(raw)[0]
+    elif raw.shape[1] == len(CLASSES):
+        probs = raw[0]
+    else:
+        raise ValueError(f"Model output width {raw.shape[1]} incompatible with {len(CLASSES)} classes")
+
+    probs = np.maximum(probs, 0.0)
+    s = probs.sum()
+    probs = probs / s if s > 0 else np.ones(len(CLASSES)) / len(CLASSES)
+
+    pred_idx = int(np.argmax(probs))
+    return {
+        "input_ok": len(missing) == 0,
+        "missing": missing,
+        "probabilities": {CLASSES[i]: float(probs[i]) for i in range(len(CLASSES))},
+        "predicted_state": CLASSES[pred_idx],
+    }
+
+# -----------------
+# FastAPI app
+# -----------------
+app = FastAPI()
+
+@app.get("/health")
+def health():
+    return {"status": "ok", "features": FEATURES, "classes": CLASSES}
+
+@app.post("/predict")
+def predict(payload: Any):
+    # Accept either {feature:value,...} or [ {feature:value,...} ]
+    if isinstance(payload, list) and len(payload) == 1 and isinstance(payload[0], dict):
+        payload = payload[0]
+    if not isinstance(payload, dict):
+        raise HTTPException(status_code=400, detail="Expected JSON object mapping feature -> value.")
+
+    try:
+        return _predict_core(payload)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))