ML Model Degradation — Detecting Data Drift in Production

# TheCodeForge — ML vs Traditional Software
# Traditional approach: human writes explicit rules
def traditional_spam_filter(email_text: str) -> str:
    """Rules written by a human. Brittle. Breaks on new spam patterns.
    Every new attack vector requires a programmer to update this manually.
    """
    if 'buy now' in email_text.lower():
        return 'spam'
    if 'click here' in email_text.lower():
        return 'spam'
    if 'free money' in email_text.lower():
        return 'spam'
    return 'not spam'

# ML approach: algorithm learns rules from labeled examples
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression

emails = [
    'buy now limited offer click here',
    'free money transfer urgent',
    'meeting agenda for tomorrow',
    'project update attached for review',
    'win a prize claim your reward',
    'quarterly report is ready'
]
labels = ['spam', 'spam', 'not spam', 'not spam', 'spam', 'not spam']

vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(emails)
model = LogisticRegression()
model.fit(X, labels)

new_email = ['urgent free offer claim now']
prediction = model.predict(vectorizer.transform(new_email))
print(f"Traditional filter: {traditional_spam_filter(new_email[0])}")
print(f"ML model prediction: {prediction[0]}")
print("ML generalises to new patterns — traditional filter only catches what it was explicitly told about")

# TheCodeForge — Three types of ML in action
import numpy as np
from sklearn.linear_model import LinearRegression
from sklearn.cluster import KMeans

np.random.seed(42)

# SUPERVISED LEARNING: predict house prices from features
# Data has labels (known prices) — algorithm learns the input-to-output mapping
X_train = np.array([[1400, 3], [1600, 3], [1700, 2], [1875, 3], [1100, 2]])
y_train = np.array([245000, 312000, 279000, 308000, 199000])
model = LinearRegression()
model.fit(X_train, y_train)
prediction = model.predict([[1500, 3]])
print(f"Supervised — Predicted price: ${prediction[0]:,.0f}")

# UNSUPERVISED LEARNING: group customers by behaviour
# No labels — algorithm discovers natural clusters in the data
X_behavior = np.array([[5, 1], [4, 0], [1, 4], [2, 5], [3, 2]])
kmeans = KMeans(n_clusters=2, random_state=42, n_init='auto')
clusters = kmeans.fit_predict(X_behavior)
print(f"Unsupervised — Cluster assignments: {clusters}")

# REINFORCEMENT LEARNING: conceptual illustration
# Agent takes actions in an environment and receives reward signals
# Full RL requires a simulation environment — see the gymnasium library
# Core loop: observe state -> select action -> receive reward -> update policy
print("Reinforcement — Agent learns optimal actions via reward signals")

# SELF-SUPERVISED LEARNING: conceptual illustration
# No human labels required — the model creates its own training signal
# Example: given 'The cat sat on the ___', predict the missing word
# This is how GPT-class models are pre-trained on billions of tokens
print("Self-supervised — Foundation models learn by predicting masked or next tokens from raw data")
print("When you fine-tune or prompt an LLM, you are building on top of self-supervised pre-training")

# TheCodeForge — Supervised Learning: Classification and Regression
import numpy as np
from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, mean_absolute_error, classification_report

np.random.seed(42)

# CLASSIFICATION: predict email spam or not spam
# Features: word_count_ratio, link_count, exclamation_marks, sender_reputation_score
X_class = np.random.rand(1000, 4)
y_class = (
    X_class[:, 0] * 0.3 +
    X_class[:, 1] * 0.5 +
    X_class[:, 2] * 0.4 -
    X_class[:, 3] * 0.6 > 0.3
).astype(int)

# stratify=y preserves class distribution in both train and test splits
X_train, X_test, y_train, y_test = train_test_split(
    X_class, y_class, test_size=0.2, random_state=42, stratify=y_class
)
clf = RandomForestClassifier(n_estimators=100, random_state=42)
clf.fit(X_train, y_train)
accuracy = accuracy_score(y_test, clf.predict(X_test))
print(f"Classification accuracy: {accuracy:.2%}")
print(classification_report(y_test, clf.predict(X_test), target_names=['not spam', 'spam']))

# REGRESSION: predict house price from features
# Features: square_feet, bedrooms, age_years, distance_to_city_km
X_reg = np.random.rand(1000, 4) * np.array([3000, 5, 50, 20])
y_reg = (
    X_reg[:, 0] * 150 +
    X_reg[:, 1] * 25000 -
    X_reg[:, 2] * 1000 -
    X_reg[:, 3] * 5000 +
    100000
)

X_train, X_test, y_train, y_test = train_test_split(
    X_reg, y_reg, test_size=0.2, random_state=42
)
reg = RandomForestRegressor(n_estimators=100, random_state=42)
reg.fit(X_train, y_train)
mae = mean_absolute_error(y_test, reg.predict(X_test))
print(f"Regression MAE: ${mae:,.0f}")

# TheCodeForge — Data Preprocessing: handling the messy reality of real datasets
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline

np.random.seed(42)

# Simulate a messy real-world dataset — missing values, mixed types, different scales
data = pd.DataFrame({
    'age': np.random.randint(18, 70, 1000).astype(float),
    'income': np.random.exponential(50000, 1000),
    'category': np.random.choice(['retail', 'tech', 'healthcare', None], 1000),
    'tenure_years': np.random.randint(0, 30, 1000).astype(float),
    'churned': np.random.choice([0, 1], 1000, p=[0.85, 0.15])
})

# Inject realistic missing values
data.loc[np.random.choice(1000, 50, replace=False), 'age'] = np.nan
data.loc[np.random.choice(1000, 30, replace=False), 'income'] = np.nan

print(f"Missing values before preprocessing:")
print(data.isnull().sum())
print(f"\nData types:\n{data.dtypes}")

# Separate features and target BEFORE any preprocessing
X = data.drop('churned', axis=1)
y = data['churned']

# Split FIRST — preprocessing sees only training data
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# Define preprocessing for each column type
numeric_features = ['age', 'income', 'tenure_years']
categorical_features = ['category']

numeric_pipeline = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),  # fill missing with median from TRAIN only
    ('scaler', StandardScaler())                     # scale to mean=0, std=1 from TRAIN only
])

categorical_pipeline = Pipeline([
    ('imputer', SimpleImputer(strategy='most_frequent')),  # fill missing categoricals
    ('encoder', OneHotEncoder(handle_unknown='ignore', sparse_output=False))
])

# Combine into a single preprocessing step
preprocessor = ColumnTransformer([
    ('num', numeric_pipeline, numeric_features),
    ('cat', categorical_pipeline, categorical_features)
])

# fit_transform on TRAIN only — transform on TEST
X_train_processed = preprocessor.fit_transform(X_train)
X_test_processed = preprocessor.transform(X_test)  # transform only, never fit_transform

print(f"\nTraining set shape after preprocessing: {X_train_processed.shape}")
print(f"Test set shape after preprocessing: {X_test_processed.shape}")
print(f"Missing values after preprocessing: {np.isnan(X_train_processed).sum()}")
print("Preprocessing complete — no data leakage, ready for model training")

# TheCodeForge — Complete ML Pipeline
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import classification_report
import joblib

# Step 1: Data Collection (simulated fraud detection dataset)
np.random.seed(42)
data = pd.DataFrame({
    'transaction_amount': np.random.exponential(100, 5000),
    'merchant_category': np.random.choice(['retail', 'food', 'travel', 'online'], 5000),
    'hour_of_day': np.random.randint(0, 24, 5000),
    'distance_from_home': np.random.exponential(10, 5000),
    'is_fraud': np.random.choice([0, 1], 5000, p=[0.97, 0.03])
})

# Step 2: Preprocessing — encode categoricals, drop nulls
data = pd.get_dummies(data, columns=['merchant_category'], drop_first=True)
data = data.dropna()

# Step 3: Feature Engineering — separate features from target
X = data.drop('is_fraud', axis=1)
y = data['is_fraud']

# Step 4: Train/Test Split — stratify to preserve the 97/3 fraud ratio
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# Step 5: Feature Scaling — fit ONLY on train, transform both
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)  # transform only — never fit_transform on test

# Step 6: Model Training
model = GradientBoostingClassifier(
    n_estimators=200,
    learning_rate=0.1,
    max_depth=5,
    random_state=42
)
model.fit(X_train_scaled, y_train)

# Step 7: Evaluation — look at recall for the fraud class specifically
# The output below reveals the class imbalance problem:
# 97% accuracy but 0% fraud recall means the model catches nothing
predictions = model.predict(X_test_scaled)
print(classification_report(y_test, predictions, target_names=['legitimate', 'fraud'], zero_division=0))

# Step 8: Save model AND scaler with matching version numbers
joblib.dump(model, 'fraud_model_v1.pkl')
joblib.dump(scaler, 'feature_scaler_v1.pkl')
print('Model v1 and scaler v1 saved — version both artifacts together')
print('WARNING: 0% fraud recall — this model needs class weight adjustment before deployment')
print('Next: fix class imbalance, configure drift monitoring, write model card')

# TheCodeForge — Your First Complete ML Model
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score
import joblib

np.random.seed(42)

# --- Step 1: Create a realistic dataset ---
n_samples = 5000
data = pd.DataFrame({
    'amount': np.random.exponential(100, n_samples),
    'hour': np.random.randint(0, 24, n_samples),
    'distance_km': np.random.exponential(10, n_samples),
    'is_fraud': np.random.choice([0, 1], n_samples, p=[0.97, 0.03])
})

X = data.drop('is_fraud', axis=1)
y = data['is_fraud']
print(f"Dataset: {len(data)} rows, fraud rate: {y.mean():.1%}")
print(f"Class distribution: {dict(y.value_counts())}\n")

# --- Step 2: Split FIRST, preprocess SECOND ---
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)
scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train)
X_test_s = scaler.transform(X_test)

# --- Step 3: Train without handling imbalance (the naive approach) ---
model_v1 = GradientBoostingClassifier(n_estimators=100, random_state=42)
model_v1.fit(X_train_s, y_train)
print("=== Model v1: No class balancing ===")
print(classification_report(y_test, model_v1.predict(X_test_s),
      target_names=['legit', 'fraud'], zero_division=0))
print(f"Confusion Matrix:\n{confusion_matrix(y_test, model_v1.predict(X_test_s))}")
print("Problem: 97% accuracy but 0% fraud recall — model is useless for its intended purpose\n")

# --- Step 4: Fix class imbalance with sample_weight ---
# Give fraud cases 30x the weight of legitimate cases during training
# This forces the model to pay attention to the minority class
weights = np.where(y_train == 1, 30.0, 1.0)
model_v2 = GradientBoostingClassifier(n_estimators=200, max_depth=4, random_state=42)
model_v2.fit(X_train_s, y_train, sample_weight=weights)

print("=== Model v2: With class balancing ===")
preds_v2 = model_v2.predict(X_test_s)
print(classification_report(y_test, preds_v2, target_names=['legit', 'fraud'], zero_division=0))
print(f"Confusion Matrix:\n{confusion_matrix(y_test, preds_v2)}")

# --- Step 5: Use AUC-ROC for a threshold-independent evaluation ---
probs_v2 = model_v2.predict_proba(X_test_s)[:, 1]
auc = roc_auc_score(y_test, probs_v2)
print(f"AUC-ROC: {auc:.4f} (0.5 = random, 1.0 = perfect)")

# --- Step 6: Cross-validate for honest accuracy ---
cv_scores = cross_val_score(model_v2, scaler.transform(X), y, cv=5, scoring='roc_auc')
print(f"Cross-validated AUC: {cv_scores.mean():.4f} (+/- {cv_scores.std():.4f})\n")

# --- Step 7: Save versioned artifacts ---
joblib.dump(model_v2, 'fraud_model_v2.pkl')
joblib.dump(scaler, 'fraud_scaler_v2.pkl')
print('Saved: fraud_model_v2.pkl, fraud_scaler_v2.pkl')
print('Next steps: configure drift monitoring, write model card, set up shadow deployment')

# TheCodeForge — Diagnosing ML Failure Modes
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import confusion_matrix

np.random.seed(42)

