Пишешь код вне ноутбука

Написание кода вне Jupyter Notebook: production-grade development

Почему обязательно писать вне ноутбука?

1. Проблемы с Jupyter Notebooks

Хотя Jupyter отлично подходит для исследований и прототипирования, он имеет серьёзные недостатки для production кода:

# ❌ Jupyter-проблема 1: Hidden state (скрытое состояние)
# Ячейки могут быть выполнены в произвольном порядке
model = train_model()  # Ячейка 5
model = load_pretrained_model()  # Ячейка 3 (выполнили позже)
# Теперь переменная model указывает на не то, что ожидали!

# ❌ Jupyter-проблема 2: Сложно версионировать
# .ipynb — это JSON, в котором смешаны код, выход, метаданные
# Git diff почти нечитаемо
# Merge конфликты адские

# ❌ Jupyter-проблема 3: Нет модульности
# Код растёт в одном файле
# Невозможно переиспользовать функции в других проектах
# Нет package structure

# ❌ Jupyter-проблема 4: Сложно тестировать
# В ноутбуке нет стандартного способа писать юнит-тесты
# Трудно CI/CD интеграция

Правильный workflow: структура проекта

ml-project/
├── src/
│   ├── __init__.py
│   ├── config.py              # Конфигурация
│   ├── data/
│   │   ├── __init__.py
│   │   ├── loaders.py         # Загрузка данных
│   │   └── preprocessing.py   # Предобработка
│   ├── models/
│   │   ├── __init__.py
│   │   ├── base.py            # Базовый класс
│   │   ├── neural_net.py      # NN модели
│   │   └── boosting.py        # Gradient Boosting
│   ├── training/
│   │   ├── __init__.py
│   │   ├── trainer.py         # Логика обучения
│   │   └── callbacks.py       # Callbacks (EarlyStopping и т.д.)
│   ├── evaluation/
│   │   ├── __init__.py
│   │   └── metrics.py         # Метрики и оценка
│   └── utils/
│       ├── __init__.py
│       ├── helpers.py         # Утилиты
│       └── logging.py         # Логирование
├── notebooks/
│   ├── 01_eda.ipynb           # Exploratory Data Analysis
│   ├── 02_feature_engineering.ipynb
│   └── 03_results.ipynb
├── tests/
│   ├── __init__.py
│   ├── test_data_loaders.py
│   ├── test_models.py
│   ├── test_training.py
│   └── test_metrics.py
├── config/
│   ├── default_config.yaml
│   └── experiments/
│       ├── exp_001.yaml
│       └── exp_002.yaml
├── scripts/
│   ├── train.py              # Основной скрипт обучения
│   ├── evaluate.py           # Оценка модели
│   ├── predict.py            # Инференс
│   └── hyperparameter_tuning.py
├── requirements.txt
├── setup.py
├── Makefile
└── README.md

Пример: модульная структура

src/data/loaders.py

from pathlib import Path
import pandas as pd
from typing import Tuple

class DataLoader:
    """Загрузка и базовая обработка данных"""
    
    def __init__(self, data_path: Path):
        self.data_path = Path(data_path)
    
    def load_csv(self, filename: str) -> pd.DataFrame:
        """Загружает CSV файл"""
        file = self.data_path / filename
        if not file.exists():
            raise FileNotFoundError(f"{file} не найден")
        return pd.read_csv(file)
    
    def load_train_test(self) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """Загружает train и test наборы"""
        train = self.load_csv('train.csv')
        test = self.load_csv('test.csv')
        return train, test

src/models/base.py

from abc import ABC, abstractmethod
import joblib
from pathlib import Path

class BaseModel(ABC):
    """Базовый класс для всех моделей"""
    
    def __init__(self, name: str):
        self.name = name
        self.model = None
    
    @abstractmethod
    def train(self, X, y, **kwargs):
        """Обучение модели"""
        pass
    
    @abstractmethod
    def predict(self, X):
        """Предсказание"""
        pass
    
    def save(self, path: Path):
        """Сохраняет модель"""
        joblib.dump(self.model, path)
    
    def load(self, path: Path):
        """Загружает модель"""
        self.model = joblib.load(path)

src/models/neural_net.py

import torch
import torch.nn as nn
from .base import BaseModel

class NeuralNetModel(BaseModel):
    """Нейронная сеть на PyTorch"""
    
    def __init__(self, input_dim: int, hidden_dim: int = 128):
        super().__init__(name="neural_net")
        self.model = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(hidden_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )
        self.criterion = nn.MSELoss()
        self.optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)
    
    def train(self, X, y, epochs=100):
        """Обучение модели"""
        X_tensor = torch.FloatTensor(X)
        y_tensor = torch.FloatTensor(y).unsqueeze(1)
        
        for epoch in range(epochs):
            self.optimizer.zero_grad()
            output = self.model(X_tensor)
            loss = self.criterion(output, y_tensor)
            loss.backward()
            self.optimizer.step()
            
            if (epoch + 1) % 10 == 0:
                print(f"Epoch {epoch+1}/{epochs}, Loss: {loss.item():.4f}")
    
    def predict(self, X):
        """Предсказание"""
        X_tensor = torch.FloatTensor(X)
        self.model.eval()
        with torch.no_grad():
            return self.model(X_tensor).numpy()

