Análise de dados do básico ao expert

# Ambiente isolado por projeto (padrão de mercado)
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install pandas numpy matplotlib seaborn jupyterlab pyarrow
pip freeze > requirements.txt    # reprodutibilidade

vendas = [1200, 980, 1450, 2100]           # lista
meta = {"jan": 1000, "fev": 1100}          # dicionário

# List/dict comprehension: a construção mais usada em dados
acima = [v for v in vendas if v > 1000]
pct_meta = {mes: v / 1000 for mes, v in meta.items()}

# f-strings para relatórios
total = sum(vendas)
print(f"Total: R$ {total:,.2f} | Média: R$ {total/len(vendas):,.1f}")

import numpy as np

precos = np.array([19.9, 45.0, 12.5, 89.9])
qtd    = np.array([100, 30, 250, 12])

receita = precos * qtd              # vetorizado: sem for!
receita_com_desc = np.where(receita > 2000, receita * 0.95, receita)

print(receita)
print(f"p90 da receita: {np.percentile(receita, 90):.0f}")

import pandas as pd

df = pd.read_csv("vendas.csv", parse_dates=["data"])

# O ritual de reconhecimento — sempre, em todo dataset novo:
df.head()
df.info()               # tipos, nulos, memória
df.describe(include="all")
df.nunique()            # cardinalidade de cada coluna

# loc = por rótulo | iloc = por posição
df.loc[df["uf"] == "SP", ["data", "receita"]]
df.iloc[:5, 0:3]

# Condições compostas: parênteses obrigatórios, & e | (não and/or)
filtro = (df["receita"] > 1000) & (df["canal"].isin(["app", "site"]))
df.loc[filtro]

# query(): mais legível para filtros longos
df.query("receita > 1000 and canal in ['app','site'] and uf != 'SP'")

df["ticket"] = df["receita"] / df["pedidos"]

# assign() permite encadear (method chaining — estilo profissional)
resumo = (
    df
    .assign(
        ticket=lambda d: d["receita"] / d["pedidos"],
        faixa=lambda d: pd.cut(d["ticket"], bins=[0, 50, 150, np.inf],
                               labels=["baixo", "médio", "alto"]),
    )
    .query("receita > 0")
    .sort_values("receita", ascending=False)
)

def diagnostico(df):
    return pd.DataFrame({
        "dtype": df.dtypes.astype(str),
        "nulos": df.isna().sum(),
        "pct_nulos": (df.isna().mean() * 100).round(1),
        "unicos": df.nunique(),
        "duplicados_col": [df[c].duplicated().sum() for c in df.columns],
        "exemplo": df.apply(lambda s: s.dropna().iloc[0] if s.notna().any() else None),
    })

diagnostico(df)
df.duplicated(subset=["pedido_id"]).sum()   # duplicatas de chave

# Estratégias em ordem de sofisticação
df["renda_ausente"] = df["renda"].isna()          # flag: sempre útil
df["renda"] = df["renda"].fillna(df["renda"].median())

# Imputação por grupo (MAR): muito mais defensável
df["renda"] = df["renda"].fillna(
    df.groupby("faixa_etaria")["renda"].transform("median")
)

# Séries temporais: propagação limitada
df["estoque"] = df["estoque"].ffill(limit=3)

q1, q3 = df["ticket"].quantile([0.25, 0.75])
iqr = q3 - q1
limites = (q1 - 1.5 * iqr, q3 + 1.5 * iqr)
outliers = df.query("ticket < @limites[0] or ticket > @limites[1]")

# Winsorização: alternativa a remover (preserva a linha, limita o valor)
df["ticket_w"] = df["ticket"].clip(*df["ticket"].quantile([0.01, 0.99]))

# Asserts baratos que salvam análises inteiras
assert df["pedido_id"].is_unique, "IDs duplicados!"
assert (df["receita"] >= 0).all(), "Receita negativa!"
assert df["data"].between("2020-01-01", pd.Timestamp.today()).all()

# Em pipelines sérios: biblioteca pandera define schemas validáveis
import pandera as pa
schema = pa.DataFrameSchema({
    "receita": pa.Column(float, pa.Check.ge(0)),
    "uf": pa.Column(str, pa.Check.isin(UFS_VALIDAS)),
})
df = schema.validate(df)

