Какие паттерны проектирования вы применяли в контексте Data Science или ML-инженерии?

В ML-проектах паттерны помогают создавать поддерживаемый, тестируемый и расширяемый код. Вот несколько, которые я использовал на практике:

1. Стратегия (Strategy) — для выбора и сравнения моделей/препроцессинга. Этот паттерн идеален, когда нужно иметь семейство взаимозаменяемых алгоритмов. Я применял его для A/B-тестирования разных методов обработки текста или финальных моделей.

from abc import ABC, abstractmethod

class FeatureEncoder(ABC):
    """Абстрактная стратегия для кодирования категориальных признаков."""
    @abstractmethod
    def encode(self, X_train, X_test=None):
        pass

class OneHotEncoderStrategy(FeatureEncoder):
    def __init__(self, handle_unknown='ignore'):
        self.encoder = OneHotEncoder(handle_unknown=handle_unknown, sparse_output=False)
    def encode(self, X_train, X_test=None):
        self.encoder.fit(X_train)
        X_train_enc = self.encoder.transform(X_train)
        X_test_enc = self.encoder.transform(X_test) if X_test is not None else None
        return X_train_enc, X_test_enc

class TargetEncoderStrategy(FeatureEncoder):
    def __init__(self, smoothing=10):
        self.smoothing = smoothing
    def encode(self, X_train, X_test=None, y_train=None):
        # ... реализация target encoding ...
        pass

# Использование
encoder_strategy = OneHotEncoderStrategy()
# encoder_strategy = TargetEncoderStrategy() # Легкая замена
X_train_processed, X_val_processed = encoder_strategy.encode(X_train_cat, X_val_cat)

2. Фабрика (Factory) — для создания конвейеров или моделей по конфигурации. Полезен при построении сложных пайплайнов, которые могут меняться в зависимости от входных данных или эксперимента.

class PipelineFactory:
    @staticmethod
    def create_pipeline(model_type, preprocessors):
        """Создает sklearn Pipeline."""
        steps = []
        for prep in preprocessors:
            steps.append((f'prep_{prep.__class__.__name__}', prep))

        if model_type == 'random_forest':
            from sklearn.ensemble import RandomForestClassifier
            model = RandomForestClassifier(n_estimators=100)
        elif model_type == 'logreg':
            from sklearn.linear_model import LogisticRegression
            model = LogisticRegression()
        else:
            raise ValueError(f"Unknown model type: {model_type}")

        steps.append((model_type, model))
        return Pipeline(steps)

# Конфигурируемое создание
my_pipeline = PipelineFactory.create_pipeline(
    model_type='random_forest',
    preprocessors=[StandardScaler(), PCA(n_components=10)]
)

3. Наблюдатель (Observer) — для логирования метрик и артефактов в процессе обучения. Я интегрировал его с MLflow или Weights & Biases для отслеживания экспериментов.

class ExperimentTracker:
    """Субъект (Subject), за которым наблюдают."""
    def __init__(self):
        self._observers = []
    def attach(self, observer):
        self._observers.append(observer)
    def notify(self, epoch, metrics):
        for obs in self._observers:
            obs.update(epoch, metrics)

class MLflowLogger:
    """Конкретный наблюдатель."""
    def update(self, epoch, metrics):
        import mlflow
        mlflow.log_metrics(metrics, step=epoch)

# Использование
tracker = ExperimentTracker()
tracker.attach(MLflowLogger())
# В цикле обучения:
for epoch in range(n_epochs):
    loss, accuracy = train_step(...)
    tracker.notify(epoch, {"train_loss": loss, "train_accuracy": accuracy})

4. Декоратор (Decorator) — для добавления сквозной функциональности. Идеален для логирования времени выполнения, кеширования результатов тяжелых вычислений или валидации входных данных.

import time
import functools
import pickle
import hashlib

def time_execution(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} executed in {elapsed:.4f} seconds")
        return result
    return wrapper

@time_execution
def compute_expensive_features(df):
    # Долгие вычисления...
    time.sleep(2)
    return df

Эти паттерны делают код чище и облегчают командную разработку и эксперименты.