[LOW] Implement Data Drift Detection for Production Models

[frankbria]

🎯 Overview

Add data drift detection to monitor changes in input data distribution that may affect model performance.

✅ Acceptance Criteria

• Statistical drift tests (KS test, PSI, etc.)
• Feature distribution monitoring
• Drift alerts and notifications
• Drift visualization dashboard
• Automatic retraining triggers
• Historical drift tracking

🏗️ Technical Requirements

• Backend: DriftDetectionService using scipy.stats
• Statistical tests: Kolmogorov-Smirnov, Population Stability Index
• Monitoring integration with deployed models
• Endpoints: GET /api/deployments/{id}/drift, POST /api/deployments/{id}/check-drift

🏷️ Labels

`low-priority`, `backend`, `ml-algorithms`, `monitoring`

⏱️ Estimated Effort

3-4 weeks

[traycerai]

Plan

Key Observations

The codebase has a solid foundation for implementing drift detection: MongoDB with Beanie ODM for data storage, FastAPI for APIs, existing monitoring infrastructure with placeholder drift detection in file:apps/backend/app/services/prediction_monitoring.py (lines 187-204), comprehensive statistical capabilities in file:apps/backend/app/services/data_processing/statistics_engine.py, and all required dependencies (scipy, numpy, pandas, statsmodels, redis). The production API serves models via /production/v1/models/{model_id}/predict and monitoring endpoints exist at /monitoring/models/{model_id}/drift but need full implementation.

Approach

Implement a comprehensive drift detection system by creating a dedicated DriftDetectionService that leverages scipy.stats for statistical tests (Kolmogorov-Smirnov, Population Stability Index, Chi-square). Store drift history in MongoDB using new Beanie document models. Extend existing monitoring endpoints to support drift checking and historical tracking. Integrate drift detection into the prediction pipeline to automatically monitor incoming data. Use Redis pub/sub for drift alerts and create frontend dashboard components for visualization. The implementation follows the existing service-oriented architecture pattern and integrates seamlessly with the current monitoring infrastructure.

Implementation Steps

1. Database Models for Drift Tracking

Create new Beanie document models in file:apps/backend/app/models/:

Create drift_detection.py:

• DriftMetrics document to store drift detection results per model
  • Fields: model_id, user_id, checked_at, drift_detected, drift_score, statistical_tests (dict with KS test, PSI, chi-square results per feature), features_with_drift (list), severity (low/medium/high), recommendation
  • Indexes on model_id, user_id, checked_at
• FeatureDriftHistory document for per-feature drift tracking
  • Fields: model_id, feature_name, timestamp, reference_distribution (dict with mean, std, quantiles), current_distribution, ks_statistic, ks_pvalue, psi_value, drift_detected
  • Indexes on model_id, feature_name, timestamp
• DriftAlert document for alert management
  • Fields: model_id, user_id, alert_type (drift_detected/threshold_exceeded), severity, message, created_at, acknowledged, acknowledged_at, metadata
  • Indexes on model_id, user_id, acknowledged, created_at
• TrainingDataStatistics document to store baseline statistics
  • Fields: model_id, feature_statistics (dict mapping feature names to statistical summaries), created_at, updated_at
  • Index on model_id

Register these models in file:apps/backend/app/main.py lifespan function (line 97-119).

2. Drift Detection Service

Create file:apps/backend/app/services/drift_detection_service.py:

Core Methods:

• calculate_psi(expected_distribution, actual_distribution, bins=10) - Calculate Population Stability Index
  • Bin both distributions, calculate PSI = Σ((actual% - expected%) × ln(actual%/expected%))
  • Return PSI value and interpretation (< 0.1 no drift, 0.1-0.25 moderate, > 0.25 significant)
• perform_ks_test(reference_data, current_data) - Kolmogorov-Smirnov test using scipy.stats.ks_2samp
  • Return statistic, p-value, and drift_detected (p-value < 0.05)
• perform_chi_square_test(reference_counts, current_counts) - Chi-square test for categorical features using scipy.stats.chisquare
  • Return statistic, p-value, drift_detected
• calculate_drift_score(feature_results) - Aggregate drift metrics across features
  • Weight by feature importance if available
  • Return overall drift score (0-1 scale)
• detect_drift_for_model(model_id, current_data_df, threshold=0.05) - Main drift detection method
  • Load training statistics from TrainingDataStatistics
  • For each feature: run appropriate statistical test (KS for numeric, chi-square for categorical)
  • Calculate PSI for all features
  • Store results in DriftMetrics and FeatureDriftHistory
  • Create DriftAlert if drift detected
  • Return comprehensive drift report
• store_training_statistics(model_id, training_data_df, feature_names) - Store baseline statistics
  • Calculate distribution statistics for each feature
  • Save to TrainingDataStatistics document
• get_drift_history(model_id, days=30) - Retrieve historical drift metrics
  • Query DriftMetrics for time range
  • Return time series data for visualization
• get_feature_drift_trends(model_id, feature_name, days=30) - Per-feature drift trends
  • Query FeatureDriftHistory for specific feature
  • Return trend data with statistical test results

Statistical Test Implementations:

• Use scipy.stats.ks_2samp for continuous features
• Use scipy.stats.chisquare for categorical features
• Implement PSI calculation manually with numpy
• Use statsmodels for additional tests if needed

3. Extend Prediction Monitoring Service

Update file:apps/backend/app/services/prediction_monitoring.py:

Enhance PredictionLog class:

• Add method get_recent_data_for_drift(model_id, limit=1000) to extract feature data from logged predictions
• Convert logged input_data to DataFrame for drift analysis

Replace placeholder detect_drift method (lines 187-204):

• Call DriftDetectionService.detect_drift_for_model()
• Pass recent prediction data from PredictionLog
• Return actual drift detection results instead of placeholder

Add new methods:

• schedule_drift_check(model_id, interval_hours=24) - Schedule periodic drift checks
• trigger_drift_alert(model_id, drift_result) - Publish alert to Redis pub/sub channel

4. API Endpoints for Drift Detection

Update file:apps/backend/app/api/routes/monitoring.py:

Enhance existing /monitoring/models/{model_id}/drift endpoint (lines 136-159):

• Replace placeholder with actual drift detection
• Add query parameters: check_now (bool), days (int for historical data)
• If check_now=true, run drift detection on recent predictions
• Otherwise, return latest drift check results from database
• Response includes: drift_detected, drift_score, features_with_drift, statistical_tests, severity, recommendation, checked_at

Add new endpoints:

• POST /api/v1/monitoring/models/{model_id}/drift/check - Manually trigger drift check
  • Accept optional request body with sample data
  • Run drift detection immediately
  • Return detailed results with per-feature breakdown
• GET /api/v1/monitoring/models/{model_id}/drift/history - Get drift history
  • Query parameters: days (default 30), feature_name (optional)
  • Return time series of drift metrics
  • Include aggregated statistics
• GET /api/v1/monitoring/models/{model_id}/drift/features - Per-feature drift analysis
  • Return drift status for each feature
  • Include distribution comparisons (reference vs current)
  • Statistical test results per feature
• GET /api/v1/monitoring/models/{model_id}/alerts - Get drift alerts
  • Query parameters: acknowledged (bool), severity (low/medium/high)
  • Return list of alerts with pagination
• PUT /api/v1/monitoring/models/{model_id}/alerts/{alert_id}/acknowledge - Acknowledge alert
  • Mark alert as acknowledged
  • Record acknowledgment timestamp

Add deployment-specific endpoints:

• GET /api/v1/deployments/{deployment_id}/drift - Alias for model drift endpoint
• POST /api/v1/deployments/{deployment_id}/check-drift - Alias for manual drift check

5. Response Schemas

Create file:apps/backend/app/schemas/drift.py:

Pydantic models:

• StatisticalTestResult - Fields: test_name, statistic, p_value, drift_detected, interpretation
• FeatureDriftResult - Fields: feature_name, drift_detected, ks_test, psi_value, chi_square_test (optional), reference_stats, current_stats
• DriftDetectionResponse - Fields: model_id, drift_detected, drift_score, severity, features_with_drift (list of FeatureDriftResult), overall_recommendation, checked_at, next_check_at
• DriftHistoryResponse - Fields: model_id, history (list of time-series points), feature_trends (dict), summary_statistics
• DriftAlertResponse - Fields: alert_id, model_id, alert_type, severity, message, created_at, acknowledged, metadata
• CheckDriftRequest - Fields: sample_data (optional list of dicts), force_check (bool)

6. Integration with Model Training

