ml-expert

name	ml-expert
version	1.0.0
description	Expert-level machine learning, deep learning, model training, and MLOps
category	ai
tags	["machine-learning","deep-learning","neural-networks","mlops","data-science"]
allowed-tools	["Read","Write","Edit","Bash(python:*)"]

Machine Learning Expert

Expert guidance for machine learning systems, deep learning, model training, deployment, and MLOps practices.

Core Concepts

Machine Learning Fundamentals

• Supervised learning (classification, regression)
• Unsupervised learning (clustering, dimensionality reduction)
• Reinforcement learning
• Feature engineering
• Model evaluation and validation
• Hyperparameter tuning

Deep Learning

• Neural networks (CNNs, RNNs, Transformers)
• Transfer learning
• Fine-tuning pre-trained models
• Attention mechanisms
• GANs (Generative Adversarial Networks)
• Autoencoders

MLOps

• Model versioning and tracking
• Experiment management
• Model deployment and serving
• Monitoring and retraining
• CI/CD for ML pipelines
• A/B testing for models

Supervised Learning

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 RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix
import joblib

class MLPipeline:
    def __init__(self):
        self.scaler = StandardScaler()
        self.model = None
        self.feature_names = None

    def prepare_data(self, X: pd.DataFrame, y: pd.Series, test_size: float = 0.2):
        """Split and scale data"""
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=test_size, random_state=42, stratify=y
        )

        # Scale features
        X_train_scaled = self.scaler.fit_transform(X_train)
        X_test_scaled = self.scaler.transform(X_test)

        self.feature_names = X.columns.tolist()

        return X_train_scaled, X_test_scaled, y_train, y_test

    def train_classifier(self, X_train, y_train, n_estimators: int = 100):
        """Train random forest classifier"""
        self.model = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=,
            random_state=,
            n_jobs=-
        )

        .model.fit(X_train, y_train)

        
        cv_scores = cross_val_score(.model, X_train, y_train, cv=)

         {
            : cv_scores.mean(),
            : cv_scores.std(),
            : ((
                .feature_names,
                .model.feature_importances_
            ))
        }

     () -> :
        
        y_pred = .model.predict(X_test)
        y_proba = .model.predict_proba(X_test)

         {
            : y_pred,
            : y_proba,
            : confusion_matrix(y_test, y_pred).tolist(),
            : classification_report(y_test, y_pred, output_dict=)
        }

     ():
        
        joblib.dump({
            : .model,
            : .scaler,
            : .feature_names
        }, path)

Deep Learning with PyTorch

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset

class NeuralNetwork(nn.Module):
    def __init__(self, input_size: int, hidden_size: int, num_classes: int):
        super().__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.3)
        self.fc2 = nn.Linear(hidden_size, hidden_size // 2)
        self.fc3 = nn.Linear(hidden_size // 2, num_classes)

    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.dropout(x)
        x = self.fc2(x)
        x = self.relu(x)
        x = self.fc3(x)
        return x

class Trainer:
    def __init__(self, model, device='cuda' if torch.cuda.is_available() else 'cpu'):
        self.model = model.to(device)
        self.device = device
        .criterion = nn.CrossEntropyLoss()
        .optimizer = optim.Adam(model.parameters(), lr=)

     () -> :
        
        .model.train()
        total_loss = 

         batch_idx, (data, target)  (dataloader):
            data, target = data.to(.device), target.to(.device)

            .optimizer.zero_grad()
            output = .model(data)
            loss = .criterion(output, target)

            loss.backward()
            .optimizer.step()

            total_loss += loss.item()

         total_loss / (dataloader)

     () -> :
        
        .model.()
        correct = 
        total = 

         torch.no_grad():
             data, target  dataloader:
                data, target = data.to(.device), target.to(.device)
                output = .model(data)
                _, predicted = torch.(output.data, )
                total += target.size()
                correct += (predicted == target).().item()

         {
            :  * correct / total,
            : total
        }

     ():
        
        history = {: [], : []}

         epoch  (epochs):
            train_loss = .train_epoch(train_loader)
            val_metrics = .evaluate(val_loader)

            history[].append(train_loss)
            history[].append(val_metrics[])

            ()

         history

Model Deployment

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np

app = FastAPI()

class PredictionRequest(BaseModel):
    features: list[float]

class PredictionResponse(BaseModel):
    prediction: int
    probability: float
    model_version: str

class ModelServer:
    def __init__(self, model_path: str):
        self.model_data = joblib.load(model_path)
        self.model = self.model_data["model"]
        self.scaler = self.model_data["scaler"]
        self.version = "1.0.0"

    def predict(self, features: np.ndarray) -> dict:
        """Make prediction"""
        # Scale features
        features_scaled = self.scaler.transform(features.reshape(1, -1))

        # Predict
        prediction = self.model.predict(features_scaled)[0]
        probability = self.model.predict_proba(features_scaled)[].()

         {
            : (prediction),
            : (probability),
            : .version
        }


model_server = ModelServer()


  ():
    :
        features = np.array(request.features)
        result = model_server.predict(features)
         PredictionResponse(**result)
     Exception  e:
         HTTPException(status_code=, detail=(e))


  ():
     {: , : model_server.version}

MLOps with MLflow

import mlflow
import mlflow.sklearn
from mlflow.tracking import MlflowClient

class MLflowExperiment:
    def __init__(self, experiment_name: str):
        mlflow.set_experiment(experiment_name)
        self.client = MlflowClient()

    def log_training_run(self, model, X_train, y_train, X_test, y_test,
                        params: dict):
        """Log training run with MLflow"""
        with mlflow.start_run():
            # Log parameters
            mlflow.log_params(params)

            # Train model
            model.fit(X_train, y_train)

            # Evaluate
            train_score = model.score(X_train, y_train)
            test_score = model.score(X_test, y_test)

            # Log metrics
            mlflow.log_metric("train_accuracy", train_score)
            mlflow.log_metric("test_accuracy", test_score)

            # Log model
            mlflow.sklearn.log_model(model, "model")

            # Log feature importance
            if hasattr(model, 'feature_importances_'):
                feature_importance = dict(enumerate(model.feature_importances_))
                mlflow.log_dict(feature_importance, "feature_importance.json")

            run_id = mlflow.active_run().info.run_id
            return run_id

    def register_model(self, run_id: str, model_name: str):
        
        model_uri = 
        mlflow.register_model(model_uri, model_name)

     ():
        
        .client.transition_model_version_stage(
            name=model_name,
            version=version,
            stage=
        )

Best Practices

Data Preparation

• Handle missing values appropriately
• Scale/normalize features
• Encode categorical variables properly
• Split data before any preprocessing
• Use stratified splits for imbalanced data
• Create validation set for hyperparameter tuning

Model Training

• Start with simple baselines
• Use cross-validation
• Monitor training and validation metrics
• Implement early stopping
• Save best model checkpoints
• Track experiments systematically

Deployment

• Version models and datasets
• Monitor model performance in production
• Implement model A/B testing
• Set up retraining pipelines
• Log predictions for analysis
• Implement fallback mechanisms

Anti-Patterns

❌ Training on test data (data leakage) ❌ No validation set for hyperparameter tuning ❌ Ignoring class imbalance ❌ Not scaling features ❌ Overfitting to training data ❌ No model versioning ❌ Missing monitoring in production

Resources

• Scikit-learn: https://scikit-learn.org/
• PyTorch: https://pytorch.org/
• TensorFlow: https://www.tensorflow.org/
• MLflow: https://mlflow.org/
• Hugging Face: https://huggingface.co/