---
title: Job Failure Prediction & Anomaly Detection API
emoji: 🔮
colorFrom: blue
colorTo: purple
sdk: docker
sdk_version: "latest"
app_file: app.py
pinned: false
---

# Job Failure Prediction & Anomaly Detection ML Service

Production-ready ML service for predicting job failures and detecting anomalies in job execution data.

## Features

- **Job Failure Prediction**: XGBoost-based classifier with SHAP explainability
- **Anomaly Detection**: Autoencoder-based unsupervised anomaly detection
- **FastAPI REST API**: Production-ready endpoints
- **Docker Support**: Containerized deployment
- **n8n Integration**: Ready-to-use workflow examples

## Quick Start

### 1. Training Models

Place your CSV files in the `data/` directory:

```bash
mkdir -p data models
# Copy your CSV files: true_export_report_20260120.csv, true_export_report_20260121.csv
```

Train the job failure prediction model:

```bash
python train_job_failure.py data/true_export_report_20260120.csv data/true_export_report_20260121.csv
```

Train the anomaly detection model:

```bash
python train_anomaly.py data/true_export_report_20260120.csv data/true_export_report_20260121.csv
```

Models will be saved to `models/` directory.

### 2. Running the Service

#### Local Development

```bash
pip install -r requirements.txt
uvicorn app:app --host 0.0.0.0 --port 8000
```

#### Docker

```bash
docker-compose up --build
```

The service will be available at `http://localhost:8000`

### 3. API Documentation

Once running, visit:
- Swagger UI: `http://localhost:8000/docs`
- ReDoc: `http://localhost:8000/redoc`

## API Endpoints

### Health Check

```bash
GET /health
```

### Job Failure Prediction

```bash
POST /predict/job-fail
Content-Type: application/json

{
  "zone": "prod",
  "job_nm": "daily_export_customer",
  "job_start_time": "2026-01-21T01:00:00",
  "duration_sec": 5400,
  "status": "SUCCESS",
  "err_msg": "",
  "explain": true
}
```

**Response:**
```json
{
  "fail_probability": 0.79,
  "risk_level": "MEDIUM",
  "top_drivers": [
    {
      "feature": "failure_rate_7",
      "shap_value": 0.30,
      "effect": "increase"
    }
  ],
  "recommended_actions": [
    "Monitor upstream dependencies and recent job history"
  ]
}
```

### Anomaly Detection

```bash
POST /detect/anomaly
Content-Type: application/json

{
  "features": {
    "duration_sec": 5400,
    "duration_zscore": 1.6,
    "err_msg_len": 0
  },
  "threshold": 0.01
}
```

**Response:**
```json
{
  "reconstruction_error": 0.0235,
  "is_anomaly": true,
  "threshold": 0.01,
  "top_drivers": [
    {
      "feature": "duration_zscore",
      "error": 0.0142
    }
  ]
}
```

## Project Structure

```
ml_service/
├── app.py                      # FastAPI application
├── preprocessing.py            # Data preprocessing and feature engineering
├── model_utils.py             # Model inference with SHAP
├── train_job_failure.py       # Training script for failure prediction
├── train_anomaly.py           # Training script for anomaly detection
├── requirements.txt           # Python dependencies
├── Dockerfile                 # Docker image definition
├── docker-compose.yml         # Docker Compose configuration
├── n8n_workflow_examples.md   # n8n integration guide
├── n8n_workflow_job_monitoring.json  # Importable n8n workflow
├── models/                    # Trained models (created after training)
│   ├── job_fail_pipeline_cpu.joblib
│   ├── anomaly_autoencoder_cpu.keras
│   ├── anomaly_scaler.joblib
│   ├── feature_schema.json
│   └── ...
└── data/                      # Training data (user-provided)
    ├── true_export_report_20260120.csv
    └── true_export_report_20260121.csv
```

## Features Engineering

The service automatically engineers the following features:

### Numeric Features
- `duration_sec`: Job duration in seconds
- `duration_zscore`: Z-score relative to rolling average
- `avg_duration_7`: 7-day rolling average duration
- `failure_rate_7`: 7-day rolling failure rate
- `err_msg_len`: Error message length
- `hour_sin`, `hour_cos`: Cyclical hour encoding

### Categorical Features
- `job_nm`: Job name
- `tasksgroup_nm`: Task group name
- `zone`: Environment/cluster
- `is_zeppelin`: Zeppelin flag
- `is_weekend`: Weekend indicator

## Model Details

### Job Failure Prediction
- **Algorithm**: XGBoost Classifier
- **Preprocessing**: StandardScaler for numeric, OneHotEncoder for categorical
- **Explainability**: SHAP values for feature importance
- **Output**: Failure probability (0-1), risk level, top drivers, recommended actions

### Anomaly Detection
- **Algorithm**: Autoencoder (TensorFlow/Keras)
- **Architecture**: Input → 64 → 32 → 64 → Output
- **Threshold**: Per-job 97th percentile or global threshold
- **Output**: Reconstruction error, anomaly flag, top contributing features

## n8n Integration

See `n8n_workflow_examples.md` for detailed integration examples and `n8n_workflow_job_monitoring.json` for an importable workflow.

### Quick n8n Setup

1. Import `n8n_workflow_job_monitoring.json` into n8n
2. Update the HTTP Request URLs to match your service endpoint
3. Configure Slack credentials (or replace with your alerting system)
4. Activate the workflow

## Risk Levels

- **MINIMAL**: `fail_probability < 0.3`
- **LOW**: `0.3 <= fail_probability < 0.5`
- **MEDIUM**: `0.5 <= fail_probability < 0.8`
- **CRITICAL**: `fail_probability >= 0.8`

## Alert Conditions

### WARNING
- `fail_probability >= 0.5 && fail_probability < 0.8`
- OR `is_anomaly === true` with moderate reconstruction error

### CRITICAL
- `fail_probability >= 0.8`
- OR `is_anomaly === true` with high reconstruction error (> 3x threshold)

## Development

### Running Tests

```bash
# Test prediction endpoint
curl -X POST http://localhost:8000/predict/job-fail \
  -H "Content-Type: application/json" \
  -d '{
    "zone": "prod",
    "job_nm": "test_job",
    "job_start_time": "2026-01-21T01:00:00",
    "duration_sec": 3600,
    "status": "SUCCESS"
  }'
```

### Model Retraining

Models should be retrained periodically as new data becomes available:

```bash
# Add new CSV files to data/ directory
python train_job_failure.py data/*.csv
python train_anomaly.py data/*.csv

# Restart service to load new models
docker-compose restart
```

## Monitoring

The service includes a health check endpoint that reports model loading status:

```bash
GET /health
```

Response:
```json
{
  "status": "healthy",
  "service": "job-failure-prediction",
  "models_loaded": {
    "predictor": true,
    "anomaly_detector": true
  }
}
```

## Production Deployment

1. **Train models** on historical data
2. **Save models** to `models/` directory
3. **Build Docker image**: `docker build -t job-ml-service .`
4. **Deploy** using docker-compose or Kubernetes
5. **Configure n8n** workflows for monitoring
6. **Set up alerting** (Slack, PagerDuty, etc.)

## License

[Your License Here]

## Support

For issues or questions, please refer to the documentation or contact the development team.