# Agregação nomeada: o padrão moderno e legível
kpis = (
    df.groupby(["uf", "canal"], observed=True)
      .agg(
          receita_total=("receita", "sum"),
          ticket_medio=("receita", "mean"),
          pedidos=("pedido_id", "nunique"),
          p90_ticket=("receita", lambda s: s.quantile(0.9)),
      )
      .reset_index()
)

df["pct_da_uf"] = df["receita"] / df.groupby("uf")["receita"].transform("sum")
df["vs_media_canal"] = df["receita"] - df.groupby("canal")["receita"].transform("mean")
df["rank_no_mes"] = df.groupby("mes")["receita"].rank(ascending=False)

# Top 3 produtos por categoria — pergunta de entrevista clássica
top3 = (df.sort_values("receita", ascending=False)
          .groupby("categoria")
          .head(3))

base = pedidos.merge(
    clientes,
    on="cliente_id",
    how="left",
    validate="m:1",        # explode se clientes tiver ID duplicado!
    indicator=True,        # coluna _merge: auditoria do join
)
base["_merge"].value_counts()
# both          98_412
# left_only        188   ← pedidos órfãos: investigar antes de seguir

# Longo → largo: a "tabela de gestor" (UF nas linhas, mês nas colunas)
matriz = df.pivot_table(
    index="uf", columns="mes", values="receita",
    aggfunc="sum", fill_value=0, margins=True, margins_name="Total"
)

# Largo → longo: formato ideal para análise e gráficos (tidy data)
tidy = matriz.drop("Total").reset_index().melt(
    id_vars="uf", var_name="mes", value_name="receita"
)

df["data"] = pd.to_datetime(df["data"], format="%d/%m/%Y")

df["ano_mes"]     = df["data"].dt.to_period("M")
df["dia_semana"]  = df["data"].dt.day_name()
df["fim_de_semana"] = df["data"].dt.dayofweek >= 5

# Aritmética de datas: idade da conta em dias
df["dias_como_cliente"] = (df["data"] - df["data_cadastro"]).dt.days

# Deslocamentos de negócio
df["vencimento"] = df["data"] + pd.offsets.BusinessDay(5)

# Padronização — o kit básico de higiene
df["cidade"] = (df["cidade"].str.strip().str.title()
                 .str.normalize("NFKD")               # remove acentos…
                 .str.encode("ascii", "ignore").str.decode("ascii"))

# Extração com regex: CNPJ no meio de texto livre
df["cnpj"] = df["descricao"].str.extract(r"(\d{2}\.?\d{3}\.?\d{3}/?\d{4}-?\d{2})")

df["contem_cancelamento"] = df["obs"].str.contains("cancel", case=False, na=False)
df[["ddd", "numero"]] = df["telefone"].str.split(")", expand=True)

import sqlalchemy as sa
engine = sa.create_engine("postgresql://user:senha@host:5432/db")

query = """
WITH mensal AS (
    SELECT cliente_id,
           date_trunc('month', data) AS mes,
           SUM(receita) AS receita
    FROM pedidos
    WHERE data >= '2025-01-01'
    GROUP BY 1, 2
)
SELECT *,
       SUM(receita)  OVER (PARTITION BY cliente_id ORDER BY mes) AS receita_acum,
       LAG(receita)  OVER (PARTITION BY cliente_id ORDER BY mes) AS receita_ant,
       ROW_NUMBER()  OVER (PARTITION BY mes ORDER BY receita DESC) AS rank_mes
FROM mensal
"""
df = pd.read_sql(query, engine)

import matplotlib.pyplot as plt
import seaborn as sns
sns.set_theme(style="whitegrid", palette="deep")

fig, ax = plt.subplots(figsize=(9, 5))
sns.lineplot(data=mensal, x="mes", y="receita", hue="canal", ax=ax)

# O que separa um gráfico amador de um profissional:
ax.set_title("Receita do canal app cresce 34% e ultrapassa o site em maio",
             loc="left", fontsize=13, fontweight="bold")   # título = conclusão!
ax.set_xlabel("")
ax.set_ylabel("Receita (R$ mil)")
ax.yaxis.set_major_formatter(lambda x, _: f"{x/1000:,.0f}")
ax.spines[["top", "right"]].set_visible(False)
ax.annotate("lançamento do cupom", xy=("2025-05", 480_000),
            xytext=(10, 15), textcoords="offset points", fontsize=9)
fig.tight_layout()
fig.savefig("receita_canal.png", dpi=150)

