import kfp
from kfp import dsl
from kfp.components import create_component_from_func

@create_component_from_func
def validate_data(input_path: str, schema_path: str) -> str:
    import pandas as pd
    from pandera import Check, Column, DataFrameSchema
    # Define validation schema
    schema = DataFrameSchema({
        "feature_a": Column(float, checks=Check.greater_than(0)),
        "feature_b": Column(int, nullable=True),
        "target": Column(int, checks=Check.isin([0, 1]))
    })
    df = pd.read_parquet(input_path)
    validated_df = schema.validate(df)
    output_path = '/tmp/validated_data.parquet'
    validated_df.to_parquet(output_path)
    return output_path

@create_component_from_func
def train_model(data_path: str, hyperparam_c: float) -> str:
    from sklearn.ensemble import RandomForestClassifier
    import joblib
    import pandas as pd
    df = pd.read_parquet(data_path)
    X, y = df.drop('target', axis=1), df['target']
    model = RandomForestClassifier(n_estimators=100, max_depth=10, random_state=42)
    model.fit(X, y)
    model_path = '/tmp/model.joblib'
    joblib.dump(model, model_path)
    return model_path

@dsl.pipeline(name='customer-churn-pipeline')
def ml_pipeline(input_path: str, schema_path: str, c_param: float = 1.0):
    validate_task = validate_data(input_path=input_path, schema_path=schema_path)
    train_task = train_model(data_path=validate_task.output, hyperparam_c=c_param)

# Compile and run
kfp.compiler.Compiler().compile(ml_pipeline, 'pipeline.yaml')

import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient()
# Log a run
with mlflow.start_run() as run:
    mlflow.log_param("alpha", 0.5)
    mlflow.log_metric("rmse", 0.87)
    mlflow.sklearn.log_model(sk_model, "model")
    run_id = run.info.run_id

# Register the model
model_name = "PropensityModel"
model_uri = f"runs:/{run_id}/model"
mv = client.create_model_version(model_name, model_uri, run_id)
print(f"Model Version: {mv.version}")

# Transition stage programmatically
client.transition_model_version_stage(
    name=model_name,
    version=mv.version,
    stage="Staging",
    archive_existing_versions=True
)

import numpy as np
from alibi_detect.cd import KSDrift
from datetime import datetime, timedelta

class ProductionDriftDetector:
    def __init__(self, reference_data, threshold=0.05):
        # Initialize detector with training/reference data
        self.detector = KSDrift(reference_data, p_val=threshold)
        self.drift_log = []

    def check_drift(self, recent_production_batch):
        """Check a batch of recent production features for drift."""
        preds = self.detector.predict(recent_production_batch)
        is_drift = preds['data']['is_drift'] == 1
        if is_drift:
            self._trigger_alert(preds)
            self._queue_retraining()
        return is_drift, preds['data']['p_val']

    def _trigger_alert(self, prediction_metadata):
        alert_msg = {
            "timestamp": datetime.utcnow().isoformat(),
            "severity": "HIGH",
            "metric": "Kolmogorov-Smirnov",
            "p_value": float(prediction_metadata['data']['p_val']),
            "features_drifted": prediction_metadata['data']['distance'].tolist()
        }
        # Send to alerting system (e.g., PagerDuty, Slack webhook)
        send_alert(alert_msg)

    def _queue_retraining(self):
        # Place a message in a queue (e.g., SQS, RabbitMQ) to trigger pipeline
        retrain_message = {"trigger": "drift", "timestamp": datetime.utcnow().isoformat()}
        queue_client.send_message(retrain_message)

# Usage
detector = ProductionDriftDetector(training_features_reference)
drift_detected, p_val = detector.check_drift(last_hour_production_features)

# main.tf - Provisioning a scalable model endpoint on AWS
resource "aws_s3_bucket" "model_artifacts" {
  bucket = "mlops-artifacts-${var.environment}"
  acl    = "private"
  versioning {
    enabled = true
  }
  server_side_encryption_configuration {
    rule {
      apply_server_side_encryption_by_default {
        sse_algorithm = "AES256"
      }
    }
  }
}

resource "aws_ecr_repository" "inference_api" {
  name                 = "model-inference-api"
  image_tag_mutability = "IMMUTABLE"
  image_scanning_configuration {
    scan_on_push = true
  }
}

resource "aws_ecs_cluster" "inference_cluster" {
  name = "model-inference-cluster"
  setting {
    name  = "containerInsights"
    value = "enabled"
  }
}

resource "aws_ecs_task_definition" "model_task" {
  family                   = "model-serving-task"
  cpu                      = "1024"
  memory                   = "2048"
  network_mode             = "awsvpc"
  requires_compatibilities = ["FARGATE"]
  execution_role_arn       = aws_iam_role.ecs_task_execution_role.arn
  container_definitions    = jsonencode([{
    name  = "model-container",
    image = "${aws_ecr_repository.inference_api.repository_url}:latest",
    portMappings = [{ containerPort = 8080, hostPort = 8080 }],
    environment = [
      { name = "MODEL_URI", value = "s3://${aws_s3_bucket.model_artifacts.bucket}/prod/model.joblib" },
      { name = "LOG_LEVEL", value = "INFO" }
    ],
    logConfiguration = {
      logDriver = "awslogs",
      options = {
        "awslogs-group"         = "/ecs/model-inference",
        "awslogs-region"        = var.aws_region,
        "awslogs-stream-prefix" = "ecs"
      }
    }
  }])
}

# feature_engineering/transforms.py
import pandas as pd
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
import joblib

class TemporalFeatureEngineer(BaseEstimator, TransformerMixin):
    """Engineer features from datetime columns."""
    def __init__(self, date_column):
        self.date_column = date_column

    def fit(self, X, y=None):
        return self

    def transform(self, X):
        X_ = X.copy()
        X_[self.date_column] = pd.to_datetime(X_[self.date_column])
        X_['day_of_week'] = X_[self.date_column].dt.dayofweek
        X_['month'] = X_[self.date_column].dt.month
        X_['is_weekend'] = X_['day_of_week'].isin([5, 6]).astype(int)
        X_['hour'] = X_[self.date_column].dt.hour
        # Cyclical encoding for hour
        X_['hour_sin'] = np.sin(2 * np.pi * X_['hour']/24)
        X_['hour_cos'] = np.cos(2 * np.pi * X_['hour']/24)
        return X_.drop(columns=[self.date_column])