# OVERFITTING DIAGNOSIS: training accuracy far exceeds test accuracy
X = np.random.rand(1000, 10)
y = (X[:, 0] + X[:, 1] > 1).astype(int)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)

# Unconstrained depth — model memorises training data including noise
overfit_model = RandomForestClassifier(n_estimators=500, max_depth=None, random_state=42)
overfit_model.fit(X_train, y_train)
train_acc = overfit_model.score(X_train, y_train)
test_acc = overfit_model.score(X_test, y_test)
print(f"Overfitting signal — Train: {train_acc:.2%}, Test: {test_acc:.2%}")
print(f"Gap of {(train_acc - test_acc):.2%} indicates overfitting — constrain depth and use cross-validation")

# FIX: constrain depth and use cross-validation for honest accuracy estimate
scores = cross_val_score(
    RandomForestClassifier(n_estimators=100, max_depth=5, random_state=42),
    X, y, cv=5
)
print(f"Cross-validation accuracy: {scores.mean():.2%} (+/- {scores.std():.2%})")

# CLASS IMBALANCE DIAGNOSIS
# The model predicts 'legitimate' for everything and achieves 97% accuracy
# This is the most dangerous failure mode — it looks correct until you read the confusion matrix
y_imbalanced = np.concatenate([np.zeros(970), np.ones(30)])
X_imbalanced = np.random.rand(1000, 5)
X_train_imb, X_test_imb, y_train_imb, y_test_imb = train_test_split(
    X_imbalanced, y_imbalanced, test_size=0.3, random_state=42
)
model = RandomForestClassifier(random_state=42)
model.fit(X_train_imb, y_train_imb)
preds = model.predict(X_test_imb)
print(f"\nImbalanced dataset — Accuracy: {model.score(X_test_imb, y_test_imb):.2%}")
print(f"Confusion Matrix:\n{confusion_matrix(y_test_imb, preds)}")
print("Bottom-left value = fraud cases the model missed — that number matters more than accuracy")

# TheCodeForge — Classical ML vs Foundation Model decision guide
# Use this as a mental checklist before choosing your approach

def recommend_ml_approach(
    data_type: str,
    has_labels: bool,
    sample_count: int,
    needs_language_understanding: bool,
    latency_requirement_ms: int
) -> str:
    """
    A simplified decision function — not a substitute for engineering judgment,
    but a useful gut-check before committing weeks to an architecture.
    """
    if data_type == 'tabular' and has_labels and sample_count > 1000:
        return (
            "Classical ML recommended: gradient boosting or random forest. "
            "Fast to train, cheap to serve, fully auditable. "
            "Start with XGBoost or LightGBM."
        )

    if needs_language_understanding or data_type in ['text', 'image', 'audio']:
        if sample_count < 500:
            return (
                "Foundation model with few-shot prompting recommended. "
                "You do not have enough data to fine-tune reliably. "
                "Use RAG if you need domain-specific knowledge."
            )
        if sample_count >= 500:
            return (
                "Fine-tuned foundation model recommended. "
                "Use LoRA or QLoRA for parameter-efficient fine-tuning. "
                "Consider smaller models first — Phi-3, Gemma-2, or Mistral variants."
            )

    if latency_requirement_ms < 50:
        return (
            "Classical ML strongly recommended. "
            "Foundation model inference rarely achieves sub-50ms P99 latency "
            "without aggressive quantisation. Consider distilled models if foundation models are required."
        )

    return "Evaluate both approaches on a small prototype before committing to architecture."

# Example decisions a team would face
print("Scenario 1: Tabular churn prediction")
print(recommend_ml_approach('tabular', True, 50000, False, 200))
print()
print("Scenario 2: Customer support ticket classification")
print(recommend_ml_approach('text', False, 200, True, 500))
print()
print("Scenario 3: Real-time pricing engine")
print(recommend_ml_approach('tabular', True, 10000, False, 30))

// io.thecodeforge — ml-ai tutorial

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_error

# Real dataset: Boston Housing (simulated)
df = pd.read_csv("boston_housing.csv")
X = df.drop("median_value", axis=1)
y = df["median_value"]

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

model = LinearRegression()
model.fit(X_train, y_train)