# Plotly: exploração interativa e relatórios HTML
import plotly.express as px
fig = px.scatter(df, x="desconto", y="margem", color="categoria",
                 size="receita", hover_data=["produto"], trendline="ols")
fig.write_html("analise_margem.html")

# Streamlit: de notebook a app interno em ~30 linhas
# streamlit run app.py

df.memory_usage(deep=True).sum() / 1e6   # MB reais

# Downcast + category: reduções de 50–90% são comuns
def otimizar(df, limite_cat=0.5):
    for col in df.select_dtypes("integer"):
        df[col] = pd.to_numeric(df[col], downcast="integer")
    for col in df.select_dtypes("float"):
        df[col] = pd.to_numeric(df[col], downcast="float")
    for col in df.select_dtypes("object"):
        if df[col].nunique() / len(df) < limite_cat:
            df[col] = df[col].astype("category")
    return df

# Melhor ainda: nem carregar o que não precisa
df = pd.read_parquet("vendas.parquet",
                     columns=["data", "uf", "receita"],
                     filters=[("ano", "=", 2025)])

# ANTES (lento): lógica condicional com apply em linhas
df["frete"] = df.apply(lambda r: 0 if r.receita > 200
                        else 15 if r.uf == "SP" else 25, axis=1)

# DEPOIS (vetorizado): np.select com condições ordenadas
cond = [df.receita > 200, df.uf.eq("SP")]
df["frete"] = np.select(cond, [0, 15], default=25)

import duckdb
# 40 GB de parquet num laptop de 16 GB de RAM? DuckDB resolve:
top = duckdb.sql("""
    SELECT uf, date_trunc('month', data) AS mes, SUM(receita) AS receita
    FROM 'dados/vendas_*.parquet'
    GROUP BY 1, 2
    ORDER BY receita DESC
""").df()      # .df() devolve um DataFrame Pandas

from scipy import stats

conv = df["converteu"]                      # série 0/1
n, p = len(conv), conv.mean()

# IC 95% para proporção (aprox. normal)
ep = np.sqrt(p * (1 - p) / n)
ic = (p - 1.96 * ep, p + 1.96 * ep)
print(f"Conversão: {p:.2%} (IC95%: {ic[0]:.2%} – {ic[1]:.2%})")

# Bootstrap: funciona para QUALQUER métrica (mediana, p90, razões…)
res = stats.bootstrap((df["ticket"].values,), np.median,
                      n_resamples=5_000, confidence_level=0.95)
print(f"Mediana do ticket: IC95% {res.confidence_interval}")

import statsmodels.formula.api as smf

# Regressão com interpretação de negócio (não é ML: é explicação)
modelo = smf.ols(
    "np.log1p(receita) ~ desconto + C(canal) + C(uf) + dias_como_cliente",
    data=df
).fit(cov_type="HC3")          # erros-padrão robustos
print(modelo.summary())

# Leitura: coef de desconto = -0.021 →
# cada ponto de desconto ↘ receita em ~2,1%, controlando canal, UF e tenure

ts = df.set_index("data").sort_index()

# Reamostragem: de eventos para grade regular
diario = ts["receita"].resample("D").sum()
mensal = ts["receita"].resample("MS").agg(["sum", "count", "mean"])

# Média móvel: suavizar sem mentir (window centrado p/ análise)
diario_mm7  = diario.rolling(7).mean()
diario_mm28 = diario.rolling(28, min_periods=20).mean()

# Comparações que o negócio pede
mom = mensal["sum"].pct_change()            # vs mês anterior
yoy = mensal["sum"].pct_change(12)          # vs mesmo mês do ano anterior
acum_12m = mensal["sum"].rolling(12).sum()  # LTM (últimos 12 meses)

from statsmodels.tsa.seasonal import STL

stl = STL(diario, period=7, robust=True).fit()
stl.plot()

# Detecção de anomalia simples e eficaz: resíduo fora de 3 desvios
resid = stl.resid
anomalias = diario[np.abs(resid) > 3 * resid.std()]

# Validação correta de séries temporais: NUNCA embaralhe o tempo
from sklearn.model_selection import TimeSeriesSplit
tscv = TimeSeriesSplit(n_splits=5, test_size=28)
for treino_idx, teste_idx in tscv.split(diario):
    ...   # treina no passado, testa no futuro, janela deslizante

from statsmodels.stats.power import NormalIndPower
from statsmodels.stats.proportion import proportion_effectsize, proportions_ztest

# 1) Tamanho de amostra: base 3% de conversão, queremos detectar +0,5 p.p.
efeito = proportion_effectsize(0.030, 0.035)
n_por_grupo = NormalIndPower().solve_power(efeito, power=0.8, alpha=0.05)
print(f"≈ {n_por_grupo:,.0f} usuários por grupo")

# 2) Análise ao final
stat, pval = proportions_ztest([conv_B, conv_A], [n_B, n_A])
lift = conv_B/n_B - conv_A/n_A
print(f"Lift: {lift:+.2%} | p = {pval:.4f}")

# Diff-in-diff via regressão (o jeito profissional)
did = smf.ols(
    "receita ~ tratado * pos_evento + C(uf) + C(mes)",
    data=painel
).fit(cov_type="cluster", cov_kwds={"groups": painel["uf"]})
# O coeficiente de tratado:pos_evento É o efeito causal estimado

from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.ensemble import HistGradientBoostingClassifier

num = ["tenure_dias", "ticket_medio", "chamados_90d"]
cat = ["plano", "canal_aquisicao", "uf"]

prep = ColumnTransformer([
    ("num", StandardScaler(), num),
    ("cat", OneHotEncoder(handle_unknown="ignore"), cat),
])
pipe = Pipeline([("prep", prep),
                 ("modelo", HistGradientBoostingClassifier())])

X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2,
                                          stratify=y, random_state=42)
pipe.fit(X_tr, y_tr)

from sklearn.metrics import (roc_auc_score, average_precision_score,
                             classification_report)

proba = pipe.predict_proba(X_te)[:, 1]
print(f"ROC-AUC: {roc_auc_score(y_te, proba):.3f}")
print(f"PR-AUC : {average_precision_score(y_te, proba):.3f}")  # melhor p/ classes raras

# A pergunta certa não é "qual a acurácia?" e sim:
# "com capacidade de contatar 500 clientes/mês, quanto churn evitamos
#  se contatarmos os 500 de maior score?" → análise de lift por decil
deciles = (pd.DataFrame({"proba": proba, "real": y_te})
             .assign(decil=lambda d: pd.qcut(d.proba, 10, labels=False))
             .groupby("decil")["real"].agg(["mean", "count"]))

import shap
explainer = shap.TreeExplainer(pipe.named_steps["modelo"])
sv = explainer(prep.transform(X_te))
shap.plots.beeswarm(sv)
# "chamados_90d ≥ 3 é o maior driver de churn; tenure protege"
# → isso vira ação de negócio, não só um score

from sklearn.cluster import KMeans

# RFM: a segmentação que todo negócio B2C entende
rfm = (df.groupby("cliente_id")
         .agg(recencia=("data", lambda s: (hoje - s.max()).days),
              frequencia=("pedido_id", "nunique"),
              valor=("receita", "sum")))

X = StandardScaler().fit_transform(np.log1p(rfm))
rfm["cluster"] = KMeans(n_clusters=4, n_init="auto", random_state=42).fit_predict(X)
rfm.groupby("cluster").median()   # perfil de cada segmento → nomeie-os!

# Estrutura de projeto que times de dados respeitam
projeto/
├── data/            # raw/ e processed/ (fora do git!)
├── notebooks/       # exploração datada
├── src/
│   ├── extracao.py
│   ├── limpeza.py   # funções puras: df entra, df sai
│   └── metricas.py
├── tests/
│   └── test_limpeza.py
├── requirements.txt
└── README.md        # como rodar, decisões, fontes

# tests/test_limpeza.py — rode com: pytest
import pandas as pd
from src.limpeza import classificar_faixa

def test_faixas_cobrem_bordas():
    entrada = pd.DataFrame({"ticket": [0, 50, 50.01, 150, 9999]})
    saida = classificar_faixa(entrada)
    assert saida["faixa"].tolist() == ["baixo","baixo","médio","médio","alto"]

def test_nao_perde_linhas():
    entrada = pd.read_parquet("tests/fixtures/amostra.parquet")
    assert len(classificar_faixa(entrada)) == len(entrada)

# O trio da automação leve:
# 1. papermill: executa notebooks com parâmetros
papermill relatorio.ipynb saida_2026-06.ipynb -p mes "2026-06"

# 2. Quarto/nbconvert: notebook → HTML/PDF apresentável
quarto render relatorio.ipynb --to html

# 3. GitHub Actions (agenda semanal) envia por e-mail/Slack
#    on: schedule: - cron: "0 10 * * 1"