from prometheus_client import Gauge, start_http_server
import time

# Initialize a Gauge metric to track model accuracy
# 'ml_model_accuracy' is the metric name, 'Model accuracy on recent predictions' is a help string
accuracy_metric = Gauge('ml_model_accuracy', 'Model accuracy on recent predictions')

def update_accuracy_metric():
    """
    This function should be called periodically to update the accuracy metric.
    In a real-world scenario, this would fetch the latest accuracy from a validation pipeline
    or a monitoring system.
    """
    # Simulate fetching accuracy (replace with your actual logic)
    latest_accuracy = 0.88  # Example: Fetch from your validation pipeline
    accuracy_metric.set(latest_accuracy)
    print(f"Updated accuracy metric to: {latest_accuracy}")

if __name__ == '__main__':
    # Start an HTTP server on port 8000 to expose the metrics
    # Prometheus will scrape metrics from http://localhost:8000/metrics
    start_http_server(8000)
    print("Prometheus metrics server started on port 8000")

    # Keep the server running and update the metric periodically
    while True:
        update_accuracy_metric()
        time.sleep(60) # Update every 60 seconds

groups:
- name: ml_performance_alerts
  rules:
  - alert: LowModelAccuracy
    # The expression that Prometheus evaluates.
    # It triggers if the 'ml_model_accuracy' metric drops below 0.85.
    expr: ml_model_accuracy < 0.85
    # 'for: 5m' means the condition must be true for 5 consecutive minutes
    # before the alert is fired. This helps reduce flapping and false positives.
    for: 5m
    labels:
      # Labels to attach to the alert, useful for routing and severity.
      severity: critical
    annotations:
      # Human-readable information about the alert.
      summary: "Model accuracy has dropped below the critical threshold."
      description: "The ML model's accuracy is currently {{ $value }}%, which is below the expected minimum of 85%."

import smtplib
from email.mime.text import MIMEText
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset
import pandas as pd

# Assume train_df and prod_df are your reference (training) and current (production) dataframes

# For demonstration purposes, let's create dummy data
train_data = {
    'feature1': [i for i in range(100)],
    'feature2': [i*2 for i in range(100)],
    'target': [i % 2 for i in range(100)]
}
prod_data = {
    'feature1': [i + 5 for i in range(100)], # Introduce a shift
    'feature2': [i*2 + 10 for i in range(100)], # Introduce a shift
    'target': [i % 2 for i in range(100)]
}
train_df = pd.DataFrame(train_data)
prod_df = pd.DataFrame(prod_data)

# Create an Evidently Report with DataDriftPreset
# This preset checks for drift in all numerical and categorical features.
drift_report = Report(metrics=[
    DataDriftPreset(),
])

# Run the report with reference and current data
try:
    drift_report.run(reference_data=train_df, current_data=prod_df)
    # Get the drift detection results as a dictionary
    report_dict = drift_report.as_dict()

    # Check if dataset drift was detected
    # The structure might vary slightly based on Evidently version and presets used.
    # This example assumes 'dataset_drift' is a boolean or a score indicating drift.
    # For DataDriftPreset, it typically indicates if ANY column drifted significantly.
    # More granular checks can be done by inspecting individual feature drift results.

    # A more robust check would iterate through specific features and their drift scores
    # For simplicity, we'll use a summary metric if available or check a key indicator.
    # In many cases, `dataset_drift` refers to whether *any* significant drift was detected.
    # Let's check for drift in a specific feature for illustration:
    # Example: Check drift for 'feature1'
    feature1_drift_info = None
    for metric in report_dict.get('metrics', []):
        if metric['metric'] == 'DataDrift':
            for result_key, result_value in metric['result'].items():
                if 'data_drift_present' in result_value and result_value['column_name'] == 'feature1':
                    feature1_drift_info = result_value
                    break
            if feature1_drift_info:
                break

    drift_detected = False
    if feature1_drift_info and feature1_drift_info.get('data_drift_present', False):
        drift_detected = True
        print(f"Data drift detected in feature1: {feature1_drift_info}")
    else:
        print("No significant data drift detected in feature1.")

    # If drift is detected, send an alert
    if drift_detected:
        print("Sending drift alert email...")
        sender_email = "your_email@example.com"
        receiver_email = "recipient_email@example.com"
        password = "your_email_password" # Consider using environment variables for sensitive info

        message = MIMEText("Alert: Significant data drift detected in ML model's production data.")

        message["Subject"] = "ML Model Alert: Data Drift Detected"
        message["From"] = sender_email
        message["To"] = receiver_email

        try:
            # Connect to the SMTP server
            smtp_server = smtplib.SMTP_SSL('smtp.gmail.com', 465) # Example for Gmail
            smtp_server.login(sender_email, password)
            smtp_server.sendmail(sender_email, receiver_email, message.as_string())
            print("Email sent successfully!")
        except Exception as e:
            print(f"Failed to send email: {e}")
        finally:
            smtp_server.quit()
    else:
        print("No drift detected, no alert sent.")

