Complete MLOps Blueprint

ml-project/
├── README.md
├── requirements.txt
├── Dockerfile
├── .gitignore
├── .dvcignore
├── dvc.yaml                    # DVC pipeline definition
├── params.yaml                 # Hyperparameters and config
├── data/
│   ├── raw/                   # Original, immutable data
│   ├── processed/             # Cleaned and processed data
│   └── external/              # Third-party datasets
├── notebooks/
│   ├── 01-exploratory-analysis.ipynb
│   ├── 02-feature-engineering.ipynb
│   └── 03-model-experiments.ipynb
├── src/
│   ├── __init__.py
│   ├── data/
│   │   ├── __init__.py
│   │   ├── extract.py         # Data extraction
│   │   ├── transform.py       # Data transformation
│   │   └── validate.py        # Data validation
│   ├── features/
│   │   ├── __init__.py
│   │   └── build_features.py  # Feature engineering
│   ├── models/
│   │   ├── __init__.py
│   │   ├── train.py          # Model training
│   │   ├── predict.py        # Model prediction
│   │   └── evaluate.py       # Model evaluation
│   └── utils/
│       ├── __init__.py
│       └── helpers.py        # Utility functions
├── tests/
│   ├── __init__.py
│   ├── test_data.py
│   ├── test_features.py
│   └── test_models.py
├── models/                    # Trained model artifacts
├── reports/                   # Analysis reports and figures
├── deployment/
│   ├── docker/
│   ├── kubernetes/
│   └── terraform/
└── mlflow_tracking/          # MLflow tracking server data

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
import pandas as pd
import joblib
from datetime import datetime
import argparse

class ModelTrainer:
    def __init__(self, config):
        self.config = config
        self.model = None
        self.X_train = None
        self.X_test = None
        self.y_train = None
        self.y_test = None
    
    def load_data(self):
        """Load and split data"""
        data_path = self.config.get('data.processed_data_path')
        df = pd.read_csv(data_path)
        
        feature_cols = (
            self.config.get('features.categorical_features', []) + 
            self.config.get('features.numerical_features', [])
        )
        target_col = self.config.get('features.target')
        
        X = df[feature_cols]
        y = df[target_col]
        
        self.X_train, self.X_test, self.y_train, self.y_test = train_test_split(
            X, y, 
            test_size=self.config.get('data.test_size', 0.2),
            random_state=self.config.get('data.random_state', 42),
            stratify=y
        )
    
    def train_model(self):
        """Train the model"""
        model_params = self.config.get('model')
        
        if model_params['algorithm'] == 'random_forest':
            self.model = RandomForestClassifier(
                n_estimators=model_params['n_estimators'],
                max_depth=model_params['max_depth'],
                min_samples_split=model_params['min_samples_split'],
                random_state=model_params['random_state']
            )
        
        self.model.fit(self.X_train, self.y_train)
    
    def evaluate_model(self):
        """Evaluate model performance"""
        y_pred = self.model.predict(self.X_test)
        
        metrics = {
            'accuracy': accuracy_score(self.y_test, y_pred),
            'precision': precision_score(self.y_test, y_pred, average='weighted'),
            'recall': recall_score(self.y_test, y_pred, average='weighted'),
            'f1': f1_score(self.y_test, y_pred, average='weighted')
        }
        
        # Cross-validation
        cv_scores = cross_val_score(
            self.model, self.X_train, self.y_train, 
            cv=self.config.get('training.cross_validation_folds', 5)
        )
        metrics['cv_accuracy_mean'] = cv_scores.mean()
        metrics['cv_accuracy_std'] = cv_scores.std()
        
        return metrics
    
    def save_model(self, model_path):
        """Save trained model"""
        joblib.dump(self.model, model_path)
    
    def run_training(self):
        """Execute complete training pipeline"""
        mlflow.set_experiment(self.config.get('mlflow.experiment_name'))
        
        with mlflow.start_run():
            # Log parameters
            mlflow.log_params(self.config.get('model'))
            mlflow.log_params(self.config.get('data'))
            
            # Load data and train
            self.load_data()
            self.train_model()
            
            # Evaluate
            metrics = self.evaluate_model()
            mlflow.log_metrics(metrics)
            
            # Save model
            model_path = f"models/model_{datetime.now().strftime('%Y%m%d_%H%M%S')}.pkl"
            self.save_model(model_path)
            mlflow.log_artifact(model_path)
            
            # Log model to MLflow Model Registry
            mlflow.sklearn.log_model(
                self.model, 
                "model",
                registered_model_name=self.config.get('mlflow.experiment_name')
            )
            
            print(f"Training completed. Metrics: {metrics}")
            return metrics

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", default="params.yaml", help="Configuration file path")
    args = parser.parse_args()
    
    config = Config(args.config)
    trainer = ModelTrainer(config)
    trainer.run_training()

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import pandas as pd
import numpy as np
from typing import List, Dict
import logging
import time
import os

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

model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: load model from registry / object store — NOT from embedded file
    global model
    try:
        model_uri = os.environ.get("MODEL_URI", "models/production_model.pkl")
        model = joblib.load(model_uri)
        logger.info(f"Model loaded successfully from {model_uri}")
    except Exception as e:
        logger.error(f"Failed to load model: {e}")
        raise
    yield
    # Shutdown: release resources
    model = None
    logger.info("Model unloaded")

# Initialize FastAPI app
app = FastAPI(
    title="ML Model API",
    description="Production ML model serving API",
    version="1.0.0",
    lifespan=lifespan
)

class PredictionRequest(BaseModel):
    features: Dict[str, float]   # keys must match training column names
    
