Como extrair sinais úteis de dados financeiros e transacionais

O que é feature engineering e por que importa

Feature engineering é o processo de transformar dados brutos em variáveis que um modelo consegue aprender. Em dados tabulares — especialmente financeiros e transacionais — a qualidade das features costuma importar mais que a escolha do algoritmo. Um XGBoost bem alimentado supera um modelo complexo com inputs pobres.

Em crédito, fraude e propensão, os dados brutos raramente chegam prontos. Você recebe transações, cadastro, histórico de pagamentos e precisa derivar sinais como "média de gastos nos últimos 90 dias", "razão entre dívida e renda" ou "quantidade de atrasos nos últimos 12 meses". Essas variáveis capturam comportamento — o que o modelo precisa para generalizar.

O desafio é fazer isso de forma reprodutível, sem leakage temporal e escalável para produção. Este post cobre as técnicas que mais aparecem em projetos reais com dados financeiros.

Tipos de features em dados financeiros

Antes de codar, organize o que você pode extrair em quatro famílias:

Agregações temporais — soma, média, máximo, contagem em janelas (7, 30, 90, 180 dias)
Razões e interações — dívida/renda, gasto/limite, pagamento mínimo/fatura
Features de recência e frequência — dias desde último pagamento, transações por semana
Encoding de categóricas — one-hot, target encoding, frequency encoding

Cada família responde a uma pergunta de negócio diferente. Agregações capturam nível de atividade; razões capturam proporção e capacidade; recência captura mudança recente de comportamento — frequentemente o sinal mais forte em churn e fraude.

Agregações temporais com pandas

Transações costumam vir em formato longo — uma linha por evento. O modelo precisa de formato largo — uma linha por cliente com colunas agregadas.

import pandas as pd

# transactions: customer_id, date, amount, category
transactions["date"] = pd.to_datetime(transactions["date"])
reference_date = transactions["date"].max()  # data de corte para snapshot

def aggregate_window(df, customer_col, date_col, value_col, days, prefix):
    cutoff = reference_date - pd.Timedelta(days=days)
    window = df[df[date_col] >= cutoff]

    agg = window.groupby(customer_col)[value_col].agg(
        count="count",
        sum="sum",
        mean="mean",
        max="max",
        std="std",
    )
    agg.columns = [f"{prefix}_{days}d_{col}" for col in agg.columns]
    return agg

features_30d = aggregate_window(transactions, "customer_id", "date", "amount", 30, "tx")
features_90d = aggregate_window(transactions, "customer_id", "date", "amount", 90, "tx")

customer_features = features_30d.join(features_90d, how="outer")

Duas decisões críticas:

Data de referência — em treino, use a data do snapshot de cada cliente; em produção, use "hoje". Nunca agregue com dados futuros
Múltiplas janelas — 30d vs 90d captura tendência; a razão entre elas (ver abaixo) é frequentemente mais preditiva que os valores absolutos

Razões e features derivadas

Razões transformam valores absolutos em sinais comparativos. Em crédito, algumas das features mais estáveis:

import numpy as np

def build_ratio_features(df):
  out = df.copy()

  # Capacidade de pagamento
  out["debt_to_income"] = out["total_debt"] / out["monthly_income"].clip(lower=1)
  out["payment_to_income"] = out["min_payment"] / out["monthly_income"].clip(lower=1)

  # Utilização de crédito
  out["credit_utilization"] = out["balance"] / out["credit_limit"].clip(lower=1)

  # Tendência de gastos: atividade recente vs histórico
  out["spend_trend"] = (
      out["tx_30d_sum"] / out["tx_90d_sum"].clip(lower=1)
  )

  # Volatilidade relativa
  out["amount_cv"] = out["tx_90d_std"] / out["tx_90d_mean"].clip(lower=1)

  return out

.clip(lower=1) evita divisão por zero — detalhe pequeno que quebra pipelines em produção quando renda ou limite são zero.

Features de delta também funcionam bem: diferença entre comportamento recente e histórico (tx_30d_mean - tx_90d_mean) sinaliza mudança abrupta — comum em inadimplência iminente.

Recência e contagem de eventos

Em dados de pagamento, "quantos dias desde o último atraso" e "quantidade de atrasos nos últimos 12 meses" costumam superar variáveis estáticas de cadastro:

payments = payments.sort_values(["customer_id", "due_date"])

payments["days_late"] = (payments["payment_date"] - payments["due_date"]).dt.days
payments["is_late"] = (payments["days_late"] > 0).astype(int)

recency = (
    payments[payments["is_late"] == 1]
    .groupby("customer_id")["due_date"]
    .max()
    .rename("last_late_date")
)

late_counts = (
    payments[payments["due_date"] >= reference_date - pd.Timedelta(days=365)]
    .groupby("customer_id")["is_late"]
    .sum()
    .rename("late_count_12m")
)