# Save the fitted pipeline for consistent training/serving
feature_pipeline = Pipeline([
    ('temporal', TemporalFeatureEngineer('timestamp')),
    # ... other steps
])
joblib.dump(feature_pipeline, 'artifacts/feature_pipeline.joblib')

# tests/test_data_validation.py
import pandas as pd
import pytest
from pandera import DataFrameSchema, Column, Check

def test_data_schema_integrity():
    """Validate that production data matches the expected schema."""
    schema = DataFrameSchema({
        "user_id": Column(str, nullable=False),
        "transaction_amount": Column(float, checks=[
            Check.greater_than_or_equal_to(0),
            Check.less_than_or_equal_to(100000)
        ]),
        "product_category": Column(str, checks=Check.isin(["Electronics", "Clothing", "Books"]))
    })
    sample_data = pd.read_parquet("tests/sample_production_data.parquet")
    # This will raise a SchemaError if validation fails
    validated_data = schema.validate(sample_data, lazy=True)
    assert len(validated_data) > 0

# tests/test_model_inference.py
def test_model_prediction_shape_and_range():
    """Ensure the model produces valid predictions."""
    model = joblib.load('artifacts/model.joblib')
    test_batch = pd.read_parquet('tests/test_batch.parquet')
    predictions = model.predict_proba(test_batch)
    # Check shape matches
    assert predictions.shape == (len(test_batch), 2)
    # Check probabilities sum to ~1
    assert np.allclose(predictions.sum(axis=1), 1.0, atol=1e-5)
    # Check probability values are within [0,1]
    assert (predictions >= 0).all() and (predictions <= 1).all()

# .github/workflows/cd-ml-pipeline.yml
name: CD Model Deployment
on:
  workflow_run:
    workflows: ["CI Model Pipeline"]
    types:
      - completed
  push:
    branches: [ main ]

jobs:
  build-and-push:
    if: ${{ github.event.workflow_run.conclusion == 'success' }}
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
        with:
          fetch-depth: 0
      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2
      - name: Log in to Container Registry
        uses: docker/login-action@v2
        with:
          registry: ${{ secrets.REGISTRY_URL }}
          username: ${{ secrets.REGISTRY_USERNAME }}
          password: ${{ secrets.REGISTRY_PASSWORD }}
      - name: Extract model metadata
        id: meta
        run: |
          MODEL_VERSION=$(python -c "import mlflow; client = mlflow.tracking.MlflowClient(); print(client.get_latest_versions('ChurnModel')[0].version)")
          echo "MODEL_VERSION=$MODEL_VERSION" >> $GITHUB_OUTPUT
      - name: Build and push Docker image
        uses: docker/build-push-action@v4
        with:
          context: .
          file: ./Dockerfile.serve
          push: true
          tags: |
            ${{ secrets.REGISTRY_URL }}/churn-model:${{ steps.meta.outputs.MODEL_VERSION }}
            ${{ secrets.REGISTRY_URL }}/churn-model:latest
          cache-from: type=gha
          cache-to: type=gha,mode=max
      - name: Deploy to Staging Kubernetes
        run: |
          kubectl set image deployment/churn-model-api \
            churn-model-container=${{ secrets.REGISTRY_URL }}/churn-model:${{ steps.meta.outputs.MODEL_VERSION }} \
            --namespace=staging
          kubectl rollout status deployment/churn-model-api --namespace=staging --timeout=300s

# prometheus/model_monitoring_rules.yml
groups:
  - name: model_observability
    rules:
      - alert: HighPredictionLatency
        expr: histogram_quantile(0.95, rate(model_inference_duration_seconds_bucket[5m])) > 0.2
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Model inference p95 latency above 200ms"
          description: "The {{ $labels.model_name }} model has a 95th percentile latency of {{ $value }}s for the last 5 minutes."

      - alert: DataDriftDetected
        expr: model_feature_drift_score > 0.15
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Significant data drift detected for {{ $labels.model_name }}"
          description: "Drift score {{ $value }} exceeds threshold. Investigate upstream data sources."

      - alert: AccuracyDegradation
        expr: (model_accuracy offset 5m) - model_accuracy < -0.05
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "Model accuracy degradation for {{ $labels.model_name }}"
          description: "Accuracy dropped by over 5% in the last 5 minutes. Current value: {{ $value }}"

# modules/ml_platform/main.tf - Production-Grade MLOps Infrastructure
variable "environment" { default = "prod" }
variable "region" { default = "us-east-1" }

# 1. Core Data & Artifact Storage
resource "aws_s3_bucket" "ml_data_lake" {
  bucket = "ml-data-lake-${var.environment}-${random_id.suffix.hex}"
  acl    = "private"
  versioning { enabled = true }
  lifecycle_rule {
    id      = "archive_old_models"
    enabled = true
    prefix  = "models/"
    transition {
      days          = 90
      storage_class = "GLACIER"
    }
  }
  server_side_encryption_configuration {
    rule {
      apply_server_side_encryption_by_default {
        sse_algorithm = "aws:kms"
      }
    }
  }
  tags = { Terraform = "true", Purpose = "MLOps" }
}

# 2. Container Registry for Model Images
resource "aws_ecr_repository" "model_registry" {
  name                 = "model-registry-${var.environment}"
  image_tag_mutability = "IMMUTABLE"
  encryption_configuration {
    encryption_type = "KMS"
  }
  image_scanning_configuration { scan_on_push = true }
}

# 3. Managed Feature Store (using AWS SageMaker Feature Store)
resource "aws_sagemaker_feature_group" "offline_store" {
  feature_group_name = "transactions-${var.environment}"
  record_identifier_feature_name  = "transaction_id"
  event_time_feature_name = "event_time"
  role_arn = aws_iam_role.sagemaker_featurestore.arn
  offline_store_config {
    s3_storage_config {
      s3_uri = "s3://${aws_s3_bucket.ml_data_lake.bucket}/feature_store/"
    }
  }
  online_store_config {
    enable_online_store = true
  }
  feature_definition {
    feature_name = "transaction_id"
    feature_type = "Integral"
  }
  feature_definition {
    feature_name = "amount"
    feature_type = "Fractional"
  }
}