Update file:apps/backend/app/services/model_service.py:

After model training completes:

• Call DriftDetectionService.store_training_statistics() to save baseline
• Store training data statistics in TrainingDataStatistics document
• Link to MLModel via model_id

In model deployment flow:

• Verify training statistics exist before allowing deployment
• Initialize drift monitoring schedule

7. Automatic Drift Monitoring

Create file:apps/backend/app/services/drift_scheduler.py:

Background task scheduler:

• Use asyncio tasks or celery for periodic drift checks
• Check all deployed models every N hours (configurable)
• For each model: fetch recent predictions, run drift detection, store results
• Publish alerts to Redis pub/sub if drift detected

Integration points:

• Start scheduler in file:apps/backend/app/main.py lifespan startup
• Subscribe to model deployment events to add models to monitoring
• Subscribe to model deactivation events to remove from monitoring

8. Alert and Notification System

Create file:apps/backend/app/services/alert_service.py:

Alert management:

• create_drift_alert(model_id, drift_result) - Create alert in database
• publish_alert_notification(alert) - Publish to Redis pub/sub channel "drift_alerts"
• get_active_alerts(user_id) - Get unacknowledged alerts for user
• acknowledge_alert(alert_id) - Mark alert as acknowledged

Redis pub/sub integration:

• Use existing Redis client from file:apps/backend/app/api/routes/production.py (lines 22-29)
• Publish to channel: drift_alerts:{user_id}
• Message format: JSON with alert details

Future extensibility:

• Add webhook support for external notifications
• Add email notifications using SMTP
• Add Slack/Teams integration

9. Frontend Dashboard Components

Create frontend components in apps/frontend/:

Drift Dashboard Page (app/monitoring/drift/page.tsx):

• Display list of deployed models with drift status
• Color-coded indicators (green/yellow/red for no/moderate/high drift)
• Summary cards: models monitored, drift detected, alerts pending
• Quick actions: trigger manual check, view details

Model Drift Detail Page (app/monitoring/drift/[modelId]/page.tsx):

• Overall drift score gauge visualization
• Time series chart showing drift score over time
• Per-feature drift breakdown table with statistical test results
• Distribution comparison charts (reference vs current) using recharts or plotly
• Alert history timeline
• Action buttons: acknowledge alerts, trigger retraining

Drift Visualization Components (components/monitoring/):

• DriftScoreGauge.tsx - Circular gauge showing drift score (0-1)
• DriftHistoryChart.tsx - Line chart with drift score over time
• FeatureDriftTable.tsx - Table with per-feature drift metrics
• DistributionComparisonChart.tsx - Overlaid histograms or KDE plots
• DriftAlertBanner.tsx - Alert notification banner

API Integration (lib/services/drift.ts):

• getDriftStatus(modelId) - Fetch current drift status
• checkDrift(modelId, sampleData?) - Trigger manual drift check
• getDriftHistory(modelId, days) - Fetch historical data
• getFeatureDrift(modelId) - Get per-feature analysis
• acknowledgeAlert(alertId) - Acknowledge alert

10. Automatic Retraining Triggers

Create file:apps/backend/app/services/retraining_service.py:

Retraining logic:

• should_trigger_retraining(drift_result, model_config) - Decision logic
  • Check drift score against threshold (configurable per model)
  • Check number of features with drift
  • Check severity level
  • Return boolean and reason
• schedule_retraining(model_id, reason) - Create retraining job
  • Store in database with status "pending"
  • Optionally trigger automatic retraining workflow
  • Notify user via alerts

Integration with drift detection:

• Call after drift detection completes
• If retraining triggered, create alert with type "retraining_recommended"
• Update model metadata with retraining recommendation

11. Configuration and Settings

Add drift detection configuration to file:apps/backend/app/config.py:

Settings:

• DRIFT_CHECK_INTERVAL_HOURS (default: 24)
• DRIFT_DETECTION_THRESHOLD (default: 0.05 for p-value)
• PSI_THRESHOLD_MODERATE (default: 0.1)
• PSI_THRESHOLD_HIGH (default: 0.25)
• MIN_SAMPLES_FOR_DRIFT_CHECK (default: 100)
• DRIFT_ALERT_RETENTION_DAYS (default: 90)
• AUTO_RETRAINING_ENABLED (default: false)
• RETRAINING_DRIFT_THRESHOLD (default: 0.7)

12. Testing

Create comprehensive tests:

Unit tests (apps/backend/tests/test_services/test_drift_detection.py):

• Test PSI calculation with known distributions
• Test KS test with synthetic data (drifted vs non-drifted)
• Test chi-square test for categorical features
• Test drift score aggregation
• Test alert creation logic

Integration tests (apps/backend/tests/test_api/test_drift_monitoring.py):

• Test drift detection endpoints
• Test drift history retrieval
• Test alert management
• Test automatic drift checking workflow

E2E tests (apps/frontend/e2e/drift-monitoring.spec.ts):

• Test drift dashboard rendering
• Test manual drift check trigger
• Test alert acknowledgment
• Test distribution visualization

13. Documentation

Update documentation:

API Documentation:

• Add drift detection endpoints to OpenAPI spec
• Include request/response examples
• Document statistical tests and thresholds

User Documentation:

• Create guide on interpreting drift metrics
• Explain when to retrain models
• Document alert configuration

Architecture Diagram

sequenceDiagram
    participant Client as Frontend Client
    participant API as Monitoring API
    participant DriftService as DriftDetectionService
    participant PredLog as PredictionLog
    participant DB as MongoDB
    participant Redis as Redis Pub/Sub
    participant Alert as AlertService

    Client->>API: GET /monitoring/models/{id}/drift
    API->>DriftService: detect_drift_for_model(model_id)
    DriftService->>DB: Load TrainingDataStatistics
    DriftService->>PredLog: get_recent_data_for_drift(model_id)
    PredLog-->>DriftService: Recent prediction data
    
    DriftService->>DriftService: Calculate KS test per feature
    DriftService->>DriftService: Calculate PSI per feature
    DriftService->>DriftService: Calculate drift score
    
    DriftService->>DB: Store DriftMetrics
    DriftService->>DB: Store FeatureDriftHistory
    
    alt Drift Detected
        DriftService->>Alert: create_drift_alert(model_id, result)
        Alert->>DB: Store DriftAlert
        Alert->>Redis: Publish to drift_alerts channel
        Redis-->>Client: WebSocket notification
    end
    
    DriftService-->>API: Drift detection result
    API-->>Client: DriftDetectionResponse
    
    Client->>API: GET /monitoring/models/{id}/drift/history
    API->>DB: Query DriftMetrics (last 30 days)
    DB-->>API: Historical drift data
    API-->>Client: DriftHistoryResponse

Loading

Data Flow

graph TD
    A[Model Training] -->|Store baseline| B[TrainingDataStatistics]
    C[Production Predictions] -->|Log data| D[PredictionLog]
    E[Drift Scheduler] -->|Periodic check| F[DriftDetectionService]
    D -->|Recent data| F
    B -->|Reference stats| F
    F -->|Statistical tests| G[KS Test / PSI / Chi-square]
    G -->|Results| H[DriftMetrics DB]
    G -->|Per-feature| I[FeatureDriftHistory DB]
    F -->|If drift detected| J[DriftAlert DB]
    J -->|Publish| K[Redis Pub/Sub]
    K -->|Notify| L[Frontend Dashboard]
    F -->|High drift| M[RetrainingService]
    M -->|Schedule| N[Retraining Job]

Loading

Key Files Summary

File	Purpose
file:apps/backend/app/models/drift_detection.py	Database models for drift tracking
file:apps/backend/app/services/drift_detection_service.py	Core drift detection logic with statistical tests
file:apps/backend/app/services/drift_scheduler.py	Background scheduler for automatic drift checks
file:apps/backend/app/services/alert_service.py	Alert management and notifications
file:apps/backend/app/services/retraining_service.py	Automatic retraining trigger logic
file:apps/backend/app/api/routes/monitoring.py	API endpoints (extend existing)
file:apps/backend/app/schemas/drift.py	Request/response schemas
file:apps/frontend/app/monitoring/drift/page.tsx	Drift dashboard UI
file:apps/frontend/lib/services/drift.ts	Frontend API client

Import In IDE

🤖 Prompt for AI Agents

## Key Observations

