Update inference/predictor.py

inference/predictor.py

@@ -1,55 +1,95 @@
-import pandas as pd
-import joblib
-
-from src.utils import get_latest_file
-from src.preprocessing import clean_and_engineer
-
-
-MODEL_DIR = "models"
-
-
-class CreditRiskPredictor:
-    def __init__(self):
-        self.model_path = get_latest_file(MODEL_DIR, "credit_model")
-        self.scaler_path = get_latest_file(MODEL_DIR, "scaler")
-        self.columns_path = get_latest_file(MODEL_DIR, "columns")
-
-        self.model = joblib.load(self.model_path)
-        self.scaler = joblib.load(self.scaler_path)
-        self.columns = joblib.load(self.columns_path)
-
-        print(f"Loaded model: {self.model_path}")
-        print(f"Loaded scaler: {self.scaler_path}")
-        print(f"Loaded columns: {self.columns_path}")
-
-    def predict(self, input_dict: dict):
-        # Convert input to DataFrame
-        df = pd.DataFrame([input_dict])
-
-        # Apply same preprocessing as training
-        df = clean_and_engineer(df)
-
-        # One-hot encode
-        df = pd.get_dummies(df, drop_first=True)
-
-        # Align columns with training
-        df = df.reindex(columns=self.columns, fill_value=0)
-
-        # Scale
-        X_scaled = self.scaler.transform(df)
-
-        # Predict
-        probability = self.model.predict_proba(X_scaled)[0][1]
-
-        # Simple credit score + rating logic
-        credit_score = int(900 - (probability * 600))
-
-        if probability < 0.3:
-            rating = "Good"
-        elif probability < 0.6:
-            rating = "Average"
-        else:
-            rating = "High Risk"
-
-        return probability, credit_score, rating
-
+import pandas as pd
+import numpy as np
+import joblib
+
+from src.utils import get_latest_file
+from src.preprocessing import clean_and_engineer
+
+MODEL_DIR = "models"
+
+
+class CreditRiskPredictor:
+    def __init__(self):
+        self.model_path = get_latest_file(MODEL_DIR, "credit_model")
+        self.scaler_path = get_latest_file(MODEL_DIR, "scaler")
+        self.columns_path = get_latest_file(MODEL_DIR, "columns")
+
+        self.model = joblib.load(self.model_path)
+        self.scaler = joblib.load(self.scaler_path)
+        self.columns = joblib.load(self.columns_path)
+
+        print(f"Loaded model: {self.model_path}")
+        print(f"Loaded scaler: {self.scaler_path}")
+        print(f"Loaded columns: {self.columns_path}")
+
+    def predict(self, input_dict: dict):
+        # -------------------------------
+        # 1. Convert input to DataFrame
+        # -------------------------------
+        df = pd.DataFrame([input_dict])
+
+        # -------------------------------
+        # 2. Apply preprocessing
+        # -------------------------------
+        df = clean_and_engineer(df)
+
+        # -------------------------------
+        # 3. One-hot encode
+        # -------------------------------
+        df = pd.get_dummies(df, drop_first=True)
+
+        # -------------------------------
+        # 4. Align columns with training
+        # -------------------------------
+        df = df.reindex(columns=self.columns, fill_value=0)
+
+        # -------------------------------
+        # 5. Scale
+        # -------------------------------
+        X_scaled = self.scaler.transform(df)
+
+        # -------------------------------
+        # 6. ORIGINAL SCORECARD LOGIC
+        # -------------------------------
+        probability, credit_score, rating = self._calculate_scorecard_output(X_scaled)
+
+        return probability, credit_score, rating
+
+    def _calculate_scorecard_output(self, X_scaled, base_score=300, scale_length=600):
+        """
+        This matches your original logic exactly:
+        x = wX + b
+        PD = sigmoid(x)
+        score = base + (1 - PD) * scale
+        rating = bucket(score)
+        """
+
+        # Linear combination
+        x = np.dot(X_scaled, self.model.coef_.T) + self.model.intercept_
+
+        # Sigmoid → default probability
+        default_probability = 1 / (1 + np.exp(-x))
+
+        # Non-default probability
+        non_default_probability = 1 - default_probability
+
+        # Credit score calculation
+        credit_score = base_score + non_default_probability.flatten() * scale_length
+        credit_score = int(credit_score[0])
+
+        # Rating bucket (same as your code)
+        rating = self._get_rating(credit_score)
+
+        return float(default_probability.flatten()[0]), credit_score, rating
+
+    def _get_rating(self, score):
+        if 300 <= score < 500:
+            return 'Poor'
+        elif 500 <= score < 650:
+            return 'Average'
+        elif 650 <= score < 750:
+            return 'Good'
+        elif 750 <= score <= 900:
+            return 'Excellent'
+        else:
+            return 'Undefined'