Projeto 2: Pipeline de Data Science do Zero ao Deploy

O Projeto 2 é um pipeline completo de Data Science — desde a coleta e limpeza de dados até o deploy de um modelo de machine learning como API em produção. Ele junta o que a série construiu em partes: o pandas de Introdução ao Data Science com Python, os modelos de Machine Learning com scikit-learn, o rastreamento de MLOps com MLflow e a API de FastAPI. O problema: prever a probabilidade de evasão escolar de um aluno com base em seu histórico acadêmico.

Visão Geral do Projeto

evasao-predictor/
├── data/
│   ├── raw/                     ← dados brutos originais
│   ├── processed/               ← dados limpos
│   └── features/                ← features engenheiradas
├── notebooks/
│   └── 01_exploracao.ipynb      ← EDA interativa
├── src/
│   └── evasao/
│       ├── __init__.py
│       ├── config.py
│       ├── data/
│       │   ├── coleta.py        ← geração/coleta de dados
│       │   ├── limpeza.py       ← preprocessamento
│       │   └── features.py      ← feature engineering
│       ├── modelos/
│       │   ├── treino.py        ← treinamento
│       │   ├── avaliacao.py     ← métricas
│       │   └── predicao.py      ← inferência
│       └── api/
│           ├── main.py          ← FastAPI
│           └── schemas.py       ← Pydantic
├── tests/
│   ├── test_limpeza.py
│   ├── test_features.py
│   └── test_api.py
├── Dockerfile
├── docker-compose.yml
└── Makefile

Etapa 1: Geração e Coleta de Dados

# src/evasao/data/coleta.py
import numpy as np
import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
import json


def gerar_dataset(n: int = 5000, seed: int = 42) -> pd.DataFrame:
    """
    Gera dataset sintético realista de alunos com risco de evasão.
    Baseado em fatores reais de estudos sobre evasão escolar no Brasil.
    """
    rng = np.random.default_rng(seed)

    # Características do aluno
    idade           = rng.integers(15, 25, n)
    sexo            = rng.choice(["M", "F"], n, p=[0.52, 0.48])
    turno           = rng.choice(["manha", "tarde", "noite"], n, p=[0.35, 0.35, 0.30])
    trabalha        = rng.choice([0, 1], n, p=[0.60, 0.40])
    renda_familiar  = rng.choice(
        ["ate_1sm", "1_3sm", "3_5sm", "acima_5sm"], n,
        p=[0.30, 0.40, 0.20, 0.10]
    )

    # Histórico acadêmico
    media_anterior  = rng.normal(6.5, 1.5, n).clip(0, 10)
    faltas_pct      = rng.beta(2, 8, n) * 50   # % de faltas
    reprovacoes     = rng.integers(0, 4, n)
    horas_estudo    = rng.normal(3, 1.5, n).clip(0, 10)

    # Engajamento
    participacao    = rng.normal(6.0, 2.0, n).clip(0, 10)
    entrega_tarefas = rng.normal(7.5, 2.0, n).clip(0, 10)
    acesso_plataforma = rng.integers(0, 30, n)   # acessos/mês

    # Notas do semestre atual
    notas_bimestres = rng.normal(
        media_anterior[:, np.newaxis],
        1.2,
        (n, 4)
    ).clip(0, 10)
    nota_b1, nota_b2, nota_b3, nota_b4 = notas_bimestres.T

    # Variável alvo — evasão (definida por regras realistas)
    score_risco = (
        1.5                                   # intercepto: ~26% de evasão
        - 0.3 * media_anterior
        + 0.4 * faltas_pct / 10
        + 0.3 * reprovacoes
        - 0.2 * horas_estudo
        - 0.2 * participacao
        + 0.3 * trabalha
        + 0.1 * (turno == "noite").astype(float)
        + rng.normal(0, 0.5, n)
    )
    prob_evasao = 1 / (1 + np.exp(-score_risco))
    # sorteio, não corte em 0.5: alunos com o mesmo perfil podem ter
    # destinos diferentes, e a probabilidade passa a significar algo
    evadiu      = rng.binomial(1, prob_evasao)

    df = pd.DataFrame({
        "id":               range(1, n + 1),
        "idade":            idade,
        "sexo":             sexo,
        "turno":            turno,
        "trabalha":         trabalha,
        "renda_familiar":   renda_familiar,
        "media_anterior":   media_anterior.round(2),
        "faltas_pct":       faltas_pct.round(2),
        "reprovacoes":      reprovacoes,
        "horas_estudo":     horas_estudo.round(2),
        "participacao":     participacao.round(2),
        "entrega_tarefas":  entrega_tarefas.round(2),
        "acesso_plataforma": acesso_plataforma,
        "nota_b1":          nota_b1.round(2),
        "nota_b2":          nota_b2.round(2),
        "nota_b3":          nota_b3.round(2),
        "nota_b4":          nota_b4.round(2),
        "evadiu":           evadiu,
    })

    return df