# 4. Kubernetes Cluster for Model Serving (EKS)
module "eks" {
  source  = "terraform-aws-modules/eks/aws"
  version = "~> 19.0"
  cluster_name    = "ml-serving-${var.environment}"
  cluster_version = "1.27"
  vpc_id     = module.vpc.vpc_id
  subnet_ids = module.vpc.private_subnets
  cluster_endpoint_private_access = true
  cluster_endpoint_public_access  = true
  eks_managed_node_groups = {
    model_serving = {
      min_size     = 1
      max_size     = 10
      desired_size = 3
      instance_types = ["m5.large", "m5.xlarge"]
      capacity_type  = "SPOT"
      labels = { workload = "model-inference" }
      taints = { dedicated = { key = "workload", value = "model-inference", effect = "NO_SCHEDULE" } }
    }
  }
  node_security_group_additional_rules = {
    ingress_allow_access_from_control_plane = {
      type                          = "ingress"
      protocol                      = "tcp"
      from_port                     = 1025
      to_port                       = 65535
      source_cluster_security_group = true
    }
  }
}

# 5. MLflow Tracking Server (Managed on ECS Fargate)
resource "aws_ecs_cluster" "mlflow" {
  name = "mlflow-tracking-${var.environment}"
}
resource "aws_ecs_task_definition" "mlflow" {
  family                   = "mlflow-server"
  cpu                      = "1024"
  memory                   = "2048"
  network_mode             = "awsvpc"
  requires_compatibilities = ["FARGATE"]
  execution_role_arn       = aws_iam_role.ecs_task_execution_role.arn
  container_definitions    = jsonencode([{
    name  = "mlflow",
    image = "ghcr.io/mlflow/mlflow:latest",
    portMappings = [{ containerPort = 5000, hostPort = 5000 }],
    environment = [
      { name = "MLFLOW_S3_ENDPOINT_URL", value = "https://s3.${var.region}.amazonaws.com" },
      { name = "AWS_DEFAULT_REGION", value = var.region },
      { name = "MLFLOW_S3_BUCKET", value = aws_s3_bucket.ml_data_lake.bucket }
    ],
    secrets = [
      { name = "DATABASE_URL", valueFrom = aws_secretsmanager_secret.rds_connection.arn }
    ]
  }])
}

# Dockerfile.serve - Production Model Serving Image
# Stage 1: Builder
FROM python:3.9-slim as builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# Stage 2: Runtime
FROM python:3.9-slim as runtime
WORKDIR /app
# Copy installed packages from builder
COPY --from=builder /root/.local /root/.local
# Ensure scripts in .local are usable
ENV PATH=/root/.local/bin:$PATH
# Create a non-root user
RUN groupadd -r modeluser && useradd -r -g modeluser modeluser
# Copy model artifacts and application code
COPY --chown=modeluser:modeluser artifacts/ ./artifacts/
COPY --chown=modeluser:modeluser serve.py .
# Security: Run as non-root user
USER modeluser
# Health check
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD python -c "import requests; requests.get('http://localhost:8080/health', timeout=2)"
# Expose port
EXPOSE 8080
# Command to run the inference server
CMD ["gunicorn", "--bind", "0.0.0.0:8080", "--workers", "2", "--threads", "4", "serve:app"]

# serve.py
from flask import Flask, request, jsonify
import pandas as pd
import joblib
import logging
from prometheus_flask_exporter import PrometheusMetrics

app = Flask(__name__)
metrics = PrometheusMetrics(app)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Load artifacts (done once at startup)
model = joblib.load('artifacts/model.joblib')
feature_pipeline = joblib.load('artifacts/feature_pipeline.joblib')

@app.route('/predict', methods=['POST'])
@metrics.do_not_track()  # Exclude from default metrics
def predict():
    try:
        data = request.get_json()
        input_df = pd.DataFrame(data['instances'])
        # Apply the exact same transformation as during training
        processed_features = feature_pipeline.transform(input_df)
        predictions = model.predict_proba(processed_features).tolist()
        logger.info(f"Processed {len(input_df)} instances.")
        return jsonify({'predictions': predictions})
    except Exception as e:
        logger.error(f"Prediction error: {str(e)}", exc_info=True)
        return jsonify({'error': str(e)}), 400

