Shadow Mode Deployment for AI Models

┌──────────────────────────────────────────────────────────────┐
│                      Load Balancer                            │
└────────────────────────┬─────────────────────────────────────┘
                         │
                         │ Production Traffic
                         │
                ┌────────▼────────┐
                │   API Gateway   │
                └────────┬────────┘
                         │
         ┌───────────────┼───────────────┐
         │               │               │
    ┌────▼────┐    ┌────▼────┐    ┌────▼────┐
    │ Primary │    │ Shadow  │    │ Metrics │
    │ Model   │    │ Model   │    │ Logger  │
    │ (v2.1)  │    │ (v2.2)  │    │         │
    └────┬────┘    └────┬────┘    └────┬────┘
         │               │               │
         │ Return to     │ Log only      │
         │ user          │ (no return)   │
         │               │               │
         └───────────────┼───────────────┘
                         │
                    ┌────▼────┐
                    │ Compare │
                    │ Outputs │
                    └────┬────┘
                         │
                    ┌────▼────┐
                    │ Database│
                    │ Metrics │
                    └─────────┘

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from typing import Optional
import time
import asyncio
from uuid import uuid4

app = FastAPI()

class PredictionRequest(BaseModel):
    text: str
    user_id: str

class PredictionResponse(BaseModel):
    prediction: str
    confidence: float
    model_version: str
    prediction_id: str

# Model loading
primary_model = load_model("v2.1")
shadow_model = load_model("v2.2")

class ShadowDeployment:
    """Manage shadow deployment."""
    
    def __init__(self, primary_model, shadow_model, shadow_percentage: float = 1.0):
        self.primary_model = primary_model
        self.shadow_model = shadow_model
        self.shadow_pct = shadow_percentage
        self.logger = ShadowLogger()
    
    async def predict_with_shadow(
        self,
        request: PredictionRequest,
        background_tasks: BackgroundTasks,
    ) -> PredictionResponse:
        """Run prediction with shadow."""
        
        prediction_id = str(uuid4())
        
        # Primary inference (blocking, returned to user)
        primary_start = time.perf_counter()
        primary_output = await self.primary_model.predict(request.text)
        primary_latency = (time.perf_counter() - primary_start) * 1000
        
        # Shadow inference (async, non-blocking)
        import random
        if random.random() < self.shadow_pct:
            # Schedule shadow prediction
            background_tasks.add_task(
                self._run_shadow_prediction,
                prediction_id=prediction_id,
                request=request,
                primary_output=primary_output,
                primary_latency=primary_latency,
            )
        
        # Return primary immediately
        return PredictionResponse(
            prediction=primary_output.label,
            confidence=primary_output.confidence,
            model_version="v2.1",
            prediction_id=prediction_id,
        )
    
    async def _run_shadow_prediction(
        self,
        prediction_id: str,
        request: PredictionRequest,
        primary_output,
        primary_latency: float,
    ) -> None:
        """Run shadow prediction (background task)."""
        
        try:
            # Shadow inference
            shadow_start = time.perf_counter()
            shadow_output = await self.shadow_model.predict(request.text)
            shadow_latency = (time.perf_counter() - shadow_start) * 1000
            
            # Compare outputs
            comparison = self._compare_outputs(primary_output, shadow_output)
            
            # Log shadow result
            await self.logger.log_shadow_prediction(
                prediction_id=prediction_id,
                user_id=request.user_id,
                input_text=request.text,
                primary_output=primary_output.label,
                primary_confidence=primary_output.confidence,
                primary_latency_ms=primary_latency,
                shadow_output=shadow_output.label,
                shadow_confidence=shadow_output.confidence,
                shadow_latency_ms=shadow_latency,
                outputs_match=comparison["match"],
                similarity_score=comparison["similarity"],
            )
        
        except Exception as e:
            # Log shadow errors but don't affect user
            print(f"Shadow prediction error: {e}")
            await self.logger.log_shadow_error(prediction_id, str(e))
    
    def _compare_outputs(self, primary, shadow) -> dict:
        """Compare primary and shadow outputs."""
        
        # Exact match
        exact_match = primary.label == shadow.label
        
        # Similarity score (for text outputs)
        from difflib import SequenceMatcher
        if hasattr(primary, "text") and hasattr(shadow, "text"):
            similarity = SequenceMatcher(
                None,
                primary.text,
                shadow.text,
            ).ratio()
        else:
            similarity = 1.0 if exact_match else 0.0
        
        return {
            "match": exact_match,
            "similarity": similarity,
        }

