import mlflow
mlflow.set_tracking_uri("http://mlflow-server:5000")
with mlflow.start_run():
    mlflow.log_param("max_depth", 5)
    mlflow.log_metric("accuracy", 0.92)
    # Log the sklearn model with a clear artifact path
    mlflow.sklearn.log_model(sk_model, "churn_model_v1")
    # Log the dataset version used for training
    mlflow.log_param("training_dataset_version", "2023-10-26_v2")

# Example GitHub Actions snippet for a validation job
name: Validate Model
on:
  workflow_dispatch:
    inputs:
      model_uri:
        description: 'URI of the model to validate'
        required: true
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout validation scripts
        uses: actions/checkout@v3
      - name: Validate Model Performance
        run: |
          python validate_model.py \
            --model-uri ${{ github.event.inputs.model_uri }} \
            --threshold-accuracy 0.90
      - name: Bias and Fairness Check
        run: |
          python check_fairness.py \
            --model-uri ${{ github.event.inputs.model_uri }} \
            --sensitive-attribute "age" \
            --threshold-disparity 0.1

# Instrumentation with Prometheus and custom business logic
from prometheus_client import Counter, Gauge, start_http_server
import logging

PREDICTION_COUNTER = Counter('model_predictions_total', 'Total predictions served')
DRIFT_SCORE_GAUGE = Gauge('feature_drift_score', 'Current computed drift score')
PREDICTION_LATENCY = Gauge('prediction_latency_seconds', 'Latency of prediction calls')

def predict(features):
    # Start latency timer
    start_time = time.time()
    PREDICTION_COUNTER.inc()

    # Calculate and record data drift
    drift_score = calculate_drift(features)
    DRIFT_SCORE_GAUGE.set(drift_score)

    # Make prediction
    result = model.predict(features)

    # Record latency
    PREDICTION_LATENCY.set(time.time() - start_time)

    # Log for audit trail (could be sent to a centralized log store)
    logging.info(f"Prediction made. Features: {features[:5]}, Result: {result}, Drift Score: {drift_score}")
    return result

# Start metrics server on port 8000
start_http_server(8000)

import pandas as pd
import numpy as np
from scipy import stats
import logging

def calculate_business_metric(features, predictions):
    """
    Mock function to calculate a business-relevant metric.
    In reality, this could be revenue, customer lifetime value, etc.
    """
    # Simple example: Assume predictions are probabilities and features contain a 'value' column.
    expected_value = np.sum(predictions * features['value'])
    return expected_value

def validate_model_performance(live_features, champion_predictions, challenger_predictions, threshold=0.05):
    """
    Validate that the challenger model's predictions aren't statistically
    significantly worse than the champion's using a paired t-test.
    This is a key reliability gate.
    """
    # Calculate a business metric for each prediction set
    champion_metric = calculate_business_metric(live_features, champion_predictions)
    challenger_metric = calculate_business_metric(live_features, challenger_predictions)

    # For a robust test, we need multiple samples. Here we simulate sample pairs.
    # In practice, you would collect metrics over a period from shadow/canary deployments.
    num_samples = 100
    champ_samples = np.random.normal(loc=champion_metric, scale=0.1, size=num_samples)
    chall_samples = np.random.normal(loc=challenger_metric, scale=0.1, size=num_samples)

    # Perform a paired t-test (or Wilcoxon signed-rank test for non-normal data)
    t_stat, p_value = stats.ttest_rel(champ_samples, chall_samples)

    logging.info(f"Validation: Champion mean={champ_samples.mean():.3f}, Challenger mean={chall_samples.mean():.3f}, p-value={p_value:.4f}")

    # Fail the deployment if challenger is significantly worse (one-tailed test)
    if p_value/2 < threshold and chall_samples.mean() < champ_samples.mean():
        raise ValueError(f"Validation FAILED: Challenger underperforms champion (p-value: {p_value:.4f}). Deployment halted.")
    return True

# Example usage in a pipeline step
try:
    validation_passed = validate_model_performance(
        live_features=production_data_sample,
        champion_predictions=champion_model.predict(production_data_sample),
        challenger_predictions=new_candidate_model.predict(production_data_sample),
        threshold=0.05
    )
    if validation_passed:
        print("Validation passed. Proceeding with canary deployment.")
except ValueError as e:
    print(e)
    # This would fail the CI/CD job, preventing automatic promotion.

import mlflow
import hashlib
import git

mlflow.set_experiment("customer_churn_production")
with mlflow.start_run(run_name="training_cycle_v2.1") as run:
    # Log parameters and metrics
    mlflow.log_param("max_depth", 10)
    mlflow.log_metric("roc_auc", 0.92)

    # Log code version for reproducibility (governance)
    repo = git.Repo(search_parent_directories=True)
    sha = repo.head.object.hexsha
    mlflow.log_param("git_commit", sha)

    # Log dataset hash for data lineage (governance)
    with open('train_dataset.csv', 'rb') as f:
        dataset_hash = hashlib.sha256(f.read()).hexdigest()
    mlflow.log_param("dataset_sha256", dataset_hash)

    # Log the model artifact
    mlflow.sklearn.log_model(clf, "model")
    # Log the preprocessing pipeline as an artifact
    mlflow.log_artifact("preprocessing_pipeline.pkl")

# Example Check: Performance Drift and Regulatory Bias
import pandas as pd
from sklearn.metrics import accuracy_score
import sys

# Load the currently deployed model (champion) and the new candidate
champion_model = mlflow.sklearn.load_model("models:/CustomerChurn/Production")
candidate_model = mlflow.sklearn.load_model(model_uri)

# Load a recent slice of production data for validation
current_data = pd.read_csv('recent_production_data.csv')
X_current, y_true = current_data.drop('churn', axis=1), current_data['churn']

# 1. Performance Gate: Fail if accuracy decays more than 5%
champion_acc = accuracy_score(y_true, champion_model.predict(X_current))
candidate_acc = accuracy_score(y_true, candidate_model.predict(X_current))
if (champion_acc - candidate_acc) > 0.05:
    print(f"FAIL: Performance decay. Champion: {champion_acc:.3f}, Candidate: {candidate_acc:.3f}")
    sys.exit(1)

# 2. Bias/Fairness Gate: Check disparate impact ratio for a sensitive attribute
from aif360.metrics import ClassificationMetric
# ... (code to load privileged/unprivileged groups) ...
# metric = ClassificationMetric(...)
# if metric.disparate_impact() < 0.8: # Example threshold per regulatory guidance
#     print("FAIL: Bias threshold exceeded.")
#     sys.exit(1)

print("PASS: All governance gates passed.")