The codebase has a solid foundation for implementing drift detection: MongoDB with Beanie ODM for data storage, FastAPI for APIs, existing monitoring infrastructure with placeholder drift detection in `file:apps/backend/app/services/prediction_monitoring.py` (lines 187-204), comprehensive statistical capabilities in `file:apps/backend/app/services/data_processing/statistics_engine.py`, and all required dependencies (scipy, numpy, pandas, statsmodels, redis). The production API serves models via `/production/v1/models/{model_id}/predict` and monitoring endpoints exist at `/monitoring/models/{model_id}/drift` but need full implementation.

## Approach

Implement a comprehensive drift detection system by creating a dedicated `DriftDetectionService` that leverages scipy.stats for statistical tests (Kolmogorov-Smirnov, Population Stability Index, Chi-square). Store drift history in MongoDB using new Beanie document models. Extend existing monitoring endpoints to support drift checking and historical tracking. Integrate drift detection into the prediction pipeline to automatically monitor incoming data. Use Redis pub/sub for drift alerts and create frontend dashboard components for visualization. The implementation follows the existing service-oriented architecture pattern and integrates seamlessly with the current monitoring infrastructure.

## Implementation Steps

### 1. Database Models for Drift Tracking

Create new Beanie document models in `file:apps/backend/app/models/`:

**Create `drift_detection.py`:**
- `DriftMetrics` document to store drift detection results per model
  - Fields: `model_id`, `user_id`, `checked_at`, `drift_detected`, `drift_score`, `statistical_tests` (dict with KS test, PSI, chi-square results per feature), `features_with_drift` (list), `severity` (low/medium/high), `recommendation`
  - Indexes on `model_id`, `user_id`, `checked_at`
- `FeatureDriftHistory` document for per-feature drift tracking
  - Fields: `model_id`, `feature_name`, `timestamp`, `reference_distribution` (dict with mean, std, quantiles), `current_distribution`, `ks_statistic`, `ks_pvalue`, `psi_value`, `drift_detected`
  - Indexes on `model_id`, `feature_name`, `timestamp`
- `DriftAlert` document for alert management
  - Fields: `model_id`, `user_id`, `alert_type` (drift_detected/threshold_exceeded), `severity`, `message`, `created_at`, `acknowledged`, `acknowledged_at`, `metadata`
  - Indexes on `model_id`, `user_id`, `acknowledged`, `created_at`
- `TrainingDataStatistics` document to store baseline statistics
  - Fields: `model_id`, `feature_statistics` (dict mapping feature names to statistical summaries), `created_at`, `updated_at`
  - Index on `model_id`

Register these models in `file:apps/backend/app/main.py` lifespan function (line 97-119).

### 2. Drift Detection Service

Create `file:apps/backend/app/services/drift_detection_service.py`:

**Core Methods:**
- `calculate_psi(expected_distribution, actual_distribution, bins=10)` - Calculate Population Stability Index
  - Bin both distributions, calculate PSI = Σ((actual% - expected%) × ln(actual%/expected%))
  - Return PSI value and interpretation (< 0.1 no drift, 0.1-0.25 moderate, > 0.25 significant)
- `perform_ks_test(reference_data, current_data)` - Kolmogorov-Smirnov test using scipy.stats.ks_2samp
  - Return statistic, p-value, and drift_detected (p-value < 0.05)
- `perform_chi_square_test(reference_counts, current_counts)` - Chi-square test for categorical features using scipy.stats.chisquare
  - Return statistic, p-value, drift_detected
- `calculate_drift_score(feature_results)` - Aggregate drift metrics across features
  - Weight by feature importance if available
  - Return overall drift score (0-1 scale)
- `detect_drift_for_model(model_id, current_data_df, threshold=0.05)` - Main drift detection method
  - Load training statistics from `TrainingDataStatistics`
  - For each feature: run appropriate statistical test (KS for numeric, chi-square for categorical)
  - Calculate PSI for all features
  - Store results in `DriftMetrics` and `FeatureDriftHistory`
  - Create `DriftAlert` if drift detected
  - Return comprehensive drift report
- `store_training_statistics(model_id, training_data_df, feature_names)` - Store baseline statistics
  - Calculate distribution statistics for each feature
  - Save to `TrainingDataStatistics` document
- `get_drift_history(model_id, days=30)` - Retrieve historical drift metrics
  - Query `DriftMetrics` for time range
  - Return time series data for visualization
- `get_feature_drift_trends(model_id, feature_name, days=30)` - Per-feature drift trends
  - Query `FeatureDriftHistory` for specific feature
  - Return trend data with statistical test results

**Statistical Test Implementations:**
- Use `scipy.stats.ks_2samp` for continuous features
- Use `scipy.stats.chisquare` for categorical features
- Implement PSI calculation manually with numpy
- Use `statsmodels` for additional tests if needed

### 3. Extend Prediction Monitoring Service

Update `file:apps/backend/app/services/prediction_monitoring.py`:

**Enhance `PredictionLog` class:**
- Add method `get_recent_data_for_drift(model_id, limit=1000)` to extract feature data from logged predictions
- Convert logged input_data to DataFrame for drift analysis

**Replace placeholder `detect_drift` method (lines 187-204):**
- Call `DriftDetectionService.detect_drift_for_model()`
- Pass recent prediction data from `PredictionLog`
- Return actual drift detection results instead of placeholder

**Add new methods:**
- `schedule_drift_check(model_id, interval_hours=24)` - Schedule periodic drift checks
- `trigger_drift_alert(model_id, drift_result)` - Publish alert to Redis pub/sub channel

### 4. API Endpoints for Drift Detection

Update `file:apps/backend/app/api/routes/monitoring.py`:

**Enhance existing `/monitoring/models/{model_id}/drift` endpoint (lines 136-159):**
- Replace placeholder with actual drift detection
- Add query parameters: `check_now` (bool), `days` (int for historical data)
- If `check_now=true`, run drift detection on recent predictions
- Otherwise, return latest drift check results from database
- Response includes: drift_detected, drift_score, features_with_drift, statistical_tests, severity, recommendation, checked_at

**Add new endpoints:**
- `POST /api/v1/monitoring/models/{model_id}/drift/check` - Manually trigger drift check
  - Accept optional request body with sample data
  - Run drift detection immediately
  - Return detailed results with per-feature breakdown
- `GET /api/v1/monitoring/models/{model_id}/drift/history` - Get drift history
  - Query parameters: `days` (default 30), `feature_name` (optional)
  - Return time series of drift metrics
  - Include aggregated statistics
- `GET /api/v1/monitoring/models/{model_id}/drift/features` - Per-feature drift analysis
  - Return drift status for each feature
  - Include distribution comparisons (reference vs current)
  - Statistical test results per feature
- `GET /api/v1/monitoring/models/{model_id}/alerts` - Get drift alerts
  - Query parameters: `acknowledged` (bool), `severity` (low/medium/high)
  - Return list of alerts with pagination
- `PUT /api/v1/monitoring/models/{model_id}/alerts/{alert_id}/acknowledge` - Acknowledge alert
  - Mark alert as acknowledged
  - Record acknowledgment timestamp

**Add deployment-specific endpoints:**
- `GET /api/v1/deployments/{deployment_id}/drift` - Alias for model drift endpoint
- `POST /api/v1/deployments/{deployment_id}/check-drift` - Alias for manual drift check

### 5. Response Schemas

Create `file:apps/backend/app/schemas/drift.py`:

**Pydantic models:**
- `StatisticalTestResult` - Fields: test_name, statistic, p_value, drift_detected, interpretation
- `FeatureDriftResult` - Fields: feature_name, drift_detected, ks_test, psi_value, chi_square_test (optional), reference_stats, current_stats
- `DriftDetectionResponse` - Fields: model_id, drift_detected, drift_score, severity, features_with_drift (list of FeatureDriftResult), overall_recommendation, checked_at, next_check_at
- `DriftHistoryResponse` - Fields: model_id, history (list of time-series points), feature_trends (dict), summary_statistics
- `DriftAlertResponse` - Fields: alert_id, model_id, alert_type, severity, message, created_at, acknowledged, metadata
- `CheckDriftRequest` - Fields: sample_data (optional list of dicts), force_check (bool)

### 6. Integration with Model Training

Update `file:apps/backend/app/services/model_service.py`:

**After model training completes:**
- Call `DriftDetectionService.store_training_statistics()` to save baseline
- Store training data statistics in `TrainingDataStatistics` document
- Link to `MLModel` via model_id

**In model deployment flow:**
- Verify training statistics exist before allowing deployment
- Initialize drift monitoring schedule

### 7. Automatic Drift Monitoring

Create `file:apps/backend/app/services/drift_scheduler.py`:

**Background task scheduler:**
- Use asyncio tasks or celery for periodic drift checks
- Check all deployed models every N hours (configurable)
- For each model: fetch recent predictions, run drift detection, store results
- Publish alerts to Redis pub/sub if drift detected

**Integration points:**
- Start scheduler in `file:apps/backend/app/main.py` lifespan startup
- Subscribe to model deployment events to add models to monitoring
- Subscribe to model deactivation events to remove from monitoring

### 8. Alert and Notification System

Create `file:apps/backend/app/services/alert_service.py`:

**Alert management:**
- `create_drift_alert(model_id, drift_result)` - Create alert in database
- `publish_alert_notification(alert)` - Publish to Redis pub/sub channel "drift_alerts"
- `get_active_alerts(user_id)` - Get unacknowledged alerts for user
- `acknowledge_alert(alert_id)` - Mark alert as acknowledged

**Redis pub/sub integration:**
- Use existing Redis client from `file:apps/backend/app/api/routes/production.py` (lines 22-29)
- Publish to channel: `drift_alerts:{user_id}`
- Message format: JSON with alert details

**Future extensibility:**
- Add webhook support for external notifications
- Add email notifications using SMTP
- Add Slack/Teams integration

### 9. Frontend Dashboard Components

Create frontend components in `apps/frontend/`:

**Drift Dashboard Page (`app/monitoring/drift/page.tsx`):**
- Display list of deployed models with drift status
- Color-coded indicators (green/yellow/red for no/moderate/high drift)
- Summary cards: models monitored, drift detected, alerts pending
- Quick actions: trigger manual check, view details

**Model Drift Detail Page (`app/monitoring/drift/[modelId]/page.tsx`):**
- Overall drift score gauge visualization
- Time series chart showing drift score over time
- Per-feature drift breakdown table with statistical test results
- Distribution comparison charts (reference vs current) using recharts or plotly
- Alert history timeline
- Action buttons: acknowledge alerts, trigger retraining

**Drift Visualization Components (`components/monitoring/`):**
- `DriftScoreGauge.tsx` - Circular gauge showing drift score (0-1)
- `DriftHistoryChart.tsx` - Line chart with drift score over time
- `FeatureDriftTable.tsx` - Table with per-feature drift metrics
- `DistributionComparisonChart.tsx` - Overlaid histograms or KDE plots
- `DriftAlertBanner.tsx` - Alert notification banner

**API Integration (`lib/services/drift.ts`):**
- `getDriftStatus(modelId)` - Fetch current drift status
- `checkDrift(modelId, sampleData?)` - Trigger manual drift check
- `getDriftHistory(modelId, days)` - Fetch historical data
- `getFeatureDrift(modelId)` - Get per-feature analysis
- `acknowledgeAlert(alertId)` - Acknowledge alert

### 10. Automatic Retraining Triggers

Create `file:apps/backend/app/services/retraining_service.py`:

**Retraining logic:**
- `should_trigger_retraining(drift_result, model_config)` - Decision logic
  - Check drift score against threshold (configurable per model)
  - Check number of features with drift
  - Check severity level
  - Return boolean and reason
- `schedule_retraining(model_id, reason)` - Create retraining job
  - Store in database with status "pending"
  - Optionally trigger automatic retraining workflow
  - Notify user via alerts

**Integration with drift detection:**
- Call after drift detection completes
- If retraining triggered, create alert with type "retraining_recommended"
- Update model metadata with retraining recommendation

### 11. Configuration and Settings

Add drift detection configuration to `file:apps/backend/app/config.py`:

**Settings:**
- `DRIFT_CHECK_INTERVAL_HOURS` (default: 24)
- `DRIFT_DETECTION_THRESHOLD` (default: 0.05 for p-value)
- `PSI_THRESHOLD_MODERATE` (default: 0.1)
- `PSI_THRESHOLD_HIGH` (default: 0.25)
- `MIN_SAMPLES_FOR_DRIFT_CHECK` (default: 100)
- `DRIFT_ALERT_RETENTION_DAYS` (default: 90)
- `AUTO_RETRAINING_ENABLED` (default: false)
- `RETRAINING_DRIFT_THRESHOLD` (default: 0.7)

### 12. Testing

Create comprehensive tests:

**Unit tests (`apps/backend/tests/test_services/test_drift_detection.py`):**
- Test PSI calculation with known distributions
- Test KS test with synthetic data (drifted vs non-drifted)
- Test chi-square test for categorical features
- Test drift score aggregation
- Test alert creation logic

**Integration tests (`apps/backend/tests/test_api/test_drift_monitoring.py`):**
- Test drift detection endpoints
- Test drift history retrieval
- Test alert management
- Test automatic drift checking workflow

**E2E tests (`apps/frontend/e2e/drift-monitoring.spec.ts`):**
- Test drift dashboard rendering
- Test manual drift check trigger
- Test alert acknowledgment
- Test distribution visualization

### 13. Documentation

Update documentation:

**API Documentation:**
- Add drift detection endpoints to OpenAPI spec
- Include request/response examples
- Document statistical tests and thresholds

**User Documentation:**
- Create guide on interpreting drift metrics
- Explain when to retrain models
- Document alert configuration

## Architecture Diagram

```mermaid
sequenceDiagram
    participant Client as Frontend Client
    participant API as Monitoring API
    participant DriftService as DriftDetectionService
    participant PredLog as PredictionLog
    participant DB as MongoDB
    participant Redis as Redis Pub/Sub
    participant Alert as AlertService

    Client->>API: GET /monitoring/models/{id}/drift
    API->>DriftService: detect_drift_for_model(model_id)
    DriftService->>DB: Load TrainingDataStatistics
    DriftService->>PredLog: get_recent_data_for_drift(model_id)
    PredLog-->>DriftService: Recent prediction data
    
    DriftService->>DriftService: Calculate KS test per feature
    DriftService->>DriftService: Calculate PSI per feature
    DriftService->>DriftService: Calculate drift score
    
    DriftService->>DB: Store DriftMetrics
    DriftService->>DB: Store FeatureDriftHistory
    
    alt Drift Detected
        DriftService->>Alert: create_drift_alert(model_id, result)
        Alert->>DB: Store DriftAlert
        Alert->>Redis: Publish to drift_alerts channel
        Redis-->>Client: WebSocket notification
    end
    
    DriftService-->>API: Drift detection result
    API-->>Client: DriftDetectionResponse
    
    Client->>API: GET /monitoring/models/{id}/drift/history
    API->>DB: Query DriftMetrics (last 30 days)
    DB-->>API: Historical drift data
    API-->>Client: DriftHistoryResponse
```

## Data Flow

```mermaid
graph TD
    A[Model Training] -->|Store baseline| B[TrainingDataStatistics]
    C[Production Predictions] -->|Log data| D[PredictionLog]
    E[Drift Scheduler] -->|Periodic check| F[DriftDetectionService]
    D -->|Recent data| F
    B -->|Reference stats| F
    F -->|Statistical tests| G[KS Test / PSI / Chi-square]
    G -->|Results| H[DriftMetrics DB]
    G -->|Per-feature| I[FeatureDriftHistory DB]
    F -->|If drift detected| J[DriftAlert DB]
    J -->|Publish| K[Redis Pub/Sub]
    K -->|Notify| L[Frontend Dashboard]
    F -->|High drift| M[RetrainingService]
    M -->|Schedule| N[Retraining Job]
```

## Key Files Summary

| File | Purpose |
|------|---------|
| `file:apps/backend/app/models/drift_detection.py` | Database models for drift tracking |
| `file:apps/backend/app/services/drift_detection_service.py` | Core drift detection logic with statistical tests |
| `file:apps/backend/app/services/drift_scheduler.py` | Background scheduler for automatic drift checks |
| `file:apps/backend/app/services/alert_service.py` | Alert management and notifications |
| `file:apps/backend/app/services/retraining_service.py` | Automatic retraining trigger logic |
| `file:apps/backend/app/api/routes/monitoring.py` | API endpoints (extend existing) |
| `file:apps/backend/app/schemas/drift.py` | Request/response schemas |
| `file:apps/frontend/app/monitoring/drift/page.tsx` | Drift dashboard UI |
| `file:apps/frontend/lib/services/drift.ts` | Frontend API client |

---

Execution Information

Branch: main
Commit: 1963a18

---

💡 Tips

Supported Commands (Inside Comments)

• Use @traycerai generate to iterate on the previous version of the implementation plan.

Supported Commands (Inside Description)

• Add @traycerai ignore anywhere in the ticket description to prevent this ticket from being processed.
• Add @traycerai branch:<branch-name> anywhere in the ticket description to specify the target branch for the implementation plan.

Community

• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[coderabbitai]

📝 CodeRabbit Plan Mode

Generate an implementation plan and prompts that you can use with your favorite coding agent.

• Create Plan

Examples

• Example 1
• Example 2

_{You can configure auto-planning by selecting labels in the issue_enrichment configuration.}

🔗 Similar Issues

Related Issues

• [HIGH] Implement Deployment Monitoring Dashboard #85
• [BLOCKER] Implement Error Analysis with Pattern Detection #81
• [BLOCKER] Implement Comprehensive Model Evaluation Dashboard with Performance Metrics #79
• [BLOCKER] Implement AI-Powered Data Issue Detection and Suggested Fixes #66
• [HIGH] Implement Prediction Confidence Scores and Explanations #83

👤 Suggested Assignees

• frankbria

---

🧪 Issue enrichment is currently in open beta.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord or schedule a call!

[traycerai]

I have updated the Implementation Plan to reflect the recent changes in the ticket.

New Plan

Key Observations

The codebase has a solid foundation for implementing drift detection: MongoDB with Beanie ODM for data storage, FastAPI for APIs, existing monitoring infrastructure with placeholder drift detection in file:apps/backend/app/services/prediction_monitoring.py (lines 187-204), comprehensive statistical capabilities in file:apps/backend/app/services/data_processing/statistics_engine.py, and all required dependencies (scipy, numpy, pandas, statsmodels, redis). The production API serves models via /production/v1/models/{model_id}/predict and monitoring endpoints exist at /monitoring/models/{model_id}/drift but need full implementation.

Approach

Implement a comprehensive drift detection system by creating a dedicated DriftDetectionService that leverages scipy.stats for statistical tests (Kolmogorov-Smirnov, Population Stability Index, Chi-square). Store drift history in MongoDB using new Beanie document models. Extend existing monitoring endpoints to support drift checking and historical tracking. Integrate drift detection into the prediction pipeline to automatically monitor incoming data. Use Redis pub/sub for drift alerts and create frontend dashboard components for visualization. The implementation follows the existing service-oriented architecture pattern and integrates seamlessly with the current monitoring infrastructure.

Implementation Steps

1. Database Models for Drift Tracking

Create new Beanie document models in file:apps/backend/app/models/:

Create drift_detection.py:

• DriftMetrics document to store drift detection results per model
  • Fields: model_id, user_id, checked_at, drift_detected, drift_score, statistical_tests (dict with KS test, PSI, chi-square results per feature), features_with_drift (list), severity (low/medium/high), recommendation
  • Indexes on model_id, user_id, checked_at
• FeatureDriftHistory document for per-feature drift tracking
  • Fields: model_id, feature_name, timestamp, reference_distribution (dict with mean, std, quantiles), current_distribution, ks_statistic, ks_pvalue, psi_value, drift_detected
  • Indexes on model_id, feature_name, timestamp
• DriftAlert document for alert management
  • Fields: model_id, user_id, alert_type (drift_detected/threshold_exceeded), severity, message, created_at, acknowledged, acknowledged_at, metadata
  • Indexes on model_id, user_id, acknowledged, created_at
• TrainingDataStatistics document to store baseline statistics
  • Fields: model_id, feature_statistics (dict mapping feature names to statistical summaries), created_at, updated_at
  • Index on model_id

Register these models in file:apps/backend/app/main.py lifespan function (line 97-119).

2. Drift Detection Service

Create file:apps/backend/app/services/drift_detection_service.py:

Core Methods:

• calculate_psi(expected_distribution, actual_distribution, bins=10) - Calculate Population Stability Index
  • Bin both distributions, calculate PSI = Σ((actual% - expected%) × ln(actual%/expected%))
  • Return PSI value and interpretation (< 0.1 no drift, 0.1-0.25 moderate, > 0.25 significant)
• perform_ks_test(reference_data, current_data) - Kolmogorov-Smirnov test using scipy.stats.ks_2samp
  • Return statistic, p-value, and drift_detected (p-value < 0.05)
• perform_chi_square_test(reference_counts, current_counts) - Chi-square test for categorical features using scipy.stats.chisquare
  • Return statistic, p-value, drift_detected
• calculate_drift_score(feature_results) - Aggregate drift metrics across features
  • Weight by feature importance if available
  • Return overall drift score (0-1 scale)
• detect_drift_for_model(model_id, current_data_df, threshold=0.05) - Main drift detection method
  • Load training statistics from TrainingDataStatistics
  • For each feature: run appropriate statistical test (KS for numeric, chi-square for categorical)
  • Calculate PSI for all features
  • Store results in DriftMetrics and FeatureDriftHistory
  • Create DriftAlert if drift detected
  • Return comprehensive drift report
• store_training_statistics(model_id, training_data_df, feature_names) - Store baseline statistics
  • Calculate distribution statistics for each feature
  • Save to TrainingDataStatistics document
• get_drift_history(model_id, days=30) - Retrieve historical drift metrics
  • Query DriftMetrics for time range
  • Return time series data for visualization
• get_feature_drift_trends(model_id, feature_name, days=30) - Per-feature drift trends
  • Query FeatureDriftHistory for specific feature
  • Return trend data with statistical test results

Statistical Test Implementations:

• Use scipy.stats.ks_2samp for continuous features
• Use scipy.stats.chisquare for categorical features
• Implement PSI calculation manually with numpy
• Use statsmodels for additional tests if needed

3. Extend Prediction Monitoring Service

Update file:apps/backend/app/services/prediction_monitoring.py:

Enhance PredictionLog class:

• Add method get_recent_data_for_drift(model_id, limit=1000) to extract feature data from logged predictions
• Convert logged input_data to DataFrame for drift analysis

Replace placeholder detect_drift method (lines 187-204):

• Call DriftDetectionService.detect_drift_for_model()
• Pass recent prediction data from PredictionLog
• Return actual drift detection results instead of placeholder

Add new methods:

• schedule_drift_check(model_id, interval_hours=24) - Schedule periodic drift checks
• trigger_drift_alert(model_id, drift_result) - Publish alert to Redis pub/sub channel

4. API Endpoints for Drift Detection

Update file:apps/backend/app/api/routes/monitoring.py:

Enhance existing /monitoring/models/{model_id}/drift endpoint (lines 136-159):

• Replace placeholder with actual drift detection
• Add query parameters: check_now (bool), days (int for historical data)
• If check_now=true, run drift detection on recent predictions
• Otherwise, return latest drift check results from database
• Response includes: drift_detected, drift_score, features_with_drift, statistical_tests, severity, recommendation, checked_at

Add new endpoints:

• POST /api/v1/monitoring/models/{model_id}/drift/check - Manually trigger drift check
  • Accept optional request body with sample data
  • Run drift detection immediately
  • Return detailed results with per-feature breakdown
• GET /api/v1/monitoring/models/{model_id}/drift/history - Get drift history
  • Query parameters: days (default 30), feature_name (optional)
  • Return time series of drift metrics
  • Include aggregated statistics
• GET /api/v1/monitoring/models/{model_id}/drift/features - Per-feature drift analysis
  • Return drift status for each feature
  • Include distribution comparisons (reference vs current)
  • Statistical test results per feature
• GET /api/v1/monitoring/models/{model_id}/alerts - Get drift alerts
  • Query parameters: acknowledged (bool), severity (low/medium/high)
  • Return list of alerts with pagination
• PUT /api/v1/monitoring/models/{model_id}/alerts/{alert_id}/acknowledge - Acknowledge alert
  • Mark alert as acknowledged
  • Record acknowledgment timestamp

Add deployment-specific endpoints:

• GET /api/v1/deployments/{deployment_id}/drift - Alias for model drift endpoint
• POST /api/v1/deployments/{deployment_id}/check-drift - Alias for manual drift check

5. Response Schemas

Create file:apps/backend/app/schemas/drift.py:

Pydantic models:

• StatisticalTestResult - Fields: test_name, statistic, p_value, drift_detected, interpretation
• FeatureDriftResult - Fields: feature_name, drift_detected, ks_test, psi_value, chi_square_test (optional), reference_stats, current_stats
• DriftDetectionResponse - Fields: model_id, drift_detected, drift_score, severity, features_with_drift (list of FeatureDriftResult), overall_recommendation, checked_at, next_check_at
• DriftHistoryResponse - Fields: model_id, history (list of time-series points), feature_trends (dict), summary_statistics
• DriftAlertResponse - Fields: alert_id, model_id, alert_type, severity, message, created_at, acknowledged, metadata
• CheckDriftRequest - Fields: sample_data (optional list of dicts), force_check (bool)

