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

А, ну вот, паттерны в машинке, ёпта. Это ж как в гараже — можно всё в кучу свалить, а можно инструмент по полочкам разложить, чтобы потом не искать полдня. Я тебе про те, что реально выручали, когда проект уже такой, что волнение ебать, а терпения ноль ебать.

Первый, бля, Стратегия (Strategy) — когда надо менять алгоритмы как перчатки. Представь, ты фичи кодируешь. Сегодня One-Hot, завтра Target Encoding, а послезавтра какой-нибудь новый метод, который на конфе вычитал. Так вот, вместо того чтобы в коде if-else городить, как манда с ушами, ты делаешь общую штуку.

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)

Суть в чём? Захотел поменять — одну строчку переписал, и всё, бля, работает. Чисто, аккуратно, и доверия ебать больше, чем к коду, где всё в одну функцию засунуто.

Второй — Фабрика (Factory). Это когда пайплайны собирать надо, а они как конструктор. Бывает же, экспериментируешь: сегодня Random Forest, завтра логистическая регрессия, а препроцессинг вообще каждый раз новый. Чтобы не писать каждый раз новый скрипт, делаешь фабрику — она тебе по названию модельку и пайплайн собирает.

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)]
)

Вот смотри, удобно же! Всё в одном месте описано, не надо по всему коду искать, где там модель инициализируется. Просто сказал «дай мне лес с масштабированием и PCA» — и получил, ядрёна вошь.

Третий — Наблюдатель (Observer). Это для трекинга экспериментов, когда метрик овердохуища. Ты же в процессе обучения логируешь всё: loss, accuracy, ещё какую-нибудь дичь. И логировать надо и в MLflow, и в консоль, а может, ещё куда-то. Вот тут наблюдатель выручает. Делаешь одного главного по трекингу, а к нему цепляешь всех, кто должен знать о твоих метриках.

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})

И всё, бля! Добавил нового логгера — прицепил его к трекеру одной строкой. Не надо лезть в сам цикл обучения и там что-то дописывать. Красота, ёперный театр!

Ну и четвёртый — Декоратор (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

Навесил @time_execution на функцию — и теперь видишь, сколько она тормозит. Прям как в машине диагностику сделал. А главное — саму функцию не трогал, она чистая осталась.

Вот эти паттерны, они реально жизнь упрощают. Код становится не такой пиздопроебибна, с ним другие люди работать могут, не проклиная тебя на чём свет стоит. И самому потом, через полгода, разобраться проще, чем в спагетти-коде, где ни хуя себе не понятно.