Why 88% of AI Projects Fail to Reach Production: The Pilot-to-Production Gap Solved

from dataclasses import dataclass
from typing import List, Dict
from enum import Enum

class ReadinessCategory(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

@dataclass
class ReadinessCheck:
    category: str
    check_name: str
    requirement: str
    priority: ReadinessCategory
    completed: bool
    notes: str = ""

class ProductionReadinessAssessment:
    """Assess if AI system is ready for production deployment"""

    def __init__(self):
        self.checks: List[ReadinessCheck] = []
        self._initialize_checks()

    def _initialize_checks(self):
        """Define comprehensive readiness checklist"""

        # Data Quality Checks
        self.checks.extend([
            ReadinessCheck(
                category="Data Quality",
                check_name="Production data availability",
                requirement="Access to real production data for testing",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
            ReadinessCheck(
                category="Data Quality",
                check_name="Data drift monitoring",
                requirement="Automated detection of distribution shifts",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
            ReadinessCheck(
                category="Data Quality",
                check_name="Data validation pipeline",
                requirement="Continuous validation of input data quality",
                priority=ReadinessCategory.HIGH,
                completed=False
            ),
        ])

        # Performance Checks
        self.checks.extend([
            ReadinessCheck(
                category="Performance",
                check_name="Latency requirements",
                requirement="p95 latency under 200ms at peak load",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
            ReadinessCheck(
                category="Performance",
                check_name="Load testing",
                requirement="Tested at 10x expected peak traffic",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
            ReadinessCheck(
                category="Performance",
                check_name="Auto-scaling",
                requirement="Automatic scaling based on load metrics",
                priority=ReadinessCategory.HIGH,
                completed=False
            ),
        ])

        # Monitoring Checks
        self.checks.extend([
            ReadinessCheck(
                category="Monitoring",
                check_name="Model performance metrics",
                requirement="Real-time tracking of accuracy/precision/recall",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
            ReadinessCheck(
                category="Monitoring",
                check_name="Business metrics",
                requirement="KPIs aligned with business objectives",
                priority=ReadinessCategory.HIGH,
                completed=False
            ),
            ReadinessCheck(
                category="Monitoring",
                check_name="Alerting system",
                requirement="Automated alerts for degradation",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
        ])

        # Integration Checks
        self.checks.extend([
            ReadinessCheck(
                category="Integration",
                check_name="API documentation",
                requirement="Complete API docs with examples",
                priority=ReadinessCategory.HIGH,
                completed=False
            ),
            ReadinessCheck(
                category="Integration",
                check_name="Error handling",
                requirement="Graceful degradation for all failure modes",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
            ReadinessCheck(
                category="Integration",
                check_name="Backward compatibility",
                requirement="Version migration strategy defined",
                priority=ReadinessCategory.MEDIUM,
                completed=False
            ),
        ])

        # Security & Compliance
        self.checks.extend([
            ReadinessCheck(
                category="Security",
                check_name="Security audit",
                requirement="Penetration testing completed",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
            ReadinessCheck(
                category="Security",
                check_name="Compliance review",
                requirement="Legal/compliance sign-off obtained",
                priority=ReadinessCategory.CRITICAL,
                completed=False
            ),
        ])

    def assess_readiness(self) -> Dict:
        """Calculate production readiness score"""

        total_checks = len(self.checks)
        completed_checks = sum(1 for c in self.checks if c.completed)

        # Weight by priority
        priority_weights = {
            ReadinessCategory.CRITICAL: 4,
            ReadinessCategory.HIGH: 3,
            ReadinessCategory.MEDIUM: 2,
            ReadinessCategory.LOW: 1
        }

        total_weight = sum(priority_weights[c.priority] for c in self.checks)
        completed_weight = sum(
            priority_weights[c.priority] for c in self.checks if c.completed
        )

        weighted_score = (completed_weight / total_weight) * 100

        # Identify blockers
        critical_incomplete = [
            c for c in self.checks
            if c.priority == ReadinessCategory.CRITICAL and not c.completed
        ]

        return {
            'overall_completion': (completed_checks / total_checks) * 100,
            'weighted_readiness_score': weighted_score,
            'checks_completed': completed_checks,
            'total_checks': total_checks,
            'critical_blockers': len(critical_incomplete),
            'blocker_details': [
                {'category': c.category, 'check': c.check_name}
                for c in critical_incomplete
            ],
            'ready_for_production': len(critical_incomplete) == 0 and weighted_score >= 80
        }

    def generate_report(self) -> str:
        """Generate human-readable readiness report"""

        assessment = self.assess_readiness()

        report = f"""
=== PRODUCTION READINESS ASSESSMENT ===

Overall Completion: {assessment['overall_completion']:.1f}%
Weighted Readiness Score: {assessment['weighted_readiness_score']:.1f}/100

Status: {'✅ READY FOR PRODUCTION' if assessment['ready_for_production'] else '❌ NOT READY'}

Critical Blockers: {assessment['critical_blockers']}
"""

        if assessment['blocker_details']:
            report += "\nMust Complete Before Production:\n"
            for blocker in assessment['blocker_details']:
                report += f"  - [{blocker['category']}] {blocker['check']}\n"

        # Group checks by category
        by_category = {}
        for check in self.checks:
            if check.category not in by_category:
                by_category[check.category] = []
            by_category[check.category].append(check)

        report += "\nChecklist by Category:\n"
        for category, checks in by_category.items():
            completed = sum(1 for c in checks if c.completed)
            total = len(checks)
            report += f"\n{category}: {completed}/{total} completed\n"
            for check in checks:
                status = "✓" if check.completed else "✗"
                priority_marker = "🔴" if check.priority == ReadinessCategory.CRITICAL else ""
                report += f"  {status} {priority_marker} {check.check_name}\n"

        return report

# Usage
assessment = ProductionReadinessAssessment()

# Mark some checks as complete
assessment.checks[0].completed = True  # Production data
assessment.checks[3].completed = True  # Latency requirements

result = assessment.assess_readiness()
print(assessment.generate_report())

print(f"\nReadiness Score: {result['weighted_readiness_score']:.1f}/100")
print(f"Production Ready: {result['ready_for_production']}")

import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Tuple
import pandas as pd

@dataclass
class DataDriftMetrics:
    feature_name: str
    train_mean: float
    production_mean: float
    drift_magnitude: float
    requires_retraining: bool

class DataDriftDetector:
    """Detect distribution shift between training and production data"""

    def __init__(self, reference_data: pd.DataFrame, threshold: float = 0.1):
        self.reference_data = reference_data
        self.threshold = threshold
        self.reference_stats = self._calculate_stats(reference_data)

    def _calculate_stats(self, data: pd.DataFrame) -> Dict:
        """Calculate distribution statistics"""
        return {
            col: {
                'mean': data[col].mean(),
                'std': data[col].std(),
                'min': data[col].min(),
                'max': data[col].max(),
                'q25': data[col].quantile(0.25),
                'q75': data[col].quantile(0.75)
            }
            for col in data.select_dtypes(include=[np.number]).columns
        }

    def detect_drift(
        self,
        production_data: pd.DataFrame
    ) -> Tuple[bool, List[DataDriftMetrics]]:
        """Detect if production data has drifted from training distribution"""

        production_stats = self._calculate_stats(production_data)
        drift_metrics = []
        significant_drift_detected = False

        for feature in production_stats.keys():
            if feature not in self.reference_stats:
                continue

            ref = self.reference_stats[feature]
            prod = production_stats[feature]

            # Calculate drift using Population Stability Index (PSI)
            psi = self._calculate_psi(
                self.reference_data[feature],
                production_data[feature]
            )

            requires_retraining = psi > self.threshold

            if requires_retraining:
                significant_drift_detected = True

            drift_metrics.append(DataDriftMetrics(
                feature_name=feature,
                train_mean=ref['mean'],
                production_mean=prod['mean'],
                drift_magnitude=psi,
                requires_retraining=requires_retraining
            ))

        return significant_drift_detected, drift_metrics

    def _calculate_psi(
        self,
        reference: pd.Series,
        production: pd.Series,
        bins: int = 10
    ) -> float:
        """Calculate Population Stability Index"""

        # Create bins from reference data
        breakpoints = np.percentile(reference, np.linspace(0, 100, bins + 1))
        breakpoints[-1] += 0.0001  # Ensure max value is included

        # Calculate proportions
        ref_counts, _ = np.histogram(reference, bins=breakpoints)
        prod_counts, _ = np.histogram(production, bins=breakpoints)

        ref_props = ref_counts / len(reference)
        prod_props = prod_counts / len(production)

        # Avoid log(0)
        ref_props = np.where(ref_props == 0, 0.0001, ref_props)
        prod_props = np.where(prod_props == 0, 0.0001, prod_props)

        # Calculate PSI
        psi = np.sum((prod_props - ref_props) * np.log(prod_props / ref_props))

        return psi

# Usage
# Training data
train_data = pd.DataFrame({
    'income': np.random.normal(50000, 15000, 1000),
    'age': np.random.normal(35, 10, 1000),
    'credit_score': np.random.normal(700, 50, 1000)
})

# Production data with drift
prod_data = pd.DataFrame({
    'income': np.random.normal(48000, 16000, 500),  # Slight drift
    'age': np.random.normal(38, 12, 500),  # Drift in mean and variance
    'credit_score': np.random.normal(695, 55, 500)  # Slight drift
})

detector = DataDriftDetector(train_data, threshold=0.1)
has_drift, metrics = detector.detect_drift(prod_data)

print(f"Significant Drift Detected: {has_drift}\n")
for metric in metrics:
    if metric.requires_retraining:
        print(f"⚠️  {metric.feature_name}:")
        print(f"   Train Mean: {metric.train_mean:.2f}")
        print(f"   Prod Mean: {metric.production_mean:.2f}")
        print(f"   PSI: {metric.drift_magnitude:.4f} (threshold: 0.1)")
        print()

from typing import List, Optional
from datetime import datetime

class DataQualityCheck:
    """Production data quality validation"""

    def __init__(self, feature_name: str, check_type: str):
        self.feature_name = feature_name
        self.check_type = check_type
        self.violations = []

    def check_missing_values(
        self,
        data: pd.Series,
        max_missing_pct: float = 0.05
    ) -> bool:
        """Check if missing value rate is acceptable"""
        missing_pct = data.isnull().sum() / len(data)

        if missing_pct > max_missing_pct:
            self.violations.append({
                'check': 'missing_values',
                'threshold': max_missing_pct,
                'actual': missing_pct,
                'severity': 'high' if missing_pct > 0.2 else 'medium'
            })
            return False

        return True

    def check_range(
        self,
        data: pd.Series,
        min_val: float,
        max_val: float
    ) -> bool:
        """Check if values are within expected range"""
        out_of_range = ((data < min_val) | (data > max_val)).sum()
        out_of_range_pct = out_of_range / len(data)

        if out_of_range_pct > 0.01:  # More than 1% out of range
            self.violations.append({
                'check': 'range_violation',
                'min': min_val,
                'max': max_val,
                'violations': int(out_of_range),
                'percentage': out_of_range_pct
            })
            return False

        return True

    def check_uniqueness(
        self,
        data: pd.Series,
        min_unique_pct: float = 0.95
    ) -> bool:
        """Check uniqueness for ID fields"""
        unique_pct = data.nunique() / len(data)

        if unique_pct < min_unique_pct:
            self.violations.append({
                'check': 'uniqueness',
                'threshold': min_unique_pct,
                'actual': unique_pct
            })
            return False

        return True

class ProductionDataValidator:
    """Validate production data before model inference"""

    def __init__(self, schema: Dict):
        self.schema = schema
        self.validation_log = []

    def validate_batch(
        self,
        data: pd.DataFrame
    ) -> Tuple[bool, List[Dict]]:
        """Validate a batch of production data"""

        all_checks_passed = True
        violations = []

        for feature, constraints in self.schema.items():
            if feature not in data.columns:
                violations.append({
                    'feature': feature,
                    'error': 'missing_column',
                    'severity': 'critical'
                })
                all_checks_passed = False
                continue

            checker = DataQualityCheck(feature, constraints['type'])

            # Type check
            if constraints['type'] == 'numeric':
                if not pd.api.types.is_numeric_dtype(data[feature]):
                    violations.append({
                        'feature': feature,
                        'error': 'type_mismatch',
                        'expected': 'numeric',
                        'actual': str(data[feature].dtype)
                    })
                    all_checks_passed = False
                    continue

                # Range check
                if 'range' in constraints:
                    min_val, max_val = constraints['range']
                    if not checker.check_range(data[feature], min_val, max_val):
                        violations.extend(checker.violations)
                        all_checks_passed = False

            # Missing value check
            if 'max_missing_pct' in constraints:
                if not checker.check_missing_values(
                    data[feature],
                    constraints['max_missing_pct']
                ):
                    violations.extend(checker.violations)
                    all_checks_passed = False

            # Uniqueness check
            if constraints.get('unique', False):
                if not checker.check_uniqueness(data[feature]):
                    violations.extend(checker.violations)
                    all_checks_passed = False

        # Log validation result
        self.validation_log.append({
            'timestamp': datetime.now(),
            'batch_size': len(data),
            'passed': all_checks_passed,
            'violations': len(violations)
        })

        return all_checks_passed, violations

# Usage
schema = {
    'user_id': {
        'type': 'string',
        'unique': True
    },
    'transaction_amount': {
        'type': 'numeric',
        'range': (0, 10000),
        'max_missing_pct': 0.01
    },
    'credit_score': {
        'type': 'numeric',
        'range': (300, 850),
        'max_missing_pct': 0.05
    }
}

validator = ProductionDataValidator(schema)

# Validate production batch
production_batch = pd.DataFrame({
    'user_id': ['U1', 'U2', 'U3', 'U2'],  # Duplicate!
    'transaction_amount': [100, 15000, 500, 200],  # Out of range!
    'credit_score': [720, 650, np.nan, 800]
})

is_valid, violations = validator.validate_batch(production_batch)

print(f"Batch Valid: {is_valid}")
if violations:
    print("\nViolations Found:")
    for v in violations:
        print(f"  - {v}")

import random
from typing import Dict, List

class ColdStartHandler:
    """Handle inference for users without historical data"""

    def __init__(self, default_model, user_models: Dict):
        self.default_model = default_model
        self.user_models = user_models
        self.cold_start_log = []

    def predict_with_fallback(
        self,
        user_id: str,
        features: Dict,
        min_data_points: int = 50
    ):
        """Make prediction with graceful fallback for cold start users"""

        # Check if user has sufficient history
        user_data_count = self._get_user_data_count(user_id)

        if user_data_count >= min_data_points:
            # Use personalized model
            return self.user_models[user_id].predict(features)

        elif user_data_count > 0:
            # Hybrid: blend personalized and global
            personal_pred = self.user_models[user_id].predict(features)
            global_pred = self.default_model.predict(features)

            # Weight by data availability
            weight = user_data_count / min_data_points
            blended = weight * personal_pred + (1 - weight) * global_pred

            self._log_cold_start('hybrid', user_id, user_data_count)
            return blended

        else:
            # Pure cold start: use global model
            self._log_cold_start('cold', user_id, 0)
            return self.default_model.predict(features)

    def _get_user_data_count(self, user_id: str) -> int:
        """Get number of data points for user"""
        # In production: query database
        return random.randint(0, 100)

    def _log_cold_start(self, strategy: str, user_id: str, data_count: int):
        """Log cold start handling for monitoring"""
        self.cold_start_log.append({
            'user_id': user_id,
            'strategy': strategy,
            'data_count': data_count,
            'timestamp': datetime.now()
        })

# Mock usage
class MockModel:
    def predict(self, features):
        return 0.75

handler = ColdStartHandler(
    default_model=MockModel(),
    user_models={'user123': MockModel()}
)

# New user prediction
pred = handler.predict_with_fallback(
    user_id='new_user_456',
    features={'amount': 100},
    min_data_points=50
)

print(f"Cold start prediction: {pred:.2f}")
print(f"Cold start cases handled: {len(handler.cold_start_log)}")

from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
from typing import Dict

class ModelPerformanceMonitor:
    """Monitor model performance degradation in production"""

    def __init__(self, baseline_metrics: Dict[str, float]):
        self.baseline_metrics = baseline_metrics
        self.degradation_threshold = 0.05  # 5% drop triggers alert
        self.performance_history = []

    def evaluate_production_performance(
        self,
        y_true: np.ndarray,
        y_pred: np.ndarray,
        timestamp: datetime = None
    ) -> Dict:
        """Evaluate current production performance"""

        current_metrics = {
            'accuracy': accuracy_score(y_true, y_pred),
            'precision': precision_score(y_true, y_pred, average='weighted', zero_division=0),
            'recall': recall_score(y_true, y_pred, average='weighted', zero_division=0),
            'f1': f1_score(y_true, y_pred, average='weighted', zero_division=0)
        }

        # Calculate degradation
        degradation = {}
        alert_triggered = False

        for metric_name, current_value in current_metrics.items():
            baseline_value = self.baseline_metrics.get(metric_name, current_value)
            degradation_pct = (baseline_value - current_value) / baseline_value if baseline_value > 0 else 0

            degradation[metric_name] = {
                'baseline': baseline_value,
                'current': current_value,
                'degradation_pct': degradation_pct,
                'alert': degradation_pct > self.degradation_threshold
            }

            if degradation_pct > self.degradation_threshold:
                alert_triggered = True

        # Log performance
        self.performance_history.append({
            'timestamp': timestamp or datetime.now(),
            'metrics': current_metrics,
            'degradation': degradation,
            'alert': alert_triggered
        })

        return {
            'current_metrics': current_metrics,
            'degradation_analysis': degradation,
            'requires_attention': alert_triggered
        }

    def generate_alert(self, analysis: Dict) -> str:
        """Generate human-readable performance alert"""

        if not analysis['requires_attention']:
            return "✅ Model performance within acceptable range"

        alert = "🚨 MODEL PERFORMANCE DEGRADATION DETECTED\n\n"

        for metric, details in analysis['degradation_analysis'].items():
            if details['alert']:
                alert += f"⚠️  {metric.upper()}:\n"
                alert += f"   Baseline: {details['baseline']:.3f}\n"
                alert += f"   Current:  {details['current']:.3f}\n"
                alert += f"   Drop:     {details['degradation_pct']:.1%}\n\n"

        alert += "RECOMMENDED ACTIONS:\n"
        alert += "1. Check for data drift\n"
        alert += "2. Analyze recent production data distribution\n"
        alert += "3. Consider model retraining\n"
        alert += "4. Review recent system changes\n"

        return alert

# Usage
baseline = {
    'accuracy': 0.92,
    'precision': 0.89,
    'recall': 0.88,
    'f1': 0.885
}

monitor = ModelPerformanceMonitor(baseline)

# Simulate degraded performance
y_true = np.array([1, 0, 1, 1, 0, 1, 0, 0, 1, 1])
y_pred = np.array([1, 0, 0, 1, 0, 1, 1, 0, 0, 1])  # Worse than baseline

analysis = monitor.evaluate_production_performance(y_true, y_pred)
print(monitor.generate_alert(analysis))

import asyncio
import time
from typing import Callable

class LoadTester:
    """Load test AI endpoints before production"""

    def __init__(self, endpoint_function: Callable):
        self.endpoint = endpoint_function
        self.results = []

    async def _make_request(self, request_id: int):
        """Simulate single request"""
        start_time = time.time()

        try:
            await self.endpoint()
            latency = time.time() - start_time
            return {'request_id': request_id, 'latency': latency, 'success': True}
        except Exception as e:
            latency = time.time() - start_time
            return {'request_id': request_id, 'latency': latency, 'success': False, 'error': str(e)}

    async def run_load_test(
        self,
        num_requests: int,
        concurrent_users: int
    ):
        """Run load test with concurrent requests"""

        print(f"Starting load test: {num_requests} requests, {concurrent_users} concurrent users")

        # Create batches of concurrent requests
        for batch_start in range(0, num_requests, concurrent_users):
            batch_size = min(concurrent_users, num_requests - batch_start)

            tasks = [
                self._make_request(batch_start + i)
                for i in range(batch_size)
            ]

            batch_results = await asyncio.gather(*tasks)
            self.results.extend(batch_results)

            # Brief pause between batches
            await asyncio.sleep(0.1)

    def analyze_results(self) -> Dict:
        """Analyze load test results"""

        latencies = [r['latency'] for r in self.results if r['success']]
        failures = [r for r in self.results if not r['success']]

        if not latencies:
            return {'error': 'No successful requests'}

        analysis = {
            'total_requests': len(self.results),
            'successful_requests': len(latencies),
            'failed_requests': len(failures),
            'success_rate': len(latencies) / len(self.results),
            'latency_p50': np.percentile(latencies, 50),
            'latency_p95': np.percentile(latencies, 95),
            'latency_p99': np.percentile(latencies, 99),
            'latency_max': max(latencies),
            'latency_mean': np.mean(latencies),
        }

        # Determine if system can handle production load
        analysis['production_ready'] = (
            analysis['success_rate'] > 0.99 and
            analysis['latency_p95'] < 0.2  # 200ms p95 latency
        )

        return analysis

# Mock async endpoint
async def mock_ai_endpoint():
    await asyncio.sleep(0.05)  # Simulate 50ms latency
    if random.random() < 0.01:  # 1% failure rate
        raise Exception("Model inference failed")

# Usage
async def run_test():
    tester = LoadTester(mock_ai_endpoint)
    await tester.run_load_test(num_requests=1000, concurrent_users=50)

    analysis = tester.analyze_results()
    print("\n=== LOAD TEST RESULTS ===")
    print(f"Success Rate: {analysis['success_rate']:.1%}")
    print(f"P50 Latency: {analysis['latency_p50']*1000:.1f}ms")
    print(f"P95 Latency: {analysis['latency_p95']*1000:.1f}ms")
    print(f"P99 Latency: {analysis['latency_p99']*1000:.1f}ms")
    print(f"\nProduction Ready: {analysis['production_ready']}")

# Run test
# asyncio.run(run_test())

from dataclasses import dataclass

@dataclass
class ScalingPolicy:
    metric_name: str
    target_value: float
    min_instances: int
    max_instances: int
    scale_up_threshold: float
    scale_down_threshold: float
    cooldown_seconds: int

class MLAutoScaler:
    """Auto-scaling for ML inference workloads"""

    def __init__(self, policy: ScalingPolicy):
        self.policy = policy
        self.current_instances = policy.min_instances
        self.last_scale_time = datetime.now()
        self.scaling_history = []

    def evaluate_scaling_decision(
        self,
        current_metric_value: float
    ) -> int:
        """
        Determine if scaling is needed
        Returns: number of instances to add/remove (positive = scale up, negative = scale down)
        """

        # Check cooldown period
        time_since_last_scale = (datetime.now() - self.last_scale_time).total_seconds()
        if time_since_last_scale < self.policy.cooldown_seconds:
            return 0  # Still in cooldown

        # Calculate how far from target
        target_ratio = current_metric_value / self.policy.target_value

        # Scale up if significantly above target
        if target_ratio > self.policy.scale_up_threshold:
            # Calculate how many instances needed
            desired_instances = int(self.current_instances * target_ratio)
            instances_to_add = min(
                desired_instances - self.current_instances,
                self.policy.max_instances - self.current_instances
            )

            if instances_to_add > 0:
                self._record_scaling_event('scale_up', instances_to_add, current_metric_value)
                self.current_instances += instances_to_add
                self.last_scale_time = datetime.now()
                return instances_to_add

        # Scale down if significantly below target
        elif target_ratio < self.policy.scale_down_threshold:
            desired_instances = max(
                int(self.current_instances * target_ratio),
                self.policy.min_instances
            )
            instances_to_remove = self.current_instances - desired_instances

            if instances_to_remove > 0:
                self._record_scaling_event('scale_down', -instances_to_remove, current_metric_value)
                self.current_instances -= instances_to_remove
                self.last_scale_time = datetime.now()
                return -instances_to_remove

        return 0  # No scaling needed

    def _record_scaling_event(self, action: str, change: int, metric_value: float):
        """Record scaling event for analysis"""
        self.scaling_history.append({
            'timestamp': datetime.now(),
            'action': action,
            'change': change,
            'instances_before': self.current_instances,
            'instances_after': self.current_instances + change,
            'metric_value': metric_value,
            'target_value': self.policy.target_value
        })

# Usage
policy = ScalingPolicy(
    metric_name="queue_depth",
    target_value=10.0,  # Target 10 requests in queue
    min_instances=2,
    max_instances=20,
    scale_up_threshold=1.5,  # Scale up if 50% above target
    scale_down_threshold=0.5,  # Scale down if 50% below target
    cooldown_seconds=300  # 5 minute cooldown
)

scaler = MLAutoScaler(policy)

# Simulate high load
current_queue_depth = 25  # Well above target of 10
scaling_decision = scaler.evaluate_scaling_decision(current_queue_depth)

if scaling_decision > 0:
    print(f"🔼 Scaling UP by {scaling_decision} instances")
elif scaling_decision < 0:
    print(f"🔽 Scaling DOWN by {abs(scaling_decision)} instances")
else:
    print("➡️  No scaling action needed")

print(f"Current instances: {scaler.current_instances}")

import re

class NotebookRefactor:
    """Refactor notebook code to production-ready modules"""

    @staticmethod
    def extract_function(notebook_code: str) -> Dict:
        """Extract reusable functions from notebook cells"""

        # Find function definitions
        function_pattern = r'def\s+(\w+)\s*\([^)]*\):'
        functions = re.findall(function_pattern, notebook_code)

        # Extract imports
        import_pattern = r'^import\s+\w+|^from\s+\w+\s+import'
        imports = re.findall(import_pattern, notebook_code, re.MULTILINE)

        return {
            'functions_found': len(functions),
            'function_names': functions,
            'imports': imports,
            'needs_refactoring': len(functions) < 3  # Too few reusable functions
        }

    @staticmethod
    def identify_hardcoded_values(code: str) -> List[Dict]:
        """Find hardcoded values that should be config"""

        issues = []

        # Find hardcoded file paths
        path_pattern = r'["\'](/[^"\']+|[A-Z]:\\[^"\']+)["\']'
        paths = re.findall(path_pattern, code)
        if paths:
            issues.append({
                'type': 'hardcoded_path',
                'count': len(paths),
                'examples': paths[:3]
            })

        # Find hardcoded numbers (magic numbers)
        # Exclude common cases like [0] or range(10)
        number_pattern = r'\b\d{4,}\b'  # Numbers with 4+ digits
        magic_numbers = re.findall(number_pattern, code)
        if magic_numbers:
            issues.append({
                'type': 'magic_numbers',
                'count': len(magic_numbers),
                'examples': magic_numbers[:3]
            })

        return issues

# Usage
notebook_code = """
import pandas as pd
data = pd.read_csv('/Users/john/data.csv')
model.fit(data, epochs=10000, batch_size=512)
"""

refactor = NotebookRefactor()
analysis = refactor.extract_function(notebook_code)
issues = refactor.identify_hardcoded_values(notebook_code)

print("Refactoring Analysis:")
print(f"  Functions found: {analysis['functions_found']}")
print(f"  Needs refactoring: {analysis['needs_refactoring']}")
print(f"\nIssues found: {len(issues)}")
for issue in issues:
    print(f"  - {issue['type']}: {issue['count']} instances")

import logging
from prometheus_client import Counter, Histogram, Gauge

class MLMonitoring:
    """Production ML monitoring with Prometheus metrics"""

    def __init__(self, model_name: str):
        self.model_name = model_name

        # Prediction metrics
        self.prediction_counter = Counter(
            'ml_predictions_total',
            'Total number of predictions',
            ['model', 'version']
        )

        self.prediction_latency = Histogram(
            'ml_prediction_latency_seconds',
            'Prediction latency',
            ['model', 'version']
        )

        self.prediction_confidence = Histogram(
            'ml_prediction_confidence',
            'Model confidence scores',
            ['model', 'version'],
            buckets=[0.5, 0.6, 0.7, 0.8, 0.9, 0.95, 0.99, 1.0]
        )

        # Model performance metrics
        self.model_accuracy = Gauge(
            'ml_model_accuracy',
            'Current model accuracy',
            ['model', 'version']
        )

        self.data_drift_score = Gauge(
            'ml_data_drift_score',
            'Data drift PSI score',
            ['model', 'feature']
        )

        # Error tracking
        self.error_counter = Counter(
            'ml_errors_total',
            'Total errors',
            ['model', 'error_type']
        )

    def record_prediction(
        self,
        version: str,
        latency: float,
        confidence: float
    ):
        """Record a single prediction"""
        self.prediction_counter.labels(
            model=self.model_name,
            version=version
        ).inc()

        self.prediction_latency.labels(
            model=self.model_name,
            version=version
        ).observe(latency)

        self.prediction_confidence.labels(
            model=self.model_name,
            version=version
        ).observe(confidence)

    def record_error(self, error_type: str):
        """Record prediction error"""
        self.error_counter.labels(
            model=self.model_name,
            error_type=error_type
        ).inc()

    def update_model_metrics(
        self,
        version: str,
        accuracy: float
    ):
        """Update model performance metrics"""
        self.model_accuracy.labels(
            model=self.model_name,
            version=version
        ).set(accuracy)

    def record_data_drift(self, feature: str, psi_score: float):
        """Record data drift for a feature"""
        self.data_drift_score.labels(
            model=self.model_name,
            feature=feature
        ).set(psi_score)

# Usage
monitor = MLMonitoring(model_name="fraud_detector")

# Record predictions
monitor.record_prediction(version="1.2.0", latency=0.045, confidence=0.92)
monitor.update_model_metrics(version="1.2.0", accuracy=0.89)
monitor.record_data_drift(feature="transaction_amount", psi_score=0.08)

print("Metrics recorded successfully")

@dataclass
class ProductionHandoff:
    """Structured handoff from data science to engineering"""

    # Model artifacts
    model_location: str
    model_version: str
    training_data_hash: str

    # Performance baselines
    baseline_accuracy: float
    baseline_latency_p95_ms: float
    expected_qps: int

    # Dependencies
    python_version: str
    dependencies_file: str
    required_env_vars: List[str]

    # Monitoring
    metrics_dashboard_url: str
    alert_recipients: List[str]
    escalation_contact: str

    # Documentation
    model_card_url: str
    api_docs_url: str
    runbook_url: str

    # Compliance
    data_privacy_review_completed: bool
    security_scan_completed: bool
    legal_approval_obtained: bool

    def validate_handoff(self) -> Tuple[bool, List[str]]:
        """Validate all handoff requirements are met"""
        issues = []

        # Check critical fields
        if not self.model_location:
            issues.append("Model location not specified")

        if self.baseline_accuracy < 0.7:
            issues.append(f"Accuracy too low: {self.baseline_accuracy:.2%}")

        if self.baseline_latency_p95_ms > 500:
            issues.append(f"Latency too high: {self.baseline_latency_p95_ms}ms")

        if not self.data_privacy_review_completed:
            issues.append("Data privacy review not completed")

        if not self.security_scan_completed:
            issues.append("Security scan not completed")

        if not self.model_card_url:
            issues.append("Model card documentation missing")

        return len(issues) == 0, issues

# Usage
handoff = ProductionHandoff(
    model_location="s3://models/fraud-v1.2.0.pkl",
    model_version="1.2.0",
    training_data_hash="abc123",
    baseline_accuracy=0.89,
    baseline_latency_p95_ms=85,
    expected_qps=500,
    python_version="3.10",
    dependencies_file="requirements.txt",
    required_env_vars=["MODEL_KEY", "DB_CONNECTION"],
    metrics_dashboard_url="https://grafana.company.com/fraud-model",
    alert_recipients=["ml-team@company.com"],
    escalation_contact="ml-lead@company.com",
    model_card_url="https://docs.company.com/models/fraud",
    api_docs_url="https://api-docs.company.com/fraud",
    runbook_url="https://wiki.company.com/fraud-runbook",
    data_privacy_review_completed=True,
    security_scan_completed=True,
    legal_approval_obtained=True
)

is_valid, issues = handoff.validate_handoff()
if is_valid:
    print("✅ Handoff validation passed - ready for production")
else:
    print("❌ Handoff validation failed:")
    for issue in issues:
        print(f"  - {issue}")

class BusinessMetricsTracker:
    """Track business KPIs alongside model metrics"""

    def __init__(self):
        self.metrics_log = []

    def record_prediction_with_business_impact(
        self,
        model_prediction: float,
        model_confidence: float,
        business_outcome: str,  # 'converted', 'churned', 'fraud_confirmed', etc.
        business_value: float  # Revenue impact, cost savings, etc.
    ):
        """Record both model and business metrics"""

        self.metrics_log.append({
            'timestamp': datetime.now(),
            'model_prediction': model_prediction,
            'model_confidence': model_confidence,
            'business_outcome': business_outcome,
            'business_value': business_value
        })

    def calculate_business_roi(self) -> Dict:
        """Calculate ROI of the AI system"""

        if not self.metrics_log:
            return {'error': 'No data'}

        # Calculate total business value generated
        total_value = sum(m['business_value'] for m in self.metrics_log)

        # Calculate lift from AI
        # Compare predictions to baseline (e.g., random or rule-based)
        ai_decisions = [m for m in self.metrics_log if m['model_confidence'] > 0.7]
        ai_value = sum(m['business_value'] for m in ai_decisions)

        return {
            'total_predictions': len(self.metrics_log),
            'high_confidence_predictions': len(ai_decisions),
            'total_business_value': total_value,
            'ai_driven_value': ai_value,
            'value_per_prediction': total_value / len(self.metrics_log) if self.metrics_log else 0
        }

# Usage
tracker = BusinessMetricsTracker()

# Fraud detection example
tracker.record_prediction_with_business_impact(
    model_prediction=0.92,  # 92% fraud probability
    model_confidence=0.95,
    business_outcome='fraud_confirmed',
    business_value=2500  # Prevented $2500 fraud loss
)

tracker.record_prediction_with_business_impact(
    model_prediction=0.15,  # 15% fraud probability
    model_confidence=0.88,
    business_outcome='legitimate',
    business_value=0  # No fraud prevented
)

roi = tracker.calculate_business_roi()
print(f"Total business value: ${roi['total_business_value']:,.2f}")
print(f"Value per prediction: ${roi['value_per_prediction']:.2f}")

class CanaryDeployment:
    """Gradual rollout with automatic rollback"""

    def __init__(self, baseline_model, candidate_model):
        self.baseline = baseline_model
        self.candidate = candidate_model
        self.traffic_split = 0.0  # Start at 0% for candidate
        self.metrics = {'baseline': [], 'candidate': []}

    def route_request(self, request):
        """Route request to baseline or candidate model"""
        if random.random() < self.traffic_split:
            # Route to candidate
            result = self.candidate.predict(request)
            self.metrics['candidate'].append(result)
            return result
        else:
            # Route to baseline
            result = self.baseline.predict(request)
            self.metrics['baseline'].append(result)
            return result

    def evaluate_canary_metrics(self) -> bool:
        """Check if candidate is performing well"""
        if len(self.metrics['candidate']) < 100:
            return True  # Need more data

        # Compare error rates
        baseline_errors = sum(1 for m in self.metrics['baseline'][-1000:] if m.get('error'))
        candidate_errors = sum(1 for m in self.metrics['candidate'][-100:] if m.get('error'))

        baseline_error_rate = baseline_errors / min(1000, len(self.metrics['baseline']))
        candidate_error_rate = candidate_errors / len(self.metrics['candidate'])

        # Rollback if candidate has 50% more errors
        if candidate_error_rate > baseline_error_rate * 1.5:
            print(f"🚨 Canary failing! Candidate error rate: {candidate_error_rate:.2%} vs baseline: {baseline_error_rate:.2%}")
            return False

        return True

    def increment_traffic(self, step: float = 0.1):
        """Gradually increase candidate traffic"""
        if self.evaluate_canary_metrics():
            self.traffic_split = min(1.0, self.traffic_split + step)
            print(f"✅ Canary healthy. Increasing traffic to {self.traffic_split:.0%}")
            return True
        else:
            print("❌ Canary unhealthy. Rolling back to baseline.")
            self.traffic_split = 0.0
            return False

# Mock usage
class MockModel:
    def predict(self, request):
        return {'prediction': 0.75, 'error': random.random() < 0.01}

deployer = CanaryDeployment(
    baseline_model=MockModel(),
    candidate_model=MockModel()
)

# Simulate gradual rollout
print("Starting canary deployment...")
for step in range(10):
    # Simulate traffic
    for _ in range(50):
        deployer.route_request({'data': 'test'})

    # Try to increase traffic
    if not deployer.increment_traffic(step=0.1):
        break

print(f"\nFinal traffic split: {deployer.traffic_split:.0%} to candidate")