Ml Engineer Skill

Name: Ml Engineer
Author: 404kidwiz

404kidwiz/claude-supercode-skills

146 installs
404kidwiz/claude-supercode-skills

Design, train, and deploy machine learning models for production systems and data-driven applications.

About

Engineer production machine learning systems. Covers model architecture selection, training and evaluation best practices, feature engineering, and strategies for deploying models at scale.

Model training
Feature engineering
Evaluation metrics
Production deployment
Data pipelines

Ml Engineer by the numbers

146 all-time installs (skills.sh)
Ranked #759 of 2,091 Data Science & ML skills by installs in the Skillselion catalog
Data as of Aug 11, 2026 (Skillselion catalog sync)

npx skills add https://github.com/404kidwiz/claude-supercode-skills --skill ml-engineer

Add your badge

Show developers this skill is listed on Skillselion. Paste this into your README.

[![Listed on Skillselion](https://skillselion.com/badge/skills/404kidwiz/claude-supercode-skills/ml-engineer.svg)](https://skillselion.com/skills/404kidwiz/claude-supercode-skills/ml-engineer)

Installs	146
Repository	404kidwiz/claude-supercode-skills ↗

What it does

Design, train, and deploy machine learning models for production systems and data-driven applications.

Files

SKILL.mdMarkdownGitHub ↗

Machine Learning Engineer

Purpose

Provides MLOps and production ML engineering expertise specializing in end-to-end ML pipelines, model deployment, and infrastructure automation. Bridges data science and production engineering with robust, scalable machine learning systems.

When to Use

Building end-to-end ML pipelines (Data → Train → Validate → Deploy)
Deploying models to production (Real-time API, Batch, or Edge)
Implementing MLOps practices (CI/CD for ML, Experiment Tracking)
Optimizing model performance (Latency, Throughput, Resource usage)
Setting up feature stores and model registries
Implementing model monitoring (Drift detection, Performance tracking)
Scaling training workloads (Distributed training)

--- ---

2. Decision Framework

Model Serving Strategy

Need to serve predictions?
│
├─ Real-time (Low Latency)?
│  │
│  ├─ High Throughput? → **Kubernetes (KServe/Seldon)**
│  ├─ Low/Medium Traffic? → **Serverless (Lambda/Cloud Run)**
│  └─ Ultra-low latency (<10ms)? → **C++/Rust Inference Server (Triton)**
│
├─ Batch Processing?
│  │
│  ├─ Large Scale? → **Spark / Ray**
│  └─ Scheduled Jobs? → **Airflow / Prefect**
│
└─ Edge / Client-side?
   │
   ├─ Mobile? → **TFLite / CoreML**
   └─ Browser? → **TensorFlow.js / ONNX Runtime Web**

Training Infrastructure

Training Environment?
│
├─ Single Node?
│  │
│  ├─ Interactive? → **JupyterHub / SageMaker Notebooks**
│  └─ Automated? → **Docker Container on VM**
│
└─ Distributed?
   │
   ├─ Data Parallelism? → **Ray Train / PyTorch DDP**
   └─ Pipeline orchestration? → **Kubeflow / Airflow / Vertex AI**

Feature Store Decision

Need	Recommendation	Rationale
Simple / MVP	No Feature Store	Use SQL/Parquet files. Overhead of FS is too high.
Team Consistency	Feast	Open source, manages online/offline consistency.
Enterprise / Managed	Tecton / Hopsworks	Full governance, lineage, managed SLA.
Cloud Native	Vertex/SageMaker FS	Tight integration if already in that cloud ecosystem.

Red Flags → Escalate to `oracle`:

"Real-time" training requirements (online learning) without massive infrastructure budget
Deploying LLMs (7B+ params) on CPU-only infrastructure
Training on PII/PHI data without privacy-preserving techniques (Federated Learning, Differential Privacy)
No validation set or "ground truth" feedback loop mechanism

--- ---

3. Core Workflows

Workflow 1: End-to-End Training Pipeline

Goal: Automate model training, validation, and registration using MLflow.

Steps:

1. Setup Tracking

    import mlflow
    import mlflow.sklearn
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.metrics import accuracy_score, precision_score

    mlflow.set_tracking_uri("http://localhost:5000")
    mlflow.set_experiment("churn-prediction-prod")

2. Training Script (`train.py`)

    def train(max_depth, n_estimators):
        with mlflow.start_run():
            # Log params
            mlflow.log_param("max_depth", max_depth)
            mlflow.log_param("n_estimators", n_estimators)
            
            # Train
            model = RandomForestClassifier(
                max_depth=max_depth, 
                n_estimators=n_estimators,
                random_state=42
            )
            model.fit(X_train, y_train)
            
            # Evaluate
            preds = model.predict(X_test)
            acc = accuracy_score(y_test, preds)
            prec = precision_score(y_test, preds)
            
            # Log metrics
            mlflow.log_metric("accuracy", acc)
            mlflow.log_metric("precision", prec)
            
            # Log model artifact with signature
            from mlflow.models.signature import infer_signature
            signature = infer_signature(X_train, preds)
            
            mlflow.sklearn.log_model(
                model, 
                "model",
                signature=signature,
                registered_model_name="churn-model"
            )
            
            print(f"Run ID: {mlflow.active_run().info.run_id}")
    
    if __name__ == "__main__":
        train(max_depth=5, n_estimators=100)

3. Pipeline Orchestration (Bash/Airflow)

    #!/bin/bash
    # Run training
    python train.py
    
    # Check if model passed threshold (e.g. via MLflow API)
    # If yes, transition to Staging

--- ---

Workflow 3: Drift Detection (Monitoring)

Goal: Detect if production data distribution has shifted from training data.

Steps:

1. Baseline Generation (During Training)

    import evidently
    from evidently.report import Report
    from evidently.metric_preset import DataDriftPreset

    # Calculate baseline profile on training data
    report = Report(metrics=[DataDriftPreset()])
    report.run(reference_data=train_df, current_data=test_df)
    report.save_json("baseline_drift.json")

2. Production Monitoring Job

    # Scheduled daily job
    def check_drift():
        # Load production logs (last 24h)
        current_data = load_production_logs()
        reference_data = load_training_data()
        
        report = Report(metrics=[DataDriftPreset()])
        report.run(reference_data=reference_data, current_data=current_data)
        
        result = report.as_dict()
        dataset_drift = result['metrics'][0]['result']['dataset_drift']
        
        if dataset_drift:
            trigger_alert("Data Drift Detected!")
            trigger_retraining()

--- ---

Workflow 5: RAG Pipeline with Vector Database

Goal: Build a production retrieval pipeline using Pinecone/Weaviate and LangChain.

Steps:

1. Ingestion (Chunking & Embedding)

    from langchain.text_splitter import RecursiveCharacterTextSplitter
    from langchain_openai import OpenAIEmbeddings
    from langchain_pinecone import PineconeVectorStore
    
    # Chunking
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    docs = text_splitter.split_documents(raw_documents)
    
    # Embedding & Indexing
    embeddings = OpenAIEmbeddings()
    vectorstore = PineconeVectorStore.from_documents(
        docs, 
        embeddings, 
        index_name="knowledge-base"
    )

2. Retrieval & Generation

    from langchain.chains import RetrievalQA
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
    )
    
    response = qa_chain.invoke("How do I reset my password?")
    print(response['result'])

3. Optimization (Hybrid Search)

Combine Dense Retrieval (Vectors) with Sparse Retrieval (BM25/Keywords).
Use Reranking (Cohere/Cross-Encoder) on the top 20 results to select best 5.

--- ---

5. Anti-Patterns & Gotchas

❌ Anti-Pattern 1: Training-Serving Skew

What it looks like:

Feature logic implemented in SQL for training, but re-implemented in Java/Python for serving.
"Mean imputation" value calculated on training set but not saved; serving uses a different default.

Why it fails:

Model behaves unpredictably in production.
Debugging is extremely difficult.

Correct approach:

Use a Feature Store or shared library for transformations.
Wrap preprocessing logic inside the model artifact (e.g., Scikit-Learn Pipeline, TensorFlow Transform).

❌ Anti-Pattern 2: Manual Deployments

What it looks like:

Data Scientist emails a .pkl file to an engineer.
Engineer manually copies it to a server and restarts the flask app.

Why it fails:

No version control.
No reproducibility.
High risk of human error.

Correct approach:

CI/CD Pipeline: Git push triggers build → test → deploy.
Model Registry: Deploy specific version hash from registry.

❌ Anti-Pattern 3: Silent Failures

What it looks like:

Model API returns 200 OK but prediction is garbage because input data was corrupted (e.g., all Nulls).
Model returns default class 0 for everything.

Why it fails:

Application keeps running, but business value is lost.
Incident detected weeks later by business stakeholders.

Correct approach:

Input Schema Validation: Reject bad requests (Pydantic/TFX).
Output Monitoring: Alert if prediction distribution shifts (e.g., if model predicts "Fraud" 0% of time for 1 hour).

--- ---

7. Quality Checklist

Reliability:

[ ] Health Checks: /health endpoint implemented (liveness/readiness).
[ ] Retries: Client has retry logic with exponential backoff.
[ ] Fallback: Default heuristic exists if model fails or times out.
[ ] Validation: Inputs validated against schema before inference.

Performance:

[ ] Latency: P99 latency meets SLA (e.g., < 100ms).
[ ] Throughput: System autoscales with load.
[ ] Batching: Inference requests batched if using GPU.
[ ] Image Size: Docker image optimized (slim base, multi-stage build).

Reproducibility:

[ ] Versioning: Code, Data, and Model versions linked.
[ ] Artifacts: Saved in object storage (S3/GCS), not local disk.
[ ] Environment: Dependencies pinned (requirements.txt / conda.yaml).

Monitoring:

[ ] Technical: Latency, Error Rate, CPU/Memory/GPU usage.
[ ] Functional: Prediction distribution, Input data drift.
[ ] Business: (If possible) Attribution of prediction to outcome.

Anti-Patterns

Training-Serving Skew

Problem: Feature logic differs between training and serving environments
Symptoms: Model performs well in testing but poorly in production
Solution: Use feature stores or embed preprocessing in model artifacts
Warning Signs: Different code paths for feature computation, hardcoded constants

Manual Deployment

Problem: Deploying models without automation or version control
Symptoms: No traceability, human errors, deployment failures
Solution: Implement CI/CD pipelines with model registry integration
Warning Signs: Email/file transfers of model files, manual server restarts

Silent Failures

Problem: Model failures go undetected
Symptoms: Bad predictions returned without error indication
Solution: Implement input validation, output monitoring, and alerting
Warning Signs: 200 OK responses with garbage data, no anomaly detection

Data Leakage

Problem: Training data contains information not available at prediction time
Symptoms: Unrealistically high training accuracy, poor generalization
Solution: Careful feature engineering and validation split review
Warning Signs: Features that would only be known after prediction

Scikit-Learn Guide

Overview

Scikit-learn is the most popular Python library for machine learning, providing simple and efficient tools for predictive data analysis.

Installation

pip install scikit-learn pandas numpy matplotlib

Quick Start

Basic Workflow

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report

# Load data
X, y = load_your_data()

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

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

# Train model
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train_scaled, y_train)