name: Model CI - Train & Validate
on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]
jobs:
  test-and-validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python 3.10
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements_dev.txt # Test dependencies
      - name: Run unit and integration tests
        run: pytest tests/unit/ tests/integration/ -v
      - name: Validate Data Schema
        run: python scripts/validate_schema.py --data-path ./data/raw --schema-file ./schema.json
      - name: Train Model
        run: python train.py --data-path ./data/processed --output-path ./model_artifact
      - name: Validate Model Performance
        run: python validate.py --model-path ./model_artifact/model.pkl --threshold-auc 0.85

  package-and-register:
    needs: test-and-validate
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Load trained artifact
        run: |
          # In practice, this would be uploaded/downloaded from a shared cache
          echo "Loading model artifact from previous job"
      - name: Log Model to MLflow Registry
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
          MLFLOW_TRACKING_USERNAME: ${{ secrets.MLFLOW_USERNAME }}
          MLFLOW_TRACKING_PASSWORD: ${{ secrets.MLFLOW_PASSWORD }}
        run: |
          python log_to_registry.py \
            --model-file ./model_artifact/model.pkl \
            --run-name "Auto-train ${{ github.sha }}" \
            --metric-auc 0.89
      - name: Build and Push Docker Serving Image
        run: |
          docker build -t my-registry.acme.com/serve-churn:${{ github.sha }} -f Dockerfile.serve .
          docker push my-registry.acme.com/serve-churn:${{ github.sha }}

# .github/workflows/ci-ml.yml
name: CI - Model Training & Validation
on: [push]
jobs:
  ci-ml:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Setup Python & Dependencies
        run: |
          make setup-environment
      - name: Run Data Integrity Tests
        run: |
          make test-data
      - name: Train Model
        run: |
          make train MODEL_PARAMS="configs/params_v2.yaml"
      - name: Validate Model
        run: |
          make validate THRESHOLD_AUC=0.88
      - name: Log to Model Registry
        if: success() && github.ref == 'refs/heads/main'
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
        run: |
          make log-model-to-registry RUN_NAME="CI_Run_${{ github.sha }}"

The accompanying `Makefile` organizes the steps:

.PHONY: train validate log-model-to-registry
setup-environment:
    pip install -r requirements.txt

train:
    python src/train.py --config $(MODEL_PARAMS) --output ./artifacts/model.pkl

validate:
    python src/validate.py \
        --model ./artifacts/model.pkl \
        --test-data ./data/processed/test.csv \
        --threshold-auc $(THRESHOLD_AUC)

log-model-to-registry:
    python src/log_model.py \
        --model-path ./artifacts/model.pkl \
        --run-name $(RUN_NAME)

If validation passes, the model artifact, its metadata (metrics, hyperparameters), and the exact Docker image are versioned and promoted to a **model registry** like MLflow.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from datetime import datetime, timedelta
import pandas as pd
import mlflow

def check_for_drift(**context):
    """Check if data drift exceeds threshold, returns True if retraining is needed."""
    # Load recent production data
    recent_data = pd.read_csv('/data/lake/production/recent.csv')
    # Load reference training data distribution (e.g., mean, std of key features)
    ref_stats = pd.read_json('/data/lake/training/reference_stats.json')

    # Simple drift check: if mean of a key feature has shifted >10%
    drift_detected = False
    for feature in ['amount', 'frequency']:
        current_mean = recent_data[feature].mean()
        ref_mean = ref_stats[feature]['mean']
        if abs(current_mean - ref_mean) / ref_mean > 0.10:
            drift_detected = True
            context['ti'].xcom_push(key='drift_feature', value=feature)

    return drift_detected

def execute_training_cycle(**context):
    """Execute the full training cycle: load data, train, validate, log."""
    # 1. Load and preprocess latest data
    df = pd.read_csv('/data/lake/training/latest.csv')
    X_train, X_test, y_train, y_test = preprocess_and_split(df)

    # 2. Train model (example using sklearn)
    from sklearn.ensemble import RandomForestClassifier
    model = RandomForestClassifier(n_estimators=100, random_state=42)
    model.fit(X_train, y_train)

    # 3. Validate
    from sklearn.metrics import roc_auc_score
    y_pred_proba = model.predict_proba(X_test)[:, 1]
    auc = roc_auc_score(y_test, y_pred_proba)

    # 4. Log to MLflow
    mlflow.set_tracking_uri("http://mlflow:5000")
    with mlflow.start_run(run_name=f"auto_retrain_{datetime.now().isoformat()}"):
        mlflow.log_param("n_estimators", 100)
        mlflow.log_metric("roc_auc", auc)
        mlflow.sklearn.log_model(model, "model")
        mlflow.log_param("trigger", context['ti'].xcom_pull(key='drift_feature', task_ids='check_drift') or 'scheduled')

    # Return the model URI and metric for downstream tasks
    model_uri = f"runs:/{mlflow.active_run().info.run_id}/model"
    context['ti'].xcom_push(key='model_uri', value=model_uri)
    context['ti'].xcom_push(key='model_auc', value=auc)
    return model_uri

def validate_and_promote(**context):
    """Validate the new model and promote if it passes gates."""
    model_uri = context['ti'].xcom_pull(key='model_uri', task_ids='train_model')
    new_auc = context['ti'].xcom_pull(key='model_auc', task_ids='train_model')

    # Load the current production model's AUC from MLflow
    client = mlflow.tracking.MlflowClient()
    prod_run = client.get_run('production_run_id_placeholder') # You would fetch this dynamically
    prod_auc = prod_run.data.metrics.get('roc_auc', 0.80)

    # Promotion Logic: New model must be at least as good (non-degradation)
    if new_auc >= prod_auc - 0.02:  # Allow 2% tolerance
        client.transition_model_version_stage(
            name="ChurnPredictor",
            version=get_version_from_uri(model_uri),
            stage="Production"
        )
        return f"Promoted {model_uri}"
    else:
        raise ValueError(f"Validation failed. New AUC {new_auc:.3f} < Prod AUC {prod_auc:.3f} (with tolerance)")

# Define the DAG
default_args = {
    'owner': 'ml_team',
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'model_retraining_pipeline',
    default_args=default_args,
    description='Automated retraining based on drift or schedule',
    schedule_interval='@weekly',  # Also triggered by drift alert
    start_date=datetime(2023, 1, 1),
    catchup=False,
) as dag:

    check_drift = PythonOperator(
        task_id='check_drift',
        python_callable=check_for_drift,
        provide_context=True,
    )

    train_model = PythonOperator(
        task_id='train_model',
        python_callable=execute_training_cycle,
        provide_context=True,
    )

    validate_promote = PythonOperator(
        task_id='validate_promote',
        python_callable=validate_and_promote,
        provide_context=True,
    )

    notify_success = SlackWebhookOperator(
        task_id='notify_slack',
        slack_webhook_conn_id='slack_webhook_default',
        message="Model retraining pipeline succeeded. New model promoted to staging for validation.",
        channel='#ml-alerts',
    )

    # Define dependencies: train only if drift is detected, or always run on schedule?
    # This setup runs train_model after check_drift, but check_drift's output can be used to skip.
    check_drift >> train_model >> validate_promote >> notify_success