def salvar_raw(df: pd.DataFrame, pasta: str = "data/raw") -> Path:
    Path(pasta).mkdir(parents=True, exist_ok=True)
    caminho = Path(pasta) / "alunos_raw.csv"
    df.to_csv(caminho, index=False)
    print(f"Dados brutos salvos: {caminho} ({len(df)} registros)")
    return caminho


if __name__ == "__main__":
    df = gerar_dataset(n=5000)
    salvar_raw(df)
    print(f"\nDistribuição da variável alvo:")
    print(df["evadiu"].value_counts(normalize=True).round(3))

A regra do alvo é a peça mais delicada de um dataset sintético, e a primeira versão errava a mão: o escore de risco tinha média perto de −2,75, e o corte determinístico em 0,5 marcava como evadidos só 10 dos 5000 alunos (0,2%). Com isso, o modelo não tinha o que aprender, e o próprio teste de distribuição do projeto, que exige entre 20% e 80%, falharia. Duas mudanças resolvem. Um intercepto de 1,5 leva a taxa a 25,9%. E o sorteio com rng.binomial(1, prob_evasao), no lugar do corte, faz alunos de perfil igual terem destinos diferentes, como na vida real. É isso que dá sentido à calibração de probabilidades usada no treino: sem o sorteio, a "probabilidade" seria só um degrau.

Etapa 2: Limpeza e Validação

# src/evasao/data/limpeza.py
import pandas as pd
import numpy as np
from pathlib import Path
from dataclasses import dataclass
from typing import Optional


@dataclass
class RelatorioLimpeza:
    n_original:    int
    n_final:       int
    nulos_removidos: int
    duplicatas:    int
    outliers:      int
    colunas_criadas: list[str]


def validar_schema(df: pd.DataFrame) -> list[str]:
    """Valida que o DataFrame tem as colunas esperadas."""
    colunas_esperadas = {
        "id", "idade", "sexo", "turno", "trabalha",
        "renda_familiar", "media_anterior", "faltas_pct",
        "reprovacoes", "horas_estudo", "participacao",
        "entrega_tarefas", "acesso_plataforma",
        "nota_b1", "nota_b2", "nota_b3", "nota_b4", "evadiu"
    }
    faltando = colunas_esperadas - set(df.columns)
    return list(faltando)


def limpar(df: pd.DataFrame) -> tuple[pd.DataFrame, RelatorioLimpeza]:
    """Pipeline completo de limpeza de dados."""
    n_original = len(df)
    df = df.copy()

    # 1. Validar schema
    faltando = validar_schema(df)
    if faltando:
        raise ValueError(f"Colunas faltando: {faltando}")

    # 2. Remover duplicatas
    n_antes       = len(df)
    df            = df.drop_duplicates(subset=["id"])
    n_duplicatas  = n_antes - len(df)

    # 3. Tratar nulos
    n_antes       = len(df)
    numericas     = df.select_dtypes(include=[np.number]).columns
    # texto é "object" no pandas 2 e "str" no 3: tudo o que não é número
    categoricas   = df.select_dtypes(exclude=[np.number]).columns

    df[numericas]  = df[numericas].fillna(df[numericas].median())
    df[categoricas] = df[categoricas].fillna(df[categoricas].mode().iloc[0])

    df             = df.dropna()
    n_nulos        = n_antes - len(df)

    # 4. Corrigir tipos
    df["idade"]         = df["idade"].astype(int)
    df["trabalha"]      = df["trabalha"].astype(int)
    df["reprovacoes"]   = df["reprovacoes"].astype(int)
    df["evadiu"]        = df["evadiu"].astype(int)

    # 5. Remover valores impossíveis — não os extremos: o aluno com 45% de
    # faltas é raro, mas é justamente o que o modelo precisa ver
    limites = {
        "media_anterior": (0, 10), "faltas_pct": (0, 100),
        "horas_estudo":   (0, 24), "participacao": (0, 10),
    }
    n_outliers = 0
    for col, (minimo, maximo) in limites.items():
        antes = len(df)
        df  = df[df[col].between(minimo, maximo)]
        n_outliers += antes - len(df)

    # 6. Normalizar categorias
    df["sexo"]  = df["sexo"].str.upper().str.strip()
    df["turno"] = df["turno"].str.lower().str.strip()

    # 7. Validar ranges
    assert df["evadiu"].isin([0, 1]).all(), "Coluna alvo com valores inválidos"
    assert (df["faltas_pct"] >= 0).all() and (df["faltas_pct"] <= 100).all()
    assert (df["media_anterior"] >= 0).all() and (df["media_anterior"] <= 10).all()

    relatorio = RelatorioLimpeza(
        n_original=       n_original,
        n_final=          len(df),
        nulos_removidos=  n_nulos,
        duplicatas=       n_duplicatas,
        outliers=         n_outliers,
        colunas_criadas=  []
    )

    print(f"Limpeza concluída:")
    print(f"  Original:   {relatorio.n_original}")
    print(f"  Final:      {relatorio.n_final}")
    print(f"  Duplicatas: {relatorio.duplicatas}")
    print(f"  Nulos:      {relatorio.nulos_removidos}")
    print(f"  Outliers:   {relatorio.outliers}")

    return df, relatorio

