Spaces:
Running
Running
File size: 5,564 Bytes
42029e4 44cd54e 42029e4 44cd54e 42029e4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | """Reconstruction-error autoencoder for unsupervised fraud detection.
We train a denoising autoencoder on (mostly) clean applications.
At inference time, applications with high reconstruction error are
flagged - including novel fraud patterns the supervised models
weren't trained on.
Loss surface is intentionally simple: MSE on normalised features.
For an MVP this beats fancier VAEs and is much easier to debug.
"""
from __future__ import annotations
from typing import Any
import numpy as np
import pandas as pd
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from sklearn.preprocessing import StandardScaler
from ..utils.logging import get_logger
from .base import FraudModel
log = get_logger(__name__)
class _AENet(nn.Module):
def __init__(self, input_dim: int, encoder_dims: list[int], dropout: float):
super().__init__()
# Build encoder
enc_layers: list[nn.Module] = []
prev = input_dim
for h in encoder_dims:
enc_layers += [nn.Linear(prev, h), nn.ReLU(), nn.Dropout(dropout)]
prev = h
self.encoder = nn.Sequential(*enc_layers)
# Symmetric decoder back to input_dim
dec_layers: list[nn.Module] = []
for h in reversed(encoder_dims[:-1]):
dec_layers += [nn.Linear(prev, h), nn.ReLU(), nn.Dropout(dropout)]
prev = h
dec_layers += [nn.Linear(prev, input_dim)]
self.decoder = nn.Sequential(*dec_layers)
def forward(self, x: torch.Tensor) -> torch.Tensor:
return self.decoder(self.encoder(x))
class AutoencoderFraudModel(FraudModel):
name = "autoencoder"
DEFAULT_PARAMS = {
"encoder_dims": [128, 64, 32],
"dropout": 0.2,
"batch_size": 512,
"epochs": 30,
"learning_rate": 1e-3,
"early_stopping_patience": 5,
"noise_std": 0.05,
}
def __init__(self, params: dict | None = None, device: str | None = None):
self.params = {**self.DEFAULT_PARAMS, **(params or {})}
self.device = torch.device(
device or ("cuda" if torch.cuda.is_available() else "cpu")
)
self.scaler = StandardScaler()
self.net: _AENet | None = None
self.score_min_ = 0.0
self.score_max_ = 1.0
self._trained = False
def fit(self, X: pd.DataFrame, y: pd.Series | None = None, **kwargs) -> "AutoencoderFraudModel":
# If labels available, train only on negatives ("clean" data) - semi-supervised.
if y is not None:
mask = y.values == 0
X_train = X.loc[mask]
log.info(
f"[AE] training on {mask.sum():,} negative samples "
f"(skipping {(~mask).sum():,} positives)"
)
else:
X_train = X
X_arr = self.scaler.fit_transform(X_train.values).astype(np.float32)
input_dim = X_arr.shape[1]
self.net = _AENet(
input_dim=input_dim,
encoder_dims=self.params["encoder_dims"],
dropout=self.params["dropout"],
).to(self.device)
optim = torch.optim.Adam(self.net.parameters(), lr=self.params["learning_rate"])
loss_fn = nn.MSELoss()
ds = TensorDataset(torch.from_numpy(X_arr))
loader = DataLoader(ds, batch_size=self.params["batch_size"], shuffle=True)
best_loss = float("inf")
patience = 0
for epoch in range(self.params["epochs"]):
self.net.train()
epoch_loss = 0.0
for (batch,) in loader:
batch = batch.to(self.device)
noisy = batch + torch.randn_like(batch) * self.params["noise_std"]
recon = self.net(noisy)
loss = loss_fn(recon, batch)
optim.zero_grad()
loss.backward()
optim.step()
epoch_loss += loss.item() * len(batch)
epoch_loss /= len(ds)
if epoch_loss < best_loss - 1e-5:
best_loss = epoch_loss
patience = 0
else:
patience += 1
if patience >= self.params["early_stopping_patience"]:
log.info(f"[AE] early stopping at epoch {epoch}, best_loss={best_loss:.5f}")
break
# Compute reconstruction-error range for normalisation
self.net.eval()
with torch.no_grad():
recon = self.net(torch.from_numpy(X_arr).to(self.device))
err = ((recon - torch.from_numpy(X_arr).to(self.device)) ** 2).mean(dim=1).cpu().numpy()
self.score_min_ = float(np.percentile(err, 1))
self.score_max_ = float(np.percentile(err, 99))
self._trained = True
log.info(
f"[AE] trained. recon_err range [{self.score_min_:.4f}, {self.score_max_:.4f}]"
)
return self
def predict_proba(self, X: pd.DataFrame) -> np.ndarray:
if not self._trained or self.net is None:
raise RuntimeError("Model not trained")
self.net.eval()
X_arr = self.scaler.transform(X.values).astype(np.float32)
with torch.no_grad():
t = torch.from_numpy(X_arr).to(self.device)
recon = self.net(t)
err = ((recon - t) ** 2).mean(dim=1).cpu().numpy()
span = max(self.score_max_ - self.score_min_, 1e-9)
prob = (err - self.score_min_) / span
return np.clip(prob, 0.0, 1.0)
def get_params(self) -> dict[str, Any]:
return {"params": self.params}
|