import mlflow
import mlflow.sklearn
from datetime import datetime

# Set the tracking server and experiment
mlflow.set_tracking_uri("http://mlflow-tracking.acme.com:5000")
mlflow.set_experiment("credit_risk_production_v2")

with mlflow.start_run(run_name=f"run_{datetime.now().strftime('%Y%m%d_%H%M%S')}") as run:
    # 1. Log Parameters (Governance: What was used?)
    params = {"max_depth": 10, "n_estimators": 200, "criterion": "gini"}
    mlflow.log_params(params)

    # 2. Log the trained model artifact
    mlflow.sklearn.log_model(
        sk_model=trained_model,
        artifact_path="credit_risk_model",
        registered_model_name="CreditRiskClassifier" # Registers it in the model registry
    )

    # 3. Log Metrics (Governance: How did it perform?)
    mlflow.log_metrics({"roc_auc": 0.945, "accuracy": 0.892, "log_loss": 0.210})

    # 4. Log Artifacts for Lineage (Governance: What code and data?)
    mlflow.log_artifact("train.py") # Training script
    mlflow.log_artifact("config/params.yaml") # Configuration
    # Log a hash of the dataset for reproducibility
    import hashlib
    with open('data/processed/train_v2.1.csv', 'rb') as f:
        file_hash = hashlib.sha256(f.read()).hexdigest()
    mlflow.log_param("training_data_sha256", file_hash)

    # 5. Log Tags for context (e.g., regulatory project code)
    mlflow.set_tag("regulatory_framework", "SR11-7")
    mlflow.set_tag("business_unit", "retail_banking")
    mlflow.set_tag("data_owner", "team-data-governance@acme.com")

print(f"Model logged with run_id: {run.info.run_id}. Full lineage captured.")

# pipeline_scripts/validate_drift.py
import sys
import pandas as pd
import numpy as np
from scipy.stats import ks_2samp
import mlflow

def calculate_psi(expected, actual, buckets=10):
    """Calculate Population Stability Index (PSI)."""
    breakpoints = np.percentile(expected, np.linspace(0, 100, buckets + 1))
    expected_percents = np.histogram(expected, breakpoints)[0] / len(expected)
    actual_percents = np.histogram(actual, breakpoints)[0] / len(actual)
    # Replace zeros to avoid division by zero in log
    expected_percents = np.clip(expected_percents, a_min=1e-10, a_max=None)
    actual_percents = np.clip(actual_percents, a_min=1e-10, a_max=None)
    psi_val = np.sum((expected_percents - actual_percents) * np.log(expected_percents / actual_percents))
    return psi_val

def main(model_uri, recent_data_path):
    # Load the candidate model
    model = mlflow.sklearn.load_model(model_uri)
    # Load training data distribution (could be loaded from stats saved during training)
    X_train_reference = pd.read_csv('data/reference/training_stats.csv')['feature_amount'].values

    # Load recent production data
    df_recent = pd.read_csv(recent_data_path)
    X_recent = df_recent['amount'].values

    # 1. KS Test for a critical feature
    ks_stat, ks_pvalue = ks_2samp(X_train_reference, X_recent)
    print(f"KS Test: stat={ks_stat:.3f}, p-value={ks_pvalue:.3e}")

    # 2. PSI for the same feature
    psi_val = calculate_psi(X_train_reference, X_recent)
    print(f"PSI: {psi_val:.3f}")

    # 3. Decision: Fail pipeline if drift is significant
    FAILURE = False
    if ks_pvalue < 0.01:  # Very strong evidence of distribution change
        print("ALERT: KS test indicates significant data drift (p < 0.01).")
        FAILURE = True
    if psi_val > 0.25:    # Common threshold for high drift
        print(f"ALERT: High PSI ({psi_val:.2f} > 0.25) indicates substantial population shift.")
        FAILURE = True

    if FAILURE:
        # In a CI pipeline, this exit code will fail the job
        sys.exit(1)
    else:
        print("Drift validation passed.")
        sys.exit(0)

if __name__ == "__main__":
    # Arguments would be passed from the CI system (e.g., GitHub Actions)
    model_uri = sys.argv[1]  # e.g., "runs:/123abc456def/model"
    recent_data_path = sys.argv[2] # e.g., "data/production/last_24h.csv"
    main(model_uri, recent_data_path)

import mlflow
import mlflow.sklearn
import git
import json
from pathlib import Path

def capture_environment():
    """Capture pip freeze output for exact environment reproducibility."""
    import subprocess
    result = subprocess.run(['pip', 'freeze'], capture_output=True, text=True)
    return result.stdout

with mlflow.start_run(run_name="fraud_model_v2_1") as run:
    # --- VERSIONING: Log all inputs and context ---
    # 1. Code Version
    repo = git.Repo(search_parent_directories=True)
    mlflow.log_param("git_commit_hash", repo.head.object.hexsha)
    mlflow.log_param("git_branch", repo.active_branch.name)

    # 2. Data Version (assuming DVC is used for data)
    try:
        import dvc.api
        data_version = dvc.api.get_url('data/processed/train.csv').split('@')[-1]
        mlflow.log_param("dvc_data_version", data_version)
    except:
        # Fallback: log a hash
        import hashlib
        with open('data/processed/train.csv', 'rb') as f:
            data_hash = hashlib.sha256(f.read()).hexdigest()
            mlflow.log_param("data_file_sha256", data_hash)

    # 3. Hyperparameters & Config
    config = {
        "model_type": "RandomForest",
        "features": ["amount", "time_of_day", "user_history_score"],
        "preprocessing": "standard_scaler_v1"
    }
    mlflow.log_dict(config, "config.json")
    mlflow.log_params({"max_depth": 15, "n_estimators": 300})

    # 4. Environment
    mlflow.log_text(capture_environment(), "requirements.txt")

    # --- LOG THE MODEL ARTIFACT ITSELF ---
    mlflow.sklearn.log_model(
        sk_model=trained_model,
        artifact_path="model",
        registered_model_name="FraudDetection",
        input_example=X_train.iloc[:5].to_dict('records'), # Example input schema
        signature=mlflow.models.infer_signature(X_train, y_train) # Auto-infer signature
    )

    # --- LOG METRICS ---
    mlflow.log_metrics({"auc": 0.972, "precision": 0.88, "recall": 0.85})

    # --- LOG ADDITIONAL ARTIFACTS FOR LINEAGE ---
    mlflow.log_artifact("src/train.py")
    mlflow.log_artifact("src/preprocess.py")
    # Log feature importance plot
    import matplotlib.pyplot as plt
    fig, ax = plt.subplots()
    ax.barh(range(len(feature_importances)), feature_importances)
    plt.savefig('feature_importance.png')
    mlflow.log_artifact('feature_importance.png')

    print(f"Model logged. Run ID: {run.info.run_id}")
    print(f"Model URI: runs:/{run.info.run_id}/model")

