Machine Learning: Monitoramento em Produção e Drift de Modelo
Última atualização: 2026-08-26
Implantar um modelo é como um avião decolando — o verdadeiro desafio é monitorar o voo e lidar com o que vier pelo caminho.
1. O que você vai aprender
- Sistemas de monitoramento de modelos: monitoramento de distribuição de previsões, monitoramento de latência, monitoramento de métricas de negócio
- Drift de dados: PSI / teste KS / divergência Jensen-Shannon, e detecção de mudanças em distribuições de features
- Drift de conceito: mudanças na relação entrada-saída, com detecção reativa e proativa
- Pipeline automatizado de re-treinamento: alerta de drift → retropreenchimento de dados → re-treinamento → validação A/B → rollout canário
- A história de guerra contra o drift do Bob: como o festival de compras Double-11 mudou drasticamente o comportamento dos usuários, e como o SalesPredict se adaptou rapidamente
2. Uma história real de um engenheiro de operações
(1) O problema: depois do Double-11, toda previsão foi por água abaixo
O modelo SalesPredict do Bob estava rodando sem problemas, com MAPE estável em 8%. Mas depois da promoção do Double-11, o erro de previsão disparou para 35% — o comportamento dos usuários tinha mudado completamente (compras por impulso, mudanças nas preferências de categoria, ciclos de compra comprimidos), enquanto o modelo ainda fazia previsões baseadas em padrões pré-promoção. A degradação do modelo após a implantação é silenciosa — quando você percebe, o dano já está feito.
(2) A solução de monitoramento de drift
Monitore continuamente a distribuição dos dados de entrada (PSI) e a acurácia das previsões. No momento em que o drift for detectado, dispare o re-treinamento automatizado.
PYTHON
def check_drift(reference_data, current_data, threshold=0.2):
psi = calculate_psi(reference_data, current_data)
if psi > threshold:
alert("Drift de dados detectado! PSI={:.3f}".format(psi))
trigger_retraining()
(3) O resultado: drift detectado em 3 dias, corrigido em uma semana
Depois que Bob implantou o monitoramento de drift, a mudança de comportamento pós-Double-11 foi detectada em 3 dias (PSI=0,45). Em uma semana, o re-treinamento + validação A/B + rollout canário estavam completos, recuperando cerca de 300 mil USD em perdas de previsão.
3. O sistema de monitoramento de modelos
(1) Uma arquitetura de monitoramento em três camadas
graph TB
L1[Camada 1: Infraestrutura<br/>Latência / Vazão / Uptime] --> L2[Camada 2: Métricas de ML<br/>Distribuição de Previsões / Estatísticas de Features]
L2 --> L3[Camada 3: Métricas de Negócio<br/>MAPE de Receita / Taxa de Conversão / Taxa de Churn]
L3 --> ALERT{Anomalia Detectada?}
ALERT -->|Sim| ACTION[Alerta → Investigar → Re-treinar]
ALERT -->|Não| CONTINUE[Continuar Monitorando]
▶ Exemplo: Dados básicos de dashboard de monitoramento
PYTHON
import numpy as np
from datetime import datetime, timedelta
class ModelMonitor:
def __init__(self):
self.metrics_history = []
def log_prediction(self, prediction, features, latency_ms):
self.metrics_history.append({
"timestamp": datetime.now(),
"prediction": prediction,
"features_mean": np.mean(features),
"latency_ms": latency_ms,
})
def check_health(self, last_n=1000):
if len(self.metrics_history) < last_n:
return "Dados insuficientes"
recent = self.metrics_history[-last_n:]
predictions = [m["prediction"] for m in recent]
latencies = [m["latency_ms"] for m in recent]
report = {
"media_previsoes": np.mean(predictions),
"desvio_previsoes": np.std(predictions),
"p95_latencia_ms": np.percentile(latencies, 95),
"contagem_amostras": len(recent),
}
return report
monitor = ModelMonitor()
for _ in range(100):
monitor.log_prediction(prediction=np.random.normal(200, 50),
features=np.random.randn(5),
latency_ms=np.random.uniform(10, 30))
print(monitor.check_health())
Saída:
TEXT
📖 Somente leitura
# Funções definidas com sucesso
| Camada de Monitoramento | Métrica | Limite de Alerta | Ferramenta |
|---|---|---|---|
| Infraestrutura | Latência da API p95 | > 100ms | Prometheus |
| Infraestrutura | Disponibilidade do serviço | < 99,9% | Grafana |
| Métricas de ML | Mudança na distribuição de previsões | PSI > 0,2 | Personalizado |
| Métricas de ML | Taxa de features ausentes | > 5% | Personalizado |
| Métricas de Negócio | MAPE de previsão | > 15% | Relatórios de negócio |
| Métricas de Negócio | Desvio de receita | > 10% | Relatórios de negócio |
4. Detecção de drift de dados
(1) PSI (Population Stability Index)
▶ Exemplo: Calculando o PSI
PYTHON
import numpy as np
def calculate_psi(reference, current, n_bins=10):
"""Calcula o Population Stability Index.
PSI < 0,1: Sem mudança significativa
PSI 0,1-0,2: Mudança moderada, investigar
PSI > 0,2: Mudança significativa, ação necessária
"""
breakpoints = np.percentile(reference, np.linspace(0, 100, n_bins + 1))
breakpoints[0] = -np.inf
breakpoints[-1] = np.inf
ref_counts = np.histogram(reference, bins=breakpoints)[0]
cur_counts = np.histogram(current, bins=breakpoints)[0]
ref_pct = ref_counts / len(reference)
cur_pct = cur_counts / len(current)
# Evitar log(0)
ref_pct = np.clip(ref_pct, 1e-6, None)
cur_pct = np.clip(cur_pct, 1e-6, None)
psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
return psi
# Simular: dados normais vs. com drift
rng = np.random.default_rng(42)
reference = rng.normal(100, 20, 10000)
# Sem drift
current_no_drift = rng.normal(100, 20, 5000)
psi_no_drift = calculate_psi(reference, current_no_drift)
# Drift moderado (média deslocada em 10)
current_moderate = rng.normal(110, 20, 5000)
psi_moderate = calculate_psi(reference, current_moderate)
# Drift severo (média deslocada em 25, efeito Double-11)
current_severe = rng.normal(125, 30, 5000)
psi_severe = calculate_psi(reference, current_severe)
print(f"Sem drift: PSI = {psi_no_drift:.3f} (OK)")
print(f"Moderado: PSI = {psi_moderate:.3f} (Investigar)")
print(f"Severo: PSI = {psi_severe:.3f} (Ação Necessária!)")
Saída:
TEXT
📖 Somente leitura
# Funções definidas com sucesso
(2) Teste KS e Divergência JS
▶ Exemplo: Comparando múltiplas métricas de drift
PYTHON
from scipy import stats
import numpy as np
def calculate_js_divergence(reference, current, n_bins=50):
"""Divergência de Jensen-Shannon (simétrica, limitada em [0, 1])."""
bins = np.linspace(min(reference.min(), current.min()),
max(reference.max(), current.max()), n_bins + 1)
p = np.histogram(reference, bins=bins, density=True)[0]
q = np.histogram(current, bins=bins, density=True)[0]
m = (p + q) / 2
js = 0.5 * stats.entropy(p, m) + 0.5 * stats.entropy(q, m)
return js
rng = np.random.default_rng(42)
ref = rng.normal(100, 20, 10000)
cur_drifted = rng.normal(115, 25, 5000)
# Teste KS
ks_stat, ks_pvalue = stats.ks_2samp(ref, cur_drifted)
# PSI
psi = calculate_psi(ref, cur_drifted)
# Divergência JS
js = calculate_js_divergence(ref, cur_drifted)
print(f"Estatística KS: {ks_stat:.4f}, valor-p: {ks_pvalue:.6f}")
print(f"PSI: {psi:.4f}")
print(f"Divergência JS: {js:.4f}")
Saída:
TEXT
📖 Somente leitura
# Funções definidas com sucesso
| Métrica | Intervalo | Limite Sem-Drift | Limite de Drift | Características |
|---|---|---|---|---|
| PSI | [0, ∞) | < 0,1 | > 0,2 | Padrão da indústria |
| Estatística KS | [0, 1] | < 0,05 | > 0,1 | Teste não paramétrico |
| Divergência JS | [0, 1] | < 0,05 | > 0,1 | Simétrica, interpretável |
| Divergência KL | [0, ∞) | < 0,1 | > 0,2 | Assimétrica |
5. Drift de conceito e re-treinamento automatizado
(1) Drift de dados vs. Drift de conceito
| Tipo | Definição | Método de Detecção | Exemplo |
|---|---|---|---|
| Drift de dados | P(X) muda — a distribuição de entrada muda | PSI/KS nas features | Gasto do usuário dobra durante o Double-11 |
| Drift de conceito | P(Y | X) muda — a relação entrada-saída muda | Aumento do erro de previsão |
| Drift de rótulo | P(Y) muda — a distribuição de saída muda | Monitoramento da distribuição de previsões | Taxa de churn sobe de 8% para 15% |
▶ Exemplo: Detecção de drift do Double-11 do Bob
PYTHON
import numpy as np
rng = np.random.default_rng(42)
# Período normal: relação gasto com anúncios → receita
n_normal = 10000
ad_spend_normal = rng.uniform(10, 100, n_normal)
revenue_normal = 50 + 0.8 * ad_spend_normal + rng.normal(0, 10, n_normal)
# Período Double-11: comportamento completamente diferente
n_promo = 5000
ad_spend_promo = rng.uniform(50, 200, n_promo) # Gasto com anúncios mais alto
revenue_promo = 200 + 1.5 * ad_spend_promo + rng.normal(0, 30, n_promo) # Coeficientes diferentes!
# Detectar drift na distribuição de gasto com anúncios
psi = calculate_psi(ad_spend_normal, ad_spend_promo)
ks_stat, ks_p = stats.ks_2samp(ad_spend_normal, ad_spend_promo)
print("=== Detecção de Drift de Dados ===")
print(f"PSI do gasto com anúncios: {psi:.3f} {'⚠️ DRIFT' if psi > 0.2 else 'OK'}")
print(f"KS do gasto com anúncios: {ks_stat:.3f} (p={ks_p:.6f}) {'⚠️ DRIFT' if ks_p < 0.05 else 'OK'}")
# Detectar drift de conceito: mesmo X, relação Y|X diferente
# Usar um modelo treinado em dados normais para prever dados promo
from sklearn.linear_model import LinearRegression
model_normal = LinearRegression()
model_normal.fit(ad_spend_normal.reshape(-1, 1), revenue_normal)
pred_promo = model_normal.predict(ad_spend_promo.reshape(-1, 1))
mape_normal_on_normal = np.mean(np.abs((revenue_normal - model_normal.predict(ad_spend_normal.reshape(-1, 1))) / revenue_normal)) * 100
mape_normal_on_promo = np.mean(np.abs((revenue_promo - pred_promo) / revenue_promo)) * 100
print(f"\n=== Detecção de Drift de Conceito ===")
print(f"MAPE (normal→normal): {mape_normal_on_normal:.1f}%")
print(f"MAPE (normal→promo): {mape_normal_on_promo:.1f}%")
print(f"Degradação de desempenho: {mape_normal_on_promo - mape_normal_on_normal:.1f}% ⚠️ DRIFT DE CONCEITO")
Saída:
TEXT
📖 Somente leitura
=== Detecção de Drift de Dados ===
(2) O pipeline automatizado de re-treinamento
▶ Exemplo: Framework de detecção de drift + re-treinamento automatizado
PYTHON
import numpy as np
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_percentage_error
class AutoRetrainPipeline:
def __init__(self, drift_threshold_psi=0.2, mape_threshold=0.15):
self.drift_threshold_psi = drift_threshold_psi
self.mape_threshold = mape_threshold
self.model = None
self.reference_data = None
def train(self, X, y):
self.model = LinearRegression()
self.model.fit(X, y)
self.reference_data = X.copy()
def check_drift(self, X_current):
drifts = []
for i in range(X_current.shape[1]):
psi = calculate_psi(self.reference_data[:, i], X_current[:, i])
drifts.append({"feature": f"feat_{i}", "psi": psi, "drift": psi > self.drift_threshold_psi})
return drifts
def check_performance(self, X, y):
y_pred = self.model.predict(X)
mape = mean_absolute_percentage_error(y, y_pred)
return mape < self.mape_threshold, mape
def auto_retrain_if_needed(self, X_current, y_current):
# Passo 1: Verificar drift
drifts = self.check_drift(X_current)
any_drift = any(d["drift"] for d in drifts)
# Passo 2: Verificar desempenho
performance_ok, mape = self.check_performance(X_current, y_current)
# Passo 3: Decisão
if not any_drift and performance_ok:
return "OK - Nenhum re-treinamento necessário"
if any_drift and not performance_ok:
# Re-treinar com dados recentes
self.model.fit(X_current, y_current)
self.reference_data = X_current.copy()
new_mape = mean_absolute_percentage_error(y_current, self.model.predict(X_current))
return f"RE-TREINADO - Drift de PSI detectado, MAPE {mape:.1%} → {new_mape:.1%}"
if any_drift:
return f"ALERTA DE DRIFT - Drift de PSI mas desempenho ainda OK (MAPE {mape:.1%})"
return f"ALERTA DE DESEMPENHO - Sem drift mas MAPE degradou para {mape:.1%}"
# Teste
pipeline = AutoRetrainPipeline()
rng = np.random.default_rng(42)
X_train = rng.uniform(10, 100, (5000, 3))
y_train = 50 + 0.8 * X_train[:, 0] + rng.normal(0, 5, 5000)
pipeline.train(X_train, y_train)
# Simular dados do Double-11
X_promo = rng.uniform(50, 200, (2000, 3))
y_promo = 200 + 1.5 * X_promo[:, 0] + rng.normal(0, 10, 2000)
result = pipeline.auto_retrain_if_needed(X_promo, y_promo)
print(result)
Saída:
TEXT
📖 Somente leitura
# Funções definidas com sucesso
❓ Perguntas Frequentes
P: PSI > 0,2 significa sempre que precisamos re-treinar? R: Não necessariamente. O PSI só detecta mudança de distribuição, não degradação de desempenho. A distribuição pode mudar enquanto o modelo permanece acurado (por exemplo, mais usuários, mas os mesmos padrões). É melhor disparar o re-treinamento com uma condição dupla: PSI + MAPE.
P: Com que frequência devemos verificar drift? R: Depende do seu ritmo de negócio — verifique de hora em hora para serviços em tempo real, diariamente para jobs em lote, e semanalmente para previsões mensais. Sempre verifique imediatamente após eventos importantes (como o Double-11).
P: Que dados devemos usar para re-treinamento? R: Recomenda-se uma janela deslizante (os 3-6 meses mais recentes de dados), ou treinamento ponderado (dando peso maior aos dados recentes). Evite usar todos os dados históricos — dados mais antigos podem refletir um conceito desatualizado.
P: Qual é mais perigoso, drift de conceito ou drift de dados? R: Drift de conceito é mais perigoso — com drift de dados apenas a distribuição de entrada muda, então o modelo ainda pode estar acurado; com drift de conceito a relação entrada-saída muda, então o modelo está garantidamente errado. Dito isso, drift de dados é frequentemente um sinal de alerta precoce de drift de conceito.
P: Re-treinamento automatizado é arriscado? R: Sim — 1) os novos dados podem ter problemas de qualidade de rótulo; 2) re-treinamento automatizado pode introduzir novos bugs; 3) ele deve passar por validação A/B antes de ir para produção. Fluxo recomendado: re-treinamento automatizado + revisão humana + validação A/B, em três etapas.
P: Como distinguimos flutuação sazonal de drift real? R: Use comparação ano a ano (este Q4 vs. Q4 do ano passado) em vez de período a período (Q4 vs. Q3). Sazonalidade é periódica (se repete todo ano); drift é estrutural (uma mudança sustentada que não se reverte).
📖 Resumo
- Arquitetura de monitoramento em três camadas: Infraestrutura (latência/disponibilidade) → Métricas de ML (distribuição de previsões/estatísticas de features) → Métricas de Negócio (MAPE/Receita)
- PSI é o padrão da indústria para detecção de drift: < 0,1 significa sem drift, 0,1-0,2 pede investigação, > 0,2 exige ação
- Drift de dados = mudança em P(X) (distribuição de entrada); drift de conceito = mudança em P(Y|X) (a relação) — este último é mais perigoso
- Pipeline automatizado de re-treinamento: detecção de drift → verificação de desempenho → re-treinamento → validação A/B → rollout canário
- Flutuação sazonal vs. drift: distinga com comparação ano a ano — sazonalidade se reverte, drift persiste
- Monitoramento + re-treinamento automatizado é o que mantém um sistema de ML confiável a longo prazo — a implantação não é a linha de chegada
📝 Exercícios
- Básico (Dificuldade ⭐): Use PSI para detectar mudança de distribuição entre dois conjuntos de dados e compare os valores de PSI para o caso sem drift versus um deslocamento de média de 10. Dica: consulte a função de cálculo de PSI na Seção 4.
- Intermediário (Dificuldade ⭐⭐): Implemente monitoramento em três camadas — para dados de previsão simulados, detecte: 1) drift de PSI; 2) degradação de MAPE; 3) anomalias de latência (p95 > 100ms). Dica: três funções de verificação independentes + um relatório combinado.
- Desafio (Dificuldade ⭐⭐⭐): Implemente um AutoRetrainPipeline completo — gere dados do período normal + período de promoção, treine um modelo inicial, detecte drift e re-treine automaticamente, compare o MAPE antes e depois do re-treinamento e produza um relatório completo de monitoramento. Dica: consulte o framework de re-treinamento automatizado na Seção 5.
← Anterior: Teste A/B | Próximo: Design do Projeto →