6. Integration with Model Training

Update file:apps/backend/app/services/model_service.py:

After model training completes:

• Call DriftDetectionService.store_training_statistics() to save baseline
• Store training data statistics in TrainingDataStatistics document
• Link to MLModel via model_id

In model deployment flow:

• Verify training statistics exist before allowing deployment
• Initialize drift monitoring schedule

7. Automatic Drift Monitoring

Create file:apps/backend/app/services/drift_scheduler.py:

Background task scheduler:

• Use asyncio tasks or celery for periodic drift checks
• Check all deployed models every N hours (configurable)
• For each model: fetch recent predictions, run drift detection, store results
• Publish alerts to Redis pub/sub if drift detected

Integration points:

• Start scheduler in file:apps/backend/app/main.py lifespan startup
• Subscribe to model deployment events to add models to monitoring
• Subscribe to model deactivation events to remove from monitoring

8. Alert and Notification System

Create file:apps/backend/app/services/alert_service.py:

Alert management:

• create_drift_alert(model_id, drift_result) - Create alert in database
• publish_alert_notification(alert) - Publish to Redis pub/sub channel "drift_alerts"
• get_active_alerts(user_id) - Get unacknowledged alerts for user
• acknowledge_alert(alert_id) - Mark alert as acknowledged

Redis pub/sub integration:

• Use existing Redis client from file:apps/backend/app/api/routes/production.py (lines 22-29)
• Publish to channel: drift_alerts:{user_id}
• Message format: JSON with alert details

Future extensibility:

• Add webhook support for external notifications
• Add email notifications using SMTP
• Add Slack/Teams integration

9. Frontend Dashboard Components

Create frontend components in apps/frontend/:

Drift Dashboard Page (app/monitoring/drift/page.tsx):

• Display list of deployed models with drift status
• Color-coded indicators (green/yellow/red for no/moderate/high drift)
• Summary cards: models monitored, drift detected, alerts pending
• Quick actions: trigger manual check, view details

Model Drift Detail Page (app/monitoring/drift/[modelId]/page.tsx):

• Overall drift score gauge visualization
• Time series chart showing drift score over time
• Per-feature drift breakdown table with statistical test results
• Distribution comparison charts (reference vs current) using recharts or plotly
• Alert history timeline
• Action buttons: acknowledge alerts, trigger retraining

Drift Visualization Components (components/monitoring/):

• DriftScoreGauge.tsx - Circular gauge showing drift score (0-1)
• DriftHistoryChart.tsx - Line chart with drift score over time
• FeatureDriftTable.tsx - Table with per-feature drift metrics
• DistributionComparisonChart.tsx - Overlaid histograms or KDE plots
• DriftAlertBanner.tsx - Alert notification banner

API Integration (lib/services/drift.ts):

• getDriftStatus(modelId) - Fetch current drift status
• checkDrift(modelId, sampleData?) - Trigger manual drift check
• getDriftHistory(modelId, days) - Fetch historical data
• getFeatureDrift(modelId) - Get per-feature analysis
• acknowledgeAlert(alertId) - Acknowledge alert

10. Automatic Retraining Triggers

Create file:apps/backend/app/services/retraining_service.py:

Retraining logic:

• should_trigger_retraining(drift_result, model_config) - Decision logic
  • Check drift score against threshold (configurable per model)
  • Check number of features with drift
  • Check severity level
  • Return boolean and reason
• schedule_retraining(model_id, reason) - Create retraining job
  • Store in database with status "pending"
  • Optionally trigger automatic retraining workflow
  • Notify user via alerts

Integration with drift detection:

• Call after drift detection completes
• If retraining triggered, create alert with type "retraining_recommended"
• Update model metadata with retraining recommendation

11. Configuration and Settings

Add drift detection configuration to file:apps/backend/app/config.py:

Settings:

• DRIFT_CHECK_INTERVAL_HOURS (default: 24)
• DRIFT_DETECTION_THRESHOLD (default: 0.05 for p-value)
• PSI_THRESHOLD_MODERATE (default: 0.1)
• PSI_THRESHOLD_HIGH (default: 0.25)
• MIN_SAMPLES_FOR_DRIFT_CHECK (default: 100)
• DRIFT_ALERT_RETENTION_DAYS (default: 90)
• AUTO_RETRAINING_ENABLED (default: false)
• RETRAINING_DRIFT_THRESHOLD (default: 0.7)

12. Testing

Create comprehensive tests:

Unit tests (apps/backend/tests/test_services/test_drift_detection.py):

• Test PSI calculation with known distributions
• Test KS test with synthetic data (drifted vs non-drifted)
• Test chi-square test for categorical features
• Test drift score aggregation
• Test alert creation logic

Integration tests (apps/backend/tests/test_api/test_drift_monitoring.py):

• Test drift detection endpoints
• Test drift history retrieval
• Test alert management
• Test automatic drift checking workflow

E2E tests (apps/frontend/e2e/drift-monitoring.spec.ts):

• Test drift dashboard rendering
• Test manual drift check trigger
• Test alert acknowledgment
• Test distribution visualization

13. Documentation

Update documentation:

API Documentation:

• Add drift detection endpoints to OpenAPI spec
• Include request/response examples
• Document statistical tests and thresholds

User Documentation:

• Create guide on interpreting drift metrics
• Explain when to retrain models
• Document alert configuration

Architecture Diagram

sequenceDiagram
    participant Client as Frontend Client
    participant API as Monitoring API
    participant DriftService as DriftDetectionService
    participant PredLog as PredictionLog
    participant DB as MongoDB
    participant Redis as Redis Pub/Sub
    participant Alert as AlertService

    Client->>API: GET /monitoring/models/{id}/drift
    API->>DriftService: detect_drift_for_model(model_id)
    DriftService->>DB: Load TrainingDataStatistics
    DriftService->>PredLog: get_recent_data_for_drift(model_id)
    PredLog-->>DriftService: Recent prediction data
    
    DriftService->>DriftService: Calculate KS test per feature
    DriftService->>DriftService: Calculate PSI per feature
    DriftService->>DriftService: Calculate drift score
    
    DriftService->>DB: Store DriftMetrics
    DriftService->>DB: Store FeatureDriftHistory
    
    alt Drift Detected
        DriftService->>Alert: create_drift_alert(model_id, result)
        Alert->>DB: Store DriftAlert
        Alert->>Redis: Publish to drift_alerts channel
        Redis-->>Client: WebSocket notification
    end
    
    DriftService-->>API: Drift detection result
    API-->>Client: DriftDetectionResponse
    
    Client->>API: GET /monitoring/models/{id}/drift/history
    API->>DB: Query DriftMetrics (last 30 days)
    DB-->>API: Historical drift data
    API-->>Client: DriftHistoryResponse

Loading

Data Flow

graph TD
    A[Model Training] -->|Store baseline| B[TrainingDataStatistics]
    C[Production Predictions] -->|Log data| D[PredictionLog]
    E[Drift Scheduler] -->|Periodic check| F[DriftDetectionService]
    D -->|Recent data| F
    B -->|Reference stats| F
    F -->|Statistical tests| G[KS Test / PSI / Chi-square]
    G -->|Results| H[DriftMetrics DB]
    G -->|Per-feature| I[FeatureDriftHistory DB]
    F -->|If drift detected| J[DriftAlert DB]
    J -->|Publish| K[Redis Pub/Sub]
    K -->|Notify| L[Frontend Dashboard]
    F -->|High drift| M[RetrainingService]
    M -->|Schedule| N[Retraining Job]

Loading

Key Files Summary

File	Purpose
file:apps/backend/app/models/drift_detection.py	Database models for drift tracking
file:apps/backend/app/services/drift_detection_service.py	Core drift detection logic with statistical tests
file:apps/backend/app/services/drift_scheduler.py	Background scheduler for automatic drift checks
file:apps/backend/app/services/alert_service.py	Alert management and notifications
file:apps/backend/app/services/retraining_service.py	Automatic retraining trigger logic
file:apps/backend/app/api/routes/monitoring.py	API endpoints (extend existing)
file:apps/backend/app/schemas/drift.py	Request/response schemas
file:apps/frontend/app/monitoring/drift/page.tsx	Drift dashboard UI
file:apps/frontend/lib/services/drift.ts	Frontend API client

Import In IDE

🤖 Prompt for AI Agents