shadow_deployment = ShadowDeployment(primary_model, shadow_model, shadow_percentage=1.0)

@app.post("/predict", response_model=PredictionResponse)
async def predict(
    request: PredictionRequest,
    background_tasks: BackgroundTasks,
):
    """Prediction endpoint with shadow."""
    
    return await shadow_deployment.predict_with_shadow(request, background_tasks)

# shadow_logger.py
from dataclasses import dataclass
from datetime import datetime, timezone
import asyncpg

@dataclass
class ShadowLog:
    """Log entry for shadow prediction."""
    
    prediction_id: str
    user_id: str
    input_text: str
    
    # Primary model
    primary_output: str
    primary_confidence: float
    primary_latency_ms: float
    
    # Shadow model
    shadow_output: str
    shadow_confidence: float
    shadow_latency_ms: float
    
    # Comparison
    outputs_match: bool
    similarity_score: float
    
    timestamp: datetime = None
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = datetime.now(timezone.utc)

class ShadowLogger:
    """Log shadow predictions for analysis."""
    
    def __init__(self, db_pool: asyncpg.Pool):
        self.db = db_pool
    
    async def log_shadow_prediction(self, **kwargs) -> None:
        """Log shadow prediction result."""
        
        log = ShadowLog(**kwargs)
        
        await self.db.execute(
            """
            INSERT INTO shadow_predictions (
                prediction_id, user_id, input_text,
                primary_output, primary_confidence, primary_latency_ms,
                shadow_output, shadow_confidence, shadow_latency_ms,
                outputs_match, similarity_score, timestamp
            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
            """,
            log.prediction_id,
            log.user_id,
            log.input_text,
            log.primary_output,
            log.primary_confidence,
            log.primary_latency_ms,
            log.shadow_output,
            log.shadow_confidence,
            log.shadow_latency_ms,
            log.outputs_match,
            log.similarity_score,
            log.timestamp,
        )
    
    async def log_shadow_error(self, prediction_id: str, error: str) -> None:
        """Log shadow prediction error."""
        
        await self.db.execute(
            """
            INSERT INTO shadow_errors (prediction_id, error_message, timestamp)
            VALUES ($1, $2, $3)
            """,
            prediction_id,
            error,
            datetime.now(timezone.utc),
        )

# Database schema
"""
CREATE TABLE shadow_predictions (
    prediction_id UUID PRIMARY KEY,
    user_id VARCHAR(255) NOT NULL,
    input_text TEXT NOT NULL,
    
    -- Primary model
    primary_output TEXT NOT NULL,
    primary_confidence FLOAT NOT NULL,
    primary_latency_ms FLOAT NOT NULL,
    
    -- Shadow model
    shadow_output TEXT NOT NULL,
    shadow_confidence FLOAT NOT NULL,
    shadow_latency_ms FLOAT NOT NULL,
    
    -- Comparison
    outputs_match BOOLEAN NOT NULL,
    similarity_score FLOAT NOT NULL,
    
    timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_shadow_predictions_timestamp ON shadow_predictions (timestamp DESC);
CREATE INDEX idx_shadow_predictions_match ON shadow_predictions (outputs_match, timestamp DESC);

CREATE TABLE shadow_errors (
    id SERIAL PRIMARY KEY,
    prediction_id UUID NOT NULL,
    error_message TEXT NOT NULL,
    timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
"""

# shadow_analysis.py
import pandas as pd
from typing import Dict

