Artigo 49 — Projeto 2: Pipeline de Data Science do Zero ao Deploy
Prof. Ricardo Matos Módulo 9 · Projetos Reais e Carreira · Artigo 49 de 52
Introdução
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. Aplicamos aqui os conceitos dos Módulos 4, 7 e 8: pandas, scikit-learn, FastAPI, MLflow, Docker e testes. 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 = (
- 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))
evadiu = (prob_evasao > 0.5).astype(int)
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))
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
categoricas = df.select_dtypes(include=["object"]).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 outliers (método IQR)
n_antes = len(df)
cols_num = ["media_anterior", "faltas_pct", "horas_estudo", "participacao"]
n_outliers = 0
for col in cols_num:
q1 = df[col].quantile(0.01)
q99 = df[col].quantile(0.99)
antes = len(df)
df = df[(df[col] >= q1) & (df[col] <= q99)]
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
Etapa 3: Feature Engineering
# src/evasao/data/features.py
import pandas as pd
import numpy as np
from sklearn.preprocessing import LabelEncoder
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
df["score_engajamento"] = (
df["participacao"] * 0.3 +
df["entrega_tarefas"] * 0.3 +
(df["acesso_plataforma"] / df["acesso_plataforma"].max()) * 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)
# One-hot para turno
dummies_turno = pd.get_dummies(
df["turno"], prefix="turno", drop_first=True
)
df = pd.concat([df, dummies_turno], axis=1)
# 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")
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,
artifact_path="modelo",
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()
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 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
app = FastAPI(
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=["*"]
)
# Carregamento do modelo
modelo = None
features = None
VERSAO_MODELO = "1.0.0"
@app.on_event("startup")
async def carregar_modelo():
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.")
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",
modelo= "Gradient Boosting Calibrated",
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 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
}
Makefile: Automatizando o Pipeline
# Makefile
.PHONY: dados limpar features treinar api testes docker
dados:
python -m evasao.data.coleta
limpar:
python -c "from evasao.data.limpeza import *; \
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 *; \
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
Testes do Pipeline
# tests/test_pipeline.py
import pytest
import pandas as pd
import numpy as np
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):
return TestClient(app)
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"
})
if resp.status_code == 200:
dados = resp.json()
assert "risco_evasao" in dados
assert 0 <= dados["risco_evasao"] <= 1
def test_validacao_campos(self, client):
resp = client.post("/prever", json={"media_anterior": 15.0})
assert resp.status_code == 422
Resumo do Projeto
Este pipeline integra:
| Etapa | Técnicas Aplicadas |
|---|---|
| Coleta | NumPy, geração sintética realista |
| Limpeza | pandas, validação de schema, IQR |
| Features | Feature engineering, encoding, composição |
| Treinamento | scikit-learn, MLflow, calibração de probabilidades |
| API | FastAPI, Pydantic, lógica de negócio interpretável |
| Testes | pytest, fixtures encadeadas, testes por etapa |
| Deploy | Docker, Makefile, automatização do pipeline |
Referências e Leituras Complementares
- 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.
Prof. Ricardo Matos — Dominando o Python em 1 Ano · Artigo 49 de 52 Próximo: Artigo 50 — Projeto 3: Bot de Automação com Scraping e Notificações
you asked
Sim