except Exception as e:
    print(f"An error occurred during Evidently report generation: {e}")

import requests
import time

def get_latest_accuracy():
    """
    Placeholder function to fetch the latest model accuracy.
    In a real scenario, this would read from a database, log file,
    or an API that exposes model performance metrics.
    """
    # Simulate fetching accuracy from a hypothetical API or database
    # For demonstration, let's just return a changing value
    current_time = time.time()
    # Simulate a drop after some time
    if current_time % 120 < 60: # Simulate accuracy around 0.9
        return 0.9
    else: # Simulate accuracy drop
        return 0.75

def send_slack_alert(message):
    """Sends a message to a Slack channel via a webhook."""
    slack_webhook_url = "YOUR_SLACK_WEBHOOK_URL" # Replace with your actual webhook URL
    payload = {"text": message}
    try:
        response = requests.post(slack_webhook_url, json=payload)
        response.raise_for_status() # Raise an exception for bad status codes
        print(f"Slack alert sent: {message}")
    except requests.exceptions.RequestException as e:
        print(f"Error sending Slack alert: {e}")

if __name__ == "__main__":
    ALERT_THRESHOLD = 0.80
    CHECK_INTERVAL_SECONDS = 300 # Check every 5 minutes

    print("Starting ML performance monitoring script...")
    while True:
        current_accuracy = get_latest_accuracy()
        print(f"Current accuracy: {current_accuracy:.2f}")

        if current_accuracy < ALERT_THRESHOLD:
            alert_message = f"⚠️ ALERT: ML model performance degradation detected! Accuracy dropped to {current_accuracy:.2f} (threshold: {ALERT_THRESHOLD:.2f})."
            send_slack_alert(alert_message)
        else:
            print("Model performance is within acceptable limits.")

        time.sleep(CHECK_INTERVAL_SECONDS)

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.datasets import make_classification
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# --- Step 1: Generate Initial Training Data ---
# Create a synthetic dataset for training.
# n_samples: number of data points
# n_features: total number of features
# n_informative: number of features that are actually useful for prediction
# n_redundant: number of features that are linear combinations of informative features
# random_state: for reproducibility
X_train, y_train = make_classification(
    n_samples=1000,
    n_features=20,
    n_informative=15,
    n_redundant=5,
    random_state=42
)
print(f"Generated training data with shape: {X_train.shape}")

# --- Step 2: Train the Model ---
# Use a RandomForestClassifier as an example model.
model = RandomForestClassifier(random_state=42)
model.fit(X_train, y_train)
print("Model trained successfully.")

# --- Step 3: Simulate Production Data with Drift ---
# Generate new data that mimics production, but with introduced drift.
# 'shift' parameter introduces a shift in the distribution of features,
# simulating concept drift or data drift.
X_prod, y_prod = make_classification(
    n_samples=1000,
    n_features=20,
    n_informative=10, # Reduced informative features can also simulate degradation
    n_redundant=10,
    shift=2.0,       # Significant shift in feature distributions
    random_state=99
)
print(f"Generated production data with shape: {X_prod.shape}")

# --- Step 4: Evaluate Performance on Both Datasets ---
# Predict using the trained model on both training and simulated production data.
train_preds = model.predict(X_train)
prod_preds = model.predict(X_prod)

# Calculate accuracy for both sets.
train_acc = accuracy_score(y_train, train_preds)
prod_acc = accuracy_score(y_prod, prod_preds)

print(f"Training Accuracy: {train_acc:.4f}")
print(f"Production Accuracy (simulated drift): {prod_acc:.4f}")

# --- Step 5: Visualize Performance Drop ---
# Create a bar chart to visually compare accuracies.
labels = ["Training Data", "Production Data (Drifted)"]
accuracies = [train_acc, prod_acc]
colors = ["green", "red"]

plt.figure(figsize=(8, 6))
plt.bar(labels, accuracies, color=colors)
plt.title("Model Accuracy Comparison: Before vs. After Simulated Drift")
plt.ylabel("Accuracy")
plt.ylim(0, 1) # Accuracy is always between 0 and 1
plt.grid(axis='y', linestyle='--', alpha=0.7)

# Display the accuracy values on top of the bars
for i, v in enumerate(accuracies):
    plt.text(i, v + 0.02, f"{v:.3f}", ha='center', va='bottom')

plt.show()

# --- Optional: Alert if Performance Drops Too Much ---
# Define a threshold for performance drop (e.g., a 10% decrease in accuracy)
performance_drop_threshold = 0.10
actual_drop = train_acc - prod_acc

if actual_drop > performance_drop_threshold:
    print(f"\n⚠️ WARNING: Model performance degradation detected!")
    print(f"  Actual drop: {actual_drop:.4f}")
    print(f"  Threshold: {performance_drop_threshold:.4f}")
    # In a real application, you would trigger an alert notification here
    # send_alert_notification(f"Model accuracy dropped by {actual_drop:.4f}!")
else:
    print("\nModel performance is stable.")