class ShadowAnalyzer:
    """Analyze shadow deployment results."""
    
    def __init__(self, db_pool):
        self.db = db_pool
    
    async def get_shadow_metrics(self, days: int = 7) -> Dict:
        """Get shadow deployment metrics."""
        
        from datetime import timedelta
        start_date = datetime.now(timezone.utc) - timedelta(days=days)
        
        # Overall statistics
        overall = await self.db.fetchrow(
            """
            SELECT 
                COUNT(*) as total_predictions,
                AVG(CASE WHEN outputs_match THEN 1 ELSE 0 END) as match_rate,
                AVG(similarity_score) as avg_similarity,
                AVG(primary_latency_ms) as avg_primary_latency,
                AVG(shadow_latency_ms) as avg_shadow_latency,
                PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY primary_latency_ms) as p95_primary_latency,
                PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY shadow_latency_ms) as p95_shadow_latency
            FROM shadow_predictions
            WHERE timestamp >= $1
            """,
            start_date,
        )
        
        # Error rate
        error_count = await self.db.fetchval(
            "SELECT COUNT(*) FROM shadow_errors WHERE timestamp >= $1",
            start_date,
        )
        
        total_predictions = overall["total_predictions"]
        error_rate = error_count / total_predictions if total_predictions > 0 else 0.0
        
        # Latency regression
        latency_regression = (
            overall["avg_shadow_latency"] - overall["avg_primary_latency"]
        ) / overall["avg_primary_latency"]
        
        return {
            "total_predictions": total_predictions,
            "match_rate": overall["match_rate"],
            "avg_similarity": overall["avg_similarity"],
            "primary_latency": {
                "avg_ms": overall["avg_primary_latency"],
                "p95_ms": overall["p95_primary_latency"],
            },
            "shadow_latency": {
                "avg_ms": overall["avg_shadow_latency"],
                "p95_ms": overall["p95_shadow_latency"],
            },
            "latency_regression_pct": latency_regression * 100,
            "error_rate": error_rate,
        }
    
    async def get_disagreement_examples(self, limit: int = 100) -> pd.DataFrame:
        """Get examples where primary and shadow disagree."""
        
        rows = await self.db.fetch(
            """
            SELECT 
                prediction_id,
                input_text,
                primary_output,
                primary_confidence,
                shadow_output,
                shadow_confidence,
                similarity_score
            FROM shadow_predictions
            WHERE outputs_match = FALSE
            ORDER BY similarity_score ASC
            LIMIT $1
            """,
            limit,
        )
        
        return pd.DataFrame([dict(r) for r in rows])
    
    async def get_quality_breakdown(self) -> Dict:
        """Get quality metrics by confidence bucket."""
        
        rows = await self.db.fetch(
            """
            SELECT 
                CASE 
                    WHEN primary_confidence < 0.5 THEN 'low'
                    WHEN primary_confidence < 0.8 THEN 'medium'
                    ELSE 'high'
                END as confidence_bucket,
                COUNT(*) as count,
                AVG(CASE WHEN outputs_match THEN 1 ELSE 0 END) as match_rate,
                AVG(similarity_score) as avg_similarity
            FROM shadow_predictions
            GROUP BY confidence_bucket
            ORDER BY confidence_bucket
            """
        )
        
        return {
            row["confidence_bucket"]: {
                "count": row["count"],
                "match_rate": row["match_rate"],
                "avg_similarity": row["avg_similarity"],
            }
            for row in rows
        }
    
    async def should_promote_shadow(self) -> tuple[bool, str]:
        """Determine if shadow model should be promoted."""
        
        metrics = await self.get_shadow_metrics(days=14)  # 2 week validation
        
        # Minimum traffic threshold
        if metrics["total_predictions"] < 10000:
            return False, f"Insufficient traffic: {metrics['total_predictions']} < 10,000"
        
        # Quality threshold
        if metrics["match_rate"] < 0.85:
            return False, f"Low match rate: {metrics['match_rate']:.1%} < 85%"
        
        # Latency regression threshold
        if metrics["latency_regression_pct"] > 50:
            return False, f"Latency regression: +{metrics['latency_regression_pct']:.0f}% > 50%"
        
        # Error rate threshold
        if metrics["error_rate"] > 0.05:
            return False, f"High error rate: {metrics['error_rate']:.1%} > 5%"
        
        # All checks passed
        return True, "All quality checks passed"

