# Dataset: Dados/marketing_campaign.csv import os import numpy as np import pandas as pd import streamlit as st import altair as alt from typing import List, Tuple from sklearn.model_selection import train_test_split from sklearn.compose import ColumnTransformer from sklearn.preprocessing import OneHotEncoder, StandardScaler from sklearn.pipeline import Pipeline from sklearn.impute import SimpleImputer from sklearn.metrics import ( roc_auc_score, accuracy_score, confusion_matrix, roc_curve, r2_score, mean_squared_error ) from sklearn.linear_model import LogisticRegression, LinearRegression import statsmodels.api as sm st.set_page_config(page_title="Inferência Estatística", layout="wide") st.title("Inferência Estatística — Reclamações de Clientes") st.caption("Escolha o **alvo** e as **variáveis explicativas** na barra lateral (esquerda) e obtenha a inferência estatística.") DATA_PATH = "Dados/marketing_campaign.csv" # ---------- Utilidades ---------- @st.cache_data(show_spinner=False) def load_csv_try(path: str) -> pd.DataFrame: """Lê CSV tentando separadores: vírgula, ponto-e-vírgula e tab.""" for sep in [",", ";", "\t"]: try: df = pd.read_csv(path, sep=sep, encoding="utf-8") if sep != "\t" and df.shape[1] == 1: continue return df except Exception: continue return pd.read_csv(path, sep=None, engine="python") def split_num_cat(df: pd.DataFrame, exclude: List[str]) -> Tuple[List[str], List[str]]: num_cols = [c for c in df.columns if c not in exclude and np.issubdtype(df[c].dtype, np.number)] cat_cols = [c for c in df.columns if c not in exclude and (df[c].dtype == "object" or df[c].dtype.name == "category")] return num_cols, cat_cols def is_binary_series(s: pd.Series) -> bool: vals = pd.unique(s.dropna()) return len(vals) == 2 or s.dtype == bool def coerce_numeric_series(s: pd.Series) -> pd.Series: """Tenta converter strings numéricas para float (lida com vírgula decimal).""" if np.issubdtype(s.dtype, np.number): return s.astype(float) tmp = s.astype(str).str.replace(r"[.\s]", "", regex=True).str.replace(",", ".", regex=False) return pd.to_numeric(tmp, errors="coerce") def engineer_features(df: pd.DataFrame) -> pd.DataFrame: """Engenharia minimalista para o dataset padrão do Kaggle.""" out = df.copy() # Tenure (dias desde Dt_Customer) if "Dt_Customer" in out.columns: out["Dt_Customer"] = pd.to_datetime(out["Dt_Customer"], errors="coerce", dayfirst=True) out["TenureDays"] = (pd.Timestamp("today").normalize() - out["Dt_Customer"]).dt.days # Total gasto (Mnt*) mnt_cols = [c for c in out.columns if c.startswith("Mnt")] if mnt_cols: out["TotalMnt"] = out[mnt_cols].sum(axis=1) # Compras totais e participações buy_cols = [c for c in ["NumWebPurchases", "NumCatalogPurchases", "NumStorePurchases"] if c in out.columns] if buy_cols: out["TotalPurchases"] = out[buy_cols].sum(axis=1) if "NumWebPurchases" in out.columns: out["OnlineShare"] = out["NumWebPurchases"] / out["TotalPurchases"].replace(0, np.nan) if "NumDealsPurchases" in out.columns: out["PromoShare"] = out["NumDealsPurchases"] / out["TotalPurchases"].replace(0, np.nan) # Ticket médio if "TotalMnt" in out.columns and "TotalPurchases" in out.columns: out["AvgTicket"] = out["TotalMnt"] / out["TotalPurchases"].replace(0, np.nan) # Diversidade de cesta (quantos tipos Mnt*>0) if mnt_cols: out["BasketDiversity"] = (out[mnt_cols] > 0).sum(axis=1) return out def build_preprocessor(num_cols: List[str], cat_cols: List[str]) -> ColumnTransformer: """Imputação + padronização (num) e OHE drop='first' (cat) para evitar colinearidade.""" num_pipe = Pipeline([ ("imp", SimpleImputer(strategy="median")), ("scaler", StandardScaler()) ]) cat_pipe = Pipeline([ ("imp", SimpleImputer(strategy="most_frequent")), ("ohe", OneHotEncoder(handle_unknown="ignore", drop="first", sparse_output=False)) ]) return ColumnTransformer([ ("num", num_pipe, num_cols), ("cat", cat_pipe, cat_cols) ]) def get_feature_names(pre: ColumnTransformer, num_cols: List[str], cat_cols: List[str]) -> List[str]: names = list(num_cols) if cat_cols: ohe = pre.named_transformers_["cat"].named_steps["ohe"] names.extend(list(ohe.get_feature_names_out(cat_cols))) return names def fit_inference(model_type: str, X_design: pd.DataFrame, y: pd.Series): """Ajusta a inferência (statsmodels): Logit p/ binário; OLS p/ contínuo.""" X_sm = sm.add_constant(X_design, has_constant="add") if model_type == "logit": res = sm.Logit(y.values, X_sm).fit(disp=False) or_vals = np.exp(res.params) or_ci = np.exp(res.conf_int()) tbl = pd.DataFrame({ "feature": res.params.index, "coef": res.params.values, "std_err": res.bse.values, "z/t": res.tvalues.values if hasattr(res, "tvalues") else res.tvalues, "p_value": res.pvalues.values, "ci_low": res.conf_int()[0].values, "ci_high": res.conf_int()[1].values, "odds_ratio": or_vals.values, "or_ci_low": or_ci[0].values, "or_ci_high": or_ci[1].values }) else: res = sm.OLS(y.values, X_sm).fit() conf = res.conf_int() tbl = pd.DataFrame({ "feature": res.params.index, "coef": res.params.values, "std_err": res.bse.values, "z/t": res.tvalues.values, "p_value": res.pvalues.values, "ci_low": conf[0].values, "ci_high": conf[1].values }) return res, tbl def recs_from_inference(tbl: pd.DataFrame, model_type: str, k: int = 5): """Gera recomendações (item e) a partir dos efeitos significativos (p<0.05), ignorando 'const'.""" df = tbl[tbl["feature"] != "const"].copy() df = df.sort_values(["p_value", "z/t"], ascending=[True, False]) core = df[df["p_value"] < 0.05].head(k) out = [] for _, r in core.iterrows(): feat = r["feature"] sign = np.sign(r["coef"]) if model_type == "logit": or_txt = f"(OR≈{r['odds_ratio']:.2f}, IC95% {r['or_ci_low']:.2f}–{r['or_ci_high']:.2f}, p={r['p_value']:.3g})" if sign > 0: out.append(f" **Reduzir exposição associada a `{feat}`** {or_txt}, pois aumento nessa variável eleva a probabilidade do alvo.") else: out.append(f" **Fortalecer fatores ligados a `{feat}`** {or_txt}, pois valores maiores reduzem a probabilidade do alvo.") else: eff = f"(β≈{r['coef']:.3g}, IC95% {r['ci_low']:.2g}–{r['ci_high']:.2g}, p={r['p_value']:.3g})" if sign > 0: out.append(f" **Mitigar o crescimento de `{feat}`** {eff}, pois contribui positivamente para o aumento do alvo.") else: out.append(f" **Aumentar `{feat}`** {eff}, pois está associado à redução do alvo.") # trilhas transversais out.append(" **Testes A/B** nas variáveis mais significativas para validar impacto causal.") out.append(" **Melhorar FCR/primeiro contato** nas causas evidenciadas pelos top fatores.") out.append(" **Feedback a Produto/Qualidade** guiado pelos efeitos com evidência estatística robusta.") return out[:k+3] # ---------- Sidebar (lado esquerdo) ---------- with st.sidebar: st.header("Configuração") if not os.path.exists(DATA_PATH): st.error(f"Arquivo não encontrado: `{DATA_PATH}`. Suba o CSV em `Dados/`.") st.stop() df_raw = load_csv_try(DATA_PATH) df_eng = engineer_features(df_raw) with st.sidebar: st.markdown("**Alvo (variável dependente):**") all_cols = df_eng.columns.tolist() # Alvo padrão fixo: Response (se existir). Caso contrário, mesma lógica de fallback. if "Response" in all_cols: default_target = "Response" else: default_target = None for c in all_cols: if is_binary_series(df_eng[c]): default_target = c; break if default_target is None: for c in all_cols: if np.issubdtype(df_eng[c].dtype, np.number): default_target = c; break target_col = st.selectbox("Alvo (y)", options=all_cols, index=all_cols.index(default_target) if default_target in all_cols else 0) # Variáveis explicativas exclude = [target_col] num_cols_all, cat_cols_all = split_num_cat(df_eng, exclude=exclude) # X padrão alinhado ao Colab (só incluir se existir na base) preferred_defaults = [ "Income", "Recency", "Education", "Marital_Status", "TenureDays", "TotalMnt", "TotalPurchases", "OnlineShare", "PromoShare", "AvgTicket", "BasketDiversity", "NumWebVisitsMonth" ] default_X = [c for c in preferred_defaults if c in (num_cols_all + cat_cols_all)] with st.sidebar: st.markdown("**Variáveis explicativas (X):**") # Se nada dos preferidos existir, cai no fallback antigo (algumas num + categ) if not default_X: engineered_first = [c for c in ["TenureDays","TotalMnt","TotalPurchases","OnlineShare","PromoShare","AvgTicket","BasketDiversity"] if c in num_cols_all] default_X = engineered_first + [c for c in num_cols_all if c not in engineered_first][:5] + cat_cols_all[:3] selected_feats = st.multiselect("Selecione X", options=(num_cols_all + cat_cols_all), default=default_X) test_size = st.slider("Proporção de teste", 0.1, 0.4, 0.2, 0.05) random_state = st.number_input("Random seed", value=42, step=1) if len(selected_feats) == 0: st.warning("Selecione pelo menos uma variável explicativa.") st.stop() # ---------- Amostra ---------- st.markdown("### Amostra dos dados") st.dataframe(df_eng[[target_col] + selected_feats].head(12), use_container_width=True) # ---------- Preparação do alvo ---------- y_raw = df_eng[target_col] # 1) tenta identificar binário diretamente is_bin = is_binary_series(y_raw) # 2) se não binário, tenta numérico (coerção segura) y_numeric_try = coerce_numeric_series(y_raw) if not is_bin else None is_numeric_ok = False if not is_bin and y_numeric_try is not None: conv_rate = y_numeric_try.notna().mean() is_numeric_ok = conv_rate >= 0.8 # 3) se não binário e não numérico, vira categórico multi-classe → one-vs-rest with st.sidebar: positive_class = None if not is_bin and not is_numeric_ok: uniq_vals = sorted(pd.unique(y_raw.dropna()).tolist(), key=lambda x: str(x)) st.markdown("**Alvo categórico com múltiplas classes**") positive_class = st.selectbox("Classe 'positiva' (one-vs-rest)", options=uniq_vals, index=0) st.caption("O modelo fará Logit para a classe escolhida vs. as demais.") # ---------- Montagem de y conforme os casos ---------- if is_bin: if not np.issubdtype(y_raw.dtype, np.number): uniq = sorted(pd.unique(y_raw.dropna()).tolist(), key=lambda x: str(x)) y = y_raw.replace({uniq[0]: 0, uniq[1]: 1}).astype(int) else: y = y_raw.astype(int) model_type = "logit" elif is_numeric_ok: y = y_numeric_try.astype(float) model_type = "ols" else: y = (y_raw == positive_class).astype(int) model_type = "logit" # Alinha df aos y válidos mask_valid = y.notna() df_model = df_eng.loc[mask_valid].copy() y = y.loc[mask_valid] X = df_model[selected_feats].copy() # ---------- Pré-processamento e treino ---------- sel_num = [c for c in selected_feats if np.issubdtype(X[c].dtype, np.number)] sel_cat = [c for c in selected_feats if (X[c].dtype == "object" or X[c].dtype.name == "category")] pre = build_preprocessor(sel_num, sel_cat) quick_est = LogisticRegression(max_iter=200) if model_type == "logit" else LinearRegression() pipe = Pipeline([("pre", pre), ("est", quick_est)]) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=test_size, random_state=random_state, stratify=y if model_type == "logit" else None ) with st.spinner("Treinando e construindo matriz de design..."): pipe.fit(X_train, y_train) pre_fit = pipe.named_steps["pre"].fit(X_train, y_train) X_train_design = pre_fit.transform(X_train) # nomes das features após OHE ohe_names = [] if sel_cat: ohe_names = list(pre_fit.named_transformers_["cat"].named_steps["ohe"].get_feature_names_out(sel_cat)) feat_names = sel_num + ohe_names X_train_df = pd.DataFrame(X_train_design, columns=feat_names) # ---------- Inferência (item e) ---------- st.markdown("## Inferência estatística") with st.spinner("Ajustando modelo de inferência (statsmodels)..."): res, infer_tbl = fit_inference(model_type, X_train_df, y_train) if model_type == "logit": if positive_class is not None: st.caption(f"Modelo: **Logit** (one-vs-rest). Classe positiva: **{positive_class}**.") else: st.caption("Modelo: **Logit** (alvo binário). Coeficientes em log-odds; exibimos **odds ratios** e IC 95%.") cols_show = ["feature","coef","std_err","z/t","p_value","ci_low","ci_high","odds_ratio","or_ci_low","or_ci_high"] else: st.caption("Modelo: **OLS** (alvo contínuo). Coeficientes, erros-padrão, estatística t e IC 95%.") cols_show = ["feature","coef","std_err","z/t","p_value","ci_low","ci_high"] st.dataframe(infer_tbl[cols_show].round(4), use_container_width=True) # ---------- Métricas ---------- st.markdown("### Desempenho do modelo") if model_type == "logit": y_proba = pipe.predict_proba(X_test)[:, 1] if hasattr(pipe.named_steps["est"], "predict_proba") else pipe.predict(X_test) y_pred = (y_proba >= 0.5).astype(int) auc = roc_auc_score(y_test, y_proba) acc = accuracy_score(y_test, y_pred) c1, c2 = st.columns(2) with c1: st.metric("AUC (ROC)", f"{auc:.3f}") with c2: st.metric("Acurácia (0.5)", f"{acc:.3f}") cm = confusion_matrix(y_test, y_pred) st.markdown("**Matriz de confusão (teste)**") st.dataframe(pd.DataFrame(cm, index=["Real 0","Real 1"], columns=["Pred 0","Pred 1"]), use_container_width=True) fpr, tpr, _ = roc_curve(y_test, y_proba) roc_data = pd.DataFrame({"fpr": fpr, "tpr": tpr}) roc_chart = alt.Chart(roc_data).mark_line().encode(x="fpr:Q", y="tpr:Q").properties(height=250, width=380) diag = alt.Chart(pd.DataFrame({"x":[0,1],"y":[0,1]})).mark_line(strokeDash=[4,4]).encode(x="x", y="y") st.altair_chart(roc_chart + diag, use_container_width=True) else: y_pred = pipe.predict(X_test) r2 = r2_score(y_test, y_pred) rmse = mean_squared_error(y_test, y_pred, squared=False) c1, c2 = st.columns(2) with c1: st.metric("R² (teste)", f"{r2:.3f}") with c2: st.metric("RMSE (teste)", f"{rmse:.3f}") # ---------- Força dos efeitos ---------- st.markdown("### Força dos efeitos (|t/z|)") eff_df = infer_tbl[infer_tbl["feature"] != "const"].copy() eff_df["effect_strength"] = eff_df["z/t"].abs() eff_chart = alt.Chart(eff_df.sort_values("effect_strength", ascending=False).head(20)).mark_bar().encode( x=alt.X("effect_strength:Q", title="|estatística t/z|"), y=alt.Y("feature:N", sort='-x', title="Variável") ).properties(height=420) st.altair_chart(eff_chart, use_container_width=True) # ---------- Predição interativa ---------- st.markdown("## Predição interativa") st.caption("Ajuste valores para X e veja a probabilidade (Logit) ou valor previsto (OLS).") with st.form("pred_form"): cols = st.columns(3) user_inputs = {} for i, col in enumerate(selected_feats): with cols[i % 3]: if col in [c for c in selected_feats if np.issubdtype(X[c].dtype, np.number)]: q1, q5, q95, q99 = X_train[col].quantile([0.01,0.05,0.95,0.99]) default_val = float(np.nan_to_num(X_train[col].median(), nan=0.0)) user_inputs[col] = st.number_input( f"{col}", value=default_val, help=f"Faixa típica ~ {q5:.2f}–{q95:.2f} (1–99%: {q1:.2f}–{q99:.2f})" ) else: opts = sorted([str(x) for x in X_train[col].dropna().unique().tolist()])[:50] user_inputs[col] = st.selectbox(f"{col}", options=opts if opts else [""], index=0 if opts else 0) submitted = st.form_submit_button("Calcular") if submitted: x_new = pd.DataFrame([user_inputs]) x_new_proc = pre_fit.transform(x_new) x_new_df = pd.DataFrame(x_new_proc, columns=feat_names) X_sm_new = sm.add_constant(x_new_df, has_constant="add") y_hat = float(res.predict(X_sm_new)[0]) if model_type == "logit": st.success(f"Probabilidade prevista do alvo: **{y_hat:.2%}**") else: st.success(f"Valor previsto do alvo: **{y_hat:.4g}**") # ---------- Recomendações (item e) ---------- st.markdown("## Recomendações estratégicas (Item e)") for r in recs_from_inference(infer_tbl, model_type=model_type, k=5): st.markdown("- " + r) st.markdown("---") st.caption("Controles na barra lateral (esquerda) • Dados: `Dados/marketing_campaign.csv` • Inferência conforme item (e).")