@app.route('/health')
def health():
    return jsonify({'status': 'healthy'}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

# k8s/base/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: model-inference-api
  labels:
    app: model-inference
spec:
  replicas: 3
  selector:
    matchLabels:
      app: model-inference
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    metadata:
      labels:
        app: model-inference
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8080"
    spec:
      containers:
      - name: model-container
        image: ${IMAGE_REPO}:${IMAGE_TAG}
        ports:
        - containerPort: 8080
        env:
        - name: MODEL_VERSION
          valueFrom:
            fieldRef:
              fieldPath: metadata.labels['app.kubernetes.io/version']
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5
        volumeMounts:
        - name: model-secrets
          mountPath: /app/secrets
          readOnly: true
      volumes:
      - name: model-secrets
        secret:
          secretName: model-credentials
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
            - matchExpressions:
              - key: workload
                operator: In
                values:
                - model-inference
---
# k8s/base/hpa.yaml - Auto-scaling based on CPU
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: model-inference-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: model-inference-api
  minReplicas: 2
  maxReplicas: 15
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

# .gitlab-ci.yml - End-to-End ML CI Pipeline
image: python:3.9-slim

variables:
  DVC_REMOTE: "s3://my-ml-bucket"
  MLFLOW_TRACKING_URI: "http://mlflow-server:5000"

stages:
  - test
  - validate
  - train
  - evaluate
  - package

cache:
  paths:
    - .venv/
  key: ${CI_COMMIT_REF_SLUG}

before_script:
  - apt-get update && apt-get install -y git curl
  - pip install virtualenv
  - virtualenv .venv
  - source .venv/bin/activate
  - pip install -r requirements-ci.txt

# STAGE 1: Code and Data Integrity Tests
unit-tests:
  stage: test
  script:
    - python -m pytest tests/unit/ -v --cov=src --cov-report=xml
  artifacts:
    reports:
      coverage_report:
        coverage_format: cobertura
        path: coverage.xml
    paths:
      - test-reports/
  allow_failure: false

data-schema-test:
  stage: validate
  script:
    - python scripts/validate_data_schema.py
  artifacts:
    when: on_failure
    paths:
      - data_validation_report.html

# STAGE 2: Data Validation and Quality Gates
data-quality:
  stage: validate
  script:
    # Use Great Expectations for comprehensive data testing
    - python -m great_expectations checkpoint run new_data_checkpoint
  artifacts:
    when: on_failure
    paths:
      - great_expectations/uncommitted/data_docs/

# STAGE 3: Model Training on a Fixed Dataset
train-model-ci:
  stage: train
  script:
    - dvc pull -r myremote  # Pull versioned dataset
    - python src/train.py --config configs/ci_config.yaml
  artifacts:
    paths:
      - models/ci_model.joblib
      - metrics/ci_metrics.json
    expire_in: 1 week
  resource_group: training-job  # Ensures only one training runs at a time

# STAGE 4: Model Evaluation against Baseline
evaluate-model:
  stage: evaluate
  dependencies:
    - train-model-ci
  script:
    - python scripts/evaluate_model.py \
        --candidate models/ci_model.joblib \
        --baseline models/production_model.joblib \
        --validation-data data/processed/validation.parquet \
        --output report.json
    # Fail if candidate performs worse than baseline by threshold
    - python scripts/check_evaluation.py --report report.json --threshold 0.02
  artifacts:
    paths:
      - evaluation_report.json
      - model_comparison_plot.png

# STAGE 5: Package Successful Model
package-model:
  stage: package
  dependencies:
    - evaluate-model
  script:
    - mlflow models build-docker \
        -m "runs:/${MLFLOW_RUN_ID}/model" \
        -n "model-serving:${CI_COMMIT_SHORT_SHA}" \
        --enable-mlserver
    - docker push "${CI_REGISTRY_IMAGE}:${CI_COMMIT_SHORT_SHA}"
  only:
    - main
    - merge_requests

import pandas as pd
import numpy as np
from scipy import stats
import json
import sys

def validate_statistical_drift(current_data_path, reference_data_path, threshold=0.05):
    """Detect drift in key numerical features using KL Divergence."""
    current = pd.read_parquet(current_data_path)
    reference = pd.read_parquet(reference_data_path)
    drift_report = {}
    for col in current.select_dtypes(include=[np.number]).columns:
        # Discretize continuous distributions for KL divergence
        current_hist, _ = np.histogram(current[col].dropna(), bins=50, density=True)
        ref_hist, _ = np.histogram(reference[col].dropna(), bins=50, density=True)
        # Add small epsilon to avoid log(0)
        kl_div = stats.entropy(current_hist + 1e-10, ref_hist + 1e-10)
        drift_report[col] = {
            'kl_divergence': float(kl_div),
            'drift_detected': kl_div > threshold
        }
        if kl_div > threshold:
            print(f"WARNING: Significant drift detected in {col}: KL={kl_div:.4f}")
    return drift_report

if __name__ == "__main__":
    # Paths would be passed as args or from environment in CI
    drift_results = validate_statistical_drift('data/raw/new_batch.parquet',
                                               'data/raw/reference.parquet')
    with open('drift_report.json', 'w') as f:
        json.dump(drift_results, f, indent=2)
    # Fail the CI job if any critical feature has drifted
    critical_features = ['amount', 'user_age', 'transaction_count']
    if any(drift_results[f]['drift_detected'] for f in critical_features):
        sys.exit(1)  # This will fail the pipeline

import json
import sys
import argparse

def evaluate_model_performance(report_path, threshold):
    """Check if candidate model meets performance criteria vs baseline."""
    with open(report_path, 'r') as f:
        report = json.load(f)
    candidate_auc = report['candidate']['metrics']['roc_auc']
    baseline_auc = report['baseline']['metrics']['roc_auc']
    improvement = candidate_auc - baseline_auc
    # Check 1: Absolute performance threshold
    if candidate_auc < 0.75:
        print(f"FAIL: Candidate AUC {candidate_auc:.3f} below minimum threshold 0.75")
        return False
    # Check 2: Significant degradation check
    if improvement < -threshold:
        print(f"FAIL: Candidate degrades by {abs(improvement):.3f} (threshold: {threshold})")
        return False
    # Check 3: Check for fairness across subgroups (simplified)
    for subgroup, metrics in report['candidate']['subgroup_metrics'].items():
        if abs(metrics['auc'] - candidate_auc) > 0.08:
            print(f"WARNING: Potential bias in subgroup {subgroup}: AUC={metrics['auc']:.3f}")
    print(f"PASS: Candidate AUC {candidate_auc:.3f}, improvement {improvement:.3f}")
    return True

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('--report', required=True, help='Path to evaluation report JSON')
    parser.add_argument('--threshold', type=float, default=0.02,
                        help='Maximum allowed degradation vs baseline')
    args = parser.parse_args()
    if not evaluate_model_performance(args.report, args.threshold):
        sys.exit(1)  # Fail the CI job

# pipeline/training_flow.py
from prefect import flow, task, get_run_logger
from prefect.blocks.system import Secret
from prefect.artifacts import create_markdown_artifact
import pandas as pd
import mlflow
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report
import joblib
import json
from datetime import datetime
import boto3

@task(retries=2, retry_delay_seconds=30)
def extract_and_validate_data(data_uri: str, schema_path: str) -> pd.DataFrame:
    """Task to pull data from source and validate against schema."""
    logger = get_run_logger()
    logger.info(f"Extracting data from {data_uri}")
    # Example: Read from S3
    s3_client = boto3.client('s3')
    bucket, key = data_uri.replace("s3://", "").split("/", 1)
    obj = s3_client.get_object(Bucket=bucket, Key=key)
    df = pd.read_parquet(obj['Body'])
    # Validate schema using pandera
    import pandera as pa
    from pandera import Check
    schema = pa.DataFrameSchema({
        "feature_a": pa.Column(float, checks=Check.greater_than(0)),
        "feature_b": pa.Column(int, nullable=True),
        "label": pa.Column(int, checks=Check.isin([0, 1]))
    })
    validated_df = schema.validate(df)
    logger.info(f"Validated {len(validated_df)} records.")
    return validated_df

@task(cache_key_fn=lambda *args, **kwargs: datetime.today().date(),
      cache_expiration=86400)
def engineer_features(raw_df: pd.DataFrame) -> pd.DataFrame:
    """Task to create features. Cached for one day to avoid recomputation."""
    df = raw_df.copy()
    # Interaction features
    df['feat_interaction'] = df['feature_a'] * df['feature_b']
    # Binning
    df['feat_a_binned'] = pd.cut(df['feature_a'], bins=5, labels=False)
    # Temporal feature (if applicable)
    if 'timestamp' in df.columns:
        df['hour'] = pd.to_datetime(df['timestamp']).dt.hour
    return df

@task
def train_model(features: pd.DataFrame, hyperparameters: dict) -> dict:
    """Training task with integrated MLflow logging."""
    logger = get_run_logger()
    X = features.drop('label', axis=1)
    y = features['label']
    X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)
    model = RandomForestClassifier(**hyperparameters, random_state=42)
    model.fit(X_train, y_train)
    # Evaluate
    val_pred = model.predict(X_val)
    report = classification_report(y_val, val_pred, output_dict=True)
    # Log to MLflow
    mlflow.set_tracking_uri(Secret.load("mlflow-tracking-uri").get())
    with mlflow.start_run():
        mlflow.log_params(hyperparameters)
        mlflow.log_metrics({
            'accuracy': report['accuracy'],
            'precision_1': report['1']['precision'],
            'recall_1': report['1']['recall']
        })
        mlflow.sklearn.log_model(model, "model")
        run_id = mlflow.active_run().info.run_id
    logger.info(f"Training complete. MLflow Run ID: {run_id}")
    return {
        'model': model,
        'metrics': report,
        'run_id': run_id,
        'feature_columns': list(X.columns)
    }

