⚡️ Saturday ML Spark – 🐤 Canary Deployment for Machine Learning Models

# 📦 Import Required Libraries
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier

from sklearn.metrics import (
    accuracy_score,
    precision_score,
    recall_score,
    f1_score
)


# =========================================================
# 🧩 Load Dataset
# =========================================================

data = load_breast_cancer()

X = pd.DataFrame(
    data.data,
    columns=data.feature_names
)

# Keep target as a Pandas Series
y = pd.Series(
    data.target,
    name="target"
)


# =========================================================
# ✂️ Split Training and Live Data
# =========================================================

X_train, X_live, y_train, y_live = train_test_split(
    X,
    y,
    test_size=0.30,
    random_state=42,
    stratify=y
)


# Reset live-data indices for clean traffic routing
X_live = X_live.reset_index(drop=True)
y_live = y_live.reset_index(drop=True)


# =========================================================
# 🤖 Train Production Model – Version 1
# =========================================================

production_model = LogisticRegression(
    max_iter=5000
)

production_model.fit(
    X_train,
    y_train
)


# =========================================================
# 🐤 Train Canary Model – Version 2
# =========================================================

canary_model = RandomForestClassifier(
    n_estimators=200,
    random_state=42
)

canary_model.fit(
    X_train,
    y_train
)


# =========================================================
# 🚦 Configure Canary Traffic
# =========================================================

CANARY_TRAFFIC_PERCENTAGE = 0.10

rng = np.random.default_rng(seed=42)

traffic_assignment = rng.choice(
    ["production", "canary"],
    size=len(X_live),
    p=[
        1 - CANARY_TRAFFIC_PERCENTAGE,
        CANARY_TRAFFIC_PERCENTAGE
    ]
)


# =========================================================
# 🛣️ Route Live Requests
# =========================================================

production_mask = (
    traffic_assignment == "production"
)

canary_mask = (
    traffic_assignment == "canary"
)

X_production = X_live.loc[production_mask]
y_production = y_live.loc[production_mask]

X_canary = X_live.loc[canary_mask]
y_canary = y_live.loc[canary_mask]


print("=== Traffic Distribution ===")
print(
    "Production Requests:",
    len(X_production)
)

print(
    "Canary Requests:",
    len(X_canary)
)


# =========================================================
# 📊 Generate Predictions
# =========================================================

production_predictions = production_model.predict(
    X_production
)

canary_predictions = canary_model.predict(
    X_canary
)


# =========================================================
# 📈 Calculate Model Metrics
# =========================================================

production_metrics = {
    "Model": "Production Model",
    "Traffic Count": len(X_production),

    "Accuracy": accuracy_score(
        y_production,
        production_predictions
    ),

    "Precision": precision_score(
        y_production,
        production_predictions,
        zero_division=0
    ),

    "Recall": recall_score(
        y_production,
        production_predictions,
        zero_division=0
    ),

    "F1 Score": f1_score(
        y_production,
        production_predictions,
        zero_division=0
    )
}


canary_metrics = {
    "Model": "Canary Model",
    "Traffic Count": len(X_canary),

    "Accuracy": accuracy_score(
        y_canary,
        canary_predictions
    ),

    "Precision": precision_score(
        y_canary,
        canary_predictions,
        zero_division=0
    ),

    "Recall": recall_score(
        y_canary,
        canary_predictions,
        zero_division=0
    ),

    "F1 Score": f1_score(
        y_canary,
        canary_predictions,
        zero_division=0
    )
}


# =========================================================
# 📋 Create Results DataFrame
# =========================================================

results = pd.DataFrame([
    production_metrics,
    canary_metrics
])

print("\n=== Canary Deployment Results ===\n")
print(results.round(4))


# =========================================================
# 📊 Visualize Traffic Allocation
# =========================================================

traffic_summary = pd.Series(
    traffic_assignment
).value_counts()

plt.figure(figsize=(6, 4))

plt.bar(
    ["Production", "Canary"],
    [
        traffic_summary.get("production", 0),
        traffic_summary.get("canary", 0)
    ]
)

plt.title("Canary Deployment Traffic Allocation")
plt.xlabel("Model")
plt.ylabel("Request Count")
plt.tight_layout()
plt.show()


# =========================================================
# 📈 Visualize Metric Comparison
# =========================================================

metric_columns = [
    "Accuracy",
    "Precision",
    "Recall",
    "F1 Score"
]

results.set_index("Model")[
    metric_columns
].plot(
    kind="bar",
    figsize=(9, 5)
)

plt.title("Production vs Canary Model Performance")
plt.ylabel("Score")
plt.ylim(0, 1.05)
plt.xticks(rotation=0)
plt.grid(
    axis="y",
    linestyle="--",
    alpha=0.5
)
plt.tight_layout()
plt.show()


# =========================================================
# 🚦 Simulate Deployment Decision
# =========================================================

production_f1 = production_metrics[
    "F1 Score"
]

canary_f1 = canary_metrics[
    "F1 Score"
]

MINIMUM_CANARY_SAMPLES = 10
MINIMUM_IMPROVEMENT = 0.00


print("\n=== Deployment Decision ===")

if len(X_canary) < MINIMUM_CANARY_SAMPLES:

    decision = (
        "Continue collecting canary traffic "
        "before making a rollout decision."
    )

elif (
    canary_f1
    >= production_f1 + MINIMUM_IMPROVEMENT
):

    decision = (
        "Increase canary traffic gradually."
    )

else:

    decision = (
        "Roll back the canary model "
        "and keep the production model active."
    )

print(decision)


# =========================================================
# 📝 Create Request-Level Deployment Logs
# =========================================================

deployment_logs = pd.DataFrame({
    "request_id": range(
        1,
        len(X_live) + 1
    ),

    "assigned_model": traffic_assignment,

    "actual": y_live
})


deployment_logs["prediction"] = np.nan


deployment_logs.loc[
    production_mask,
    "prediction"
] = production_predictions


deployment_logs.loc[
    canary_mask,
    "prediction"
] = canary_predictions


deployment_logs["prediction"] = (
    deployment_logs["prediction"]
    .astype(int)
)


deployment_logs["correct"] = (
    deployment_logs["actual"]
    ==
    deployment_logs["prediction"]
)


print("\n=== Deployment Logs Preview ===\n")
print(deployment_logs.head(10))


# =========================================================
# 💾 Save Results
# =========================================================

results.to_csv(
    "canary_model_metrics.csv",
    index=False
)

deployment_logs.to_csv(
    "canary_deployment_logs.csv",
    index=False
)

print(
    "\nModel metrics saved to canary_model_metrics.csv"
)

print(
    "Deployment logs saved to canary_deployment_logs.csv"
)


# =========================================================
# ✅ Final Summary
# =========================================================

print("\n" + "=" * 48)
print(" Canary Deployment Simulation Completed")
print("=" * 48)

print(
    f"Production Traffic: "
    f"{len(X_production)} requests"
)

print(
    f"Canary Traffic: "
    f"{len(X_canary)} requests"
)

print(
    f"Production F1: "
    f"{production_f1:.4f}"
)

print(
    f"Canary F1: "
    f"{canary_f1:.4f}"
)

print(
    f"Decision: {decision}"
)