recency_features = pd.DataFrame({
    "days_since_last_late": (reference_date - recency).dt.days,
    "late_count_12m": late_counts,
})

days_since_last_late com valor alto (ou missing tratado como "nunca atrasou") é um sinal forte de bom pagador. Missingness aqui é informação — clientes sem histórico de atraso são diferentes de quem atrasou recentemente.

Encoding de variáveis categóricas

Categóricas de alta cardinalidade (cidade, merchant, SKU) não funcionam bem com one-hot — explodem a dimensionalidade. Duas alternativas comuns:

Target encoding — substitui cada categoria pela média do target naquela categoria, com suavização para evitar overfitting:

def target_encode(series, target, smoothing=10):
    global_mean = target.mean()
    agg = target.groupby(series).agg(["mean", "count"])
    smooth = (agg["count"] * agg["mean"] + smoothing * global_mean) / (agg["count"] + smoothing)
    return series.map(smooth)

# Aplicar APENAS no fold de treino em validação cruzada
df["city_risk"] = target_encode(df["city"], df["default"])

Frequency encoding — substitui pela frequência da categoria no dataset. Simples e eficaz para fraude (merchants raros podem ser mais suspeitos).

Em produção, persista o mapeamento de encoding (dict ou transformer scikit-learn) — categorias novas devem cair em um valor default, não gerar erro.

Seleção de features e validação

Com dezenas ou centenas de features derivadas, o próximo passo é filtrar o que realmente ajuda:

from sklearn.feature_selection import mutual_info_classif
import xgboost as xgb

# Importância por mutual information (não linear)
mi_scores = mutual_info_classif(X_train, y_train, random_state=42)
mi_ranking = pd.Series(mi_scores, index=X_train.columns).sort_values(ascending=False)

# Importância por XGBoost após treino
model = xgb.XGBClassifier(
    n_estimators=300,
    max_depth=5,
    scale_pos_weight=len(y_train[y_train==0]) / len(y_train[y_train==1]),
    random_state=42,
)
model.fit(X_train, y_train)

importance = pd.Series(
    model.feature_importances_, index=X_train.columns
).sort_values(ascending=False)

Use as duas visões em conjunto: mutual information detecta relações não lineares antes do treino; importância do XGBoost reflete o que o modelo efetivamente usou. Features que aparecem no topo de ambas são candidatas fortes.

Evite selecionar features olhando o conjunto de teste — use validação cruzada ou um holdout separado para seleção.

Pipeline reprodutível

Para produção, encapsule a engenharia de features em funções ou transformers que rodam igual em treino e inferência:

from sklearn.base import BaseEstimator, TransformerMixin

class FinancialFeatureBuilder(BaseEstimator, TransformerMixin):
    def __init__(self, reference_date=None):
        self.reference_date = reference_date

    def fit(self, X, y=None):
        self.reference_date_ = self.reference_date or X["date"].max()
        return self

    def transform(self, X):
        # aplicar agregações, razões e recência
        return build_all_features(X, self.reference_date_)

# Integrar no pipeline completo
from sklearn.pipeline import Pipeline

full_pipeline = Pipeline([
    ("features", FinancialFeatureBuilder()),
    ("classifier", xgb.XGBClassifier(...)),
])

Transformers customizados garantem que a mesma lógica roda em batch scoring e na API — sem copiar código de notebook para produção.

Além do básico: pontos de atenção

Armadilhas frequentes em feature engineering financeiro:

Leakage temporal — incluir transações posteriores à data do target invalida o modelo; sempre defina um reference_date por snapshot
Agregar antes do split — se o split é por cliente, agregue por cliente; se é temporal, agregue até a data de corte de cada período
Overfitting em target encoding — aplique dentro de folds de CV; encoding global vaza o target
Features instáveis — variáveis com alta variância entre meses degradam em produção; monitore distribuição
Redundância — tx_30d_sum e tx_30d_mean correlacionam com tx_30d_count; XGBoost tolera, mas logistic regression sofre

Boas práticas:

Documente a definição de cada feature em linguagem de negócio ("dívida total dividida pela renda mensal")
Versione o código de feature engineering junto com o modelo
Valide com split temporal, não aleatório — comportamento de crédito muda com sazonalidade e macroeconomia

Conclusão

Feature engineering é onde modelos tabulares ganham ou perdem em dados financeiros. Agregações temporais, razões, recência e encoding bem feitos extraem comportamento que algoritmos conseguem generalizar.

O fluxo prático: parta das perguntas de negócio, derive features em janelas temporais sem leakage, encode categóricas com cuidado, selecione com mutual information e importância do modelo, e encapsule tudo em um transformer reprodutível. Combinado com técnicas de balanceamento e deploy que cobrimos nos posts anteriores, você tem a base completa para um modelo de crédito ou fraude em produção.