from mlflow.tracking import MlflowClient
from datetime import datetime

client = MlflowClient()

# Simulate an approval and promotion action
model_name = "FraudDetection"
model_version = 5
user = "ml-engineer-jane"
action = "APPROVED_FOR_PRODUCTION"
note = "Passed all validation gates and received sign-off from compliance."

# 1. Transition the model stage (this is an auditable event in MLflow)
client.transition_model_version_stage(
    name=model_name,
    version=model_version,
    stage="Production",
    archive_existing_versions=True
)

# 2. Add a custom tag or description for the audit log (MLflow stores this)
client.set_model_version_tag(
    name=model_name,
    version=model_version,
    key="audit_log",
    value=f"{datetime.utcnow().isoformat()}Z | {user} | {action} | {note}"
)

# 3. Query the audit trail
model_version_details = client.get_model_version(name=model_name, version=model_version)
print(f"Current Stage: {model_version_details.current_stage}")
print(f"Audit Tag: {model_version_details.tags.get('audit_log', 'No tag')}")
print(f"Last Modified: {model_version_details.last_updated_timestamp}")

# scripts/approval_gates.py
import json
import sys
import requests  # For calling external audit services

def load_validation_results(results_path='validation_results.json'):
    with open(results_path) as f:
        return json.load(f)

def check_bias_fairness(metrics, model_uri):
    """Call an external fairness audit service or library."""
    # Example using the AI Fairness 360 toolkit
    # from aif360.metrics import ClassificationMetric
    # ... compute disparate impact, statistical parity difference ...
    # return pass/fail
    # For this example, we mock a check
    bias_score = metrics.get('disparate_impact_ratio', 1.0)
    # Threshold: between 0.8 and 1.2 is often considered acceptable
    if bias_score < 0.8 or bias_score > 1.25:
        return False, f"Bias fairness check failed. Disparate Impact Ratio: {bias_score:.2f}"
    return True, "Bias check passed."

def main():
    metrics = load_validation_results()

    # Gate 1: Performance Thresholds
    if metrics['test_accuracy'] < 0.85:
        print(f"FAIL: Accuracy {metrics['test_accuracy']} < 0.85")
        sys.exit(1)
    if metrics['roc_auc'] < 0.90:
        print(f"FAIL: AUC {metrics['roc_auc']} < 0.90")
        sys.exit(1)

    # Gate 2: Bias & Fairness (Critical for Governance)
    bias_passed, bias_msg = check_bias_fairness(metrics, model_uri=sys.argv[1] if len(sys.argv)>1 else None)
    if not bias_passed:
        print(f"FAIL: {bias_msg}")
        sys.exit(1)

    # Gate 3: Operational Metrics (e.g., model size, latency)
    if metrics.get('model_size_mb', 0) > 500:
        print(f"FAIL: Model size {metrics['model_size_mb']}MB exceeds 500MB limit.")
        sys.exit(1)
    if metrics.get('p99_latency_ms', 1000) > 200:
        print(f"WARN: High latency: {metrics['p99_latency_ms']}ms. Requires review.")
        # This could be a warning that doesn't fail but requires manual approval.

    # Gate 4: Custom Business Logic
    expected_profit_lift = metrics.get('estimated_profit_lift', 0)
    if expected_profit_lift < 0.01: # 1% minimum lift
        print(f"FAIL: Insufficient business lift: {expected_profit_lift:.2%}. Deployment not justified.")
        sys.exit(1)

    print("SUCCESS: All automated approval gates passed.")
    print("The pipeline will now pause for MANUAL APPROVAL before production deployment.")
    # In a system like GitLab CI, this exit code 0 would allow the pipeline to proceed to a manual 'approval' job.
    # In Jenkins, you might use the 'input' step.

if __name__ == "__main__":
    main()

# k8s-rbac-ml-roles.yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: ml-staging
  name: ml-data-scientist
rules:
- apiGroups: [""]
  resources: ["pods", "services"]
  verbs: ["get", "list", "watch"] # Can view jobs
- apiGroups: ["kubeflow.org"]
  resources: ["scheduledworkflows"]
  verbs: ["create", "get"] # Can submit training jobs
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: ml-production
  name: ml-engineer
rules:
- apiGroups: [""]
  resources: ["configmaps", "secrets"]
  verbs: ["get", "list"]
- apiGroups: ["apps"]
  resources: ["deployments"]
  verbs: ["get", "list", "watch", "update", "patch"] # Can update model deployments
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  namespace: ml-production
  name: ml-approver
rules:
- apiGroups: ["argoproj.io"]
  resources: ["applications"]
  verbs: ["get", "patch"]
- apiGroups: [""]
  resources: ["events"]
  verbs: ["create"] # Can approve by patching the ArgoCD Application resource

import mlflow
import pandas as pd
from sklearn.model_selection import train_test_split

mlflow.set_experiment("fraud_detection_monitored")
with mlflow.start_run():
    # Load and version data
    df = pd.read_csv('data/processed/train_2023_q4.csv')
    mlflow.log_param("data_file", "train_2023_q4.csv")
    mlflow.log_param("data_rows", len(df))

    # Train model
    X = df.drop('is_fraud', axis=1)
    y = df['is_fraud']
    X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)
    model = train_model(X_train, y_train)

    # Log model
    mlflow.sklearn.log_model(model, "model", registered_model_name="FraudClassifier")

    # Log validation metrics as baseline for future monitoring
    val_metrics = evaluate_model(model, X_val, y_val)
    mlflow.log_metrics(val_metrics)

    # Log a sample of validation predictions to establish a baseline distribution
    val_predictions = model.predict_proba(X_val)[:, 1]
    pd.Series(val_predictions).to_csv('baseline_predictions.csv', index=False)
    mlflow.log_artifact('baseline_predictions.csv')

# monitoring/drift_detector.py
import numpy as np
import pandas as pd
from scipy import stats
import mlflow
from datetime import datetime, timedelta
import logging
from database import fetch_recent_predictions  # Your function to get live data

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

def calculate_psi(expected, actual, buckets=10):
    """Calculate Population Stability Index."""
    # Ensure no zero counts to avoid log(0)
    eps = 1e-10
    breakpoints = np.percentile(expected, np.linspace(0, 100, buckets + 1))
    expected_counts, _ = np.histogram(expected, breakpoints)
    actual_counts, _ = np.histogram(actual, breakpoints)
    expected_percents = expected_counts / len(expected) + eps
    actual_percents = actual_counts / len(actual) + eps
    psi = np.sum((expected_percents - actual_percents) * np.log(expected_percents / actual_percents))
    return psi