@task
def validate_model(training_output: dict, validation_criteria: dict) -> bool:
    """Validation gate: checks performance against business criteria."""
    metrics = training_output['metrics']
    # Criterion 1: Minimum accuracy
    if metrics['accuracy'] < validation_criteria['min_accuracy']:
        return False
    # Criterion 2: Minimum precision for positive class (e.g., fraud)
    if metrics['1']['precision'] < validation_criteria['min_precision']:
        return False
    # Criterion 3: Maximum disparity between subgroups (simplified fairness)
    # In practice, this would use detailed subgroup analysis
    return True

@task
def register_model(training_output: dict, stage: str = "Staging"):
    """Register the model in MLflow Model Registry."""
    client = mlflow.tracking.MlflowClient()
    model_name = "ProductionClassifier"
    # Create model version
    mv = client.create_model_version(
        name=model_name,
        source=f"runs:/{training_output['run_id']}/model",
        run_id=training_output['run_id']
    )
    # Transition stage
    client.transition_model_version_stage(
        name=model_name,
        version=mv.version,
        stage=stage
    )
    # Log artifact
    create_markdown_artifact(
        key="model-training-report",
        markdown=f"""
        # Model Training Report
        **Run ID:** {training_output['run_id']}
        **Accuracy:** {training_output['metrics']['accuracy']:.3f}
        **Features Used:** {', '.join(training_output['feature_columns'][:5])}...
        """,
        description="Summary of the latest model training run"
    )
    return mv.version

@flow(name="automated-model-training", flow_run_name="training-{date}")
def automated_training_pipeline(
    data_uri: str = "s3://my-bucket/data/latest.parquet",
    hyperparams: dict = {"n_estimators": 200, "max_depth": 15}
):
    """Main Prefect flow orchestrating the training pipeline."""
    logger = get_run_logger()
    logger.info("Starting automated training pipeline.")
    # 1. Data Extraction & Validation
    raw_data = extract_and_validate_data(data_uri, "schemas/training_schema.yaml")
    # 2. Feature Engineering
    features = engineer_features(raw_data)
    # 3. Model Training with MLflow logging
    training_result = train_model(features, hyperparams)
    # 4. Validation Gate
    validation_passed = validate_model(training_result, {
        'min_accuracy': 0.85,
        'min_precision': 0.90
    })
    if not validation_passed:
        logger.error("Model validation failed. Pipeline stopped.")
        raise ValueError("Model did not meet validation criteria")
    # 5. Register Model
    model_version = register_model(training_result, "Staging")
    logger.info(f"Pipeline successful. Model version {model_version} registered.")
    return model_version

# Deploy as a scheduled flow
if __name__ == "__main__":
    # Run on a schedule (e.g., weekly) or via API trigger
    automated_training_pipeline.serve(
        name="weekly-retraining",
        cron="0 2 * * 0",  # Run at 2 AM every Sunday
        tags=["training", "production"]
    )

# monitoring/drift_detector.py
import numpy as np
from scipy import stats
from typing import Dict, List, Tuple
import warnings
warnings.filterwarnings('ignore')