class PredictionResponse(BaseModel):
    prediction: int
    probability: List[float]
    model_version: str
    processing_time: float

class HealthResponse(BaseModel):
    status: str
    model_loaded: bool
    timestamp: str

# Health check endpoint
@app.get("/health", response_model=HealthResponse)
async def health_check():
    return HealthResponse(
        status="healthy" if model is not None else "unhealthy",
        model_loaded=model is not None,
        timestamp=pd.Timestamp.now().isoformat()
    )

# Prediction endpoint
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    if model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    
    start_time = time.time()
    
    try:
        # Convert input to DataFrame
        input_data = pd.DataFrame([request.features])
        
        # Make prediction
        prediction = model.predict(input_data)[0]
        probability = model.predict_proba(input_data)[0].tolist()
        
        processing_time = time.time() - start_time
        
        # Log prediction
        logger.info(f"Prediction made: {prediction}, Processing time: {processing_time:.3f}s")
        
        return PredictionResponse(
            prediction=int(prediction),
            probability=probability,
            model_version="1.0.0",
            processing_time=processing_time
        )
        
    except Exception as e:
        logger.error(f"Prediction failed: {e}")
        raise HTTPException(status_code=400, detail=f"Prediction failed: {str(e)}")

# Batch prediction endpoint
@app.post("/predict/batch")
async def predict_batch(requests: List[PredictionRequest]):
    if model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    
    try:
        # Convert all inputs to DataFrame
        features_list = [req.features for req in requests]
        input_data = pd.DataFrame(features_list)
        
        # Make predictions
        predictions = model.predict(input_data)
        probabilities = model.predict_proba(input_data)
        
        # Format responses
        responses = []
        for i, (pred, prob) in enumerate(zip(predictions, probabilities)):
            responses.append({
                "id": i,
                "prediction": int(pred),
                "probability": prob.tolist()
            })
        
        return responses
        
    except Exception as e:
        logger.error(f"Batch prediction failed: {e}")
        raise HTTPException(status_code=400, detail=f"Batch prediction failed: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# .github/workflows/ml-pipeline.yml
name: ML Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
        - uses: actions/checkout@v4
    
    - name: Set up Python
      uses: actions/setup-python@v5         with:
        python-version: "3.11"
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest pytest-cov flake8
    
    - name: Lint with flake8
      run: |
        flake8 src/ --count --select=E9,F63,F7,F82 --show-source --statistics
        flake8 src/ --count --max-complexity=10 --max-line-length=100 --statistics
    
    - name: Run tests
      run: |
        pytest tests/ --cov=src/ --cov-report=xml
    
    - name: Upload coverage to Codecov
      uses: codecov/codecov-action@v4   
  train:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
    - uses: actions/checkout@v4
    
    - name: Set up Python
      uses: actions/setup-python@v5
      with:
        python-version: "3.11"
    
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
    
    - name: Setup DVC
      uses: iterative/setup-dvc@v2       
    - name: Pull data with DVC
      run: dvc pull
      env:
        AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
        AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
    
    - name: Train model
      run: python src/models/train.py
      env:
        MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
    
    - name: Run model tests
      run: pytest tests/test_models.py

  deploy:
    needs: [test, train]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
    - uses: actions/checkout@v4
    
    - name: Build Docker image
      run: |
        docker build -t ml-model-api:${{ github.sha }} .
        docker tag ml-model-api:${{ github.sha }} ml-model-api:latest
    
    - name: Scan image for vulnerabilities
      run: |
                docker run --rm -v /var/run/docker.sock:/var/run/docker.sock \
          aquasec/trivy:latest image --exit-code 1 --severity CRITICAL \
          ml-model-api:${{ github.sha }}
    
    - name: Deploy to staging
      run: |
        # Deploy to staging environment
        kubectl apply -f deployment/kubernetes/staging/
    
    - name: Run integration tests
      run: |
        # Run integration tests against staging
        pytest tests/integration/
    
    - name: Deploy to production
      if: success()
      run: |
        # Deploy to production environment
        kubectl apply -f deployment/kubernetes/production/

# monitoring/prometheus.yml
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'ml-model-api'
    static_configs:
      - targets: ['ml-model-api-service:80']
    metrics_path: '/metrics'
    scrape_interval: 30s

# monitoring/grafana-dashboard.json
{
  "dashboard": {
    "title": "ML Model Monitoring",
    "panels": [
      {
        "title": "Prediction Rate",
        "type": "timeseries",
        "datasource": { "type": "prometheus" },
        "targets": [
          {
            "expr": "rate(ml_predictions_total[5m])",
            "legendFormat": "Predictions/sec"
          }
        ]
      },
      {
        "title": "Prediction Latency",
        "type": "timeseries",
        "datasource": { "type": "prometheus" },
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(ml_prediction_duration_seconds_bucket[5m]))",
            "legendFormat": "p95 latency"
          },
          {
            "expr": "histogram_quantile(0.50, rate(ml_prediction_duration_seconds_bucket[5m]))",
            "legendFormat": "p50 (median)"
          }
        ]
      },
      {
        "title": "Model Accuracy",
        "type": "stat",
        "datasource": { "type": "prometheus" },
        "targets": [
          {
            "expr": "ml_model_accuracy",
            "legendFormat": "Current Accuracy"
          }
        ]
      },
      {
        "title": "Data Drift Score",
        "type": "timeseries",
        "datasource": { "type": "prometheus" },
        "targets": [
          {
            "expr": "ml_data_drift_score",
            "legendFormat": "Drift Score"
          }
        ]
      }
    ]
  }
}