Duas decisões dessa etapa mudaram em relação à primeira versão. A remoção de "outliers", apresentada como IQR, na verdade cortava 1% de cada ponta de quatro colunas. Medido: 246 alunos removidos (4,9%), nenhum com valor impossível, e com evasão de 36,2% entre eles, contra 25,9% no geral. O corte jogava fora justamente os casos mais arriscados, que o modelo mais precisa ver. Agora só saem valores fora dos limites possíveis. A outra mudança é a detecção de colunas de texto: select_dtypes(include=["object"]) ainda funciona no pandas 3, mas emite um Pandas4Warning avisando que as colunas str deixarão de ser incluídas; exclude=[np.number] funciona nas duas versões.

Etapa 3: Feature Engineering

# src/evasao/data/features.py
import pandas as pd
import numpy as np

ACESSOS_MES_MAX = 30   # teto de acessos/mês considerado no score


def criar_features(df: pd.DataFrame) -> pd.DataFrame:
    """Cria e transforma features para o modelo."""
    df = df.copy()

    # ── Features derivadas ────────────────────────────
    # Média dos bimestres
    df["media_bimestres"] = df[["nota_b1", "nota_b2", "nota_b3", "nota_b4"]].mean(axis=1)

    # Tendência de notas (b4 - b1)
    df["tendencia_notas"] = df["nota_b4"] - df["nota_b1"]

    # Variância das notas — inconsistência
    df["variancia_notas"] = df[["nota_b1", "nota_b2", "nota_b3", "nota_b4"]].var(axis=1)

    # Score de engajamento composto
    # Acessos normalizados por uma constante fixa, não pelo máximo do lote:
    # a API chama esta função com um aluno só, e o máximo seria ele mesmo
    acesso_norm = (df["acesso_plataforma"] / ACESSOS_MES_MAX).clip(upper=1)
    df["score_engajamento"] = (
        df["participacao"] * 0.3 +
        df["entrega_tarefas"] * 0.3 +
        acesso_norm * 10 * 0.2 +
        df["horas_estudo"] * 0.2
    )

    # Risco por faltas
    df["risco_faltas"] = pd.cut(
        df["faltas_pct"],
        bins=[-1, 10, 20, 30, 100],
        labels=[0, 1, 2, 3]
    ).astype(int)

    # Situação acadêmica anterior
    df["reprovado_anterior"] = (df["reprovacoes"] > 0).astype(int)
    df["multiplas_reprovacoes"] = (df["reprovacoes"] >= 2).astype(int)

    # Faixa etária
    df["faixa_etaria"] = pd.cut(
        df["idade"],
        bins=[14, 17, 20, 100],
        labels=["adolescente", "jovem", "adulto"]
    )

    # ── Encoding de categorias ────────────────────────
    # Ordinal para renda
    mapa_renda = {
        "ate_1sm": 0, "1_3sm": 1,
        "3_5sm": 2,   "acima_5sm": 3
    }
    df["renda_ordinal"] = df["renda_familiar"].map(mapa_renda)

    # Turno: colunas explícitas. get_dummies(drop_first=True) numa linha só
    # descartaria o único turno presente, e o aluno da noite viraria "manhã"
    df["turno_tarde"] = (df["turno"] == "tarde").astype(int)
    df["turno_noite"] = (df["turno"] == "noite").astype(int)

    # Binary para sexo
    df["sexo_bin"] = (df["sexo"] == "M").astype(int)

    # Faixa etária
    dummies_idade = pd.get_dummies(
        df["faixa_etaria"], prefix="faixa", drop_first=True
    )
    df = pd.concat([df, dummies_idade], axis=1)

    # ── Features finais para o modelo ─────────────────
    features = [
        # Acadêmicas
        "media_anterior", "media_bimestres", "tendencia_notas",
        "variancia_notas", "faltas_pct", "reprovacoes",
        "reprovado_anterior", "multiplas_reprovacoes",
        "risco_faltas",
        # Comportamentais
        "horas_estudo", "participacao", "entrega_tarefas",
        "acesso_plataforma", "score_engajamento",
        # Sociodemográficas
        "idade", "trabalha", "renda_ordinal", "sexo_bin",
        # Dummies
        "turno_noite", "turno_tarde",
        "faixa_jovem", "faixa_adulto"
    ]

    # Filtra apenas colunas existentes
    features_existentes = [f for f in features if f in df.columns]

    return df[features_existentes + ["evadiu"]]


def salvar_features(df: pd.DataFrame, pasta: str = "data/features") -> None:
    import pathlib
    pathlib.Path(pasta).mkdir(parents=True, exist_ok=True)
    df.to_parquet(f"{pasta}/features.parquet", index=False)
    print(f"Features salvas: {len(df)} registros, {len(df.columns)-1} features")