# Predict
y_pred = model.predict(X_test_scaled)

# Evaluate
accuracy = accuracy_score(y_test, y_pred)
print(f"Accuracy: {accuracy:.2%}")

Model Selection Guide

Classification Models

Random Forest

When: Most classification tasks, good starting point
Pros: Handles non-linear data, robust to overfitting
Cons: Can be slow on large datasets

from sklearn.ensemble import RandomForestClassifier

model = RandomForestClassifier(
    n_estimators=100,
    max_depth=10,
    random_state=42
)

Gradient Boosting

When: High accuracy required, structured data
Pros: Best performance on many tasks
Cons: Longer training time, sensitive to overfitting

from sklearn.ensemble import GradientBoostingClassifier

model = GradientBoostingClassifier(
    n_estimators=100,
    learning_rate=0.1,
    max_depth=5
)

Logistic Regression

When: Binary classification, interpretable results
Pros: Fast, interpretable, good baseline

from sklearn.linear_model import LogisticRegression

model = LogisticRegression(
    penalty='l2',
    C=1.0,
    random_state=42
)

SVM

When: Small to medium datasets, clear margin separation
Pros: Effective in high dimensions
Cons: Slow on large datasets

from sklearn.svm import SVC

model = SVC(
    kernel='rbf',
    C=1.0,
    gamma='scale'
)

Regression Models

Linear Regression

When: Simple linear relationships, interpretable
Pros: Fast, interpretable

from sklearn.linear_model import LinearRegression

model = LinearRegression()

Random Forest Regressor

When: Non-linear relationships, robust model needed
Pros: Handles complex patterns

from sklearn.ensemble import RandomForestRegressor

model = RandomForestRegressor(n_estimators=100)

Gradient Boosting Regressor

When: High accuracy required
Pros: State-of-the-art for tabular data

from sklearn.ensemble import GradientBoostingRegressor

model = GradientBoostingRegressor(n_estimators=100)

Preprocessing Pipeline

Complete Pipeline

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer

# Numeric preprocessing
numeric_features = ['age', 'income']
numeric_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# Categorical preprocessing
categorical_features = ['city', 'gender']
categorical_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

# Combine transformers
preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer, numeric_features),
        ('cat', categorical_transformer, categorical_features)
    ])

# Full pipeline
model = Pipeline(steps=[
    ('preprocessor', preprocessor),
    ('classifier', RandomForestClassifier())
])

model.fit(X_train, y_train)

Feature Scaling

from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler

# StandardScaler (z-score normalization)
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# MinMaxScaler (0-1 range)
scaler = MinMaxScaler()
X_scaled = scaler.fit_transform(X)

# RobustScaler (handles outliers)
scaler = RobustScaler()
X_scaled = scaler.fit_transform(X)

Encoding Categorical Variables

from sklearn.preprocessing import LabelEncoder, OneHotEncoder

# Label encoding for ordinal data
le = LabelEncoder()
y_encoded = le.fit_transform(y)

# One-hot encoding for nominal data
ohe = OneHotEncoder(handle_unknown='ignore')
X_encoded = ohe.fit_transform(X_categorical)

Cross-Validation

K-Fold Cross-Validation

from sklearn.model_selection import cross_val_score

model = RandomForestClassifier(n_estimators=100)
scores = cross_val_score(model, X, y, cv=5)

print(f"CV Scores: {scores}")
print(f"Mean Score: {scores.mean():.2%} (+/- {scores.std() * 2:.2%})")

Stratified K-Fold (for classification)

from sklearn.model_selection import StratifiedKFold, cross_val_score

skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
scores = cross_val_score(model, X, y, cv=skf)

Grid Search CV

from sklearn.model_selection import GridSearchCV

param_grid = {
    'n_estimators': [50, 100, 200],
    'max_depth': [10, 20, None],
    'min_samples_split': [2, 5, 10]
}

grid_search = GridSearchCV(
    RandomForestClassifier(),
    param_grid,
    cv=5,
    scoring='accuracy',
    n_jobs=-1
)

grid_search.fit(X_train, y_train)

print(f"Best params: {grid_search.best_params_}")
print(f"Best score: {grid_search.best_score_:.2%}")

Feature Engineering

Creating New Features

import pandas as pd

# Interaction features
X['feature_product'] = X['feature1'] * X['feature2']

# Polynomial features
from sklearn.preprocessing import PolynomialFeatures
poly = PolynomialFeatures(degree=2)
X_poly = poly.fit_transform(X)

# Binning
X['age_bin'] = pd.cut(X['age'], bins=[0, 18, 35, 50, 100], labels=['child', 'young', 'middle', 'senior'])

Feature Selection

from sklearn.feature_selection import SelectKBest, f_classif

# Select top k features
selector = SelectKBest(f_classif, k=10)
X_selected = selector.fit_transform(X, y)

# Get selected feature names
selected_features = [X.columns[i] for i in selector.get_support(indices=True)]

Model Evaluation

Classification Metrics

from sklearn.metrics import (
    accuracy_score, precision_score, recall_score,
    f1_score, confusion_matrix, classification_report,
    roc_auc_score, roc_curve
)

# Basic metrics
accuracy = accuracy_score(y_test, y_pred)
precision = precision_score(y_test, y_pred, average='weighted')
recall = recall_score(y_test, y_pred, average='weighted')
f1 = f1_score(y_test, y_pred, average='weighted')

# Confusion matrix
cm = confusion_matrix(y_test, y_pred)

# Classification report
report = classification_report(y_test, y_pred)

# ROC AUC
from sklearn.preprocessing import label_binarize
y_bin = label_binarize(y_test, classes=[0, 1, 2])
roc_auc = roc_auc_score(y_bin, y_pred_proba, multi_class='ovr')

Regression Metrics

from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score

mse = mean_squared_error(y_test, y_pred)
rmse = np.sqrt(mse)
mae = mean_absolute_error(y_test, y_pred)
r2 = r2_score(y_test, y_pred)

Model Persistence

Save and Load Models

import joblib

# Save model
joblib.dump(model, 'model.pkl')
joblib.dump(scaler, 'scaler.pkl')

# Load model
loaded_model = joblib.load('model.pkl')
loaded_scaler = joblib.load('scaler.pkl')

# Make predictions
y_pred = loaded_model.predict(loaded_scaler.transform(X_test))

Best Practices

1. Always scale features: Required for many models 2. Use cross-validation: More reliable than single train/test split 3. Handle class imbalance: Use class weights or resampling 4. Start simple: Begin with Random Forest, then try complex models 5. Validate on hold-out set: Don't use test data for tuning 6. Save preprocessing: Store scalers/encoders with model 7. Monitor feature importance: Understand what drives predictions 8. Check for overfitting: Compare train and validation scores

Common Issues

Overfitting

Symptoms: High train accuracy, low test accuracy
Solutions:
Reduce model complexity
Add regularization
Increase training data
Use cross-validation

Underfitting

Symptoms: Low accuracy on both train and test
Solutions:
Increase model complexity
Add more features
Reduce regularization
Train longer

Class Imbalance

Symptoms: Poor performance on minority class
Solutions:
Use class_weight parameter
Resample data (SMOTE, random undersampling)
Use appropriate metrics (F1, AUC)

Resources

"""
Scikit-learn Training Pipeline
Production-ready ML pipeline with scikit-learn
"""

import logging
import joblib
import json
import yaml
from pathlib import Path
from typing import Dict, List, Any, Optional, Union
from dataclasses import dataclass, asdict

try:
    import numpy as np
    import pandas as pd
    from sklearn.model_selection import train_test_split, cross_val_score
    from sklearn.preprocessing import StandardScaler, LabelEncoder
    from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
    from sklearn.metrics import (
        accuracy_score, precision_score, recall_score,
        f1_score, classification_report, confusion_matrix
    )
except ImportError:
    raise ImportError("scikit-learn required: pip install scikit-learn")

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


@dataclass
class ModelConfig:
    model_type: str = "random_forest"
    test_size: float = 0.2
    random_state: int = 42
    scaling: bool = True
    encoding: bool = True

    # Model hyperparameters
    n_estimators: int = 100
    max_depth: Optional[int] = None
    min_samples_split: int = 2
    learning_rate: float = 0.1

    @classmethod
    def from_yaml(cls, path: Union[str, Path]) -> 'ModelConfig':
        with open(path, 'r') as f:
            config = yaml.safe_load(f)
        return cls(**config)