class StatisticalDriftDetector:
    def __init__(self, reference_data: np.ndarray, alpha: float = 0.05):
        self.reference = reference_data
        self.alpha = alpha
        self.drift_history = []

    def kolmogorov_smirnov_test(self, current_batch: np.ndarray, feature_name: str) -> Dict:
        """Perform KS test for a single feature."""
        stat, p_value = stats.ks_2samp(self.reference, current_batch)
        return {
            'feature': feature_name,
            'statistic': stat,
            'p_value': p_value,
            'drift_detected': p_value < self.alpha
        }

    def population_stability_index(self, current_batch: np.ndarray, feature_name: str,
                                   bins: int = 10) -> Dict:
        """Calculate PSI for a single feature."""
        # Create bins based on reference distribution
        quantiles = np.linspace(0, 1, bins + 1)
        ref_bin_edges = np.percentile(self.reference, quantiles * 100)
        ref_bin_edges[0], ref_bin_edges[-1] = -np.inf, np.inf
        # Bin counts
        ref_counts, _ = np.histogram(self.reference, bins=ref_bin_edges)
        current_counts, _ = np.histogram(current_batch, bins=ref_bin_edges)
        # Normalize to proportions
        ref_prop = ref_counts / len(self.reference)
        current_prop = current_counts / len(current_batch)
        # Avoid division by zero
        ref_prop = np.where(ref_prop == 0, 1e-10, ref_prop)
        current_prop = np.where(current_prop == 0, 1e-10, current_prop)
        # PSI calculation
        psi = np.sum((current_prop - ref_prop) * np.log(current_prop / ref_prop))
        return {
            'feature': feature_name,
            'psi': psi,
            'severity': 'low' if psi < 0.1 else 'medium' if psi < 0.25 else 'high'
        }

    def detect_multivariate_drift(self, current_data: pd.DataFrame,
                                  reference_data: pd.DataFrame) -> Dict:
        """Detect drift using a classifier-based method (more robust for high-dim)."""
        from sklearn.ensemble import RandomForestClassifier
        from sklearn.model_selection import cross_val_score
        # Label reference as 0, current as 1
        combined = pd.concat([reference_data, current_data])
        labels = np.array([0] * len(reference_data) + [1] * len(current_data))
        # Shuffle
        indices = np.random.permutation(len(combined))
        combined_shuffled = combined.iloc[indices]
        labels_shuffled = labels[indices]
        # Train a classifier to distinguish between the two
        clf = RandomForestClassifier(n_estimators=50, max_depth=5)
        scores = cross_val_score(clf, combined_shuffled, labels_shuffled,
                                 cv=3, scoring='roc_auc')
        mean_auc = scores.mean()
        # If a classifier can easily tell them apart (AUC >> 0.5), drift exists
        return {
            'multivariate_drift_score': mean_auc,
            'drift_detected': mean_auc > 0.7  # Threshold can be tuned
        }

# Usage in pipeline validation gate
def validate_for_drift(training_features, production_sample):
    detector = StatisticalDriftDetector(training_features['important_feature'])
    ks_result = detector.kolmogorov_smirnov_test(production_sample['important_feature'],
                                                 'important_feature')
    if ks_result['drift_detected']:
        # Trigger alert and potentially halt promotion
        send_alert(f"Drift detected in important_feature: p={ks_result['p_value']:.4f}")
        return False
    return True

# Initialize Git and DVC
git init
dvc init
# Configure remote storage (e.g., AWS S3, Google Cloud Storage)
dvc remote add -d myremote s3://my-ml-bucket/dvc-store
# Create a .dvcignore file (similar to .gitignore)
echo "*.log" >> .dvcignore
echo "tmp/" >> .dvcignore

# Add and version a raw dataset directory
dvc add data/raw
# This creates a data/raw.dvc pointer file
git add data/raw.dvc .gitignore
git commit -m "Track raw dataset v1.0"
# Push data to remote storage
dvc push

# dvc.yaml
stages:
  prepare:
    cmd: python src/prepare.py --input data/raw --output data/prepared
    deps:
      - src/prepare.py
      - data/raw
    params:
      - prepare.split_ratio
      - prepare.random_seed
    outs:
      - data/prepared:
          cache: true
          persist: false  # Will be tracked by DVC

  featurize:
    cmd: python src/featurize.py --input data/prepared --output data/features
    deps:
      - src/featurize.py
      - data/prepared
    params:
      - featurize.scaling_method
    outs:
      - data/features

  train:
    cmd: python src/train.py --features data/features --model models/classifier.pkl
    deps:
      - src/train.py
      - data/features
    params:
      - train.n_estimators
      - train.max_depth
    outs:
      - models/classifier.pkl
    metrics:
      - metrics/accuracy.json:
          cache: false  # Metrics are not cached, just tracked

# Parameters are defined in a separate file for easy tuning
# params.yaml
prepare:
  split_ratio: 0.8
  random_seed: 42
featurize:
  scaling_method: standard
train:
  n_estimators: 100
  max_depth: 10

# Execute the entire pipeline. DVC runs only changed stages.
dvc repro
# This generates a `dvc.lock` file capturing the exact state of all outputs.
# Commit the pipeline definition and lock file.
git add dvc.yaml params.yaml dvc.lock
git commit -m "Experiment v1: n_estimators=100, max_depth=10"

# Edit params.yaml
sed -i 's/n_estimators: 100/n_estimators: 200/' params.yaml
# DVC will rerun only the 'train' stage, as its parameters changed.
dvc repro
# Compare metrics between the two experiments
dvc metrics diff
# Output shows changes in accuracy, loss, etc.

# src/train.py
import mlflow
import dvc.api
import joblib
from sklearn.ensemble import RandomForestClassifier

def train():
    # Read parameters from DVC's params.yaml via its API
    params = dvc.api.params_show()
    train_params = params['train']
    # Start MLflow run
    with mlflow.start_run():
        # Log parameters from DVC
        mlflow.log_params(train_params)
        # Load DVC-tracked features
        X, y = load_features('data/features')
        model = RandomForestClassifier(**train_params)
        model.fit(X, y)
        accuracy = model.score(X, y)
        # Log metric to MLflow
        mlflow.log_metric("accuracy", accuracy)
        # Log the model to MLflow
        mlflow.sklearn.log_model(model, "model")
        # Also save locally for DVC to track
        joblib.dump(model, 'models/classifier.pkl')
        # Write metric for DVC to track
        with open('metrics/accuracy.json', 'w') as f:
            json.dump({'accuracy': accuracy}, f)

from mlflow.tracking import MlflowClient
client = MlflowClient()
# Search for the best run
best_run = client.search_runs(
    experiment_ids=["0"],
    filter_string="metrics.accuracy > 0.9",
    order_by=["metrics.accuracy DESC"]
)[0]
# Register the model
model_uri = f"runs:/{best_run.info.run_id}/model"
model_version = client.create_model_version(
    name="ProductionClassifier",
    source=model_uri,
    run_id=best_run.info.run_id
)
# Transition stage
client.transition_model_version_stage(
    name="ProductionClassifier",
    version=model_version.version,
    stage="Staging",
    archive_existing_versions=False
)