# Usage
analyzer = ShadowAnalyzer(db_pool)

# Daily shadow report
metrics = await analyzer.get_shadow_metrics(days=7)
print(f"Shadow Metrics (last 7 days):")
print(f"  Total predictions: {metrics['total_predictions']:,}")
print(f"  Match rate: {metrics['match_rate']:.1%}")
print(f"  Avg similarity: {metrics['avg_similarity']:.3f}")
print(f"  Latency regression: {metrics['latency_regression_pct']:+.1f}%")
print(f"  Error rate: {metrics['error_rate']:.2%}")

# Check if ready for promotion
should_promote, reason = await analyzer.should_promote_shadow()
print(f"\nPromotion: {'✅ Ready' if should_promote else '❌ Not ready'}")
print(f"Reason: {reason}")

# Analyze disagreements
disagreements = await analyzer.get_disagreement_examples(limit=20)
print(f"\nTop 20 disagreements:")
print(disagreements[["input_text", "primary_output", "shadow_output", "similarity_score"]])

# shadow_monitoring.py
from prometheus_client import Counter, Histogram, Gauge

# Shadow metrics
shadow_predictions_total = Counter(
    "shadow_predictions_total",
    "Total shadow predictions",
    ["status"],  # success, error
)

shadow_match_rate = Gauge(
    "shadow_match_rate",
    "Percentage of shadow outputs matching primary",
)

shadow_latency_seconds = Histogram(
    "shadow_latency_seconds",
    "Shadow prediction latency",
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0],
)

shadow_similarity_score = Histogram(
    "shadow_similarity_score",
    "Shadow output similarity to primary",
    buckets=[0.5, 0.6, 0.7, 0.8, 0.9, 0.95, 1.0],
)

# Cost tracking
async def calculate_shadow_cost(days: int = 7) -> Dict:
    """Calculate cost of shadow deployment."""
    
    # Get shadow prediction count
    metrics = await analyzer.get_shadow_metrics(days)
    
    # Estimate cost
    predictions_count = metrics["total_predictions"]
    avg_latency_ms = metrics["shadow_latency"]["avg_ms"]
    
    # Assume GPU instance cost
    gpu_hourly_cost = 1.50  # g4dn.xlarge
    inference_per_second = 1000 / avg_latency_ms
    
    # Calculate GPU hours needed
    total_seconds = predictions_count / inference_per_second
    gpu_hours = total_seconds / 3600
    
    total_cost = gpu_hours * gpu_hourly_cost
    
    return {
        "predictions": predictions_count,
        "gpu_hours": gpu_hours,
        "total_cost_usd": total_cost,
        "cost_per_prediction": total_cost / predictions_count,
    }

# Alerting
from dataclasses import dataclass

@dataclass
class ShadowAlert:
    """Alert condition for shadow deployment."""
    
    name: str
    condition: callable
    severity: str  # "warning", "critical"

async def check_shadow_alerts() -> List[str]:
    """Check for shadow alert conditions."""
    
    metrics = await analyzer.get_shadow_metrics(days=1)
    
    alerts = []
    
    # Alert conditions
    if metrics["match_rate"] < 0.70:
        alerts.append(f"🚨 Critical: Low match rate {metrics['match_rate']:.1%}")
    
    if metrics["error_rate"] > 0.10:
        alerts.append(f"🚨 Critical: High error rate {metrics['error_rate']:.1%}")
    
    if metrics["latency_regression_pct"] > 100:
        alerts.append(f"⚠️  Warning: Latency 2x slower (+{metrics['latency_regression_pct']:.0f}%)")
    
    if metrics["total_predictions"] < 100:
        alerts.append("⚠️  Warning: Low shadow traffic")
    
    return alerts

# Scheduled alert check
async def shadow_alert_monitor():
    """Monitor shadow and send alerts."""
    
    alerts = await check_shadow_alerts()
    
    if alerts:
        # Send to Slack/PagerDuty
        message = "Shadow Deployment Alerts:\n" + "\n".join(alerts)
        await send_alert(message)

# promotion.py
from datetime import datetime, timedelta

