Shadow Deployment Framework

Production ML serving system with automatic rollback based on statistical degradation detection.

Problem

When you deploy a new ML model, you cannot know if it's better without serving it to real users. Manual rollback takes time and requires human intervention. This framework runs champion and challenger models in parallel, automatically detecting and rolling back degradation within 1 minute without human involvement.

What It Does

Dual Model Serving: Champion (production) and challenger (candidate) run simultaneously
Shadow Mode: Challenger predictions logged but never served to users
Automatic Detection: Statistical tests (Welch's t-test, K-S, Chi-square) identify degradation
Fast Rollback: Rollback to champion-only within 60 seconds of detection
Zero User Impact: Async logging; challenger latency never blocks responses

Key Features

Detection Metrics

Latency: Compare mean latency between models (Welch's t-test)
Prediction Distribution Shift: KS test for output distribution changes
Null Rate: Chi-square test for increase in error/null predictions
Feature Importance Drift: Detect changes in what features models rely on (Chi-square + KL divergence)
Threshold: p-value < 0.01 (per-test) + 15% relative degradation required
Multiple Test Requirement: Chi-square (null rate) + at least one other test must fail to trigger rollback (reduces false positives)

Model Support

SASRec: Neural recommendation model with pretrained weights
Scikit-learn: RandomForest, GradientBoosting, LogisticRegression, SVM, etc.
PyTorch: Any PyTorch model (neural networks, transformers)
Extensible: Base class ModelServer for custom model types

Feature Importance Analysis

Gradient-based: For neural networks (PyTorch models)
SHAP: For tree models and others (if shap library installed)
Built-in: sklearn model.feature_importances_ as fallback
Drift Detection: Detect when models shift which features they rely on

Online Learning

Shadow Data Collection: Store predictions during shadow deployment
Retraining Trigger: Automatic scheduling based on time and data volume
Performance Gaps: Identify where challenger differs most from champion
Incremental Learning: Use shadow data to retrain challenger

Cost Optimization

ROI Calculator: Compute ROI of deploying new model
Break-Even Analysis: Find accuracy improvement threshold for deployment
Risk Accounting: Factor in detection power and deployment cost
Cost-Aware Decisions: Automatically decide deploy/hold based on economics

Architecture

User Request
    |
    v
[FastAPI] -> Champion (serve immediately)
           -> Challenger (shadow, non-blocking)
    |
    v
[Redis] -> Prediction logging (async, batched)
    |
    v
[Aggregator] -> Metrics (every 60s)
    |
    v
[Detector] -> Statistical tests
    |
    v
[Rollback Manager] -> Automatic rollback if degraded

Configuration

detection:
  window_seconds: 180             # Aggregation window (3x larger for power)
  grace_period_seconds: 300       # 5 min before first check
  p_value_threshold: 0.01         # Stricter significance (5x)
  degradation_threshold: 0.15     # 15% relative degradation (3x higher)

Traffic Assumptions: Configured for ~50 req/sec sustained traffic. At 180s window, that's ~9,000 predictions per aggregation. Adjust window based on your traffic:

<10 req/sec: Increase window or accept lower power
200 req/sec: Can reduce window to 60s

Validation Results

"The Finding" Test: Deploy degraded SASRec model (15% null rate + 30ms latency)

Metric	Value
Detection Latency	0.27s
Tests Triggered	Latency (Welch), Distribution (K-S), Null Rate (Chi-square)
Conjunction Rule	Chi-square + 2 other tests -> Rollback (pass)
Champion Latency	1.2ms
Challenger Latency	25.2ms
p-values	2.1e-05 (latency), 1.1e-05 (dist), 3.9e-4 (null)

All three statistical tests detected the degradation within the first second.

Quick Start

Local Development (with Redis)

# Install dependencies
pip install -e .

# Optional: Install SHAP for advanced feature importance
pip install shap

# Start Redis
docker run -d -p 6379:6379 redis:7-alpine

# Run tests
pytest tests/ -v

# Run validation test (deploy degraded SASRec, measure detection)
python tests/the_finding_validation.py

# Run multi-model example (scikit-learn RandomForest)
python examples/multi_model_deployment.py

Docker Compose

docker-compose up --build

# In another terminal:
curl http://localhost:8000/health

# Load challenger
curl -X POST http://localhost:8000/models/load-challenger \
  -H "Content-Type: application/json" \
  -d '{"version": "v2.0.0", "degraded": true}'

# Send predictions
curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"feature_0": 0.1, "feature_1": 0.2, "feature_2": 0.3}'

# Check status
curl http://localhost:8000/status
curl http://localhost:8000/metrics
curl http://localhost:8000/history

API Reference

Core Endpoints

POST /predict

Serve champion prediction + shadow challenger
Request: {"feature_0": float, "feature_1": float, ...}
Response: {"prediction": float, "latency_ms": float, "model_version": str}

POST /models/load-challenger

Load new challenger model
Body: {"version": "v2.0.0", "degraded": false}

POST /models/promote-challenger

Promote challenger to champion (manual approval)

Monitoring Endpoints

GET /status

Current deployment state (versions, rollback status)

GET /metrics

Champion vs challenger metrics (latency, distribution, null rate)

GET /history?limit=100

Rollback event history

GET /health

Health check (Redis connectivity)

The Smoke Test

Validates the complete detection and rollback flow:

python tests/smoke_test.py

Load champion model (LinearMockModel v1.0.0)
Generate 100 baseline predictions
Load degraded challenger (DegradedMockModel v2.0.0)
Generate 150 shadow predictions
Run statistical tests
Verify degradation detected
Confirm rollback within SLA

Expected Output:

SHADOW DEPLOYMENT - SMOKE TEST
============================================================

TEST RESULTS
{
  "status": "success",
  "detection_latency_ms": 45.2,
  "detection_latency_within_sla": true,
  "rollback_time_seconds": 3.1,
  "rollback_time_within_sla": true,
  "false_positive_rate": 0.0,
  "rollback_reason": "Challenger degradation detected",
  "metrics": {
    "test_results": [
      {"test": "welch_ttest", "is_degraded": true, ...},
      {"test": "chi_square", "is_degraded": true, ...}
    ]
  }
}

PASS: Degradation detected and rollback triggered within SLA
  - Detection Latency: 45ms (pass)
  - Rollback Time: 3.1s (pass)
  - False Positive Rate: 0.0%
============================================================

Performance Characteristics

Metric	Value	Notes
Grace Period	5 min	No checks during model warm-up
Detection Latency After Grace Period	210s	180s aggregation window + 30s detection cycle
Rollback Time	5-15s	Server polls Redis every 5s
Total Time (deployment to rollback)	~8.5 min	5min grace + 180s detection + ~30s detection + ~10s rollback
False Positive Rate (95% CI)	0-30%	Tested on 10 identical models with 0 false triggers; rule-of-three gives upper bound
Detection (in tested scenario)	100%	Observed in one test with large deliberate degradation (error rates + latency)
User Latency Impact	0ms added	Async logging; no blocking
Detection Power (Latency+Null Rate)	89% (5% latency + 100% null-rate increase); 29% (5% latency + 50% increase)	Measured via scripts/power_calc.py
Limiting Factor	Chi-square (null rate) test	Latency & distribution tests have 100% power; conjunction rule is conservative
Blindspot	Subtle null-rate increase (<50%, 1%-1.5%) with latency <5%	Requires longer window or relax conjunction rule

Key Findings (from power analysis and testing):

Conjunction rule (Chi-square + latency) provides strong false-positive protection
Latency and distribution tests have 100% power even for 3-5% degradation
Chi-square (null rate) is the limiting factor: 28% power for 50% increase (1%-1.5%), 90% for 100% increase (1%-2%)
Measured joint power (via Monte Carlo): 89% for 5% latency + 100% null-rate increase; 29% for 5% latency + 50% null-rate increase
FPR on 10 identical models: 0/10, 95% CI upper bound is 30% per rule-of-three (requires n~300 for confident <1% claim)
Cannot detect pure accuracy loss or subtle (<50%) null-rate increases without longer windows

Design Decisions

Async Logging: Challenger latency never impacts users (fire-and-forget)
Statistical Testing: Avoids false positives from random noise
Grace Period: Gives challenger 5 min to stabilize before evaluation
Fast Rollback: Server polls Redis every 5s for rollback flag
Redis-Based: Simple state management, no complex databases

Advanced Features

Multi-Model Support

The framework supports multiple model architectures out of the box:

Supported Model Types:

Neural Networks: PyTorch models (SASRec, transformers, etc.)
Tree-based: RandomForest, GradientBoosting, XGBoost
Linear Models: LogisticRegression, Ridge, Lasso
Custom Models: Extend ModelServer base class

Using Scikit-learn Models:

from sklearn.ensemble import RandomForestClassifier
from src.models.sklearn_models import SklearnModelServer

rf = RandomForestClassifier(n_estimators=100)
rf.fit(X_train, y_train)

champion = SklearnModelServer(
    version="v1.0",
    model=rf,
    model_type="RandomForest",
    feature_names=["feature_0", "feature_1", ...]
)

result = champion.predict({"X": feature_vector})
print(f"Prediction: {result.value}, Latency: {result.latency_ms}ms")

Using Custom Models:

from src.models.base import ModelServer, PredictionResult

class MyModelServer(ModelServer):
    def __init__(self, version, model):
        super().__init__(version, model_type="MyCustomModel")
        self.model = model
    
    def predict(self, features):
        return PredictionResult(
            value=prediction,
            latency_ms=latency,
            is_null=False
        )
    
    def _get_feature_count(self):
        return 20  # Number of features

Feature Importance Drift Detection

Automatically detect when models shift their reliance on different features.

Using SHAP (Recommended):

from src.detection.feature_importance import compute_feature_importance_sklearn

# Requires: pip install shap
importances = compute_feature_importance_sklearn(
    model=my_model,
    X=training_data,
    feature_names=["feature_0", "feature_1", ...],
    use_shap=True,  # Use SHAP (TreeExplainer or KernelExplainer)
    background_samples=100
)

# Output: {"feature_0": 0.45, "feature_1": 0.30, ...}

Gradient-based (for Neural Networks):

from src.detection.feature_importance import compute_feature_importance_sasrec

importances = compute_feature_importance_sasrec(
    model=torch_model,
    user_history=[10, 20, 30],
    candidate_ids=list(range(1, 201)),
    num_samples=10
)

Built-in Feature Importances:

importances = compute_feature_importance_sklearn(
    model=my_model,
    X=training_data,
    use_shap=False  # Use model.feature_importances_
)

Drift Detection:

from src.detection.statistical_tests import feature_importance_drift

is_degraded, result = feature_importance_drift(
    champion_importances={"feat_a": 0.5, "feat_b": 0.3, "feat_c": 0.2},
    challenger_importances={"feat_a": 0.2, "feat_b": 0.4, "feat_c": 0.4},
    p_threshold=0.01,
    degradation_threshold=0.15
)

if is_degraded:
    print("Feature importance distribution shifted significantly")
    print(f"KL divergence: {result['kl_divergence']:.4f}")
    print(f"p-value: {result['p_value']:.2e}")

Online Learning

Collect shadow deployment data and automatically schedule retraining.

Data Collection:

from src.learning.online_trainer import ShadowDataCollector

collector = ShadowDataCollector(redis_client, retention_days=7)

collected = await collector.collect_from_predictions(
    predictions=[
        {
            "champion_pred": 0.95,
            "challenger_pred": 0.92,
            "champion_latency_ms": 10,
            "challenger_latency_ms": 15,
            "features_hash": "abc123"
        },
    ],
    labels=[1, 0, 1, ...]  # Optional: ground truth labels
)

print(f"Collected {collected} samples")

Retraining Scheduling:

from src.learning.online_trainer import ModelRetrainingScheduler

scheduler = ModelRetrainingScheduler(
    redis_client,
    retraining_interval_hours=24,
    min_samples_for_retraining=1000
)

if await scheduler.should_retrain():
    data = await scheduler.prepare_retraining_data()
    print(f"Available samples: {data['total_samples']}")
    print(f"Mean performance gap: {data['mean_performance_gap']:.4f}")
    
    # Retrain your model here
    # ...
    
    scheduler.record_retraining_completion(metrics)

Cost Optimization

Make data-driven deployment decisions based on cost and risk.

ROI Calculation:

from src.deployment.cost_optimizer import (
    DeploymentCosts,
    ModelMetrics,
    DeploymentROICalculator
)

costs = DeploymentCosts(
    champion_serving_cost=100.0,      # Cost to serve champion ($/hour)
    dual_serving_cost=180.0,          # Cost to serve both models ($/hour)
    bad_deployment_cost=50000.0,      # Revenue loss from bad deployment
    detection_latency_penalty=500.0   # Cost per second of detection latency
)

metrics = ModelMetrics(
    champion_accuracy=0.92,
    challenger_accuracy=0.94,
    champion_latency_p99_ms=50,
    challenger_latency_p99_ms=55,
    null_rate_champion=0.01,
    null_rate_challenger=0.02,
    detection_power=0.95,
    detection_latency_seconds=5.0
)

calculator = DeploymentROICalculator(
    costs=costs,
    hours_until_deployment=24,
    expected_annual_revenue=1e6
)

roi = calculator.calculate_roi(metrics)

print(f"Should deploy: {roi['should_deploy']}")
print(f"ROI: {roi['roi_percentage']:.1f}%")
print(f"Net benefit: ${roi['net_benefit_usd']:.2f}")

Break-Even Analysis:

breakeven = calculator.break_even_analysis(metrics)

print(f"Accuracy improvement needed: {breakeven['breakeven_accuracy_improvement_percent']:.2f}%")
print(f"Current improvement: {(metrics.challenger_accuracy - metrics.champion_accuracy) * 100:.2f}%")
print(f"Above break-even: {breakeven['is_above_breakeven']}")

Cost-Aware Decision Making:

from src.deployment.cost_optimizer import CostAwareDecisionMaker

decision_maker = CostAwareDecisionMaker(calculator)

decision = decision_maker.decide_deployment(
    detection_result=detection_output,
    metrics=metrics
)

print(f"Decision: {decision['decision']}")  # DEPLOY, HOLD, or REJECT
print(f"Reason: {decision['reason']}")
print(f"Confidence: {decision['confidence']}")  # high, medium, low

Project Structure

shadow-deploy/
├── README.md                    # This file
├── pyproject.toml               # Dependencies
├── docker-compose.yml           # Local dev environment
├── Dockerfile                   # Production image
│
├── src/
│   ├── config.py                # Configuration loading
│   ├── logging_setup.py         # Structured logging
│   ├── server.py                # FastAPI app + main loop
│   │
│   ├── models/
│   │   ├── base.py              # ModelServer ABC
│   │   ├── mock.py              # Mock models
│   │   ├── real_sasrec.py       # Real SASRec with degradation
│   │   ├── sklearn_models.py    # Scikit-learn adapters
│   │   └── registry.py          # Model versioning
│   │
│   ├── storage/
│   │   ├── redis.py             # Redis client
│   │   └── mock_redis.py        # In-memory mock for testing
│   │
│   ├── logging/
│   │   └── prediction_logger.py # Async logging
│   │
│   ├── metrics/
│   │   └── aggregator.py        # Metric aggregation
│   │
│   ├── detection/
│   │   ├── statistical_tests.py # Welch's t, K-S, Chi-square, Feature Importance
│   │   ├── detector.py          # Decision engine
│   │   └── feature_importance.py # Feature importance computation
│   │
│   ├── learning/                # Online learning
│   │   └── online_trainer.py    # Data collection + retraining scheduler
│   │
│   └── deployment/              # Cost optimization
│       └── cost_optimizer.py    # ROI calculation + decision making
│
├── tests/
│   ├── smoke_test.py            # End-to-end validation
│   └── the_finding_validation.py # Production validation with real SASRec
│
└── examples/
    └── multi_model_deployment.py # Complete workflow with sklearn

Technical Approach

Why This Matters

Conjunction Requirement: Requires both Chi-square (null rate) AND at least one other test (latency or distribution) to fail before rollback. This is more conservative than either test alone.
Production-Ready: Non-blocking async logging, automatic recovery without human intervention
Observable Degradation: Designed for operational failures (errors, latency) with built-in signals
Zero User Impact: Challenger latency never blocks production responses

Limitations

Limited statistical power for subtle degradation (<5%) without larger sample sizes
Cannot detect pure model accuracy loss without ground-truth labels
Requires operational signals (errors, latency) to trigger; cannot detect distribution shift alone
Test independence assumption not empirically validated (may inflate FWER slightly)

Statistical Power Analysis

Detection power measured via Monte Carlo simulation (1000+ trials):

Welch's t-test (Latency):

100% power at 3% degradation
100% power at 5% degradation
45% power at 1% degradation

Kolmogorov-Smirnov test (Distribution):

High power for large distribution shifts
Variable power depending on shift magnitude

Chi-square test (Null Rate):

90% power for 1% to 2% null-rate increase (100% increase from baseline)
28% power for 1% to 1.5% null-rate increase (50% increase from baseline)

Joint Power (Latency + Null Rate):

89% measured power for 5% latency + 100% null-rate increase
29% measured power for 5% latency + 50% null-rate increase

The Chi-square test is the limiting factor due to typically low baseline null rates. Power scales with window size and traffic volume.

Implementation Notes

Complete workflow example:

import asyncio
from sklearn.ensemble import RandomForestClassifier

async def full_workflow():
    # 1. Setup
    config = get_config()
    redis_client = MockRedisClient(config)
    model_registry = ModelRegistry()
    
    # 2. Train models
    rf = RandomForestClassifier(n_estimators=100)
    rf.fit(X_train, y_train)
    
    # 3. Create model servers
    champion = SklearnModelServer("v1.0", rf)
    challenger = DegradedSklearnModelServer("v2.0", rf, degradation_amount=30)
    
    model_registry.set_champion(champion)
    model_registry.set_challenger(challenger)
    
    # 4. Log predictions
    prediction_logger = PredictionLogger(redis_client, config.logging)
    await prediction_logger.start()
    
    for i, X in enumerate(X_test):
        champ_result = champion.predict({"X": X})
        chal_result = challenger.predict({"X": X})
        await prediction_logger.log_prediction(
            request_id=f"req_{i}",
            features={"X": X},
            champion_pred=champ_result.value,
            champion_latency=champ_result.latency_ms,
            challenger_pred=chal_result.value,
            challenger_latency=chal_result.latency_ms,
        )
    
    # 5. Detect degradation
    detector = RollbackDetector(redis_client, ...)
    decision = await detector.check_and_decide()
    
    # 6. Collect shadow data
    collector = ShadowDataCollector(redis_client)
    collected = await collector.collect_from_predictions(predictions)
    
    # 7. Calculate ROI
    roi = roi_calculator.calculate_roi(metrics)
    
    # 8. Make deployment decision
    if decision and decision.should_rollback:
        print("Degradation detected: REJECT")
    elif roi['should_deploy']:
        print("No degradation, ROI positive: DEPLOY")
    else:
        print("No degradation, ROI negative: HOLD")
    
    await prediction_logger.stop()

asyncio.run(full_workflow())

User Implementation Next Steps

Load Your Model: Replace mock models with your actual ML model
Calibrate Thresholds: Run power analysis on your model to measure FPR on stable versions
Adjust Window: Based on your traffic volume, tune window_seconds (current: 180s for ~50 req/sec)
Deploy to Staging: Run for 1-2 weeks on staging traffic to validate on real patterns
Set Up Monitoring: Wire /health and /metrics endpoints to your monitoring system
Production Deployment: Deploy with your calibrated thresholds and monitoring

References

Welch's t-test: https://en.wikipedia.org/wiki/Welch%27s_t-test
Kolmogorov-Smirnov test: https://en.wikipedia.org/wiki/Kolmogorov%E2%80%93Smirnov_test
Statistical hypothesis testing: https://en.wikipedia.org/wiki/Statistical_hypothesis_testing
Feature monitoring for ML: https://arxiv.org/abs/2201.08614
SHAP: https://github.com/slundberg/shap

License

MIT

Name		Name	Last commit message	Last commit date
examples		examples
scripts		scripts
src		src
tests		tests
.gitignore		.gitignore
Dockerfile		Dockerfile
README.md		README.md
docker-compose.yml		docker-compose.yml
pyproject.toml		pyproject.toml

Latest commit

History

Folders and files

Repository files navigation

Shadow Deployment Framework

Problem

What It Does

Key Features

Detection Metrics

Model Support

Feature Importance Analysis

Online Learning

Cost Optimization

Architecture

Configuration

Validation Results

Quick Start

Local Development (with Redis)

Docker Compose

API Reference

Core Endpoints

Monitoring Endpoints

The Smoke Test

Performance Characteristics

Design Decisions

Advanced Features

Multi-Model Support

Feature Importance Drift Detection

Online Learning

Cost Optimization

Project Structure

Technical Approach

Why This Matters

Limitations

Statistical Power Analysis

Implementation Notes

User Implementation Next Steps

References

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages