machine-learning

name	machine-learning
description	Machine learning best practices and patterns
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"domain-specific"}

What I do

• Design and implement ML pipelines
• Handle data preprocessing and feature engineering
• Select appropriate models for the problem
• Evaluate models with proper metrics
• Prevent overfitting and ensure generalization
• Deploy models to production
• Monitor model performance (MLOps)
• Implement hyperparameter tuning

When to use me

When implementing machine learning solutions or ML pipelines.

ML Pipeline Structure

ml/
├── data/
│   ├── raw/              # Raw data files
│   ├── processed/        # Processed features
│   └── features/        # Feature definitions
├── models/              # Saved models
├── notebooks/           # Jupyter notebooks
├── src/
│   ├── data/
│   │   ├── loaders.py
│   │   ├── preprocessing.py
│   │   └── feature_engineering.py
│   ├── models/
│   │   ├── base.py
│   │   ├── classifier.py
│   │   └── regressor.py
│   ├── training/
│   │   ├── trainer.py
│   │   ├── evaluation.py
│   │   └── hyperparameter_tuning.py
│   └── serving/
│       ├── inference.py
│       └── deployment.py
├── tests/
│   ├── test_data.py
│   ├── test_models.py
│   └── test_training.py
├── configs/
│   ├── model_config.yaml
│   └── training_config.yaml
└── requirements-ml.txt

Data Loading and Preprocessing

import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from typing import Tuple, List


class DataLoader:
    """Handle data loading and initial preprocessing."""
    
    def __init__(self, data_path: str) -> None:
        self.data_path = data_path
    
    def load_data(self) -> pd.DataFrame:
        """Load data from file."""
        if self.data_path.endswith('.csv'):
            return pd.read_csv(self.data_path)
        elif self.data_path.endswith('.parquet'):
            return pd.read_parquet(self.data_path)
        else:
            raise ValueError(f"Unsupported file format: {self.data_path}")
    
    def clean_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """Handle missing values and outliers."""
        df = df.copy()
        
        
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        df[numeric_cols] = df[numeric_cols].fillna(df[numeric_cols].median())
        
        categorical_cols = df.select_dtypes(include=[]).columns
         col  categorical_cols:
            df[col] = df[col].fillna(df[col].mode()[])
        
        
         col  numeric_cols:
            Q1 = df[col].quantile()
            Q3 = df[col].quantile()
            IQR = Q3 - Q1
            df[col] = df[col].clip(Q1 -  * IQR, Q3 +  * IQR)
        
         df
    
     () -> [pd.DataFrame, ...]:
        
        X = df.drop(columns=[target_col])
        y = df[target_col]
        
        X_train, X_test, y_train, y_test = train_test_split(
            X, y,
            test_size=test_size,
            stratify=y,
            random_state=random_state,
        )
        
        X_train, X_val, y_train, y_val = train_test_split(
            X_train, y_train,
            test_size=val_size / ( - test_size),
            stratify=y_train,
            random_state=random_state,
        )
        
         X_train, X_val, X_test, y_train, y_val, y_test

Feature Engineering

from sklearn.base import BaseEstimator, TransformerMixin


class FeatureEngineer(BaseEstimator, TransformerMixin):
    """Custom feature engineering pipeline."""
    
    def __init__(self) -> None:
        self.numeric_features = None
        self.categorical_features = None
        self.label_encoders = {}
    
    def fit(self, X: pd.DataFrame, y=None) -> 'FeatureEngineer':
        """Learn feature-specific transformations."""
        self.numeric_features = X.select_dtypes(
            include=['int64', 'float64']
        ).columns.tolist()
        
        self.categorical_features = X.select_dtypes(
            include=['object', 'category']
        ).columns.tolist()
        
        for col in self.categorical_features:
            le = LabelEncoder()
            le.fit(X[col].astype(str))
            self.label_encoders[col] = le
        
        return self
    
    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
        """Apply feature transformations."""
        X = X.copy()
        
        # Encode categorical variables
        for col in self.categorical_features:
            X[col] = X[col].astype()
            X[col] = .label_encoders[col].transform(X[col])
        
        
        X = ._create_interactions(X)
        
        
        X = ._create_polynomial_features(X)
        
         X
    
     () -> pd.DataFrame:
        
         (.numeric_features) >= :
            top_features = .numeric_features[:]
            
             i, f1  (top_features):
                 f2  top_features[i+:]:
                    X[] = X[f1] * X[f2]
        
         X
    
     () -> pd.DataFrame:
        
         col  .numeric_features[:]:
            X[] = X[col] ** 
        
         X

Model Training

import mlflow
import mlflow.sklearn
from sklearn.model_selection import cross_val_score
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score,
    f1_score, roc_auc_score, confusion_matrix
)
from typing import Dict, Any


class ModelTrainer:
    """Train and evaluate ML models with MLflow tracking."""
    
    def __init__(
        self,
        experiment_name: str,
        model_type: str = 'classifier'
    ) -> None:
        self.experiment_name = experiment_name
        self.model_type = model_type
        self.best_model = None
        self.best_score = 0
    
    def train(
        self,
        model,
        X_train: pd.DataFrame,
        y_train: pd.Series,
        X_val: pd.DataFrame,
        y_val: pd.Series,
        params: Dict[str, Any]
    ) -> Dict[str, float]:
        """
        Train model with logging to MLflow.
        """
        mlflow.set_experiment(self.experiment_name)
        
        with mlflow.start_run():
            # Log parameters
            mlflow.log_params(params)
            
            # Train
            model.fit(X_train, y_train)
            
            # Evaluate
            train_metrics = .evaluate(model, X_train, y_train, )
            val_metrics = .evaluate(model, X_val, y_val, )
            
            
             metric, value  train_metrics.items():
                mlflow.log_metric(, value)
             metric, value  val_metrics.items():
                mlflow.log_metric(, value)
            
            
            mlflow.sklearn.log_model(model, )
            
            
             val_metrics[] > .best_score:
                .best_score = val_metrics[]
                .best_model = model
            
             val_metrics
    
     () -> [, ]:
        
        y_pred = model.predict(X)
        y_proba = model.predict_proba(X)[:, ]  (model, )  
        
        metrics = {
            : accuracy_score(y, y_pred),
            : precision_score(y, y_pred, average=),
            : recall_score(y, y_pred, average=),
            : f1_score(y, y_pred, average=),
        }
        
         y_proba   :
            metrics[] = roc_auc_score(y, y_proba)
        
         metrics
    
     () -> [, ]:
        
        scores = cross_val_score(model, X, y, cv=cv, scoring=)
        
         {
            : scores.mean(),
            : scores.std(),
        }

Hyperparameter Tuning

from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
from scipy.stats import uniform, randint


def tune_hyperparameters(
    model,
    param_distributions: Dict[str, Any],
    X: pd.DataFrame,
    y: pd.Series,
    n_iter: int = 50,
    cv: int = 5,
    scoring: str = 'f1_weighted'
) -> GridSearchCV:
    """
    Random search for hyperparameter optimization.
    """
    search = RandomizedSearchCV(
        model,
        param_distributions=param_distributions,
        n_iter=n_iter,
        cv=cv,
        scoring=scoring,
        random_state=42,
        n_jobs=-1,
        verbose=1,
    )
    
    search.fit(X, y)
    
    print(f"Best score: {search.best_score_:.4f}")
    print(f"Best params: {search.best_params_}")
    
    return search.best_estimator_, search.best_params_, search.best_score_


# Example parameter distributions
param_distributions = {
    'n_estimators': randint(100, 500),
    'max_depth': randint(3, 15),
    'min_samples_split': randint(2, 20),
    'min_samples_leaf': randint(1, 10),
    'learning_rate': uniform(0.01, 0.3),
    : uniform(, ),
    : uniform(, ),
}

Model Serving

import joblib
import json
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List


app = FastAPI(title='ML Model API')


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


class PredictionResponse(BaseModel):
    prediction: int
    probability: float
    confidence: float


# Load model at startup
model = joblib.load('models/best_model.pkl')
scaler = joblib.load('models/scaler.pkl')


@app.post('/predict', response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    """Make a single prediction."""
    try:
        features = np.array(request.features).reshape(1, -1)
        features = scaler.transform(features)
        
        prediction = int(model.predict(features)[0])
        probability = float(model.predict_proba(features)[0][prediction])
        
        confidence = float(max(model.predict_proba(features)[0]))
        
        return PredictionResponse(
            prediction=prediction,
            probability=probability,
            confidence=confidence,
        )
     Exception  e:
         HTTPException(status_code=, detail=(e))



  ():
    
     {: }