Custom functions and pipelines

Question

I'm not really used to working with pipelines, so I'm wondering how can I use custom functions and pipelines.

Situation: I want to fill some missing values with the mean but using groups based on other feature. That's why I'm using this custom function:

def replaceNullFromGroup(From, To, variable, by):

    # 1. Create aggregation from train dataset
    From_grp = From.groupby(by)[variable].median().reset_index()

    # 2. Merge dataframes
    To_merged = To.merge(From_grp, on=by, suffixes=['_test', '_train'], how = "left")

    # 3. Create dictionaries
    to_cols = [col for col in To_merged.columns if 'test' in col]
    from_cols = [col for col in To_merged.columns if 'train' in col]
    dict_cols =dict(zip(to_cols, from_cols))

    # 4. Replace null values
    for to_col, from_col  in dict_cols.items():
        To_merged[to_col] = np.where(To_merged[to_col].isnull(), 
                                     To_merged[from_col], 
                                     To_merged[to_col])

    # 5. Clean up dataframe    
    To_merged.drop(from_col, axis=1, inplace=True)
    To_merged.columns = To_merged.columns.str.replace('_test', '')
    return To_merged

Variables meaning:

• From: Dataframe where I'm taking the information (Train dataset)
• To: Dataframe where I will fill the missing values (Train and test dataset)
• variable: variable with missing values
• by: Variables I'm using to make groups

Can I use this function in a pipeline so I can use cross validation avoiding data leakage?

Thank you very much

Answer 1

To include this logic into a pipeline you have to create a custom transformer. You need to ask yourself:

1. [INIT] Are there any parameters in my logic?
   • The variable you want to impute and the category you want this imputation to be based on.
2. [FIT] What part of the logic is related to computing what the transformation will be?
   • When you compute the median() by groups and store the data somehow for later transform.
3. [TRANSFORM] What part of the logic transforms the data, given the parameters (in 1) and the setting that was made (in 2)?
   • When you get the parameters (access specific key in the dictionary) to retrieve what was the mean for that group, and then fill the missing value with this.

Here is an example :

from sklearn.base import BaseEstimator, TransformerMixin

class CustomImputer(BaseEstimator, TransformerMixin) : 
     def __init__(self, variable, by) : 
          #self.something enables you to include the passed parameters
          #as object attributes and use it in other methods of the class
          self.variable = variable
          self.by = by
          return self

     def fit(self, X, y=None) : 
          self.map = X.groupby(self.by)[variable].mean()
          #self.map become an attribute that is, the map of values to
          #impute in function of index (corresponding table, like a dict)
          return self

     def transform(self, X, y=None) : 
          X[variable] = X[variable].fillna(value = X[by].map(self.map))
          #Change the variable column. If the value is missing, value should 
          #be replaced by the mapping of column "by" according to the map you
          #created in fit method (self.map)
          return X

Now, it can be included in any pipeline :

#Minimal example, you could include this imputer in columns transformer to 
#apply it multiple time
pipeline = Pipeline(steps = [('myImputer', CustomImputer('variabletofill',
                                                         'based_on_variable'),
                              ('model', LinearRegression())])

y_pred = pipeline.fit(X_train, y_train).predict(X_test)

As you can see, the mapping is computed only based on train data. Then it is re-used to impute the missing values. It is data-leakage-proof. Here is a good article that explains how to create a custom transformer.

Hope this helps