class MLModelTrainer:
    def __init__(self, config: ModelConfig):
        self.config = config
        self.model = None
        self.scaler = None
        self.label_encoder = None
        self.feature_names = None

    def load_data(
        self,
        filepath: Union[str, Path],
        target_column: str,
        feature_columns: Optional[List[str]] = None
    ) -> tuple:
        logger.info(f"Loading data from {filepath}")

        if filepath.suffix == '.csv':
            data = pd.read_csv(filepath)
        elif filepath.suffix in ['.xlsx', '.xls']:
            data = pd.read_excel(filepath)
        else:
            raise ValueError(f"Unsupported file format: {filepath.suffix}")

        if feature_columns is None:
            feature_columns = [col for col in data.columns if col != target_column]

        X = data[feature_columns]
        y = data[target_column]

        logger.info(f"Features: {X.shape}, Target: {y.shape}")
        return X, y

    def preprocess_features(self, X_train: pd.DataFrame, X_test: Optional[pd.DataFrame] = None) -> tuple:
        X_train_processed = X_train.copy()
        X_test_processed = X_test.copy() if X_test is not None else None

        self.feature_names = X_train.columns.tolist()

        # Encode categorical features
        if self.config.encoding:
            X_train_processed = self._encode_features(X_train_processed, fit=True)
            if X_test_processed is not None:
                X_test_processed = self._encode_features(X_test_processed, fit=False)

        # Scale features
        if self.config.scaling:
            X_train_processed, X_test_processed = self._scale_features(
                X_train_processed, X_test_processed
            )

        return X_train_processed, X_test_processed

    def _encode_features(self, X: pd.DataFrame, fit: bool) -> pd.DataFrame:
        cat_columns = X.select_dtypes(include=['object']).columns.tolist()

        if not cat_columns:
            return X

        X_encoded = X.copy()

        for col in cat_columns:
            if fit:
                encoder = LabelEncoder()
                X_encoded[col] = encoder.fit_transform(X[col].astype(str))
                if not hasattr(self, 'encoders'):
                    self.encoders = {}
                self.encoders[col] = encoder
            else:
                if col in self.encoders:
                    X_encoded[col] = self.encoders[col].transform(X[col].astype(str))

        return X_encoded

    def _scale_features(
        self,
        X_train: pd.DataFrame,
        X_test: Optional[pd.DataFrame]
    ) -> tuple:
        if self.scaler is None:
            self.scaler = StandardScaler()

        X_train_scaled = pd.DataFrame(
            self.scaler.fit_transform(X_train),
            columns=X_train.columns,
            index=X_train.index
        )

        X_test_scaled = None
        if X_test is not None:
            X_test_scaled = pd.DataFrame(
                self.scaler.transform(X_test),
                columns=X_test.columns,
                index=X_test.index
            )

        return X_train_scaled, X_test_scaled

    def preprocess_target(self, y_train: pd.Series, y_test: Optional[pd.Series] = None) -> tuple:
        if y_train.dtype == 'object':
            if self.label_encoder is None:
                self.label_encoder = LabelEncoder()
                y_train_encoded = self.label_encoder.fit_transform(y_train)
            else:
                y_train_encoded = self.label_encoder.transform(y_train)

            y_test_encoded = None
            if y_test is not None:
                y_test_encoded = self.label_encoder.transform(y_test)

            return y_train_encoded, y_test_encoded

        return y_train, y_test

    def train_model(self, X_train: pd.DataFrame, y_train: pd.Series):
        logger.info(f"Training {self.config.model_type} model")

        if self.config.model_type == "random_forest":
            self.model = RandomForestClassifier(
                n_estimators=self.config.n_estimators,
                max_depth=self.config.max_depth,
                min_samples_split=self.config.min_samples_split,
                random_state=self.config.random_state
            )
        elif self.config.model_type == "gradient_boosting":
            self.model = GradientBoostingClassifier(
                n_estimators=self.config.n_estimators,
                learning_rate=self.config.learning_rate,
                max_depth=self.config.max_depth,
                random_state=self.config.random_state
            )
        else:
            raise ValueError(f"Unknown model type: {self.config.model_type}")

        self.model.fit(X_train, y_train)
        logger.info("Model training completed")

    def evaluate_model(
        self,
        X_test: pd.DataFrame,
        y_test: pd.Series
    ) -> Dict[str, Any]:
        logger.info("Evaluating model")

        y_pred = self.model.predict(X_test)

        metrics = {
            'accuracy': accuracy_score(y_test, y_pred),
            'precision': precision_score(y_test, y_pred, average='weighted', zero_division=0),
            'recall': recall_score(y_test, y_pred, average='weighted', zero_division=0),
            'f1_score': f1_score(y_test, y_pred, average='weighted', zero_division=0)
        }

        # Cross-validation
        cv_scores = cross_val_score(self.model, X_test, y_test, cv=5)
        metrics['cv_mean'] = cv_scores.mean()
        metrics['cv_std'] = cv_scores.std()

        # Classification report
        metrics['classification_report'] = classification_report(y_test, y_pred)

        # Confusion matrix
        cm = confusion_matrix(y_test, y_pred)
        metrics['confusion_matrix'] = cm.tolist()

        logger.info(f"Accuracy: {metrics['accuracy']:.2%}")
        return metrics

    def save_model(self, filepath: Union[str, Path]):
        model_data = {
            'model': self.model,
            'scaler': self.scaler,
            'label_encoder': self.label_encoder,
            'feature_names': self.feature_names,
            'config': asdict(self.config)
        }

        Path(filepath).parent.mkdir(parents=True, exist_ok=True)
        joblib.dump(model_data, filepath)
        logger.info(f"Model saved to {filepath}")

    def load_model(self, filepath: Union[str, Path]):
        model_data = joblib.load(filepath)

        self.model = model_data['model']
        self.scaler = model_data['scaler']
        self.label_encoder = model_data['label_encoder']
        self.feature_names = model_data['feature_names']
        self.config = ModelConfig(**model_data['config'])

        logger.info(f"Model loaded from {filepath}")

    def predict(self, X: pd.DataFrame) -> np.ndarray:
        X_processed = self._encode_features(X.copy(), fit=False)
        if self.scaler:
            X_processed = pd.DataFrame(
                self.scaler.transform(X_processed),
                columns=X_processed.columns,
                index=X_processed.index
            )

        return self.model.predict(X_processed)

    def get_feature_importance(self) -> Dict[str, float]:
        if not hasattr(self.model, 'feature_importances_'):
            return {}

        importance = self.model.feature_importances_
        return dict(zip(self.feature_names, importance))


def main():
    config = ModelConfig(model_type="random_forest", n_estimators=50)

    trainer = MLModelTrainer(config)

    # Create sample data
    np.random.seed(42)
    n_samples = 1000
    X = pd.DataFrame({
        'feature1': np.random.randn(n_samples),
        'feature2': np.random.randn(n_samples),
        'feature3': np.random.randn(n_samples),
        'feature4': np.random.randn(n_samples)
    })
    y = pd.Series(np.random.randint(0, 2, n_samples), name='target')

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

    # Preprocess
    X_train_processed, X_test_processed = trainer.preprocess_features(X_train, X_test)

    # Train
    trainer.train_model(X_train_processed, y_train)

    # Evaluate
    metrics = trainer.evaluate_model(X_test_processed, y_test)
    print("Metrics:", json.dumps({k: v for k, v in metrics.items() if k != 'classification_report'}, indent=2))

    # Feature importance
    importance = trainer.get_feature_importance()
    print("\nFeature Importance:", importance)


if __name__ == "__main__":
    main()

"""
Hyperparameter Tuning with Optuna
Automated hyperparameter optimization
"""