# Find the Git commit hash for the experiment you want
git log --oneline
# Checkout that commit
git checkout abc123
# DVC will restore the corresponding data and model files
dvc checkout

# validation/model_validator.py
import numpy as np
import pandas as pd
from typing import Dict, Tuple
import json

class ModelReleaseValidator:
    def __init__(self, champion_model, validation_dataset, config):
        self.champion = champion_model
        self.val_data = validation_dataset
        self.config = config  # Contains thresholds, fairness definitions

    def performance_validation(self, candidate_model) -> Tuple[bool, Dict]:
        """Validate candidate against champion on key metrics."""
        X_val, y_val = self.val_data
        # Champion predictions
        champ_preds = self.champion.predict_proba(X_val)[:, 1]
        champ_auc = roc_auc_score(y_val, champ_preds)
        # Candidate predictions
        cand_preds = candidate_model.predict_proba(X_val)[:, 1]
        cand_auc = roc_auc_score(y_val, cand_preds)
        # Calculate improvement/degradation
        auc_diff = cand_auc - champ_auc
        passes = auc_diff >= self.config['min_auc_improvement']
        report = {
            'champion_auc': champ_auc,
            'candidate_auc': cand_auc,
            'auc_diff': auc_diff,
            'threshold': self.config['min_auc_improvement'],
            'passed': passes
        }
        return passes, report

    def fairness_validation(self, candidate_model, sensitive_features) -> Tuple[bool, Dict]:
        """Check for disparate impact across sensitive groups."""
        X_val, y_val = self.val_data
        groups = {}
        for group_name, group_mask in sensitive_features.items():
            group_X = X_val[group_mask]
            group_y = y_val[group_mask]
            if len(group_y) == 0:
                continue
            group_preds = candidate_model.predict_proba(group_X)[:, 1]
            group_auc = roc_auc_score(group_y, group_preds)
            groups[group_name] = group_auc
        # Calculate disparity (max difference in AUC)
        auc_values = list(groups.values())
        max_disparity = max(auc_values) - min(auc_values)
        passes = max_disparity <= self.config['max_auc_disparity']
        return passes, {'group_aucs': groups, 'max_disparity': max_disparity}

    def inference_performance_validation(self, candidate_model) -> Tuple[bool, Dict]:
        """Ensure model meets latency and throughput requirements."""
        import time
        X_sample = self.val_data[0][:100]  # Sample of 100
        # Warm-up
        _ = candidate_model.predict_proba(X_sample[:10])
        # Latency test
        latencies = []
        for i in range(100):
            start = time.perf_counter()
            _ = candidate_model.predict_proba(X_sample[i:i+1])
            latencies.append(time.perf_counter() - start)
        p95_latency = np.percentile(latencies, 95)
        passes = p95_latency <= self.config['max_p95_latency_ms'] / 1000
        return passes, {'p95_latency_ms': p95_latency * 1000}

    def validate_all(self, candidate_model, sensitive_features) -> Dict:
        """Run all validation checks."""
        results = {}
        # Performance
        perf_ok, perf_report = self.performance_validation(candidate_model)
        results['performance'] = {'passed': perf_ok, 'details': perf_report}
        # Fairness
        fair_ok, fair_report = self.fairness_validation(candidate_model, sensitive_features)
        results['fairness'] = {'passed': fair_ok, 'details': fair_report}
        # Inference
        infer_ok, infer_report = self.inference_performance_validation(candidate_model)
        results['inference'] = {'passed': infer_ok, 'details': infer_report}
        # Overall
        results['all_passed'] = all([perf_ok, fair_ok, infer_ok])
        return results

# Usage in CD pipeline
validator = ModelReleaseValidator(champion_model, validation_data, config)
validation_results = validator.validate_all(candidate_model, sensitive_attrs)
if not validation_results['all_passed']:
    raise ValidationError(f"Model validation failed: {json.dumps(validation_results, indent=2)}")

# canary-release.yaml
apiVersion: flagger.app/v1beta1
kind: Canary
metadata:
  name: fraud-model
  namespace: production
spec:
  targetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: fraud-model-deployment
  service:
    port: 9898
    gateways:
      - public-gateway
    hosts:
      - fraud-model.example.com
  analysis:
    interval: 30s
    threshold: 5
    maxWeight: 50
    stepWeight: 10
    metrics:
      - name: request-success-rate
        thresholdRange:
          min: 99
        interval: 30s
      - name: request-duration
        thresholdRange:
          max: 500
        interval: 30s
      - name: model-precision  # Custom ML metric
        thresholdRange:
          min: 0.95
        interval: 1m
        query: |
          avg_over_time(
            model_predictions{model="fraud",status="success"}[1m]
          )
    webhooks:
      - name: load-test
        type: pre-rollout
        url: http://loadtester.default/
        timeout: 30s
        metadata:
          cmd: "hey -z 1m -q 10 -c 2 http://fraud-model-canary.production/"
      - name: acceptance-test
        type: rollout
        url: http://acceptance-tester.default/
        timeout: 30s
        metadata:
          cmd: "python acceptance_test.py --host fraud-model-canary.production"

# monitoring/scheduled_drift_check.py
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset
from evidently.test_suite import TestSuite
from evidently.test_preset import DataDriftTestPreset
import pandas as pd
from datetime import datetime, timedelta
import schedule
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def run_drift_analysis():
    """Scheduled job to compute data drift."""
    logger.info(f"Running drift analysis at {datetime.utcnow()}")
    # 1. Load reference data (e.g., last month's production data)
    reference_data = pd.read_parquet('s3://bucket/monitoring/reference.parquet')
    # 2. Load current production data (e.g., last hour)
    current_data = fetch_recent_production_data(hours=1)
    # 3. Generate drift report
    data_drift_report = Report(metrics=[DataDriftPreset()])
    data_drift_report.run(reference_data=reference_data, current_data=current_data)
    # 4. Save report
    report_path = f'reports/drift_{datetime.utcnow().strftime("%Y%m%d_%H%M")}.html'
    data_drift_report.save_html(report_path)
    # 5. Run statistical tests
    drift_test_suite = TestSuite(tests=[DataDriftTestPreset()])
    drift_test_suite.run(reference_data=reference_data, current_data=current_data)
    # 6. Check results and alert
    if not drift_test_suite.as_dict()['summary']['all_passed']:
        logger.warning("Data drift detected!")
        send_alert(drift_test_suite.as_dict())
        # Optionally trigger retraining
        trigger_retraining_pipeline()
    logger.info("Drift analysis complete.")

