Сегодня разбираемся, как создавать собственные преобразователи 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, мы поможем вам прокачать навыки или с самого начала освоить профессию, востребованную в любое время:
Выбрать другую востребованную профессию.
