Как писать преобразователи данных в Sklearn

Сегодня разбираемся, как создавать собственные преобразователи Sklearn, позволяющие интегрировать практически любую функцию или преобразование данных в классы конвейера Sklearn. Подробности под катом к старту флагманского курса по Data Science.

Зачем?

Только один вызов fit, и один — predict — насколько это было бы здорово? Вы получаете данные, обучаете конвейер единожды, и он заботится о предварительной обработке, инжиниринге признаков, моделировании. Всё, что нужно сделать вам, — это вызвать fit.

Какой же конвейер для этого достаточно мощный? В Sklearn много преобразователей, но не для всех ситуаций предварительной обработки. Итак, наш конвейер — несбыточная мечта, или нет? Точно нет.

Что такое конвейеры Sklearn?

Вот простой конвейер, который заполняет пропущенные значения числами, масштабирует их и обучает XGBRegressor на X, y:

from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer
import xgboost as xgb

xgb_pipe = make_pipeline(
                SimpleImputer(strategy='mean'),
                StandardScaler(),
                xgb.XGBRegressor()
            )

_ = xgb_pipe.fit(X, y)

В этом посте я в мельчайших подробностях рассказал о конвейерах Sklearn и об их преимуществах.

Самое заметное преимущество конвейеров — способность объединять все этапы предварительной обработки и моделирования в единственный оценщик, предотвращать утечку данных и не вызывать fit на наборах данных для валидации. А ещё конвейер — это бонус в виде краткого, воспроизводимого, модульного кода.

Но вся эта идея атомарных, аккуратных конвейеров ломается, как только нужно выполнять операции, которые не встроены в Sklearn как функции оценки, например:

• Извлечь шаблоны регулярных выражений для очистки текстовых данных нужно.

• Объединить существующие функции в одну, исходя из знаний предметной области.

Чтобы сохранить все преимущества конвейеров, нужен способ интегрировать в Sklearn пользовательскую предварительную обработку и логику инжиниринга признаков. Здесь и вступают в игру пользовательские преобразователи.

Интеграция простых функций через FunctionTransformer

В сентябрьском конкурсе TPS 2021 на Kaggle одна из идей — добавить количества пропущенных [в сроке данных] значений как новый признак — значительно повысила производительность модели. Эта операция не реализована в Sklearn, поэтому напишем функцию, которая отработает после импорта данных:

tps_df = pd.read_csv("data/train.csv")
tps_df.head()

Источник данных: Kaggle

>>> tps_df.shape
(957919, 120)

>>> # Find the number of missing values across rows
>>> tps_df.isnull().sum(axis=1)
0         1
1         0
2         5
3         2
4         8
         ..
957914    0
957915    4
957916    0
957917    1
957918    4
Length: 957919, dtype: int64

Эта функция принимает DataFrame и реализует операцию:

def num_missing_row(X: pd.DataFrame, y=None):
    # Calculate some metrics across rows
    num_missing = X.isnull().sum(axis=1)
    num_missing_std = X.isnull().std(axis=1)

    # Add the above series as a new feature to the df
    X["#missing"] = num_missing
    X["num_missing_std"] = num_missing_std

    return X

Добавим в функцию в конвейер — передадим её в FunctionTransformer:

from sklearn.preprocessing import FunctionTransformer

num_missing_estimator = FunctionTransformer(num_missing_row)

При передаче пользовательской функции в FunctionTransformer создаётся оценщик с методами fit, transform и fit_transform:

# Check number of columns before
print(f"Number of features before preprocessing: {len(tps_df.columns)}")

# Apply the custom estimator
tps_df = num_missing_estimator.transform(tps_df)
print(f"Number of features after preprocessing: {len(tps_df.columns)}")

------------------------------------------------------

Number of features before preprocessing: 120
Number of features after preprocessing: 122

Итак, у нас есть простая функция, поэтому нет необходимости вызывать fit: она просто возвращает нетронутую оценку. Единственное требование FunctionTransformer состоит в том, что передаваемая функция должна принимать данные в своём первом аргументе. При желании, если в функции нужен целевой массив, можно передать и его:

# FunctionTransformer signature
def custom_function(X, y=None):
    ...

estimator = FunctionTransformer(custom_function)  # no errors

custom_pipeline = make_pipeline(StandardScaler(), estimator, xgb.XGBRegressor())
custom_pipeline.fit(X, y)

FunctionTransformer также принимает инверсию переданной функции на случай, если понадобится отменить изменения:

def custom_function(X, y=None):
    ...

def inverse_of_custom(X, y=None):
    ...

estimator = FunctionTransformer(func=custom_function, inverse_func=inverse_of_custom)

Подробности о других аргументах смотрите в документации.

Интеграция сложных шагов предварительной обработки

Один из самых распространённых вариантов масштабирования искажённых данных — это логарифмическое преобразование. Но если функция содержит хотя бы один 0, преобразование с помощью np.log или PowerTransformer вернёт ошибку.

Для обхода особенности особенности участники соревнований Kaggle ко всем образцам данных добавляют 1, и лишь затем применяют преобразование. Если преобразование выполняется в целевом массиве, то потребуется обратное преобразование, для которого после прогнозирования нужно использовать экспоненциальную функцию и вычесть 1:

y_transformed = np.log(y + 1)

_ = model.fit(X, y_transformed)
preds = np.exp(model.predict(X, y_transformed) - 1)

Работает, но осталась та же проблема — мы не можем включить код в конвейер из коробки. Конечно, можно обратиться к новому другу — FunctionTransformer, но он не подходит для сложных этапов предварительной обработки, таких как этот.