import logging
from typing import Dict, Any, Optional, Callable, List
from dataclasses import dataclass
import joblib
import numpy as np

try:
    import optuna
    from optuna.pruners import MedianPruner
    from optuna.samplers import TPESampler
    import optuna.visualization as vis
except ImportError:
    raise ImportError("optuna required: pip install optuna")

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


@dataclass
class TuningConfig:
    n_trials: int = 100
    timeout: Optional[int] = None
    direction: str = "maximize"
    metric_name: str = "accuracy"
    study_name: str = "optimization"

    # Pruning
    enable_pruning: bool = True
    n_startup_trials: int = 10

    # Sampler
    sampler_type: str = "tpe"  # "tpe", "random", "grid"

    @classmethod
    def from_dict(cls, config: Dict[str, Any]) -> 'TuningConfig':
        return cls(**config)


class HyperparameterTuner:
    def __init__(self, config: TuningConfig):
        self.config = config
        self.study = None
        self.best_params = None
        self.best_value = None

    def create_study(self, objective: Callable, storage_path: Optional[str] = None):
        logger.info(f"Creating study: {self.config.study_name}")

        sampler = self._get_sampler()

        pruner = None
        if self.config.enable_pruning:
            pruner = MedianPruner(
                n_startup_trials=self.config.n_startup_trials,
                n_warmup_steps=5
            )

        self.study = optuna.create_study(
            study_name=self.config.study_name,
            direction=self.config.direction,
            sampler=sampler,
            pruner=pruner,
            storage=f"sqlite:///{storage_path}" if storage_path else None,
            load_if_exists=True
        )

    def _get_sampler(self):
        if self.config.sampler_type == "tpe":
            return TPESampler(seed=42)
        elif self.config.sampler_type == "random":
            return optuna.samplers.RandomSampler(seed=42)
        else:
            return optuna.samplers.GridSampler()

    def optimize(self, objective: Callable):
        logger.info(
            f"Starting optimization: {self.config.n_trials} trials, "
            f"direction={self.config.direction}"
        )

        self.study.optimize(
            objective,
            n_trials=self.config.n_trials,
            timeout=self.config.timeout,
            show_progress_bar=True
        )

        self.best_params = self.study.best_params
        self.best_value = self.study.best_value

        logger.info(f"Best value: {self.best_value}")
        logger.info(f"Best params: {self.best_params}")

    def suggest_hyperparameters(self, trial: optuna.Trial, param_space: Dict[str, Any]) -> Dict[str, Any]:
        params = {}

        for param_name, param_config in param_space.items():
            param_type = param_config['type']

            if param_type == 'float':
                if param_config.get('log', False):
                    params[param_name] = trial.suggest_float(
                        param_name,
                        param_config['low'],
                        param_config['high'],
                        log=True
                    )
                else:
                    params[param_name] = trial.suggest_float(
                        param_name,
                        param_config['low'],
                        param_config['high']
                    )

            elif param_type == 'int':
                if param_config.get('log', False):
                    params[param_name] = trial.suggest_int(
                        param_name,
                        param_config['low'],
                        param_config['high'],
                        log=True
                    )
                else:
                    params[param_name] = trial.suggest_int(
                        param_name,
                        param_config['low'],
                        param_config['high']
                    )

            elif param_type == 'categorical':
                params[param_name] = trial.suggest_categorical(
                    param_name,
                    param_config['choices']
                )

            elif param_type == 'discrete_uniform':
                params[param_name] = trial.suggest_discrete_uniform(
                    param_name,
                    param_config['low'],
                    param_config['high'],
                    param_config['q']
                )

        return params

    def get_best_trial(self) -> optuna.trial.FrozenTrial:
        return self.study.best_trial

    def get_trials_dataframe(self):
        return self.study.trials_dataframe()

    def plot_optimization_history(self, save_path: Optional[str] = None):
        fig = vis.plot_optimization_history(self.study)
        if save_path:
            fig.write_html(save_path)
        return fig

    def plot_param_importances(self, save_path: Optional[str] = None):
        fig = vis.plot_param_importances(self.study)
        if save_path:
            fig.write_html(save_path)
        return fig

    def plot_parallel_coordinate(self, save_path: Optional[str] = None):
        fig = vis.plot_parallel_coordinate(self.study)
        if save_path:
            fig.write_html(save_path)
        return fig

    def save_study(self, filepath: str):
        joblib.dump(self.study, filepath)
        logger.info(f"Study saved to {filepath}")

    def load_study(self, filepath: str):
        self.study = joblib.load(filepath)
        self.best_params = self.study.best_params
        self.best_value = self.study.best_value
        logger.info(f"Study loaded from {filepath}")