Essa função é chamada em dois lugares: no treino, com milhares de alunos, e na API, com um aluno só. Tudo o que ela calcula a partir do conjunto, e não da linha, muda de valor entre os dois. A primeira versão tinha dois casos assim. O acesso_plataforma / acesso_plataforma.max() numa linha só dá sempre 1 — e dá 0/0, NaN, quando o aluno tem zero acessos, o que derrubava a previsão com 500. E o get_dummies(drop_first=True) numa linha só descarta o único turno presente, e o aluno do noturno chegava ao modelo como se fosse da manhã. Medido em 300 alunos: 6 geravam NaN, e nos outros a probabilidade da API diferia da do treino em até 0,12. Com a constante fixa e as colunas de turno explícitas, a diferença caiu a zero. O nome disso é training-serving skew, e a defesa é uma regra simples: transformação de feature que depende do conjunto é aprendida no treino e salva junto do modelo, como faz o StandardScaler dentro do Pipeline.

Etapa 4: Treinamento com MLflow

# src/evasao/modelos/treino.py
import mlflow
import mlflow.sklearn
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split, StratifiedKFold, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.ensemble import (
    GradientBoostingClassifier,
    RandomForestClassifier
)
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
    f1_score, roc_auc_score, precision_score,
    recall_score, accuracy_score, classification_report
)
from sklearn.calibration import CalibratedClassifierCV
import joblib
from pathlib import Path


def carregar_features(caminho: str = "data/features/features.parquet") -> tuple:
    df = pd.read_parquet(caminho)
    X  = df.drop(columns=["evadiu"])
    y  = df["evadiu"]
    print(f"Dataset: {X.shape[0]} amostras, {X.shape[1]} features")
    print(f"Evasão: {y.mean():.1%} positivos")
    return X, y


def treinar_experimento(
    X_train, X_test, y_train, y_test,
    nome: str, modelo, params: dict
) -> dict:
    """Treina um modelo e registra no MLflow."""

    with mlflow.start_run(run_name=nome):
        mlflow.log_params(params)
        mlflow.log_param("n_treino", len(X_train))
        mlflow.log_param("n_teste",  len(X_test))
        mlflow.log_param("features", list(X_train.columns))

        # Pipeline com calibração de probabilidades
        pipe = Pipeline([
            ("scaler", StandardScaler()),
            ("clf",    CalibratedClassifierCV(modelo, cv=5, method="isotonic"))
        ])
        pipe.fit(X_train, y_train)

        # Predições
        y_pred = pipe.predict(X_test)
        y_prob = pipe.predict_proba(X_test)[:, 1]

        # Métricas
        metricas = {
            "accuracy":  round(accuracy_score(y_test, y_pred), 4),
            "f1":        round(f1_score(y_test, y_pred), 4),
            "precision": round(precision_score(y_test, y_pred), 4),
            "recall":    round(recall_score(y_test, y_pred), 4),
            "roc_auc":   round(roc_auc_score(y_test, y_prob), 4),
        }

        # CV
        cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
        cv_f1 = cross_val_score(pipe, X_train, y_train, cv=cv, scoring="f1")
        metricas["cv_f1_mean"] = round(cv_f1.mean(), 4)
        metricas["cv_f1_std"]  = round(cv_f1.std(),  4)

        mlflow.log_metrics(metricas)

        # Registro
        assinatura = mlflow.models.infer_signature(X_train, y_pred)
        mlflow.sklearn.log_model(
            pipe,
            name="modelo",
            skops_trusted_types=[
                "sklearn.calibration._CalibratedClassifier",
                "sklearn.tree._tree.Tree",
            ],
            registered_model_name=f"evasao_{nome.lower().replace(' ', '_')}",
            signature=assinatura,
            input_example=X_train.iloc[:3]
        )

        run_id = mlflow.active_run().info.run_id
        print(f"\n[{nome}] F1={metricas['f1']:.4f} "
              f"AUC={metricas['roc_auc']:.4f} "
              f"CV_F1={metricas['cv_f1_mean']:.4f}±{metricas['cv_f1_std']:.4f}")

        return {"run_id": run_id, "pipe": pipe, **metricas}


def executar_pipeline_treinamento():
    mlflow.set_tracking_uri("sqlite:///mlflow_evasao.db")
    mlflow.set_experiment("predicao_evasao_escolar")

    X, y = carregar_features()
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

    experimentos = [
        (
            "Logistic Regression",
            LogisticRegression(max_iter=1000, class_weight="balanced"),
            {"C": 1.0, "class_weight": "balanced"}
        ),
        (
            "Random Forest",
            RandomForestClassifier(
                n_estimators=200, max_depth=10,
                class_weight="balanced", random_state=42
            ),
            {"n_estimators": 200, "max_depth": 10}
        ),
        (
            "Gradient Boosting",
            GradientBoostingClassifier(
                n_estimators=200, learning_rate=0.05,
                max_depth=4, subsample=0.8, random_state=42
            ),
            {"n_estimators": 200, "learning_rate": 0.05}
        ),
    ]

    resultados = []
    for nome, modelo, params in experimentos:
        r = treinar_experimento(
            X_train, X_test, y_train, y_test,
            nome, modelo, params
        )
        resultados.append({"nome": nome, **r})

    # Seleciona melhor modelo pelo AUC
    melhor = max(resultados, key=lambda x: x["roc_auc"])
    print(f"\nMelhor modelo: {melhor['nome']} (AUC={melhor['roc_auc']:.4f})")

    # Salva modelo final
    Path("modelos").mkdir(exist_ok=True)
    joblib.dump(melhor["pipe"], "modelos/modelo_evasao.pkl")
    joblib.dump(list(X_train.columns), "modelos/features.pkl")

    return melhor


if __name__ == "__main__":
    executar_pipeline_treinamento()

O parquet exige o pyarrow instalado, que precisa estar no requirements.txt. No MLflow 3, o log_model grava em formato skops e recusa tipos que não foram declarados confiáveis: a calibração acrescenta _CalibratedClassifier, e as árvores, Tree — o skops.io.get_untrusted_types lista exatamente o que falta. Sobre as métricas: o melhor modelo, a regressão logística, tem AUC de 0,70, mas F1 de apenas 0,25, e o motivo é o limiar. As probabilidades calibradas refletem a taxa real, de 26%, e poucas passam de 0,5. Medido no conjunto de teste: com limiar 0,5, o modelo identifica só 16% dos alunos que evadem; com 0,3, identifica 56%, sinalizando 36% da turma. Num sistema de alerta, o limiar é uma decisão de negócio, pesando o custo de uma conversa desnecessária contra o de um aluno perdido, e não o 0,5 padrão do predict.

Etapa 5: API de Predição

# src/evasao/api/schemas.py
from pydantic import BaseModel, Field
from typing import Optional


class DadosAluno(BaseModel):
    # Acadêmicas
    media_anterior:   float = Field(..., ge=0, le=10)
    nota_b1:          float = Field(..., ge=0, le=10)
    nota_b2:          float = Field(..., ge=0, le=10)
    nota_b3:          float = Field(..., ge=0, le=10)
    nota_b4:          float = Field(..., ge=0, le=10)
    faltas_pct:       float = Field(..., ge=0, le=100)
    reprovacoes:      int   = Field(..., ge=0, le=10)

    # Comportamentais
    horas_estudo:     float = Field(..., ge=0, le=24)
    participacao:     float = Field(..., ge=0, le=10)
    entrega_tarefas:  float = Field(..., ge=0, le=10)
    acesso_plataforma: int  = Field(..., ge=0)

    # Sociodemográficas
    idade:            int   = Field(..., ge=14, le=60)
    trabalha:         int   = Field(..., ge=0, le=1)
    renda_familiar:   str   = Field(..., pattern="^(ate_1sm|1_3sm|3_5sm|acima_5sm)$")
    sexo:             str   = Field(..., pattern="^[MF]$")
    turno:            str   = Field(..., pattern="^(manha|tarde|noite)$")

    model_config = {
        "json_schema_extra": {
            "example": {
                "media_anterior": 6.5, "nota_b1": 6.0,
                "nota_b2": 6.5, "nota_b3": 7.0, "nota_b4": 7.5,
                "faltas_pct": 15.0, "reprovacoes": 0,
                "horas_estudo": 3.0, "participacao": 7.0,
                "entrega_tarefas": 8.0, "acesso_plataforma": 12,
                "idade": 17, "trabalha": 0,
                "renda_familiar": "1_3sm", "sexo": "F", "turno": "manha"
            }
        }
    }


class PrevisaoResposta(BaseModel):
    risco_evasao:        float
    classificacao:       str
    fatores_risco:       list[str]
    recomendacoes:       list[str]
    confianca:           str


class SaudeResposta(BaseModel):
    status:      str
    modelo:      str
    versao:      str
    n_features:  int

# src/evasao/api/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import joblib
import numpy as np
import pandas as pd
from pathlib import Path
from .schemas import DadosAluno, PrevisaoResposta, SaudeResposta
from evasao.data.features import criar_features


# Carregamento do modelo
modelo     = None
features   = None
VERSAO_MODELO = "1.0.0"


@asynccontextmanager
async def lifespan(app: FastAPI):
    global modelo, features
    try:
        modelo   = joblib.load("modelos/modelo_evasao.pkl")
        features = joblib.load("modelos/features.pkl")
        print(f"Modelo carregado: {len(features)} features")
    except FileNotFoundError:
        print("Modelo não encontrado — execute o pipeline de treinamento.")
    yield


app = FastAPI(
    lifespan=lifespan,
    title="API de Predição de Evasão Escolar",
    description="Modelo ML para identificar alunos em risco de evasão",
    version="1.0.0"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"]
)