src/training/trainer.py

from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, r2_score
import logging

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

class ModelTrainer:
    """Управление обучением модели"""
    
    def __init__(self, model, config: dict):
        self.model = model
        self.config = config
    
    def train_and_evaluate(self, X, y, test_size=0.2):
        """Обучает модель и оценивает её"""
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=test_size, random_state=42
        )
        
        logger.info(f"Обучение {self.model.name}...")
        self.model.train(X_train, y_train, epochs=self.config['epochs'])
        
        logger.info("Оценка модели...")
        predictions = self.model.predict(X_test)
        mse = mean_squared_error(y_test, predictions)
        r2 = r2_score(y_test, predictions)
        
        logger.info(f"MSE: {mse:.4f}")
        logger.info(f"R2: {r2:.4f}")
        
        return self.model, {'mse': mse, 'r2': r2}

scripts/train.py — главный скрипт

import argparse
from pathlib import Path
import yaml

from src.data.loaders import DataLoader
from src.models.neural_net import NeuralNetModel
from src.training.trainer import ModelTrainer

def main(config_path: str):
    # Загружаем конфигурацию
    with open(config_path) as f:
        config = yaml.safe_load(f)
    
    # Загружаем данные
    loader = DataLoader(config['data_path'])
    X, y = loader.load_data()
    
    # Создаём и обучаем модель
    model = NeuralNetModel(
        input_dim=X.shape[1],
        hidden_dim=config['model']['hidden_dim']
    )
    
    trainer = ModelTrainer(model, config['training'])
    trained_model, metrics = trainer.train_and_evaluate(X, y)
    
    # Сохраняем модель
    model.save(Path(config['output_path']) / 'model.pkl')
    print(f"Модель сохранена: {metrics}")

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--config', default='config/default_config.yaml')
    args = parser.parse_args()
    
    main(args.config)

tests/test_models.py — юнит-тесты

import pytest
import numpy as np
from src.models.neural_net import NeuralNetModel

def test_model_initialization():
    """Тест инициализации модели"""
    model = NeuralNetModel(input_dim=10)
    assert model.model is not None

def test_model_output_shape():
    """Тест формы выхода"""
    model = NeuralNetModel(input_dim=10)
    X = np.random.randn(5, 10)
    output = model.predict(X)
    assert output.shape == (5, 1)

def test_model_training():
    """Тест обучения"""
    model = NeuralNetModel(input_dim=5)
    X = np.random.randn(10, 5)
    y = np.random.randn(10)
    model.train(X, y, epochs=5)
    # Если обучение не выкинуло exception — OK
    assert True

Makefile для удобства

.PHONY: install test train clean

install:
	pip install -r requirements.txt

test:
	pytest tests/ -v --cov=src

lint:
	flake8 src/ tests/
	typye src/

train:
	python scripts/train.py --config config/default_config.yaml

clean:
	find . -type d -name __pycache__ -exec rm -rf {} +
	find . -type f -name "*.pyc" -delete

Workflow: от ноутбука к production

Шаг 1: Исследование в Jupyter

# notebooks/01_eda.ipynb
df = pd.read_csv('data/raw/train.csv')
df.describe()
df.plot(kind='hist')
# ... визуализация, анализ

Шаг 2: Извлекаем код в модули

# src/data/preprocessing.py — вынимаем логику из ноутбука
def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    df = df.dropna()
    df['age'] = df['age'].fillna(df['age'].median())
    return df

Шаг 3: Пишем тесты

# tests/test_preprocessing.py
def test_clean_data():
    df = pd.DataFrame({'age': [25, None, 30]})
    result = clean_data(df)
    assert len(result) == 3  # NaN заполнены

Шаг 4: Интегрируем в pipeline

# scripts/train.py
df = loader.load_csv('train.csv')
df = preprocess(df)  # Используем функцию из src/
model.train(df)

Best Practices

Инструменты

# Poetry для зависимостей (лучше, чем pip)
poetry add scikit-learn pandas

# pytest для тестов
pytest tests/ -v

# Black для форматирования
black src/

# pylint/flake8 для linting
flake8 src/

# mypy для type checking
mypy src/

# MLflow для отслеживания экспериментов
mlflow run .

Вывод

Написание кода вне ноутбука — это не просто best practice, это необходимость для production ML систем. Это обеспечивает:

• Воспроизводимость результатов
• Версионирование и контроль качества
• Тестируемость и надёжность
• Модульность и переиспользование
• Удобную интеграцию в CI/CD
• Масштабируемость систем

От этого напрямую зависит успех проекта и качество production системы.