def monitor_model(model_name, feature_name='transaction_amount', hours_back=24):
    """Main monitoring function for a single feature."""
    # 1. Load the baseline (training) distribution from MLflow
    client = mlflow.tracking.MlflowClient()
    # Get the run that created the current production model (simplified)
    prod_run = client.search_runs(
        experiment_ids=["1"],
        filter_string=f"tags.mlflow.log-model.history.models.name = '{model_name}' AND attributes.status = 'FINISHED'",
        order_by=["attributes.start_time DESC"]
    )[0]
    baseline_path = client.download_artifacts(prod_run.info.run_id, "baseline_predictions.csv")
    baseline_data = pd.read_csv(baseline_path)[feature_name].dropna().values

    # 2. Fetch recent production data
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=hours_back)
    recent_df = fetch_recent_predictions(start_time, end_time)
    recent_data = recent_df[feature_name].dropna().values

    if len(recent_data) < 100:
        logger.warning(f"Insufficient recent data points ({len(recent_data)}). Skipping drift check.")
        return

    # 3. Calculate Drift Metrics
    # Kolmogorov-Smirnov test
    ks_stat, ks_pvalue = stats.ks_2samp(baseline_data, recent_data)
    # Population Stability Index
    psi_val = calculate_psi(baseline_data, recent_data)

    logger.info(f"Drift Check for '{feature_name}': KS p-value={ks_pvalue:.3e}, PSI={psi_val:.3f}")

    # 4. Evaluate against thresholds and alert
    alert = False
    if ks_pvalue < 0.01:  # Strong evidence of distribution change
        logger.error(f"ALERT: Significant KS drift detected for {feature_name}. p-value: {ks_pvalue:.3e}")
        alert = True
    if psi_val > 0.2:      # Moderate to high population shift
        logger.error(f"ALERT: High PSI ({psi_val:.2f}) for {feature_name}.")
        alert = True

    # 5. (Optional) Log metrics back to MLflow for time-series tracking
    with mlflow.start_run(run_name=f"monitor_{datetime.now().isoformat()}", nested=True):
        mlflow.log_metrics({
            f"drift_ks_pvalue_{feature_name}": ks_pvalue,
            f"drift_psi_{feature_name}": psi_val
        })
        mlflow.set_tag("model_monitored", model_name)

    if alert:
        # Trigger remediation workflow: e.g., send to Slack, create Jira ticket, trigger retraining pipeline
        trigger_alert(
            model_name=model_name,
            feature=feature_name,
            metrics={'ks_pvalue': ks_pvalue, 'psi': psi_val}
        )

if __name__ == "__main__":
    # Monitor multiple key features
    for feature in ['transaction_amount', 'time_since_last_transaction', 'user_age']:
        monitor_model("FraudClassifier", feature_name=feature)

# monitoring_suite/production_monitor.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from evidently.report import Report
from evidently.metrics import DataDriftTable, ClassificationQualityMetric
from evidently.metric_preset import DataQualityPreset
import prometheus_client
from prometheus_client import Gauge, push_to_gateway
import logging
from database_connector import get_production_features, get_ground_truth_labels  # Your custom connectors

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

# Define Prometheus metrics for push or pull model
MODEL_ACCURACY = Gauge('model_accuracy', 'Current accuracy on recent data', ['model_name'])
FEATURE_DRIFT_PSI = Gauge('feature_drift_psi', 'PSI value for a feature', ['model_name', 'feature_name'])
ALERT_COUNTER = Gauge('model_alert_total', 'Total number of alerts triggered', ['model_name', 'alert_type'])

def calculate_metrics(model_name, window_hours=24):
    """Fetch data, calculate metrics, and push to monitoring system."""
    logger.info(f"Starting monitoring run for {model_name}")

    # 1. Fetch Data
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(hours=window_hours)

    # Get production inference data (features and predictions)
    prod_data = get_production_features(model_name, start_time, end_time)
    # In some cases, you may have delayed ground truth (e.g., fraud label confirmed after 24h)
    ground_truth = get_ground_truth_labels(model_name, start_time, end_time)

    if prod_data.empty or ground_truth.empty:
        logger.warning("Insufficient data for monitoring cycle.")
        return

    # Merge predictions with ground truth on a common key (e.g., request_id)
    evaluation_data = pd.merge(prod_data, ground_truth, on='request_id', how='inner')
    if evaluation_data.empty:
        logger.warning("No ground truth available for evaluation yet.")
        # Proceed with drift detection only
        run_drift_only = True
    else:
        run_drift_only = False

    # 2. Load Reference Data (from training or a known good period)
    # This could be loaded from a parquet file, feature store, or MLflow
    reference_data = pd.read_parquet(f'reference_data/{model_name}_reference.parquet')

    # 3. Generate Evidently Reports
    report = Report(metrics=[
        DataQualityPreset(), 
        DataDriftTable(),
        ClassificationQualityMetric() if not run_drift_only else None
    ])
    # For drift, we compare current production *features* to reference features
    report.run(reference_data=reference_data.drop('target', axis=1, errors='ignore'), 
               current_data=prod_data.drop('prediction', axis=1, errors='ignore'))

    # 4. Extract Metrics and Push to Prometheus
    report_dict = report.as_dict()

    # Push Data Drift Metrics
    drift_metrics = report_dict['metrics'][1]  # DataDriftTable index
    for feature_name, stats in drift_metrics['result']['drift_by_columns'].items():
        psi_value = stats.get('psi', 0)
        FEATURE_DRIFT_PSI.labels(model_name=model_name, feature_name=feature_name).set(psi_value)
        logger.debug(f"Feature {feature_name} PSI: {psi_value}")

    # Check for overall dataset drift and alert
    dataset_drift = drift_metrics['result']['dataset_drift']
    if dataset_drift:
        logger.error(f"DATA DRIFT ALERT for {model_name}. Dataset drift detected.")
        ALERT_COUNTER.labels(model_name=model_name, alert_type='data_drift').inc()
        # Trigger external alert (e.g., PagerDuty, Slack)
        send_alert(f"Data Drift Alert - {model_name}", f"Evidently detected dataset drift.")

    if not run_drift_only:
        # Push Performance Metrics
        quality_metrics = report_dict['metrics'][2]  # ClassificationQualityMetric
        accuracy = quality_metrics['result']['accuracy']
        MODEL_ACCURACY.labels(model_name=model_name).set(accuracy)

        # Alert on performance degradation (e.g., accuracy drop > 5%)
        baseline_accuracy = 0.92  # Retrieved from model registry
        if accuracy < baseline_accuracy - 0.05:
            logger.error(f"PERFORMANCE ALERT for {model_name}. Accuracy dropped to {accuracy:.3f}.")
            ALERT_COUNTER.labels(model_name=model_name, alert_type='performance_degradation').inc()
            send_alert(f"Performance Alert - {model_name}", 
                       f"Accuracy dropped to {accuracy:.3f} from baseline {baseline_accuracy:.3f}.")

    # 5. Push all metrics to Prometheus Pushgateway (for batch jobs) or expose via HTTP server
    prometheus_client.push_to_gateway('localhost:9091', job=f'model_monitor_{model_name}', 
                                       registry=prometheus_client.REGISTRY)
    logger.info(f"Monitoring run completed for {model_name}. Metrics pushed.")

