Probabilistic anomaly detection using a Variational Autoencoder (VAE). Learns the latent distribution of normal data during training, then detects anomalies by computing reconstruction probability at inference time. Provides calibrated uncertainty quantification alongside anomaly scores through Monte Carlo sampling in the latent space.
Input x
|
v
Encoder: FC(input_dim, 128) -> ReLU -> FC(128, 64) -> ReLU
| |
v v
FC(64, latent_dim) = mu FC(64, latent_dim) = log_var
| |
+---> Reparameterization Trick <--------+
| z = mu + eps * exp(0.5 * log_var)
v
Decoder: FC(latent_dim, 64) -> ReLU -> FC(64, 128) -> ReLU -> FC(128, input_dim) -> Sigmoid
|
v
Reconstruction x_hat
Loss = Reconstruction Loss (BCE/MSE) + beta * KL Divergence
| Feature | VAE (this project) | AE + Isolation Forest (anomaly-detection-system) |
|---|---|---|
| Anomaly Score | Reconstruction probability | Reconstruction error + isolation score |
| Uncertainty | Native (posterior sampling) | Requires calibration |
| Interpretability | Moderate (latent space) | High (feature importance) |
| Training Stability | Requires KL annealing | Straightforward |
| Threshold Setting | Probability-based | Percentile-based (simpler) |
| Robustness | KL collapse risk | More robust |
Decision: The simpler Autoencoder + Isolation Forest ensemble was chosen for the primary project due to better interpretability, more stable training, and fewer hyperparameters to tune. The VAE approach provides richer uncertainty estimates but requires careful KL annealing and monitoring to avoid posterior collapse.
- Python 3.10+
- PyTorch 2.0+
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt# Start API server
uvicorn src.api:app --host 0.0.0.0 --port 8000 --reload
# Generate demo data and train
python -m src.data_generator
python -m src.trainer --config configs/config.yaml| Method | Endpoint | Description |
|---|---|---|
| POST | /detect |
Anomaly detection with uncertainty estimates |
| POST | /encode |
Get latent space representation |
| POST | /train |
Train a new model on uploaded data |
| GET | /health |
Service health check |
import requests
import numpy as np
# Detect anomalies
data = np.random.randn(10, 30).tolist()
response = requests.post(
"http://localhost:8000/detect",
json={"data": data}
)
results = response.json()
for i, r in enumerate(results["results"]):
print(f"Sample {i}: anomaly={r['is_anomaly']}, score={r['anomaly_score']:.4f}")vae-anomaly-detection/
├── src/
│ ├── vae_model.py # VAE architecture
│ ├── anomaly_detector.py # Detection pipeline
│ ├── trainer.py # Training with KL annealing
│ ├── data_generator.py # Synthetic data generation
│ └── api.py # FastAPI application
├── configs/
│ └── config.yaml # Configuration
├── requirements.txt
└── README.md
MIT License