def fetch_recent_production_data(hours: int):
    """Fetch recent production features from data lake."""
    # This would query your feature store or data warehouse
    # Example using a hypothetical feature store client
    from feature_store_client import Client
    fs = Client()
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=hours)
    features = fs.get_historical_features(
        entity_df=pd.DataFrame({
            'event_timestamp': pd.date_range(start_time, end_time, freq='1min')
        }),
        feature_refs=['user:*']  # Get all user features
    )
    return features.to_pandas()

# Schedule to run every hour
schedule.every().hour.do(run_drift_analysis)
while True:
    schedule.run_pending()
    time.sleep(60)

# prometheus/alert_rules.yml
groups:
  - name: ml_observability
    rules:
      - alert: ConceptDriftDetected
        expr: |
          increase(
            model_prediction_discrepancy{model="churn_prediction"}[1h]
          ) > 0.15
        for: 30m
        labels:
          severity: critical
          team: ml-ops
        annotations:
          summary: "Concept drift detected for {{ $labels.model }}"
          description: |
            Prediction distribution discrepancy has increased by {{ $value }} over the last hour.
            Current production accuracy: {{ query model_accuracy{model="churn_prediction"} | first | value | printf "%.3f" }}.
            Investigate data pipeline or trigger retraining.
          runbook_url: "https://wiki.company.com/runbook/ml-drift-response"
          dashboard_url: "https://grafana.company.com/d/ml-monitoring"

      - alert: HighInferenceLatency
        expr: |
          histogram_quantile(0.95, rate(model_inference_duration_seconds_bucket[5m])) > 0.5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "High inference latency for {{ $labels.model }}"
          description: "p95 latency is {{ $value }}s (threshold: 0.5s). Check model server resources."

      - alert: PredictionSkew
        expr: |
          abs(
            avg_over_time(model_prediction_mean[1h]) - 
            avg_over_time(model_prediction_mean[6h] offset 1h)
          ) > 0.1
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Prediction distribution shift for {{ $labels.model }}"
          description: "Mean prediction changed by {{ $value }}. Possible data drift."

{
  "Comment": "Automated Retraining Pipeline Triggered by Drift",
  "StartAt": "EvaluateDriftSeverity",
  "States": {
    "EvaluateDriftSeverity": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.drift_score",
          "NumericGreaterThan": 0.25,
          "Next": "TriggerImmediateRetraining"
        },
        {
          "Variable": "$.drift_score",
          "NumericGreaterThan": 0.1,
          "Next": "ScheduleRetraining"
        }
      ],
      "Default": "LogAndContinue"
    },
    "TriggerImmediateRetraining": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sagemaker:createTrainingJob.sync",
      "Parameters": {
        "TrainingJobName": "retrain-$.model_name-$.timestamp",
        "AlgorithmSpecification": {
          "TrainingImage": "$.training_image_uri",
          "TrainingInputMode": "File"
        },
        "RoleArn": "${SageMakerExecutionRole}",
        "InputDataConfig": [
          {
            "ChannelName": "training",
            "DataSource": {
              "S3DataSource": {
                "S3DataType": "S3Prefix",
                "S3Uri": "$.training_data_uri",
                "S3DataDistributionType": "FullyReplicated"
              }
            }
          }
        ],
        "OutputDataConfig": {
          "S3OutputPath": "s3://my-ml-bucket/training-output/"
        },
        "ResourceConfig": {
          "InstanceType": "ml.m5.xlarge",
          "InstanceCount": 1,
          "VolumeSizeInGB": 30
        },
        "StoppingCondition": {
          "MaxRuntimeInSeconds": 3600
        }
      },
      "Next": "UpdateModelRegistry"
    },
    "ScheduleRetraining": {
      "Type": "Task",
      "Resource": "arn:aws:states:::events:putEvents",
      "Parameters": {
        "Entries": [
          {
            "Source": "mlops.drift",
            "DetailType": "ScheduledRetraining",
            "Detail": {
              "model_name": "$.model_name",
              "schedule_time": "$.next_retraining_time"
            }
          }
        ]
      },
      "Next": "SendNotification"
    },
    "UpdateModelRegistry": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:update-model-registry",
      "Next": "RunValidation"
    },
    "RunValidation": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "model-validator",
        "Payload": {
          "model_uri.$": "$.ModelArtifacts.S3ModelArtifacts"
        }
      },
      "Next": "DeployIfValid"
    },
    "DeployIfValid": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.Payload.validation_passed",
          "BooleanEquals": true,
          "Next": "CanaryDeploy"
        }
      ],
      "Default": "Rollback"
    },
    "CanaryDeploy": {
      "Type": "Task",
      "Resource": "arn:aws:states:::eks:runJob.sync",
      "Parameters": {
        "ClusterName": "ml-serving",
        "Manifest": {
          "apiVersion": "flagger.app/v1beta1",
          "kind": "Canary",
          "metadata": {
            "name": "$.model_name",
            "namespace": "production"
          },
          "spec": {
            "analysis": {
              "interval": "1m",
              "threshold": 5,
              "iterations": 10
            }
          }
        }
      },
      "End": true
    },
    "Rollback": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:rollback-model",
      "End": true
    },
    "LogAndContinue": {
      "Type": "Pass",
      "Result": {
        "message": "Drift within acceptable limits, no action taken."
      },
      "End": true
    },
    "SendNotification": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "${AlertTopicARN}",
        "Message": {
          "default": "Moderate drift detected. Retraining scheduled.",
          "email": "Drift alert for model: $.model_name. Score: $.drift_score. Retraining scheduled for $.next_retraining_time."
        }
      },
      "End": true
    }
  }
}