Como detetar drift de características em Machine Learning: passo a passo
Detetar drift de características em Machine Learning é útil para perceber quando os dados em produção mudam relativamente aos dados de treino, evitando perda de desempenho. Este tutorial mostra, passo a passo, como calcular medidas simples de drift por característica e sinalizar mudanças relevantes com Python.
Pré-requisitos
- Python 3.8+ instalado
- Pandas e scipy instalados (pip install pandas scipy)
- Dataset de treino e de produção como ficheiros CSV
- Conhecimentos básicos de Python e estatística descritiva
Passo 1: Porquê detetar drift de características?
O drift de características (feature drift) acontece quando a distribuição de uma coluna muda ao longo do tempo. Identificar drift ajuda a decidir se é necessário retrain, ajustar pré-processamento ou sinalizar alertas. Vamos usar testes estatísticos e medidas práticas para cada característica.
Passo 2: Preparar dados e carregar bibliotecas
Carregue os CSV do conjunto de treino e do conjunto de produção e selecione as colunas que quer monitorizar. Mantemos apenas colunas numéricas neste exemplo; para categóricas usamos comparações por frequências.
import pandas as pd
from scipy.stats import ks_2samp, chi2_contingency
train = pd.read_csv('train.csv')
prod = pd.read_csv('prod.csv')
# Lista de colunas a monitorizar (numéricas e categóricas)
num_cols = ['age','salary']
cat_cols = ['region','product_type']
Passo 3: Calcular drift para características numéricas (Kolmogorov-Smirnov)
O teste de Kolmogorov-Smirnov compara duas amostras para ver se vêm da mesma distribuição. Não assume normalidade. Usamos p-value e uma medida de efeito (D) para sinalizar drift.
def detect_numeric_drift(train_ser, prod_ser, alpha=0.05):
# Remove NaNs
a = train_ser.dropna()
b = prod_ser.dropna()
if len(a) < 20 or len(b) < 20:
return {'p_value': None, 'D': None, 'drift': 'insufficient_data'}
stat, p = ks_2samp(a, b)
drift = 'drift' if p < alpha else 'no_drift'
return {'p_value': float(p), 'D': float(stat), 'drift': drift}
results_num = {}
for col in num_cols:
results_num[col] = detect_numeric_drift(train[col], prod[col])
print(results_num)
Passo 4: Calcular drift para características categóricas (teste qui-quadrado)
Para categóricas com contagens, comparamos tabelas de frequência. Se houver categorias novas ou removidas, isto indica drift qualitativo.
def detect_categorical_drift(train_ser, prod_ser, alpha=0.05):
a = train_ser.fillna('NULL')
b = prod_ser.fillna('NULL')
# Frequências alinhadas por categoria
freq = pd.concat([a.value_counts(), b.value_counts()], axis=1, sort=False).fillna(0)
freq.columns = ['train', 'prod']
chi2, p, _, _ = chi2_contingency(freq.values)
different_categories = set(b.unique()) - set(a.unique())
drift = 'drift' if p < alpha or len(different_categories) > 0 else 'no_drift'
return {'p_value': float(p), 'different_categories': list(different_categories), 'drift': drift}
results_cat = {}
for col in cat_cols:
results_cat[col] = detect_categorical_drift(train[col], prod[col])
print(results_cat)
Passo 5: Medidas complementares e thresholds práticos
Além de testes estatísticos, é útil calcular diferenças de média, mediana e proporções para priorizar features. Defina thresholds práticos (ex.: diferença média > 10% ou D > 0.1).
def numeric_statistics(train_ser, prod_ser):
t = train_ser.dropna()
p = prod_ser.dropna()
stats = {
'train_mean': float(t.mean()),
'prod_mean': float(p.mean()),
'mean_rel_diff': abs(t.mean() - p.mean()) / (abs(t.mean()) + 1e-9)
}
return stats
summary = {}
for col in num_cols:
summary[col] = {**results_num[col], **numeric_statistics(train[col], prod[col])}
print(summary)
Passo 6: Automatizar e gerar relatório
Junte tudo numa função que itere por colunas, aplique testes e escreva um CSV de alerta com colunas sinalizadas por prioridade.
def drift_report(train_df, prod_df, num_cols, cat_cols):
rows = []
for col in num_cols:
r = detect_numeric_drift(train_df[col], prod_df[col])
stats = numeric_statistics(train_df[col], prod_df[col])
priority = 'high' if r['drift']=='drift' and stats['mean_rel_diff'] > 0.1 else 'low'
rows.append({'feature': col, 'type': 'numeric', **r, **stats, 'priority': priority})
for col in cat_cols:
r = detect_categorical_drift(train_df[col], prod_df[col])
priority = 'high' if r['drift']=='drift' else 'low'
rows.append({'feature': col, 'type': 'categorical', **r, 'priority': priority})
return pd.DataFrame(rows)
report = drift_report(train, prod, num_cols, cat_cols)
report.to_csv('drift_report.csv', index=False)
print(report)
Verificar o resultado
Abra drift_report.csv: deve listar cada característica com p_value, indicador de drift e priority. Verifique colunas marcadas como high: inspeccione histogramas ou boxplots para confirmar visualmente. Se vários features com high, considere retrain, recalibrar thresholds, ou ajustar ETL.
Conclusão
Este método simples combina testes estatísticos (KS e qui-quadrado) com medidas práticas para detetar drift de características em Machine Learning. Próximos passos: integrar isto num pipeline de monitorização em produção, adicionar testes por janela temporal e usar técnicas de explicabilidade para entender causas. Dica: comece a monitorizar gradualmente e ajuste thresholds conforme o custo de falso positivo/negativo no seu cenário.