How to train an anomaly detector in PyTorch

Anomaly detection often starts with examples of normal behavior and flags samples that reconstruct badly. In PyTorch, a small autoencoder can learn the normal feature pattern and turn reconstruction error into a score for telemetry, tabular metrics, or sensor-like rows.

Synthetic normal data keeps the training loop, threshold calculation, and scoring logic testable before any private dataset is involved. The model trains only on normal samples, calculates a threshold from training reconstruction errors, and compares a normal probe batch with a shifted outlier batch.

Use the same structure with real data after scaling features consistently and keeping known anomalies out of the normal-only training slice. The checkpoint stores both the model state and the threshold so a later scoring run can make the same normal-versus-suspect decision.

1. Create the training script.

   train_anomaly_detector.py

   from pathlib import Path
    
   import torch
   from torch import nn
   from torch.utils.data import DataLoader, TensorDataset
    
    
   torch.manual_seed(7)
    
   FEATURE_COUNT = 6
   BATCH_SIZE = 32
   EPOCHS = 40
   CHECKPOINT = Path("anomaly_detector.pt")
    
    
   class Autoencoder(nn.Module):
       def __init__(self, feature_count: int) -> None:
           super().__init__()
           self.encoder = nn.Sequential(
               nn.Linear(feature_count, 4),
               nn.ReLU(),
               nn.Linear(4, 2),
           )
           self.decoder = nn.Sequential(
               nn.Linear(2, 4),
               nn.ReLU(),
               nn.Linear(4, feature_count),
           )
    
       def forward(self, inputs: torch.Tensor) -> torch.Tensor:
           return self.decoder(self.encoder(inputs))
    
    
   def score_samples(model: nn.Module, samples: torch.Tensor) -> torch.Tensor:
       model.eval()
       with torch.inference_mode():
           reconstructed = model(samples)
           return torch.mean((samples - reconstructed) ** 2, dim=1)
    
    
   def main() -> None:
       normal_samples = 0.20 * torch.randn(512, FEATURE_COUNT)
       loader = DataLoader(
           TensorDataset(normal_samples),
           batch_size=BATCH_SIZE,
           shuffle=True,
       )
    
       model = Autoencoder(FEATURE_COUNT)
       loss_fn = nn.MSELoss()
       optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
    
       for epoch in range(1, EPOCHS + 1):
           model.train()
           running_loss = 0.0
    
           for (batch,) in loader:
               optimizer.zero_grad()
               reconstructed = model(batch)
               loss = loss_fn(reconstructed, batch)
               loss.backward()
               optimizer.step()
               running_loss += loss.item() * batch.size(0)
    
           if epoch in (1, 10, 20, 30, 40):
               print(f"epoch={epoch:02d} train_loss={running_loss / len(normal_samples):.6f}")
    
       train_scores = score_samples(model, normal_samples)
       threshold = train_scores.mean() + 3 * train_scores.std()
    
       normal_probe = 0.20 * torch.randn(16, FEATURE_COUNT)
       outlier_probe = normal_probe + 3.0
       normal_score_mean = score_samples(model, normal_probe).mean()
       outlier_score_mean = score_samples(model, outlier_probe).mean()
    
       torch.save(
           {
               "feature_count": FEATURE_COUNT,
               "threshold": threshold,
               "model_state": model.state_dict(),
           },
           CHECKPOINT,
       )
    
       print(f"threshold={threshold:.6f}")
       print(f"normal_score_mean={normal_score_mean:.6f}")
       print(f"outlier_score_mean={outlier_score_mean:.6f}")
       decision = bool(outlier_score_mean > threshold and normal_score_mean < threshold)
       print(f"anomaly_decision={decision}")
       print(f"checkpoint={CHECKPOINT}")
    
    
   if __name__ == "__main__":
       main()

   The threshold uses the training reconstruction-error mean plus three standard deviations. Tune that rule against validation data before turning the score into an alert.

2. Run the training script.

   $ python train_anomaly_detector.py
   epoch=01 train_loss=0.097516
   epoch=10 train_loss=0.040477
   epoch=20 train_loss=0.040476
   epoch=30 train_loss=0.040521
   epoch=40 train_loss=0.040619
   threshold=0.110219
   normal_score_mean=0.042106
   outlier_score_mean=8.940481
   anomaly_decision=True
   checkpoint=anomaly_detector.pt

   The normal probe mean should be below threshold, while the shifted outlier probe should be above it.

3. Create a checkpoint scoring script.

   score_anomaly_checkpoint.py

   import torch
    
   from train_anomaly_detector import Autoencoder, score_samples
    
    
   checkpoint = torch.load("anomaly_detector.pt", weights_only=True)
   model = Autoencoder(checkpoint["feature_count"])
   model.load_state_dict(checkpoint["model_state"])
    
   samples = torch.tensor(
       [
           [0.05, -0.04, 0.02, 0.01, -0.03, 0.04],
           [3.10, 2.85, 3.25, 2.95, 3.05, 3.15],
       ],
       dtype=torch.float32,
   )
   scores = score_samples(model, samples)
   threshold = checkpoint["threshold"]
    
   print(f"normal_sample_score={scores[0]:.6f}")
   print(f"suspect_sample_score={scores[1]:.6f}")
   print(f"threshold={threshold:.6f}")
   decision = bool(scores[0] < threshold and scores[1] > threshold)
   print(f"loaded_checkpoint_decision={decision}")

   weights_only=True restricts torch.load to tensor weights and simple metadata needed for this checkpoint.

4. Run the checkpoint smoke test.

   $ python score_anomaly_checkpoint.py
   normal_sample_score=0.001285
   suspect_sample_score=9.290407
   threshold=0.110219
   loaded_checkpoint_decision=True