#!/usr/bin/env python # coding: utf-8 """ Streamlit – Churn (Logistic Regression) for Hugging Face Spaces --------------------------------------------------------------- - Loads "Dados/Churn_Modelling.csv" - Lets the user choose features (numeric / categorical) - Trains Logistic Regression - Shows coefficients, odds ratios, and quick interpretations - Provides an interactive control panel to simulate a customer's probability of churn Obs.: Esta versão atende ao item (a) da Tarefa 5: modelagem com Regressão Logística e interpretação dos coeficientes/odds ratio. """ import os import numpy as np import pandas as pd import streamlit as st from pathlib import Path from sklearn.model_selection import train_test_split from sklearn.preprocessing import OneHotEncoder, StandardScaler from sklearn.compose import ColumnTransformer from sklearn.linear_model import LogisticRegression from sklearn.pipeline import Pipeline # ----------------------------- # Page config # ----------------------------- st.set_page_config( page_title="Churn – Regressão Logística (PPCA/UnB)", layout="wide", initial_sidebar_state="expanded" ) st.title("Churn – Regressão Logística (PPCA/UnB)") st.caption("Item (a) – Modelagem da Retenção de Clientes e interpretação de coeficientes/odds ratio.") # Botão para limpar cache do loader e revarrer o repositório st.sidebar.button("🔄 Reescanear CSV", on_click=lambda: st.cache_data.clear()) # ----------------------------- # Data loader (cache) – robusto para HF Spaces # ----------------------------- @st.cache_data def load_data(): from pathlib import Path import pandas as _pd # Candidate roots a varrer roots = [] try: roots.append(Path(__file__).parent) except Exception: pass roots += [Path.cwd(), Path("."), Path("/home/user/app")] # ADIÇÃO: raízes comuns em Spaces roots += [Path("/app"), Path("/app/src")] # Caminhos explícitos rápidos fast_candidates = [ Path("Dados/Churn_Modelling.csv"), Path("./Dados/Churn_Modelling.csv"), Path("/mnt/data/Dados/Churn_Modelling.csv"), Path("Churn_Modelling.csv"), Path("./Churn_Modelling.csv"), ] # Função simples de "sniff" de delimitador def _detect_sep(sample_lines): if any(";" in line for line in sample_lines): return ";" if any("\t" in line for line in sample_lines): return "\t" return "," # 1) Tentar candidatos explícitos for pth in fast_candidates: try: if pth.exists(): text = pth.read_text(encoding="utf-8", errors="ignore") sample = text.splitlines()[:5] sep = _detect_sep(sample) df_ = _pd.read_csv(pth, sep=sep) return df_, str(pth) except Exception: pass # 2) Busca recursiva case-insensitive pelo nome targets = [] seen_dirs = set() for root in roots: if root.exists() and str(root) not in seen_dirs: seen_dirs.add(str(root)) for p in root.rglob("*"): try: if p.is_file() and p.name.lower() == "churn_modelling.csv": targets.append(p) except Exception: continue # Preferir caminho dentro de 'Dados/' targets.sort( key=lambda p: ( 0 if ("Dados" in str(p.parent) or "dados" in str(p.parent)) else 1, len(str(p)) ) ) for pth in targets: try: text = pth.read_text(encoding="utf-8", errors="ignore") sample = text.splitlines()[:5] sep = _detect_sep(sample) df_ = _pd.read_csv(pth, sep=sep) return df_, str(pth) except Exception: continue # Não achou return _pd.DataFrame(), "caminhos não encontrados" df, data_info = load_data() if df.empty: st.error("Não foi possível carregar **Churn_Modelling.csv** nos caminhos padrão.") with st.expander("Diagnóstico rápido", expanded=True): st.write("**Caminho de trabalho atual (cwd):**", os.getcwd()) try: st.write("**Arquivos na raiz:**", os.listdir(".")) except Exception as e: st.write("Falha ao listar raiz:", e) dados_dir = Path("Dados") if dados_dir.exists(): try: st.write("**Arquivos em `Dados/`:**", os.listdir(dados_dir)) except Exception as e: st.write("Falha ao listar `Dados/`:", e) st.caption( "Se `Dados/Churn_Modelling.csv` não aparecer acima, suba o CSV para o repositório do Space " "com exatamente esse caminho e nome (case-sensitive)." ) # Alternativa 1: upload local (pode falhar em Spaces públicos com 403) st.info("**Alternativa:** faça upload do CSV abaixo para testar agora (não persiste no repositório).") up = st.file_uploader("Envie Churn_Modelling.csv", type=["csv"]) if up is not None: # Tentar separar por vírgula, depois ponto-e-vírgula e tab, se necessário try: df = pd.read_csv(up) except Exception: up.seek(0) try: df = pd.read_csv(up, sep=";") except Exception: up.seek(0) df = pd.read_csv(up, sep="\t") data_info = "via upload do usuário" # Alternativa 2: URL direta st.info("**Alternativa 2 (URL):** informe um link direto (RAW) para o CSV (ex.: GitHub Raw ou Hugging Face Datasets) e clique em **Carregar via URL**.") url_csv = st.text_input("URL direta do CSV (https://...)") load_url = st.button("Carregar via URL") if load_url and url_csv: try: df = pd.read_csv(url_csv) data_info = f"via URL: {url_csv}" except Exception: try: df = pd.read_csv(url_csv, sep=";") data_info = f"via URL (sep=';'): {url_csv}" except Exception: try: df = pd.read_csv(url_csv, sep="\t") data_info = f"via URL (sep='\\t'): {url_csv}" except Exception as e2: st.error(f"Falha ao carregar via URL: {e2}") st.stop() else: # Se nem upload nem URL foram usados e df continua vazio, parar aqui if df.empty: st.stop() st.success(f"Dataset carregado de: `{data_info}`") # Normalizar nomes de colunas df.columns = [c.strip() for c in df.columns] # ----------------------------- # Target and candidate features (dataset padrão do Kaggle) # ----------------------------- TARGET = "Exited" # 1 = saiu, 0 = permaneceu candidates_num = [ c for c in [ "CreditScore", "Age", "Tenure", "Balance", "NumOfProducts", "HasCrCard", "IsActiveMember", "EstimatedSalary" ] if c in df.columns ] candidates_cat = [c for c in ["Geography", "Gender"] if c in df.columns] # Sidebar: feature selection & model hyperparams st.sidebar.header("Configuração do Modelo") use_num = st.sidebar.multiselect( "Variáveis numéricas", options=candidates_num, default=[c for c in ["Age", "Balance", "NumOfProducts", "IsActiveMember"] if c in candidates_num] ) use_cat = st.sidebar.multiselect( "Variáveis categóricas", options=candidates_cat, default=[c for c in ["Geography", "Gender"] if c in candidates_cat] ) test_size = st.sidebar.slider("Proporção de teste", 0.1, 0.4, 0.2, 0.05) reg_strength = st.sidebar.slider("Força de regularização (C)", 0.05, 5.0, 1.0, 0.05) class_balanced = st.sidebar.checkbox("Class weight = 'balanced' (útil se desbalanceado)", value=True) max_iter = st.sidebar.slider("Max iter", 200, 2000, 1000, 100) train_btn = st.sidebar.button("Treinar modelo") # ----------------------------- # Quick EDA block (compact) # ----------------------------- st.subheader("Visão rápida do conjunto de dados") col_a, col_b = st.columns([2, 1]) with col_a: st.dataframe(df.sample(min(10, len(df))), use_container_width=True) with col_b: if TARGET in df.columns: n1 = int(df[TARGET].sum()) n0 = int((1 - df[TARGET]).sum()) st.metric("Clientes que saíram (1)", n1) st.metric("Clientes que ficaram (0)", n0) # ----------------------------- # Training # ----------------------------- def build_pipeline(num_cols, cat_cols, C=1.0, class_weight=None, max_iter=1000): preprocess = ColumnTransformer( transformers=[ ("num", StandardScaler(with_mean=True, with_std=True), num_cols), ("cat", OneHotEncoder(drop="first", handle_unknown="ignore"), cat_cols), ], remainder="drop" ) lr = LogisticRegression( C=C, penalty="l2", solver="lbfgs", max_iter=max_iter, class_weight=class_weight, n_jobs=None ) pipe = Pipeline(steps=[("prep", preprocess), ("clf", lr)]) return pipe def get_feature_names(preprocess, num_cols, cat_cols): names = [] if num_cols: names.extend(num_cols) if cat_cols: ohe = preprocess.named_transformers_["cat"] cat_names = ohe.get_feature_names_out(cat_cols).tolist() names.extend(cat_names) return names if train_btn: if not use_num and not use_cat: st.warning("Selecione pelo menos **uma** variável explicativa (numérica ou categórica).") st.stop() cols_needed = [TARGET] + use_num + use_cat df_model = df[cols_needed].dropna().copy() X = df_model[use_num + use_cat] y = df_model[TARGET] cw = "balanced" if class_balanced else None pipe = build_pipeline(use_num, use_cat, C=reg_strength, class_weight=cw, max_iter=max_iter) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=test_size, random_state=42, stratify=y ) pipe.fit(X_train, y_train) # ------------------------- # Coefficients & Odds Ratios # ------------------------- lr = pipe.named_steps["clf"] preprocess = pipe.named_steps["prep"] feat_names = get_feature_names(preprocess, use_num, use_cat) coefs = lr.coef_.ravel() odds = np.exp(coefs) coef_table = pd.DataFrame({ "Variável": feat_names, "Coeficiente (β)": coefs, "Odds Ratio (e^β)": odds }).sort_values(by="Odds Ratio (e^β)", ascending=False) st.subheader("Coeficientes e Odds Ratio") st.write( "Interpretação: mantendo as demais variáveis constantes, um aumento de uma unidade na variável " "(ou mudança para a categoria indicada) multiplica as *odds* de churn por `e^β`. " "Se `e^β > 1`, o risco de churn aumenta; se `< 1`, diminui." ) st.dataframe(coef_table, use_container_width=True, height=380) # Acurácia simples (para referência rápida no item a) acc = pipe.score(X_test, y_test) st.info(f"**Acurácia (holdout)**: {acc:.3f} | Amostras de treino: {len(X_train)} | Amostras de teste: {len(X_test)}") # ------------------------- # Interactive prediction # ------------------------- st.subheader("Simulação: probabilidade de churn para um perfil de cliente") with st.expander("Abrir painel de controle do cliente", expanded=True): inputs = {} cols = st.columns(2) # Numeric controls for i, col in enumerate(use_num): with cols[i % 2]: vmin = float(np.nanmin(df[col])) if np.isfinite(df[col]).all() else 0.0 vmax = float(np.nanmax(df[col])) if np.isfinite(df[col]).all() else 1.0 vmean = float(np.nanmean(df[col])) if np.isfinite(df[col]).all() else (vmin + vmax) / 2.0 step = (vmax - vmin) / 100.0 if vmax > vmin else 1.0 inputs[col] = st.number_input( f"{col}", value=round(vmean, 2), step=step, min_value=vmin, max_value=vmax, format="%.2f" ) # Categorical controls for i, col in enumerate(use_cat): with cols[i % 2]: opts = sorted([o for o in df[col].dropna().unique().tolist()]) default_idx = 0 if opts else None inputs[col] = st.selectbox(f"{col}", options=opts, index=default_idx if default_idx is not None else 0) # Compose a single-row DataFrame if inputs: row = pd.DataFrame([inputs]) proba = float(pipe.predict_proba(row)[0, 1]) st.metric("Probabilidade de churn (sair do banco)", f"{proba:.1%}") st.caption("Dica: ajuste os controles e observe como a probabilidade muda.") # ------------------------- # Textual help / interpretation # ------------------------- st.subheader("Como interpretar os coeficientes") st.markdown(""" - **Sinal de β**: positivo ⇒ aumenta as *odds* de churn; negativo ⇒ reduz. - **Magnitude**: valores maiores em módulo indicam maior impacto, dado o mesmo escalonamento. - **Odds Ratio `e^β`**: fator multiplicativo nas *odds*. Ex.: `e^β = 1,30` ⇒ as *odds* aumentam **30%**. - Em variáveis **categóricas**, o β refere-se à **categoria de referência vs. a categoria exibida** (depois do one-hot com `drop='first'`). """) else: st.info("Selecione as variáveis na barra lateral e clique em **Treinar modelo** para começar.") # ----------------------------- # Footer # ----------------------------- st.markdown("---") st.caption("PPCA/UnB • Tarefa 5 – Item (a) • Regressão Logística + Odds Ratio • Feito para rodar em Hugging Face Spaces (Streamlit).")