def send_alert(alert_title, alert_message):
    """Function to send alerts to various channels."""
    # Example: Send to Slack webhook
    import requests
    webhook_url = "https://hooks.slack.com/services/..."
    slack_msg = {"text": f"*{alert_title}*:\n{alert_message}"}
    try:
        requests.post(webhook_url, json=slack_msg)
    except Exception as e:
        logger.error(f"Failed to send Slack alert: {e}")
    # Could also integrate with PagerDuty, OpsGenie, etc.

if __name__ == "__main__":
    # Run for all active production models
    active_models = ["FraudClassifier", "ChurnPredictor", "RecommendationEngine_v2"]
    for model in active_models:
        calculate_metrics(model, window_hours=24)

# remediation/drift_detection_setup.py
import numpy as np
import pandas as pd
from alibi_detect.cd import MMDDrift, KSDrift, CVMDrift
from alibi_detect.saving import save_detector, load_detector
from alibi_detect.utils.saving import save_detector
import pickle
import mlflow
import logging

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

def initialize_drift_detector(model_name, X_reference, detector_type='mmd', **kwargs):
    """
    Initialize and save a drift detector for a model.

    Args:
        model_name: Name of the model (for saving).
        X_reference: Reference data (numpy array or pandas DataFrame).
        detector_type: 'mmd', 'ks', or 'cvm'.
        **kwargs: Additional arguments for the detector (e.g., p_val threshold).
    """
    if detector_type == 'mmd':
        # MMD (Maximum Mean Discrepancy) is good for multivariate drift
        detector = MMDDrift(X_reference, p_val=0.05, **kwargs)
    elif detector_type == 'ks':
        # KS for univariate per feature
        detector = KSDrift(X_reference, p_val=0.05, **kwargs)
    elif detector_type == 'cvm':
        # Cramer-von Mises test
        detector = CVMDrift(X_reference, p_val=0.05, **kwargs)
    else:
        raise ValueError(f"Detector type {detector_type} not supported.")

    # Save the detector to disk for later use in the monitoring pipeline
    detector_path = f'detectors/{model_name}_{detector_type}.pkl'
    save_detector(detector, detector_path)
    logger.info(f"Drift detector saved to {detector_path}")

    # Also log it as an MLflow artifact for lineage
    with mlflow.start_run(run_name=f"drift_detector_init_{model_name}"):
        mlflow.log_artifact(detector_path)
        mlflow.log_param("detector_type", detector_type)
        mlflow.log_param("reference_data_shape", X_reference.shape)

    return detector_path

def check_drift_and_remediate(model_name, X_current, trigger_retraining=True):
    """
    Load the detector, check for drift, and trigger remediation if needed.
    """
    detector_path = f'detectors/{model_name}_mmd.pkl'  # Assume MMD for multivariate
    detector = load_detector(detector_path)

    # Predict drift
    preds = detector.predict(X_current, return_p_val=True, return_distance=True)

    is_drift = preds['data']['is_drift']
    p_val = preds['data']['p_val']
    distance = preds['data']['distance']

    logger.info(f"Drift check: is_drift={is_drift}, p_val={p_val:.4f}, distance={distance:.4f}")

    if is_drift:
        logger.warning(f"DRIFT DETECTED for model {model_name}. p-value: {p_val}")
        ALERT_COUNTER.labels(model_name=model_name, alert_type='drift_detected').inc()

        # 1. Send Immediate Alert
        send_alert(
            title=f"Drift Alert - {model_name}",
            message=f"Multivariate drift detected (p={p_val:.3e}, distance={distance:.3f})."
        )

        # 2. Log the drifted data for analysis
        drift_data_path = f'drifted_data/{model_name}_{datetime.now().strftime("%Y%m%d_%H%M%S")}.parquet'
        pd.DataFrame(X_current).to_parquet(drift_data_path)
        logger.info(f"Drifted data snapshot saved to {drift_data_path}")

        # 3. Trigger Automated Remediation Workflow
        if trigger_retraining:
            trigger_retraining_pipeline(model_name, drift_alert=True)

        # 4. (Optional) Model Rollback or Traffic Shift
        # In a champion-challenger setup, you could automatically shift traffic away from the drifted model.
        # shift_traffic_to_challenger(model_name)

    return is_drift, p_val

def trigger_retraining_pipeline(model_name, drift_alert=False):
    """Trigger a retraining pipeline (e.g., by calling an Airflow DAG or GitHub Actions workflow)."""
    logger.info(f"Triggering retraining pipeline for {model_name}.")

    # Method 1: Call a pre-defined API endpoint that starts the pipeline
    # import requests
    # response = requests.post('http://airflow-webserver:8080/api/v1/dags/model_retrain_dag/dagRuns',
    #                          json={"conf": {"model_name": model_name, "trigger": "drift_alert"}})

    # Method 2: Directly invoke a CLI command (if monitor runs in the same environment)
    # import subprocess
    # cmd = f"python orchestration/trigger_retrain.py --model {model_name}"
    # subprocess.run(cmd, shell=True)

    # Method 3: Write to a message queue (e.g., Redis, SQS) that the orchestrator listens to
    import redis
    r = redis.Redis(host='localhost', port=6379, db=0)
    r.publish('retrain_requests', f"{model_name}:{datetime.utcnow().isoformat()}")

    logger.info(f"Retraining request for {model_name} published.")

# Example Usage in a Scheduled Monitoring Job
if __name__ == "__main__":
    # 1. Load reference data (do this once during setup)
    # ref_data = pd.read_parquet('reference_data/churn_model_ref.parquet').values
    # initialize_drift_detector('ChurnModel', ref_data, detector_type='mmd')

    # 2. In the monitoring job, load recent production data
    recent_data = fetch_last_24h_data('ChurnModel')
    recent_data_np = recent_data[['feature1', 'feature2', 'feature3']].values

    # 3. Check for drift and remediate
    drift_detected, p_val = check_drift_and_remediate('ChurnModel', recent_data_np, trigger_retraining=True)