## Key Observations

The codebase has a solid foundation for implementing drift detection: MongoDB with Beanie ODM for data storage, FastAPI for APIs, existing monitoring infrastructure with placeholder drift detection in `file:apps/backend/app/services/prediction_monitoring.py` (lines 187-204), comprehensive statistical capabilities in `file:apps/backend/app/services/data_processing/statistics_engine.py`, and all required dependencies (scipy, numpy, pandas, statsmodels, redis). The production API serves models via `/production/v1/models/{model_id}/predict` and monitoring endpoints exist at `/monitoring/models/{model_id}/drift` but need full implementation.

## Approach

Implement a comprehensive drift detection system by creating a dedicated `DriftDetectionService` that leverages scipy.stats for statistical tests (Kolmogorov-Smirnov, Population Stability Index, Chi-square). Store drift history in MongoDB using new Beanie document models. Extend existing monitoring endpoints to support drift checking and historical tracking. Integrate drift detection into the prediction pipeline to automatically monitor incoming data. Use Redis pub/sub for drift alerts and create frontend dashboard components for visualization. The implementation follows the existing service-oriented architecture pattern and integrates seamlessly with the current monitoring infrastructure.

## Implementation Steps

### 1. Database Models for Drift Tracking

Create new Beanie document models in `file:apps/backend/app/models/`:

**Create `drift_detection.py`:**
- `DriftMetrics` document to store drift detection results per model
  - Fields: `model_id`, `user_id`, `checked_at`, `drift_detected`, `drift_score`, `statistical_tests` (dict with KS test, PSI, chi-square results per feature), `features_with_drift` (list), `severity` (low/medium/high), `recommendation`
  - Indexes on `model_id`, `user_id`, `checked_at`
- `FeatureDriftHistory` document for per-feature drift tracking
  - Fields: `model_id`, `feature_name`, `timestamp`, `reference_distribution` (dict with mean, std, quantiles), `current_distribution`, `ks_statistic`, `ks_pvalue`, `psi_value`, `drift_detected`
  - Indexes on `model_id`, `feature_name`, `timestamp`
- `DriftAlert` document for alert management
  - Fields: `model_id`, `user_id`, `alert_type` (drift_detected/threshold_exceeded), `severity`, `message`, `created_at`, `acknowledged`, `acknowledged_at`, `metadata`
  - Indexes on `model_id`, `user_id`, `acknowledged`, `created_at`
- `TrainingDataStatistics` document to store baseline statistics
  - Fields: `model_id`, `feature_statistics` (dict mapping feature names to statistical summaries), `created_at`, `updated_at`
  - Index on `model_id`

Register these models in `file:apps/backend/app/main.py` lifespan function (line 97-119).

### 2. Drift Detection Service

Create `file:apps/backend/app/services/drift_detection_service.py`:

**Core Methods:**
- `calculate_psi(expected_distribution, actual_distribution, bins=10)` - Calculate Population Stability Index
  - Bin both distributions, calculate PSI = Σ((actual% - expected%) × ln(actual%/expected%))
  - Return PSI value and interpretation (< 0.1 no drift, 0.1-0.25 moderate, > 0.25 significant)
- `perform_ks_test(reference_data, current_data)` - Kolmogorov-Smirnov test using scipy.stats.ks_2samp
  - Return statistic, p-value, and drift_detected (p-value < 0.05)
- `perform_chi_square_test(reference_counts, current_counts)` - Chi-square test for categorical features using scipy.stats.chisquare
  - Return statistic, p-value, drift_detected
- `calculate_drift_score(feature_results)` - Aggregate drift metrics across features
  - Weight by feature importance if available
  - Return overall drift score (0-1 scale)
- `detect_drift_for_model(model_id, current_data_df, threshold=0.05)` - Main drift detection method
  - Load training statistics from `TrainingDataStatistics`
  - For each feature: run appropriate statistical test (KS for numeric, chi-square for categorical)
  - Calculate PSI for all features
  - Store results in `DriftMetrics` and `FeatureDriftHistory`
  - Create `DriftAlert` if drift detected
  - Return comprehensive drift report
- `store_training_statistics(model_id, training_data_df, feature_names)` - Store baseline statistics
  - Calculate distribution statistics for each feature
  - Save to `TrainingDataStatistics` document
- `get_drift_history(model_id, days=30)` - Retrieve historical drift metrics
  - Query `DriftMetrics` for time range
  - Return time series data for visualization
- `get_feature_drift_trends(model_id, feature_name, days=30)` - Per-feature drift trends
  - Query `FeatureDriftHistory` for specific feature
  - Return trend data with statistical test results

**Statistical Test Implementations:**
- Use `scipy.stats.ks_2samp` for continuous features
- Use `scipy.stats.chisquare` for categorical features
- Implement PSI calculation manually with numpy
- Use `statsmodels` for additional tests if needed

### 3. Extend Prediction Monitoring Service

Update `file:apps/backend/app/services/prediction_monitoring.py`:

**Enhance `PredictionLog` class:**
- Add method `get_recent_data_for_drift(model_id, limit=1000)` to extract feature data from logged predictions
- Convert logged input_data to DataFrame for drift analysis

**Replace placeholder `detect_drift` method (lines 187-204):**
- Call `DriftDetectionService.detect_drift_for_model()`
- Pass recent prediction data from `PredictionLog`
- Return actual drift detection results instead of placeholder

**Add new methods:**
- `schedule_drift_check(model_id, interval_hours=24)` - Schedule periodic drift checks
- `trigger_drift_alert(model_id, drift_result)` - Publish alert to Redis pub/sub channel

### 4. API Endpoints for Drift Detection

Update `file:apps/backend/app/api/routes/monitoring.py`:

**Enhance existing `/monitoring/models/{model_id}/drift` endpoint (lines 136-159):**
- Replace placeholder with actual drift detection
- Add query parameters: `check_now` (bool), `days` (int for historical data)
- If `check_now=true`, run drift detection on recent predictions
- Otherwise, return latest drift check results from database
- Response includes: drift_detected, drift_score, features_with_drift, statistical_tests, severity, recommendation, checked_at

**Add new endpoints:**
- `POST /api/v1/monitoring/models/{model_id}/drift/check` - Manually trigger drift check
  - Accept optional request body with sample data
  - Run drift detection immediately
  - Return detailed results with per-feature breakdown
- `GET /api/v1/monitoring/models/{model_id}/drift/history` - Get drift history
  - Query parameters: `days` (default 30), `feature_name` (optional)
  - Return time series of drift metrics
  - Include aggregated statistics
- `GET /api/v1/monitoring/models/{model_id}/drift/features` - Per-feature drift analysis
  - Return drift status for each feature
  - Include distribution comparisons (reference vs current)
  - Statistical test results per feature
- `GET /api/v1/monitoring/models/{model_id}/alerts` - Get drift alerts
  - Query parameters: `acknowledged` (bool), `severity` (low/medium/high)
  - Return list of alerts with pagination
- `PUT /api/v1/monitoring/models/{model_id}/alerts/{alert_id}/acknowledge` - Acknowledge alert
  - Mark alert as acknowledged
  - Record acknowledgment timestamp

**Add deployment-specific endpoints:**
- `GET /api/v1/deployments/{deployment_id}/drift` - Alias for model drift endpoint
- `POST /api/v1/deployments/{deployment_id}/check-drift` - Alias for manual drift check

### 5. Response Schemas

Create `file:apps/backend/app/schemas/drift.py`:

**Pydantic models:**
- `StatisticalTestResult` - Fields: test_name, statistic, p_value, drift_detected, interpretation
- `FeatureDriftResult` - Fields: feature_name, drift_detected, ks_test, psi_value, chi_square_test (optional), reference_stats, current_stats
- `DriftDetectionResponse` - Fields: model_id, drift_detected, drift_score, severity, features_with_drift (list of FeatureDriftResult), overall_recommendation, checked_at, next_check_at
- `DriftHistoryResponse` - Fields: model_id, history (list of time-series points), feature_trends (dict), summary_statistics
- `DriftAlertResponse` - Fields: alert_id, model_id, alert_type, severity, message, created_at, acknowledged, metadata
- `CheckDriftRequest` - Fields: sample_data (optional list of dicts), force_check (bool)

### 6. Integration with Model Training

Update `file:apps/backend/app/services/model_service.py`:

**After model training completes:**
- Call `DriftDetectionService.store_training_statistics()` to save baseline
- Store training data statistics in `TrainingDataStatistics` document
- Link to `MLModel` via model_id