class ShadowPromotion:
    """Manage shadow model promotion."""
    
    VALIDATION_CRITERIA = {
        "min_predictions": 10000,
        "min_match_rate": 0.85,
        "max_latency_regression_pct": 50,
        "max_error_rate": 0.05,
        "min_validation_days": 14,
    }
    
    def __init__(self, db_pool):
        self.db = db_pool
        self.analyzer = ShadowAnalyzer(db_pool)
    
    async def evaluate_promotion(self) -> dict:
        """Evaluate if shadow model is ready for promotion."""
        
        # Get shadow start date
        shadow_start = await self.db.fetchval(
            "SELECT MIN(timestamp) FROM shadow_predictions"
        )
        
        days_running = (datetime.now(timezone.utc) - shadow_start).days
        
        # Get metrics
        metrics = await self.analyzer.get_shadow_metrics(days=days_running)
        
        # Check each criterion
        checks = {}
        
        checks["traffic_volume"] = {
            "passed": metrics["total_predictions"] >= self.VALIDATION_CRITERIA["min_predictions"],
            "value": metrics["total_predictions"],
            "threshold": self.VALIDATION_CRITERIA["min_predictions"],
        }
        
        checks["match_rate"] = {
            "passed": metrics["match_rate"] >= self.VALIDATION_CRITERIA["min_match_rate"],
            "value": metrics["match_rate"],
            "threshold": self.VALIDATION_CRITERIA["min_match_rate"],
        }
        
        checks["latency"] = {
            "passed": metrics["latency_regression_pct"] <= self.VALIDATION_CRITERIA["max_latency_regression_pct"],
            "value": metrics["latency_regression_pct"],
            "threshold": self.VALIDATION_CRITERIA["max_latency_regression_pct"],
        }
        
        checks["error_rate"] = {
            "passed": metrics["error_rate"] <= self.VALIDATION_CRITERIA["max_error_rate"],
            "value": metrics["error_rate"],
            "threshold": self.VALIDATION_CRITERIA["max_error_rate"],
        }
        
        checks["validation_duration"] = {
            "passed": days_running >= self.VALIDATION_CRITERIA["min_validation_days"],
            "value": days_running,
            "threshold": self.VALIDATION_CRITERIA["min_validation_days"],
        }
        
        # All checks must pass
        all_passed = all(check["passed"] for check in checks.values())
        
        return {
            "ready_for_promotion": all_passed,
            "checks": checks,
            "metrics": metrics,
        }
    
    async def promote_to_production(self, shadow_version: str) -> None:
        """Promote shadow model to production."""
        
        print(f"🚀 Promoting shadow model {shadow_version} to production...")
        
        # Update model version in production
        await self._update_production_model(shadow_version)
        
        # Archive shadow predictions
        await self._archive_shadow_logs()
        
        # Send notification
        await self._notify_promotion(shadow_version)
        
        print(f"✅ Shadow model {shadow_version} promoted to production")
    
    async def _update_production_model(self, version: str) -> None:
        """Update production model version."""
        # Update Kubernetes deployment, model registry, etc.
        pass
    
    async def _archive_shadow_logs(self) -> None:
        """Archive shadow logs to cold storage."""
        # Move to S3/GCS for long-term analysis
        pass
    
    async def _notify_promotion(self, version: str) -> None:
        """Notify team of promotion."""
        # Send Slack message, create deployment record
        pass

# Automated promotion pipeline
async def automated_promotion_check():
    """Check if shadow ready for promotion."""
    
    promotion = ShadowPromotion(db_pool)
    evaluation = await promotion.evaluate_promotion()
    
    if evaluation["ready_for_promotion"]:
        print("✅ Shadow model ready for promotion!")
        
        # Require human approval for production promotion
        approval = input("Promote to production? (yes/no): ")
        if approval.lower() == "yes":
            await promotion.promote_to_production("v2.2")
    else:
        print("❌ Shadow model not ready for promotion")
        print("\nFailed checks:")
        for name, check in evaluation["checks"].items():
            if not check["passed"]:
                print(f"  {name}: {check['value']} (threshold: {check['threshold']})")

# Run daily
# asyncio.run(automated_promotion_check())