def preparar_entrada(dados: DadosAluno) -> pd.DataFrame:
    """Converte dados de entrada para DataFrame com features."""
    raw = pd.DataFrame([{
        "id":                 0,
        "idade":              dados.idade,
        "sexo":               dados.sexo,
        "turno":              dados.turno,
        "trabalha":           dados.trabalha,
        "renda_familiar":     dados.renda_familiar,
        "media_anterior":     dados.media_anterior,
        "faltas_pct":         dados.faltas_pct,
        "reprovacoes":        dados.reprovacoes,
        "horas_estudo":       dados.horas_estudo,
        "participacao":       dados.participacao,
        "entrega_tarefas":    dados.entrega_tarefas,
        "acesso_plataforma":  dados.acesso_plataforma,
        "nota_b1":            dados.nota_b1,
        "nota_b2":            dados.nota_b2,
        "nota_b3":            dados.nota_b3,
        "nota_b4":            dados.nota_b4,
        "evadiu":             0   # placeholder
    }])

    df_feat = criar_features(raw)

    # Garante as mesmas colunas do treino na mesma ordem
    for col in features:
        if col not in df_feat.columns:
            df_feat[col] = 0
    df_feat = df_feat.drop(columns=["evadiu"], errors="ignore")
    return df_feat[features]


def identificar_fatores_risco(dados: DadosAluno, prob: float) -> list[str]:
    """Identifica principais fatores de risco baseado nos dados."""
    fatores = []

    if dados.faltas_pct > 25:
        fatores.append(f"Alto índice de faltas ({dados.faltas_pct:.1f}%)")
    if dados.media_anterior < 5.0:
        fatores.append(f"Média anterior baixa ({dados.media_anterior:.1f})")
    if dados.reprovacoes >= 2:
        fatores.append(f"Múltiplas reprovações ({dados.reprovacoes})")
    if dados.horas_estudo < 1.5:
        fatores.append("Poucas horas de estudo por dia")
    if dados.participacao < 4.0:
        fatores.append("Baixa participação nas aulas")
    if dados.entrega_tarefas < 5.0:
        fatores.append("Baixa entrega de tarefas")
    if dados.trabalha and dados.turno == "noite":
        fatores.append("Trabalha e estuda no período noturno")
    if dados.acesso_plataforma < 5:
        fatores.append("Poucos acessos à plataforma de ensino")

    return fatores[:5]   # top 5 fatores


def gerar_recomendacoes(fatores: list[str], prob: float) -> list[str]:
    """Gera recomendações personalizadas."""
    recomendacoes = []

    if prob >= 0.7:
        recomendacoes.append("Agendar reunião urgente com orientação pedagógica")
        recomendacoes.append("Contatar responsáveis para alinhamento")

    for fator in fatores:
        if "faltas" in fator.lower():
            recomendacoes.append("Investigar causas das faltas e oferecer suporte")
        if "média" in fator.lower() or "reprovac" in fator.lower():
            recomendacoes.append("Encaminhar para reforço escolar")
        if "estudo" in fator.lower() or "participação" in fator.lower():
            recomendacoes.append("Promover engajamento com tutoria individualizada")
        if "plataforma" in fator.lower():
            recomendacoes.append("Verificar acesso à internet e dispositivos")

    return list(dict.fromkeys(recomendacoes))[:4]   # deduplica, top 4


@app.get("/health", response_model=SaudeResposta)
async def health():
    return SaudeResposta(
        status=     "healthy" if modelo else "degraded",
        # o modelo que foi salvo, e não um nome fixo no código
        modelo=     type(modelo[-1].estimator).__name__ if modelo else "nenhum",
        versao=     VERSAO_MODELO,
        n_features= len(features) if features else 0
    )


@app.post("/prever", response_model=PrevisaoResposta)
async def prever_evasao(dados: DadosAluno):
    if modelo is None:
        raise HTTPException(status_code=503, detail="Modelo não disponível.")

    try:
        X    = preparar_entrada(dados)
        prob = float(modelo.predict_proba(X)[0, 1])

        if prob >= 0.70:
            classificacao = "ALTO RISCO"
        elif prob >= 0.40:
            classificacao = "RISCO MODERADO"
        else:
            classificacao = "BAIXO RISCO"

        if prob >= 0.85 or prob <= 0.15:
            confianca = "Alta"
        elif prob >= 0.65 or prob <= 0.35:
            confianca = "Média"
        else:
            confianca = "Baixa"

        fatores        = identificar_fatores_risco(dados, prob)
        recomendacoes  = gerar_recomendacoes(fatores, prob)

        return PrevisaoResposta(
            risco_evasao=    round(prob, 4),
            classificacao=   classificacao,
            fatores_risco=   fatores,
            recomendacoes=   recomendacoes,
            confianca=       confianca
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/prever/lote")
async def prever_lote(lista: list[DadosAluno]):
    if modelo is None:
        raise HTTPException(status_code=503, detail="Modelo não disponível.")
    if not lista:
        raise HTTPException(status_code=400, detail="Lote vazio.")
    if len(lista) > 500:
        raise HTTPException(status_code=400, detail="Máximo 500 alunos por lote.")

    resultados = []
    for dados in lista:
        X    = preparar_entrada(dados)
        prob = float(modelo.predict_proba(X)[0, 1])
        resultados.append({
            "risco_evasao":  round(prob, 4),
            "classificacao": "ALTO" if prob >= 0.7 else ("MODERADO" if prob >= 0.4 else "BAIXO")
        })

    total     = len(resultados)
    alto_risco = sum(1 for r in resultados if r["classificacao"] == "ALTO")

    return {
        "total":      total,
        "alto_risco": alto_risco,
        "pct_risco":  round(alto_risco / total * 100, 1),
        "previsoes":  resultados
    }

Três ajustes na API. O modelo é carregado no lifespan, que substituiu o @app.on_event("startup"), obsoleto no FastAPI. O /health informa o tipo do modelo realmente salvo; a primeira versão dizia "Gradient Boosting Calibrated" fixo no código, embora o melhor modelo fosse a regressão logística. E o lote vazio é recusado com 400, em vez de estourar numa divisão por zero no cálculo do percentual.

Makefile: Automatizando o Pipeline

# Makefile
# As linhas de comando começam com TAB, não com espaços
export PYTHONPATH := src

.PHONY: dados limpar features treinar pipeline api testes docker mlflow-ui all

dados:
	python -m evasao.data.coleta

limpar:
	mkdir -p data/processed
	python -c "from evasao.data.limpeza import limpar; \
	           import pandas as pd; \
	           df, _ = limpar(pd.read_csv('data/raw/alunos_raw.csv')); \
	           df.to_csv('data/processed/alunos_limpos.csv', index=False)"

features:
	python -c "from evasao.data.features import criar_features, salvar_features; \
	           import pandas as pd; \
	           df = criar_features(pd.read_csv('data/processed/alunos_limpos.csv')); \
	           salvar_features(df)"

treinar:
	python -m evasao.modelos.treino

pipeline: dados limpar features treinar
	@echo "Pipeline completo executado!"

api:
	uvicorn evasao.api.main:app --reload --port 8001

testes:
	pytest tests/ -v --cov=src --cov-report=term-missing

docker:
	docker compose up --build -d

mlflow-ui:
	mlflow ui --backend-store-uri sqlite:///mlflow_evasao.db --port 5001

all: pipeline testes

Um Makefile exige TAB no início de cada linha de comando; com espaços, o make para antes de começar, com Makefile:5: *** missing separator. Copiado de uma página web, o TAB costuma virar espaços, e vale conferir no editor. O export PYTHONPATH := src faz o python -m evasao... encontrar o pacote, e o mkdir -p cria a pasta de saída da limpeza, que não existia. Com essas mudanças, make pipeline roda do começo ao fim.

Testes do Pipeline

# tests/test_pipeline.py
import pytest
import pandas as pd
import numpy as np
from pathlib import Path
from evasao.data.coleta import gerar_dataset
from evasao.data.limpeza import limpar, validar_schema
from evasao.data.features import criar_features
from fastapi.testclient import TestClient
from evasao.api.main import app


@pytest.fixture(scope="session")
def dataset_raw():
    return gerar_dataset(n=500, seed=42)


@pytest.fixture(scope="session")
def dataset_limpo(dataset_raw):
    df, _ = limpar(dataset_raw)
    return df


@pytest.fixture(scope="session")
def dataset_features(dataset_limpo):
    return criar_features(dataset_limpo)


class TestColeta:
    def test_gera_colunas_corretas(self, dataset_raw):
        assert "evadiu" in dataset_raw.columns
        assert len(dataset_raw) == 500

    def test_distribuicao_balanceada(self, dataset_raw):
        pct = dataset_raw["evadiu"].mean()
        assert 0.2 <= pct <= 0.8, f"Distribuição desequilibrada: {pct:.1%}"

    def test_sem_ids_duplicados(self, dataset_raw):
        assert dataset_raw["id"].nunique() == len(dataset_raw)


class TestLimpeza:
    def test_sem_nulos_apos_limpeza(self, dataset_limpo):
        assert dataset_limpo.isnull().sum().sum() == 0

    def test_schema_valido(self, dataset_limpo):
        faltando = validar_schema(dataset_limpo)
        assert faltando == [], f"Colunas faltando: {faltando}"

    def test_target_binario(self, dataset_limpo):
        assert set(dataset_limpo["evadiu"].unique()).issubset({0, 1})

    def test_faltas_validas(self, dataset_limpo):
        assert (dataset_limpo["faltas_pct"] >= 0).all()
        assert (dataset_limpo["faltas_pct"] <= 100).all()


class TestFeatures:
    def test_media_bimestres_criada(self, dataset_features):
        assert "media_bimestres" in dataset_features.columns

    def test_sem_nulos_nas_features(self, dataset_features):
        assert dataset_features.isnull().sum().sum() == 0

    def test_score_engajamento_valido(self, dataset_features):
        assert (dataset_features["score_engajamento"] >= 0).all()

    def test_n_features_esperado(self, dataset_features):
        n_features = len(dataset_features.columns) - 1   # exclui target
        assert n_features >= 15, f"Poucas features: {n_features}"


class TestAPI:
    @pytest.fixture
    def client(self):
        if not Path("modelos/modelo_evasao.pkl").exists():
            pytest.skip("sem modelo treinado: rode `make pipeline` antes")
        with TestClient(app) as c:     # o "with" roda o lifespan
            yield c

    def test_health(self, client):
        resp = client.get("/health")
        assert resp.status_code == 200

    def test_prever_aluno_risco_alto(self, client):
        resp = client.post("/prever", json={
            "media_anterior": 3.5, "nota_b1": 3.0,
            "nota_b2": 3.5, "nota_b3": 4.0, "nota_b4": 3.0,
            "faltas_pct": 40.0, "reprovacoes": 3,
            "horas_estudo": 0.5, "participacao": 2.0,
            "entrega_tarefas": 3.0, "acesso_plataforma": 2,
            "idade": 19, "trabalha": 1,
            "renda_familiar": "ate_1sm", "sexo": "M", "turno": "noite"
        })
        assert resp.status_code == 200
        dados = resp.json()
        assert dados["risco_evasao"] >= 0.7
        assert dados["classificacao"] == "ALTO RISCO"

    def test_validacao_campos(self, client):
        resp = client.post("/prever", json={"media_anterior": 15.0})
        assert resp.status_code == 422

O teste da API mudou de propósito. A primeira versão fazia if resp.status_code == 200: antes de conferir o resultado, e como o TestClient era criado sem with, o modelo nunca era carregado: a API respondia 503, o if pulava tudo, e o teste passava sem verificar nada — medido. Um teste que passa quando o sistema está fora do ar não testa o sistema. Agora ele exige o 200 e um risco de pelo menos 0,7 para o aluno de perfil extremo, e é pulado, de forma visível, quando não há modelo treinado. Com tudo corrigido, os 14 testes passam.

O pipeline percorre o caminho inteiro de um projeto de dados: gera o conjunto, limpa e valida, cria as features, treina e compara três modelos com o MLflow, salva o melhor e o serve numa API que devolve o risco, os fatores e as recomendações para cada aluno. O Makefile encadeia as etapas, e os testes conferem cada uma. É a mesma arquitetura em camadas de Projeto 1: API REST Completa com FastAPI, PostgreSQL e Docker, aplicada a um modelo em vez de um cadastro.

Os defeitos que este projeto tinha são os típicos de ciência de dados, e quase nenhum levanta exceção. Um alvo com 0,2% de positivos produz um modelo que parece funcionar. Uma limpeza que corta os extremos descarta os alunos mais arriscados. Uma função de features que depende do lote dá uma resposta no treino e outra na API. E um teste com if passa com o sistema fora do ar. Em cada caso, o que revelou o problema foi olhar os números: a taxa do alvo, quem saiu na limpeza, a mesma previsão pelos dois caminhos, e o que o teste de fato conferia.

Fontes e leituras recomendadas

• scikit-learn Pipelines — https://scikit-learn.org/stable/modules/pipeline.html
• Calibrated Classifiers — https://scikit-learn.org/stable/modules/calibration.html
• MLflow Projects — https://mlflow.org/docs/latest/projects.html
• pandas Parquet — https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.to_parquet.html
• GÉRON, Aurélien. Hands-On Machine Learning. 3. ed. O'Reilly Media, 2022. Cap. 2 — pipeline completo de ML do início ao fim.
• HUYEN, Chip. Designing Machine Learning Systems. O'Reilly Media, 2022. Cap. 4–6 — feature engineering e pipelines de dados em produção.

Exercícios

Exercício 1

A primeira versão do gerador de dados definia evadiu = (prob_evasao > 0.5), e o conjunto de 5000 alunos saiu com 10 evasões. Um modelo que responde "não evade" para todo mundo acerta 99,8% das vezes. Por que a acurácia é inútil aqui, e o que estava errado na regra?

Exercício 2

Uma aluna com zero acessos à plataforma é enviada à API, que responde 500 com Input X contains NaN. No treino, os alunos com zero acessos foram processados sem problema. O mesmo acontece, sem erro, com os alunos do noturno: a API os trata como se fossem da manhã. O que as duas situações têm em comum?

Exercício 3

O relatório de limpeza mostra "Outliers: 246", e o comentário diz "método IQR". Não havia nenhum valor impossível nos dados. Entre os 246 removidos, a evasão era de 36,2%, contra 25,9% no conjunto todo. Que efeito essa limpeza tem no modelo?

Exercício 4

O melhor modelo tem AUC de 0,70, mas o F1 no teste é 0,25, e a escola reclama que o sistema quase nunca sinaliza ninguém. Medido: com o limiar padrão de 0,5, o modelo identifica 16% dos alunos que evadem; com 0,3, identifica 56%. Qual limiar usar?

Exercício 5

O teste test_prever_aluno_risco_alto da primeira versão passou em todas as execuções do CI por meses. Um dia descobriu-se que o arquivo do modelo nunca tinha sido gerado no ambiente de CI. O teste começava com if resp.status_code == 200:. O que ele estava testando?