import mlflow
import json

with mlflow.start_run():
    # Log the core model
    mlflow.sklearn.log_model(sk_model, "model", registered_model_name="RevenueForecaster")

    # Log input schema for validation and documentation
    input_schema = {
        "features": ["sales_last_month", "marketing_spend", "seasonality_index"],
        "required": True,
        "type": "array",
        "items": {"type": "number"}
    }
    mlflow.log_dict(input_schema, "input_schema.json")

    # Log example payload for testing
    example_input = {"features": [15000.0, 5000.0, 0.8]}
    mlflow.log_dict(example_input, "example_input.json")

# pipeline_scripts/final_validation.py
def validate_model_for_production(model_uri, validation_dataset_path):
    import pandas as pd
    import mlflow
    from my_metrics import calculate_business_kpi

    model = mlflow.pyfunc.load_model(model_uri)
    test_data = pd.read_csv(validation_dataset_path)

    # 1. Performance Validation
    predictions = model.predict(test_data)
    accuracy = accuracy_score(test_data['target'], predictions)
    assert accuracy > 0.85, f"Accuracy {accuracy} below 0.85 threshold."

    # 2. Business Logic Validation
    expected_kpi_lift = calculate_business_kpi(predictions, test_data)
    assert expected_kpi_lift > 0.02, f"Business KPI lift {expected_kpi_lift:.2%} insufficient."

    # 3. Fairness & Bias Validation (using external library)
    bias_report = run_fairness_audit(model, test_data, sensitive_attr='region')
    assert bias_report['disparate_impact'] >= 0.8, "Fairness check failed."

    print("All production validation gates passed.")
    return True

-- Example alert logic for monitoring performance drift (run hourly)
WITH current_perf AS (
    SELECT 
        model_id,
        DATE_TRUNC('hour', prediction_timestamp) as hour_bucket,
        AVG(CASE WHEN actual_label IS NOT NULL THEN (prediction = actual_label)::INT ELSE NULL END) as hourly_accuracy
    FROM prediction_monitoring_logs
    WHERE prediction_timestamp > CURRENT_TIMESTAMP - INTERVAL '2 hours'
        AND actual_label IS NOT NULL -- Only where ground truth is available
    GROUP BY 1, 2
),
baseline_perf AS (
    SELECT model_id, target_accuracy
    FROM model_performance_baselines
    WHERE is_current = TRUE
)
SELECT 
    c.model_id,
    c.hour_bucket,
    c.hourly_accuracy,
    b.target_accuracy,
    (b.target_accuracy - c.hourly_accuracy) as accuracy_drop
FROM current_perf c
JOIN baseline_perf b ON c.model_id = b.model_id
WHERE c.hourly_accuracy < b.target_accuracy - 0.05 -- Alert threshold: 5% drop
HAVING COUNT(*) > 100  -- Ensure sufficient sample size
ORDER BY accuracy_drop DESC;

-- Query to calculate average lead time per model in the last 30 days
SELECT 
    model_name,
    COUNT(DISTINCT deployment_id) as num_deployments,
    AVG(TIMESTAMPDIFF(HOUR, training_finished_at, production_deployment_at)) as avg_lead_time_hours,
    PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY TIMESTAMPDIFF(HOUR, training_finished_at, production_deployment_at)) as median_lead_time_hours
FROM model_deployment_metadata
WHERE production_deployment_at > CURRENT_DATE - INTERVAL '30' DAY
    AND training_finished_at IS NOT NULL
GROUP BY model_name
ORDER BY avg_lead_time_hours DESC;

# monitoring/psi_calculator.py
import pandas as pd
import numpy as np
from scipy import stats
import prometheus_client
from prometheus_client import Gauge

PSI_GAUGE = Gauge('feature_psi', 'Population Stability Index by feature', ['model_name', 'feature'])

def calculate_psi_for_feature(expected, actual, buckets=10, epsilon=1e-10):
    """Calculate PSI for a single feature vector."""
    breakpoints = np.percentile(expected, np.linspace(0, 100, buckets + 1))
    # Ensure unique breakpoints to avoid empty bins
    breakpoints = np.unique(breakpoints)
    if len(breakpoints) < 2:
        return 0.0  # Not enough variation to calculate PSI

    expected_percents, _ = np.histogram(expected, bins=breakpoints)
    actual_percents, _ = np.histogram(actual, bins=breakpoints)

    # Normalize to percentages and add epsilon
    expected_percents = expected_percents / len(expected) + epsilon
    actual_percents = actual_percents / len(actual) + epsilon

    psi = np.sum((expected_percents - actual_percents) * np.log(expected_percents / actual_percents))
    return psi

def monitor_all_features(model_name, reference_df, current_df):
    """Calculate PSI for all numeric features and alert if high."""
    high_drift_features = []
    for col in reference_df.select_dtypes(include=[np.number]).columns:
        if col in current_df.columns:
            psi_val = calculate_psi_for_feature(
                reference_df[col].dropna().values,
                current_df[col].dropna().values
            )
            PSI_GAUGE.labels(model_name=model_name, feature=col).set(psi_val)

            # Alert logic: PSI > 0.2 suggests moderate drift, > 0.3 high drift
            if psi_val > 0.25:
                high_drift_features.append((col, psi_val))
                print(f"ALERT: High PSI for {col}: {psi_val:.3f}")

    return high_drift_features

# Usage in a scheduled job
if __name__ == "__main__":
    ref_data = pd.read_parquet('reference_data.parquet')
    curr_data = fetch_last_24h_production_data()

    drifting_features = monitor_all_features('ChurnModel', ref_data, curr_data)
    if drifting_features:
        trigger_alert(f"High feature drift in {len(drifting_features)} features.")

# business_impact/ab_test_analysis.py
import pandas as pd
import numpy as np
from scipy import stats
import matplotlib.pyplot as plt