**In model deployment flow:**
- Verify training statistics exist before allowing deployment
- Initialize drift monitoring schedule

### 7. Automatic Drift Monitoring

Create `file:apps/backend/app/services/drift_scheduler.py`:

**Background task scheduler:**
- Use asyncio tasks or celery for periodic drift checks
- Check all deployed models every N hours (configurable)
- For each model: fetch recent predictions, run drift detection, store results
- Publish alerts to Redis pub/sub if drift detected

**Integration points:**
- Start scheduler in `file:apps/backend/app/main.py` lifespan startup
- Subscribe to model deployment events to add models to monitoring
- Subscribe to model deactivation events to remove from monitoring

### 8. Alert and Notification System

Create `file:apps/backend/app/services/alert_service.py`:

**Alert management:**
- `create_drift_alert(model_id, drift_result)` - Create alert in database
- `publish_alert_notification(alert)` - Publish to Redis pub/sub channel "drift_alerts"
- `get_active_alerts(user_id)` - Get unacknowledged alerts for user
- `acknowledge_alert(alert_id)` - Mark alert as acknowledged

**Redis pub/sub integration:**
- Use existing Redis client from `file:apps/backend/app/api/routes/production.py` (lines 22-29)
- Publish to channel: `drift_alerts:{user_id}`
- Message format: JSON with alert details

**Future extensibility:**
- Add webhook support for external notifications
- Add email notifications using SMTP
- Add Slack/Teams integration

### 9. Frontend Dashboard Components

Create frontend components in `apps/frontend/`:

**Drift Dashboard Page (`app/monitoring/drift/page.tsx`):**
- Display list of deployed models with drift status
- Color-coded indicators (green/yellow/red for no/moderate/high drift)
- Summary cards: models monitored, drift detected, alerts pending
- Quick actions: trigger manual check, view details

**Model Drift Detail Page (`app/monitoring/drift/[modelId]/page.tsx`):**
- Overall drift score gauge visualization
- Time series chart showing drift score over time
- Per-feature drift breakdown table with statistical test results
- Distribution comparison charts (reference vs current) using recharts or plotly
- Alert history timeline
- Action buttons: acknowledge alerts, trigger retraining

**Drift Visualization Components (`components/monitoring/`):**
- `DriftScoreGauge.tsx` - Circular gauge showing drift score (0-1)
- `DriftHistoryChart.tsx` - Line chart with drift score over time
- `FeatureDriftTable.tsx` - Table with per-feature drift metrics
- `DistributionComparisonChart.tsx` - Overlaid histograms or KDE plots
- `DriftAlertBanner.tsx` - Alert notification banner

**API Integration (`lib/services/drift.ts`):**
- `getDriftStatus(modelId)` - Fetch current drift status
- `checkDrift(modelId, sampleData?)` - Trigger manual drift check
- `getDriftHistory(modelId, days)` - Fetch historical data
- `getFeatureDrift(modelId)` - Get per-feature analysis
- `acknowledgeAlert(alertId)` - Acknowledge alert

### 10. Automatic Retraining Triggers

Create `file:apps/backend/app/services/retraining_service.py`:

**Retraining logic:**
- `should_trigger_retraining(drift_result, model_config)` - Decision logic
  - Check drift score against threshold (configurable per model)
  - Check number of features with drift
  - Check severity level
  - Return boolean and reason
- `schedule_retraining(model_id, reason)` - Create retraining job
  - Store in database with status "pending"
  - Optionally trigger automatic retraining workflow
  - Notify user via alerts

**Integration with drift detection:**
- Call after drift detection completes
- If retraining triggered, create alert with type "retraining_recommended"
- Update model metadata with retraining recommendation

### 11. Configuration and Settings

Add drift detection configuration to `file:apps/backend/app/config.py`:

**Settings:**
- `DRIFT_CHECK_INTERVAL_HOURS` (default: 24)
- `DRIFT_DETECTION_THRESHOLD` (default: 0.05 for p-value)
- `PSI_THRESHOLD_MODERATE` (default: 0.1)
- `PSI_THRESHOLD_HIGH` (default: 0.25)
- `MIN_SAMPLES_FOR_DRIFT_CHECK` (default: 100)
- `DRIFT_ALERT_RETENTION_DAYS` (default: 90)
- `AUTO_RETRAINING_ENABLED` (default: false)
- `RETRAINING_DRIFT_THRESHOLD` (default: 0.7)

### 12. Testing

Create comprehensive tests:

**Unit tests (`apps/backend/tests/test_services/test_drift_detection.py`):**
- Test PSI calculation with known distributions
- Test KS test with synthetic data (drifted vs non-drifted)
- Test chi-square test for categorical features
- Test drift score aggregation
- Test alert creation logic

**Integration tests (`apps/backend/tests/test_api/test_drift_monitoring.py`):**
- Test drift detection endpoints
- Test drift history retrieval
- Test alert management
- Test automatic drift checking workflow

**E2E tests (`apps/frontend/e2e/drift-monitoring.spec.ts`):**
- Test drift dashboard rendering
- Test manual drift check trigger
- Test alert acknowledgment
- Test distribution visualization

### 13. Documentation

Update documentation:

**API Documentation:**
- Add drift detection endpoints to OpenAPI spec
- Include request/response examples
- Document statistical tests and thresholds

**User Documentation:**
- Create guide on interpreting drift metrics
- Explain when to retrain models
- Document alert configuration

## Architecture Diagram

```mermaid
sequenceDiagram
    participant Client as Frontend Client
    participant API as Monitoring API
    participant DriftService as DriftDetectionService
    participant PredLog as PredictionLog
    participant DB as MongoDB
    participant Redis as Redis Pub/Sub
    participant Alert as AlertService

    Client->>API: GET /monitoring/models/{id}/drift
    API->>DriftService: detect_drift_for_model(model_id)
    DriftService->>DB: Load TrainingDataStatistics
    DriftService->>PredLog: get_recent_data_for_drift(model_id)
    PredLog-->>DriftService: Recent prediction data
    
    DriftService->>DriftService: Calculate KS test per feature
    DriftService->>DriftService: Calculate PSI per feature
    DriftService->>DriftService: Calculate drift score
    
    DriftService->>DB: Store DriftMetrics
    DriftService->>DB: Store FeatureDriftHistory
    
    alt Drift Detected
        DriftService->>Alert: create_drift_alert(model_id, result)
        Alert->>DB: Store DriftAlert
        Alert->>Redis: Publish to drift_alerts channel
        Redis-->>Client: WebSocket notification
    end
    
    DriftService-->>API: Drift detection result
    API-->>Client: DriftDetectionResponse
    
    Client->>API: GET /monitoring/models/{id}/drift/history
    API->>DB: Query DriftMetrics (last 30 days)
    DB-->>API: Historical drift data
    API-->>Client: DriftHistoryResponse
```

## Data Flow

```mermaid
graph TD
    A[Model Training] -->|Store baseline| B[TrainingDataStatistics]
    C[Production Predictions] -->|Log data| D[PredictionLog]
    E[Drift Scheduler] -->|Periodic check| F[DriftDetectionService]
    D -->|Recent data| F
    B -->|Reference stats| F
    F -->|Statistical tests| G[KS Test / PSI / Chi-square]
    G -->|Results| H[DriftMetrics DB]
    G -->|Per-feature| I[FeatureDriftHistory DB]
    F -->|If drift detected| J[DriftAlert DB]
    J -->|Publish| K[Redis Pub/Sub]
    K -->|Notify| L[Frontend Dashboard]
    F -->|High drift| M[RetrainingService]
    M -->|Schedule| N[Retraining Job]
```

## Key Files Summary

| File | Purpose |
|------|---------|
| `file:apps/backend/app/models/drift_detection.py` | Database models for drift tracking |
| `file:apps/backend/app/services/drift_detection_service.py` | Core drift detection logic with statistical tests |
| `file:apps/backend/app/services/drift_scheduler.py` | Background scheduler for automatic drift checks |
| `file:apps/backend/app/services/alert_service.py` | Alert management and notifications |
| `file:apps/backend/app/services/retraining_service.py` | Automatic retraining trigger logic |
| `file:apps/backend/app/api/routes/monitoring.py` | API endpoints (extend existing) |
| `file:apps/backend/app/schemas/drift.py` | Request/response schemas |
| `file:apps/frontend/app/monitoring/drift/page.tsx` | Drift dashboard UI |
| `file:apps/frontend/lib/services/drift.ts` | Frontend API client |

---

Execution Information

Branch: main
Commit: 1963a18