Update model_utils.py

atualizando model_utils.py

model_utils.py

@@ -1 +1,107 @@
-#
+# model_utils.py
+import pandas as pd
+import numpy as np
+from sklearn.model_selection import train_test_split
+from sklearn.preprocessing import StandardScaler, OneHotEncoder
+from sklearn.impute import SimpleImputer
+from sklearn.compose import ColumnTransformer
+from sklearn.pipeline import Pipeline
+from sklearn.linear_model import LogisticRegression
+from imblearn.over_sampling import SMOTE
+from typing import Dict, Any, List, Tuple, Union
+
+# Definir as colunas que seu modelo espera
+ALL_FEATURES = ['Idade', 'Saldo_Conta', 'Numero_Produtos', 'Tempo_Cliente_Meses',
+                'Genero', 'Cidade', 'Renda_Anual_USD', 'Membro_Ativo', 'Tem_Cartao_Credito']
+TARGET_COLUMN = 'Churn'
+
+class ChurnModelPipeline:
+    def __init__(self):
+        self.model = None
+        self.preprocessor = None
+        self.feature_names_out = None # Nomes das features após o pré-processamento
+
+    def _build_preprocessor(self, X: pd.DataFrame) -> ColumnTransformer:
+        # Identificar features numéricas e categóricas com base no dataframe X
+        numeric_features = X.select_dtypes(include=np.number).columns.tolist()
+        categorical_features = X.select_dtypes(include='object').columns.tolist()
+
+        numeric_transformer = Pipeline(steps=[
+            ('imputer', SimpleImputer(strategy='median')),
+            ('scaler', StandardScaler())
+        ])
+
+        categorical_transformer = Pipeline(steps=[
+            ('imputer', SimpleImputer(strategy='most_frequent')),
+            ('onehot', OneHotEncoder(handle_unknown='ignore'))
+        ])
+
+        preprocessor = ColumnTransformer(
+            transformers=[
+                ('num', numeric_transformer, numeric_features),
+                ('cat', categorical_transformer, categorical_features)
+            ],
+            remainder='passthrough'
+        )
+        return preprocessor
+
+    def train(self, df: pd.DataFrame) -> None:
+        X = df[ALL_FEATURES]
+        y = df[TARGET_COLUMN]
+
+        # 1. Dividir em treino e teste (estratificado para Churn)
+        X_train, _, y_train, _ = train_test_split(
+            X, y, test_size=0.2, random_state=42, stratify=y
+        )
+
+        # 2. Construir e ajustar o pré-processador
+        self.preprocessor = self._build_preprocessor(X_train)
+        X_train_processed = self.preprocessor.fit_transform(X_train)
+        
+        # Obter nomes das features após o pré-processamento
+        numeric_f = X_train.select_dtypes(include=np.number).columns.tolist()
+        categorical_f = X_train.select_dtypes(include='object').columns.tolist()
+        
+        self.feature_names_out = numeric_f + list(self.preprocessor.named_transformers_['cat']['onehot'].get_feature_names_out(categorical_f))
+
+
+        # 3. Balanceamento de Classes com SMOTE
+        smote = SMOTE(random_state=42)
+        X_train_resampled, y_train_resampled = smote.fit_resample(X_train_processed, y_train)
+
+        # 4. Treinar o modelo
+        # Usamos LogisticRegression, similar ao seu relatório
+        self.model = LogisticRegression(random_state=42, solver='liblinear', C=0.1, max_iter=500)
+        self.model.fit(X_train_resampled, y_train_resampled)
+        print("Modelo de Churn treinado com sucesso!")
+    
+    def predict_churn(self, input_data: pd.DataFrame) -> Tuple[int, float]:
+        if self.model is None or self.preprocessor is None:
+            raise RuntimeError("Modelo ou pré-processador não treinados. Chame .train() primeiro.")
+        
+        # Validar se as colunas necessárias estão presentes
+        if not all(col in input_data.columns for col in ALL_FEATURES):
+            missing_cols = [col for col in ALL_FEATURES if col not in input_data.columns]
+            raise ValueError(f"Dados de entrada brutos não contêm todas as features esperadas: {missing_cols}. Features esperadas: {ALL_FEATURES}")
+
+        # Garantir a ordem das colunas e que todas as features estejam presentes, mesmo que preenchidas com NaN
+        input_data_ordered = input_data[ALL_FEATURES]
+
+        # Aplicar o mesmo pré-processamento usado no treino
+        X_processed = self.preprocessor.transform(input_data_ordered)
+        
+        # Fazer a previsão
+        prediction = self.model.predict(X_processed)[0]
+        # Obter a probabilidade da classe positiva (Churn=1)
+        probability_churn = self.model.predict_proba(X_processed)[0][1]
+        
+        return int(prediction), float(probability_churn)
+
+# Funções auxiliares para Gradio
+def get_model_coefficients(model_pipeline: ChurnModelPipeline) -> pd.DataFrame:
+    if model_pipeline.model and hasattr(model_pipeline.model, 'coef_') and model_pipeline.feature_names_out:
+        coefs = model_pipeline.model.coef_[0] if model_pipeline.model.coef_.ndim > 1 else model_pipeline.model.coef_
+        coef_df = pd.DataFrame({'Feature': model_pipeline.feature_names_out, 'Coeficiente': coefs})
+        coef_df['Odds_Ratio'] = np.exp(coef_df['Coeficiente'])
+        return coef_df.sort_values(by='Odds_Ratio', ascending=False)
+    return pd.DataFrame()