def analyze_ab_test_results(control_group_csv, treatment_group_csv, primary_kpi='conversion_rate'):
    """Analyze the results of an A/B test for a model deployment."""
    control = pd.read_csv(control_group_csv)
    treatment = pd.read_csv(treatment_group_csv)

    control_kpi = control[primary_kpi]
    treatment_kpi = treatment[primary_kpi]

    # Calculate lift
    lift = (treatment_kpi.mean() - control_kpi.mean()) / control_kpi.mean()

    # Perform a statistical significance test (e.g., t-test)
    t_stat, p_value = stats.ttest_ind(treatment_kpi, control_kpi, equal_var=False)

    # Calculate confidence interval for the difference
    diff_mean = treatment_kpi.mean() - control_kpi.mean()
    se = np.sqrt(treatment_kpi.var()/len(treatment_kpi) + control_kpi.var()/len(control_kpi))
    conf_int = stats.t.interval(0.95, df=min(len(treatment_kpi), len(control_kpi))-1, loc=diff_mean, scale=se)

    print(f"Business Impact Analysis for Model Deployment:")
    print(f"  Control {primary_kpi}: {control_kpi.mean():.4f}")
    print(f"  Treatment {primary_kpi}: {treatment_kpi.mean():.4f}")
    print(f"  Relative Lift: {lift:.2%}")
    print(f"  Absolute Difference: {diff_mean:.4f} (95% CI: [{conf_int[0]:.4f}, {conf_int[1]:.4f}])")
    print(f"  Statistical Significance (p-value): {p_value:.4f}")

    if p_value < 0.05 and lift > 0:
        print(f"  CONCLUSION: Deployment successful. Significant positive lift detected.")
        return True, lift
    else:
        print(f"  CONCLUSION: No statistically significant positive impact detected.")
        return False, lift

# Example call
is_successful, measured_lift = analyze_ab_test_results(
    'ab_test_control_week.csv',
    'ab_test_treatment_week.csv',
    primary_kpi='purchase_amount'
)

# advanced_monitoring/streaming_drift_detector.py
from alibi_detect.cd import MMDDriftOnline
from alibi_detect.saving import load_detector
import numpy as np
import threading
import queue
import logging
from prometheus_client import Counter, Gauge

logger = logging.getLogger(__name__)
DRIFT_ALERT_COUNTER = Counter('streaming_drift_alerts_total', 'Total drift alerts from streaming detector')
DRIFT_SCORE = Gauge('streaming_drift_score', 'Current online drift score')

class StreamingDriftMonitor:
    def __init__(self, model_name, detector_config_path, window_size=100, drift_threshold=0.05):
        self.detector = load_detector(detector_config_path)
        self.window_size = window_size
        self.drift_threshold = drift_threshold
        self.feature_buffer = []
        self.lock = threading.Lock()
        self.model_name = model_name

    def add_prediction_features(self, feature_vector):
        """Add a single prediction's feature vector to the monitoring buffer."""
        with self.lock:
            self.feature_buffer.append(feature_vector)
            if len(self.feature_buffer) >= self.window_size:
                # Process the window
                self._check_window()

    def _check_window(self):
        """Check the last window of features for drift."""
        window_data = np.array(self.feature_buffer[-self.window_size:])

        # Online detectors update their state and return a test statistic and threshold
        # preds = self.detector.predict(window_data)
        # For this example, we use a simplified check
        from scipy import stats
        # Compare window to reference distribution (loaded during init)
        # This is a placeholder for the actual online detection logic
        ks_stat, p_value = stats.ks_2samp(self.detector.reference_data[:,0], window_data[:,0])

        DRIFT_SCORE.set(ks_stat)

        if p_value < self.drift_threshold:
            logger.warning(f"Streaming drift alert for {self.model_name}. KS stat: {ks_stat:.3f}, p: {p_value:.3e}")
            DRIFT_ALERT_COUNTER.inc()
            self._trigger_micro_remediation(window_data)

        # Clear buffer after processing (or keep a sliding window)
        # self.feature_buffer = []

    def _trigger_micro_remediation(self, drifted_window):
        """Immediate, lightweight remediation for critical drift."""
        # 1. Log the drifted window for immediate analysis
        log_drifted_data(drifted_window, self.model_name)
        # 2. Possibly scale down traffic to this model instance
        # 3. Send high-priority alert
        send_priority_alert(
            f"REALTIME DRIFT - {self.model_name}",
            "Significant feature drift detected in live traffic window."
        )
        # 4. Optionally, trigger a fast retraining pipeline on a subset of data

# Integration into a FastAPI inference endpoint
from fastapi import FastAPI, Request
import json
app = FastAPI()
monitor = StreamingDriftMonitor("MyLiveModel", "detectors/online_mmd_config.pkl")

@app.post("/predict")
async def predict(request: Request):
    data = await request.json()
    features = np.array(data['features']).reshape(1, -1)

    # 1. Make prediction (your existing model call)
    prediction = model.predict(features)

    # 2. Asynchronously add features to the drift monitor (non-blocking)
    # Use a thread pool or task queue in production
    threading.Thread(target=monitor.add_prediction_features, args=(features.flatten(),)).start()

    return {"prediction": prediction.tolist()}

*Measurable Benefit:* This shifts from batch-based drift detection (e.g., hourly) to near real-time, potentially reducing the exposure time to a degraded model from hours to minutes. It enables *just-in-time retraining* or traffic shifting, optimizing resource use and maintaining stricter performance SLAs.

# kserve_traffic_splitting.yaml (simplified)
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: my-model
spec:
  predictor:
    canaryTrafficPercent: 5
    model:
      modelFormat:
        name: sklearn
      storageUri: gs://my-bucket/models/champion # Champion model
      env:
        - name: MODEL_NAME
          value: champion
    canary:
      model:
        modelFormat:
          name: sklearn
        storageUri: gs://my-bucket/models/candidate_v2 # Canary model
        env:
          - name: MODEL_NAME
            value: candidate_v2

3.  Instrument both model endpoints to log predictions, latencies, and business-specific metrics (e.g., conversion rate, add-to-cart) to a central analytics database or stream processor (e.g., Kafka to Snowflake).
4.  Define a decision rule in an automated analysis job that runs periodically (e.g., every 30 minutes). The rule should evaluate key metrics:

# canary_analysis.py
def evaluate_canary_performance(canary_metrics, champion_metrics, duration_hours=4):
    # Rule 1: Non-inferiority on primary business KPI (e.g., conversion rate)
    kpi_lift = canary_metrics['conversion_rate'] - champion_metrics['conversion_rate']
    if kpi_lift < -0.005:  # Canary must not be worse than 0.5%
        return "FAIL", f"Business KPI degradation: {kpi_lift:.3%}"

    # Rule 2: No regression in operational metrics (e.g., p99 latency)
    latency_increase = canary_metrics['p99_latency_ms'] - champion_metrics['p99_latency_ms']
    if latency_increase > 20:  # Latency increase > 20ms is unacceptable
        return "FAIL", f"Latency increase: {latency_increase:.0f}ms"

    # Rule 3: No significant increase in error rates
    if canary_metrics['error_rate'] > champion_metrics['error_rate'] + 0.001:
        return "FAIL", f"Error rate increase."

    return "PASS", f"Canary meets all criteria. KPI lift: {kpi_lift:+.3%}"

5.  Automate the full rollout or rollback based on this analysis. If "PASS", the pipeline updates the serving configuration to route 100% of traffic to the canary (now the new champion). If "FAIL", it triggers an automatic rollback by updating the configuration back to 100% champion and sends an alert for investigation.