def create_rf_param_space() -> Dict[str, Any]:
    return {
        'n_estimators': {
            'type': 'int',
            'low': 10,
            'high': 500
        },
        'max_depth': {
            'type': 'int',
            'low': 5,
            'high': 50
        },
        'min_samples_split': {
            'type': 'int',
            'low': 2,
            'high': 20
        },
        'min_samples_leaf': {
            'type': 'int',
            'low': 1,
            'high': 10
        },
        'max_features': {
            'type': 'categorical',
            'choices': ['sqrt', 'log2', None]
        }
    }


def create_gb_param_space() -> Dict[str, Any]:
    return {
        'n_estimators': {
            'type': 'int',
            'low': 50,
            'high': 500
        },
        'learning_rate': {
            'type': 'float',
            'low': 0.01,
            'high': 0.3,
            'log': True
        },
        'max_depth': {
            'type': 'int',
            'low': 3,
            'high': 20
        },
        'subsample': {
            'type': 'float',
            'low': 0.5,
            'high': 1.0
        }
    }


def main():
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.datasets import make_classification
    from sklearn.model_selection import cross_val_score
    from sklearn.model_selection import train_test_split

    # Create sample data
    X, y = make_classification(
        n_samples=1000,
        n_features=20,
        n_informative=15,
        n_classes=2,
        random_state=42
    )

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

    # Define objective function
    def objective(trial):
        tuner = HyperparameterTuner(TuningConfig())
        params = tuner.suggest_hyperparameters(trial, create_rf_param_space())

        model = RandomForestClassifier(**params, random_state=42)

        scores = cross_val_score(model, X_train, y_train, cv=5, scoring='accuracy')
        return scores.mean()

    # Run optimization
    config = TuningConfig(n_trials=50, study_name="rf_optimization")
    tuner = HyperparameterTuner(config)
    tuner.create_study(objective)
    tuner.optimize(objective)

    print(f"\nBest parameters: {tuner.best_params}")
    print(f"Best accuracy: {tuner.best_value:.2%}")

    # Visualizations
    tuner.plot_optimization_history("optimization_history.html")
    tuner.plot_param_importances("param_importances.html")

    # Train final model with best params
    best_model = RandomForestClassifier(**tuner.best_params, random_state=42)
    best_model.fit(X_train, y_train)

    test_accuracy = best_model.score(X_test, y_test)
    print(f"Test accuracy: {test_accuracy:.2%}")


if __name__ == "__main__":
    main()

Related skills

Microsoft FoundryDeploy, evaluate, and continuously improve Microsoft Foundry agents from a single agent interface.513k1.4k

Ai Research ReproductionOrchestrate trustworthy, auditable reproduction of deep learning repositories directly from their READMEs.199k511

Run TrainSafely execute selected deep learning training commands with standardized evidence capture.199k511

Explore RunSafely run isolated exploratory experiments with clear recording and conservative selection before committing changes.199k511

Paper Context ResolverFetch precise reproduction-critical details like dataset splits, preprocessing steps, or evaluation protocols from the original academic paper when the repo README leav141k511

Repo Intake And PlanScan unfamiliar AI research repositories and receive a minimal, trustworthy reproduction target before investing significant time.140k511

Data Science & MLllmautomation

About

Ml Engineer by the numbers

Add your badge

What it does

Files

Machine Learning Engineer

Purpose

When to Use

2. Decision Framework

Model Serving Strategy

Training Infrastructure

Feature Store Decision

3. Core Workflows

Workflow 1: End-to-End Training Pipeline

Workflow 3: Drift Detection (Monitoring)

Workflow 5: RAG Pipeline with Vector Database

5. Anti-Patterns & Gotchas

❌ Anti-Pattern 1: Training-Serving Skew

❌ Anti-Pattern 2: Manual Deployments

❌ Anti-Pattern 3: Silent Failures

7. Quality Checklist

Anti-Patterns

Training-Serving Skew

Manual Deployment

Silent Failures

Data Leakage

Scikit-Learn Guide

Overview

Installation

Quick Start

Basic Workflow

Model Selection Guide

Classification Models

Regression Models

Preprocessing Pipeline

Complete Pipeline

Feature Scaling

Encoding Categorical Variables

Cross-Validation

K-Fold Cross-Validation

Stratified K-Fold (for classification)

Grid Search CV

Feature Engineering

Creating New Features

Feature Selection

Model Evaluation

Classification Metrics

Regression Metrics

Model Persistence

Save and Load Models

Best Practices

Common Issues

Overfitting

Underfitting

Class Imbalance

Resources

Related skills

This week in AI coding