""" api.py - FastAPI for Hugging Face Spaces Deploy this as a Hugging Face Space (SDK: Docker or Gradio-FastAPI) """ from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import List, Dict, Optional import pandas as pd import numpy as np import joblib import os app = FastAPI( title="Student Dropout Prediction API", description="4-Model Ensemble for Predicting Student Dropout Risk", version="3.0.0" ) # ─── CORS (required so your website can call this API) ─────────────────────── app.add_middleware( CORSMiddleware, allow_origins=["*"], # Change to your website domain in production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ─── Global model state ────────────────────────────────────────────────────── rf_model = xgb_model = lr_model = kmeans_model = scaler = feature_cols = None @app.on_event("startup") async def load_models(): global rf_model, xgb_model, lr_model, kmeans_model, scaler, feature_cols rf_model = joblib.load('models/dropout_randomforest_model.pkl') xgb_model = joblib.load('models/dropout_xgboost_model.pkl') lr_model = joblib.load('models/dropout_logistic_model.pkl') kmeans_model = joblib.load('models/dropout_kmeans_model.pkl') scaler = joblib.load('models/dropout_scaler.pkl') feature_cols = joblib.load('models/dropout_feature_columns.pkl') print("All 4 models loaded.") # ─── Schemas ───────────────────────────────────────────────────────────────── class StudentData(BaseModel): Admission_Grade: float = Field(..., ge=60, le=200) First_Sem_Grade: float = Field(..., ge=0, le=20) Second_Sem_Grade: float = Field(..., ge=0, le=20) Attendance_Percentage: float = Field(..., ge=0, le=100) Curricular_Units_1st_Sem_Credited: int = Field(..., ge=0, le=30) Curricular_Units_2nd_Sem_Credited: int = Field(..., ge=0, le=30) Tuition_Fees_Up_to_Date: int = Field(..., ge=0, le=1) # 0=ok, 1=delinquent Scholarship_Holder: int = Field(..., ge=0, le=1) Debtor: int = Field(..., ge=0, le=1) Displaced: int = Field(..., ge=0, le=1) Age: int = Field(..., ge=16, le=70) Unemployment_Rate: float = Field(..., ge=0, le=30) Inflation_Rate: float = Field(..., ge=-5, le=20) GDP: float = Field(..., ge=-10, le=10) class PredictionResponse(BaseModel): student_id: Optional[str] dropout_risk_percentage: float risk_level: str # HIGH / MEDIUM / LOW prediction: str # Dropout / Graduate confidence: float random_forest_prediction: str xgboost_prediction: str logistic_prediction: str cluster_id: int cluster_risk_level: str top_risk_factors: List[Dict] all_probabilities: Dict[str, float] # ─── Feature engineering (must match training) ─────────────────────────────── def engineer_features(raw: dict) -> dict: d = dict(raw) d['Grade_Average'] = (d['First_Sem_Grade'] + d['Second_Sem_Grade']) / 2 d['Grade_Trend'] = d['Second_Sem_Grade'] - d['First_Sem_Grade'] d['Financial_Burden'] = min(d['Tuition_Fees_Up_to_Date'] + d['Debtor'], 2) d['Academic_Momentum'] = ( d['Curricular_Units_1st_Sem_Credited'] + d['Curricular_Units_2nd_Sem_Credited'] ) / 60 return d # ─── Core prediction logic ─────────────────────────────────────────────────── def predict_dropout(student: StudentData, student_id: str = None) -> PredictionResponse: # Build feature dict with engineered features raw = student.dict() feat = engineer_features(raw) df = pd.DataFrame({col: [feat[col]] for col in feature_cols}) scaled = scaler.transform(df) rf_p = rf_model.predict(scaled)[0] rf_pr = rf_model.predict_proba(scaled)[0, 1] xgb_p = xgb_model.predict(scaled)[0] xgb_pr = xgb_model.predict_proba(scaled)[0, 1] lr_p = lr_model.predict(scaled)[0] lr_pr = lr_model.predict_proba(scaled)[0, 1] cluster = int(kmeans_model.predict(scaled)[0]) # Soft-voting ensemble (same as training) ensemble_proba = (rf_pr + xgb_pr) / 2 ensemble_pred = 1 if ensemble_proba >= 0.5 else 0 risk_pct = round(ensemble_proba * 100, 2) if risk_pct > 60: risk_level = "HIGH" elif risk_pct > 30: risk_level = "MEDIUM" else: risk_level = "LOW" # Top risk factors from LR coefficients coefs = lr_model.coef_[0] top_idx = np.argsort(np.abs(coefs))[-5:][::-1] top_factors = [ { "factor": feature_cols[i], "impact": round(float(coefs[i]), 4), "direction": "risk" if coefs[i] > 0 else "protective" } for i in top_idx ] # Cluster risk labels (derived from training analysis) cluster_map = {0: "HIGH", 1: "MEDIUM-HIGH", 2: "LOW", 3: "MEDIUM"} return PredictionResponse( student_id = student_id, dropout_risk_percentage = risk_pct, risk_level = risk_level, prediction = "Dropout" if ensemble_pred == 1 else "Graduate", confidence = round(float(max(ensemble_proba, 1 - ensemble_proba)), 3), random_forest_prediction= "Dropout" if rf_p == 1 else "Graduate", xgboost_prediction = "Dropout" if xgb_p == 1 else "Graduate", logistic_prediction = "Dropout" if lr_p == 1 else "Graduate", cluster_id = cluster, cluster_risk_level = cluster_map.get(cluster, "MEDIUM"), top_risk_factors = top_factors, all_probabilities = { "Graduate": round(float(1 - ensemble_proba), 3), "Dropout": round(float(ensemble_proba), 3), } ) # ─── Endpoints ─────────────────────────────────────────────────────────────── @app.get("/") def root(): return { "message": "Student Dropout Prediction API v3", "models": ["Random Forest", "XGBoost", "Logistic Regression", "K-Means"], "ensemble": "Soft voting (RF + XGBoost)", "endpoints": ["/health", "/predict", "/predict-batch", "/model-info"] } @app.get("/health") def health(): return { "status": "healthy", "models_loaded": rf_model is not None, "version": "3.0.0" } @app.get("/model-info") def model_info(): return { "random_forest": {"n_estimators": 500, "max_depth": 15, "class_weight": "balanced"}, "xgboost": {"n_estimators": 400, "learning_rate": 0.05, "max_depth": 7}, "logistic_regression":{"C": 0.5, "penalty": "l2", "class_weight": "balanced"}, "kmeans": {"n_clusters": 4, "n_init": 20}, "ensemble": "soft voting (average RF + XGBoost probabilities)", "feature_count": 18, # 14 raw + 4 engineered } @app.post("/predict", response_model=PredictionResponse) def predict_single(student: StudentData): """Predict dropout risk for a single student""" try: return predict_dropout(student) except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.post("/predict-batch") def predict_batch(students: List[StudentData]): """Predict dropout risk for a list of students (max 500)""" if len(students) > 500: raise HTTPException(status_code=400, detail="Max 500 students per batch") try: preds = [ predict_dropout(s, student_id=f"STU_{i+1}").dict() for i, s in enumerate(students) ] return { "total_students": len(preds), "high_risk_count": sum(1 for p in preds if p['dropout_risk_percentage'] > 60), "medium_risk_count": sum(1 for p in preds if 30 < p['dropout_risk_percentage'] <= 60), "low_risk_count": sum(1 for p in preds if p['dropout_risk_percentage'] <= 30), "predictions": preds, } except Exception as e: raise HTTPException(status_code=400, detail=str(e))