predictions = model.predict(X_test)
error = mean_absolute_error(y_test, predictions)

print(f"First model MAE: ${error:.2f}k")

// io.thecodeforge — ml-ai tutorial

// This is pseudocode for the PartyRock workflow, not raw Python.
// Represents a 3-node chain you'd build visually.

WORKFLOW_CHAIN = {
    "node_1": {
        "model": "Claude Instant",
        "prompt": "Extract all product names from this customer transcript.",
        "input": "{{user_query}}"
    },
    "node_2": {
        "model": "Claude Instant",
        "prompt": "Categorize each product name: ['Electronics', 'Apparel', 'Food'].",
        "input": "{{node_1.output}}"
    },
    "node_3": {
        "model": "Claude Haiku",
        "prompt": "Check if any extracted product names are on the recall list. Output 'RECALL' or 'CLEAR'.",
        "input": "{{node_2.output}}"
    }
}

// Run: PartyRock invokes each node sequentially.

// io.thecodeforge — ml-ai tutorial

from sklearn.cluster import KMeans
import numpy as np

# 100 customer records: [annual_spend, session_count]
X = np.random.rand(100, 2) * 100

kmeans = KMeans(n_clusters=3, random_state=42, n_init=10)
kmeans.fit(X)
predictions = kmeans.predict(X)

print("Cluster assignments for first 10 customers:")
print(predictions[:10])
print("\nCentroids:")
print(kmeans.cluster_centers_)

// io.thecodeforge — ml-ai tutorial

from sklearn.decomposition import PCA
from sklearn.datasets import make_classification

# Fake high-dimensional data: 1000 samples, 50 features
X, y = make_classification(n_samples=1000, n_features=50, random_state=42)

pca = PCA(n_components=2)
X_reduced = pca.fit_transform(X)

print(f"Original shape:  {X.shape}")
print(f"Reduced shape:   {X_reduced.shape}")
print(f"Explained variance ratio: {pca.explained_variance_ratio_}")
print(f"Total variance retained: {sum(pca.explained_variance_ratio_):.2%}")

// io.thecodeforge — ml-ai tutorial

import numpy as np
from collections import Counter

class KNN:
    def __init__(self, k=3):
        self.k = k
    
    def fit(self, X, y):
        self.X_train = np.array(X)
        self.y_train = np.array(y)
    
    def predict(self, X):
        predictions = []
        for x in X:
            distances = np.linalg.norm(self.X_train - x, axis=1)
            k_indices = np.argsort(distances)[:self.k]
            k_labels = self.y_train[k_indices]
            predictions.append(Counter(k_labels).most_common(1)[0][0])
        return np.array(predictions)

# demo
X = [[1,2],[2,3],[3,1],[6,5],[7,7]]
y = [0,0,0,1,1]
model = KNN(k=3)
model.fit(X, y)
print(model.predict([[4,4]]))

// io.thecodeforge — ml-ai tutorial

from sklearn.naive_bayes import GaussianNB
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

model = GaussianNB()
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
print(f"Accuracy: {accuracy_score(y_test, y_pred):.2f}")
print(f"Predicted class probs: {model.predict_proba(X_test[:1])}")

// io.thecodeforge — ml-ai tutorial
from sklearn.linear_model import LinearRegression
import numpy as np

X = np.array([[1], [2], [3], [4]])
y = np.array([2.1, 4.0, 5.9, 8.2])

model = LinearRegression()
model.fit(X, y)
print(f'Slope: {model.coef_[0]:.2f}, Intercept: {model.intercept_:.2f}')
prediction = model.predict([[5]])
print(f'Prediction for x=5: {prediction[0]:.2f}')

// io.thecodeforge — ml-ai tutorial
from sklearn.neighbors import KNeighborsClassifier
import numpy as np

X_train = np.array([[1, 2], [2, 3], [3, 1], [6, 5], [7, 7], [8, 6]])
y_train = np.array([0, 0, 0, 1, 1, 1])

knn = KNeighborsClassifier(n_neighbors=3)
knn.fit(X_train, y_train)
X_new = np.array([[5, 4]])
print(f'Predicted class: {knn.predict(X_new)[0]}')