Вместо этого напишем собственный класс преобразователя и создадим функции fit, transform вручную. В конце концов у нас снова будет Sklearn-совместимый оценщик. Начнём:

from sklearn.base import BaseEstimator, TransformerMixin

class CustomLogTransformer(BaseEstimator, TransformerMixin):
    pass

Сначала мы создаём класс, который наследуется от BaseEstimator и TransformerMixin из sklearn.base. Наследование этих классов позволяет конвейерам Sklearn распознавать классы как пользовательские оценщики.

Напишем метод __init__, где инициализируем экземпляр PowerTransformer:

from sklearn.preprocessing import PowerTransformer

class CustomLogTransformer(BaseEstimator, TransformerMixin):
    def __init__(self):
        self._estimator = PowerTransformer()

Напишем fit, где добавляем 1 ко всем признакам в данных и обучаем PowerTransformer:

class CustomLogTransformer(BaseEstimator, TransformerMixin):
    def __init__(self):
        self._estimator = PowerTransformer()

    def fit(self, X, y=None):
        X_copy = np.copy(X) + 1
        self._estimator.fit(X_copy)

        return self

Метод fit должен возвращать сам преобразователь, это делается возвратом self. И проверим то, что мы написали:

custom_log = CustomLogTransformer()

>>> custom_log.fit(tps_df)
CustomLogTransformer()

Пока работает как положено.

У нас есть transform, где после добавления 1 к переданным данным вызывается transform из класса PowerTransformer:

class CustomLogTransformer(BaseEstimator, TransformerMixin):
    def __init__(self):
        self._estimator = PowerTransformer()

    def fit(self, X, y=None):
        X_copy = np.copy(X) + 1
        self._estimator.fit(X_copy)

        return self

    def transform(self, X):
        X_copy = np.copy(X) + 1

        return self._estimator.transform(X_copy)

Проверим его по-другому:

custom_log = CustomLogTransformer()
custom_log.fit(tps_df)

transformed_tps = custom_log.transform(tps_df)

>>> transformed_tps[:5, :5]
array([[ 0.48908946, -2.17126787, -1.79124946, -0.52828469,         nan],
       [ 0.38660665, -0.29384644,  1.31313666,  0.1901713 , -0.34236997],
       [-0.04286469, -0.05047097, -1.16463754,  0.95459266,  1.71830766],
       [-0.584329  , -1.5743182 , -1.02444525, -0.15117546,  0.46952437],
       [-0.87027925, -0.13045462, -0.10489176, -0.36806683,  1.21317668]])

Работает как надо. Как я уже говорил, нам нужен метод отмены преобразования:

class CustomLogTransformer(BaseEstimator, TransformerMixin):
    def __init__(self):
        self._estimator = PowerTransformer()

    def fit(self, X, y=None):
        X_copy = np.copy(X) + 1
        self._estimator.fit(X_copy)

        return self

    def transform(self, X):
        X_copy = np.copy(X) + 1

        return self._estimator.transform(X_copy)

    def inverse_transform(self, X):
        X_reversed = self._estimator.inverse_transform(np.copy(X))

        return X_reversed - 1

Вместо inverse_transform можно было воспользоваться np.exp. Теперь проведём окончательную проверку:

custom_log = CustomLogTransformer()

tps_transformed = custom_log.fit_transform(tps_df)
tps_inversed = custom_log.inverse_transform(tps_transformed)

Но подождите! Мы не писали _fit_transform_ — откуда она взялась?

Это просто — когда вы наследуетесь от _BaseEstimator_ и _TransformerMixin_, то метод _fit_transform_ получаете просто так.

После обратного преобразования можно сравнить его с исходными данными:

>>> tps_df.values[:5, 5]
array([0.35275, 0.17725, 0.25997, 0.4293 , 0.34079])

>>> tps_inversed[:5, 5]
array([0.35275, 0.17725, 0.25997, 0.4293 , 0.34079])

Теперь у нас есть собственный преобразователь. Давайте соберём весь код воедино:

from sklearn.metrics import roc_auc_score
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline

xgb_pipe = make_pipeline(
    FunctionTransformer(num_missing_row),
    SimpleImputer(strategy="constant", fill_value=-99999),
    CustomLogTransformer(),
    xgb.XGBClassifier(
        n_estimators=1000, tree_method="gpu_hist", objective="binary:logistic"
    ),
)

X, y = tps_df.drop("claim", axis=1), tps_df[["claim"]].values.flatten()
split = train_test_split(X, y, test_size=0.33, random_state=1121218)
X_train, X_test, y_train, y_test = split
xgb_pipe.fit(X_train, y_train)
preds = xgb_pipe.predict_proba(X_test)

>>> roc_auc_score(y_test, preds[:, 1])
0.7986831816726399

Несмотря на то что преобразование нанесло вред модели, мы заставили наш конвейер работать!

Говоря коротко, сигнатура пользовательского класса преобразователя должна быть такой:

class CustomTransformer(BaseEstimator, TransformerMixin):
    def __init__(self):
        pass

    def fit(self):
        pass

    def transform(self):
        pass

    def inverse_transform(self):
        pass

Так вы получаете fit_transform без всяких усилий. Если не нужны методы __init__, fit, transform или inverse_transform, не используйте их, родительские классы Sklearn позаботятся обо всём. Логика этих методов полностью зависит от ваших нужд.

А пока вы осваиваете преобразования в Sklearn, мы поможем вам прокачать навыки или с самого начала освоить профессию, востребованную в любое время:

• Профессия Data Scientist

• Профессия Fullstack-разработчик на Python

Выбрать другую востребованную профессию.