""" Description: Contains feature processing pipelines used for preprocessing raw data into a format that can be used. """ import pandas as pd import numpy as np # utils etc import re import json # tensorflow imports import tensorflow as tf # scikitlearn from sklearn.preprocessing import ( # Scalers StandardScaler, MinMaxScaler, # Encoders OneHotEncoder, OrdinalEncoder ) from sklearn.impute import ( SimpleImputer ) from sklearn.compose import ColumnTransformer from sklearn.pipeline import Pipeline from sklearn.feature_selection import ( SelectPercentile ) # internal from forecasting_toolkit.feature_preprocessing.timeseries.transformers import ( sin_transformer, cos_transformer, periodic_spline_transformer ) """ Feature Processing Pipeline Factory v1 """ def feature_preprocessing_pipeline_factory_v1(categorical_cols, numerical_float_cols, time_cols, bool_cols, text_cols): """ Feature processing pipeline (v1) params: - categorical_cols (list[str]) - list of categorical features - numerical_float_cols (list[str]) - list of float cols, - time_cols (list[str]) - list of time features, such as ISOWEEK, YEAR. - bool_cols (list[str]) - list of boolean feature - has_trackname_feature (bool) - adds trackname features returns: - pipeline (sklearn.Pipeline) """ # categorical feature categorical_features_pipeline = [ ("categorical_features", OneHotEncoder(), categorical_cols) ] # count cols time_cols = [ ("time_features_minmax", MinMaxScaler(), time_cols), ("time_features_onehot", OneHotEncoder(), time_cols) ] # numerical features numerical_floats_features_pipeline = [ ("float_features", StandardScaler(), numerical_float_cols) ] # bool features bool_features_pipeline = [ ("boolean_features", OneHotEncoder(), bool_cols), ] # text features text_features_pipeline = [] if 'TRACKNAME_LENGTH' in text_cols: text_features_pipeline.append(("trackname_length_feature", MinMaxScaler(), ['TRACKNAME_LENGTH'])) # column transformers column_transform_fns = ColumnTransformer(transformers=categorical_features_pipeline + \ time_cols + \ numerical_floats_features_pipeline +\ bool_features_pipeline + \ text_features_pipeline, remainder = 'drop' ) return Pipeline(steps=[("feature_processor", column_transform_fns)]) """ Feature Processing Pipeline Factory v2 """ def feature_preprocessing_pipeline_factory_v2( numerical_float_cols, time_cols, bool_cols, snapshot_year = 'SNAPSHOT_YEAR', snapshot_day_of_week_col = 'SNAPSHOY_DAY_OF_WEEK', snapshot_day_of_year_col = 'SNAPSHOT_DAY_OF_YEAR', snapshot_month_col = 'SNAPSHOT_MONTH', release_year = 'RELEASE_YEAR', release_month_col = 'RELEASE_MONTH', release_week_col = 'RELEASE_WEEKISO', release_day_of_week_col = 'RELEASE_DOW', categorical_cols=['ARTIST_ID', 'LABEL_ID', 'RELEASE_GENREID'] ): """ Feature processing pipeline (v2) Some features: - Encodes time cyclical nature of time using sin and cos functions - handles missing data for categoricals, floats and bools params: - categorical_cols (list[str]) - list of categorical features - numerical_float_cols (list[str]) - list of float cols, - time_cols (list[str]) - list of time features, such as DAY_OF_WEEK, YEAR, etc. - bool_cols (list[str]) - list of boolean feature - snapshot_year (str) - snapshot year target column - snapshot_day_of_week_col (str) - snapshot year target column - snapshot_day_of_year_col (str) - snapshot_day_of_year_col - snapshot_month_col (str) - snapshot_month_col - release_year (str) - release_year - release_month_col (str) - release_month_col - release_week_col (str) - release_week_col - release_day_of_week_col (str) - release_day_of_week_col - categorical_cols (List[str]) - list of categorical columns returns: - pipeline (sklearn.Pipeline) """ # categorical feature categorical_features_pipeline = [ ("categorical_features", Pipeline(steps=[ ("not_available_imputer", SimpleImputer(strategy="constant", fill_value='N/A')), ("ordinal_encoder", OrdinalEncoder(handle_unknown='use_encoded_value', unknown_value=-1, encoded_missing_value=np.nan)), ("one_hot_encoder", OneHotEncoder(handle_unknown='infrequent_if_exist', min_frequency=10)) ]), categorical_cols ) ] # Time Feature Encoding time_cols = [ ("time_features_minmax", MinMaxScaler(), time_cols), ("time_features_onehot", Pipeline(steps=[ ("ordinal_encoder", OrdinalEncoder(handle_unknown='use_encoded_value', unknown_value=-1, encoded_missing_value=np.nan)), ("one_hot_encoder", OneHotEncoder(handle_unknown='infrequent_if_exist', min_frequency=5)) ]), time_cols ), # SNAPSHOT # day of week ("day_of_week_sin", sin_transformer(7), [snapshot_day_of_week_col]), ("day_of_week_cos", cos_transformer(7), [snapshot_day_of_week_col]), # year ("year_sin", sin_transformer(7), [snapshot_year]), ("year_cos", cos_transformer(7), [snapshot_year]), # day of year ("day_of_year_sin", sin_transformer(366), [snapshot_day_of_year_col]), ("day_of_year_cos", cos_transformer(366), [snapshot_day_of_year_col]), # month ("month_sin", sin_transformer(30), [snapshot_month_col]), ("month_cos", cos_transformer(30), [snapshot_month_col]), # RELEASE # year ("release_year_sin", sin_transformer(12), [release_year]), ("release_year_cos", cos_transformer(12), [release_year]), # month ("release_month_col_sin", sin_transformer(30), [release_month_col]), ("release_month_col_cos", cos_transformer(30), [release_month_col]), # week ("release_week_col_sin", sin_transformer(56), [release_week_col]), ("release_week_col_cos", cos_transformer(56), [release_week_col]), # day of week ("release_day_of_week_col_sin", sin_transformer(7), [release_day_of_week_col]), ("release_day_of_week_col_cos", cos_transformer(7), [release_day_of_week_col]), ] # numercial floats feature pipeline numerical_floats_features_pipeline = [ ("float_features", Pipeline(steps=[ ("mean_imputation", SimpleImputer(strategy="mean")), ("standardize_flaot", StandardScaler()), ("rescale", MinMaxScaler(feature_range=(-1,1))) ]), numerical_float_cols ) ] # bool features bool_features_pipeline = [ ("boolean_features", Pipeline(steps=[ ("not_available_imputer", SimpleImputer(strategy="constant", fill_value='N/A')), ("one_hot_encoder", OneHotEncoder(handle_unknown='infrequent_if_exist', min_frequency=5)) ]), bool_cols), ] # column transformers column_transform_fns = ColumnTransformer(transformers=categorical_features_pipeline + \ time_cols + \ numerical_floats_features_pipeline +\ bool_features_pipeline, remainder = 'drop' ) return Pipeline(steps=[("feature_processor", column_transform_fns)]) """ Feature Processing Pipeline Factory v3 """ def feature_preprocessing_pipeline_factory_v3(numerical_float_cols, time_cols, bool_cols, snapshot_year = 'SNAPSHOT_YEAR', snapshot_day_of_week_col = 'SNAPSHOY_DAY_OF_WEEK', snapshot_day_of_year_col = 'SNAPSHOT_DAY_OF_YEAR', snapshot_month_col = 'SNAPSHOT_MONTH', release_year = 'RELEASE_YEAR', release_month_col = 'RELEASE_MONTH', release_week_col = 'RELEASE_WEEKISO', release_day_of_week_col = 'RELEASE_DOW', categorical_cols=['ARTIST_ID', 'LABEL_ID', 'RELEASE_GENREID'], use_feature_selection=False ): """ Feature processing pipeline (v3) This includes features such as spline features on time columns params: - categorical_cols (list[str]) - list of categorical features - numerical_float_cols (list[str]) - list of float cols, - time_cols (list[str]) - list of time features, such as DAY_OF_WEEK, YEAR, etc. - bool_cols (list[str]) - list of boolean feature - snapshot_year (str) - snapshot year target column - snapshot_day_of_week_col (str) - snapshot year target column - snapshot_day_of_year_col (str) - snapshot_day_of_year_col - snapshot_month_col (str) - snapshot_month_col - release_year (str) - release_year - release_month_col (str) - release_month_col - release_week_col (str) - release_week_col - release_day_of_week_col (str) - release_day_of_week_col - use_feature_selection (bool) - Feature Selection returns: - pipeline (sklearn.Pipeline) """ steps = [] # categorical feature categorical_features_pipeline = [ ("categorical_features", OneHotEncoder(), categorical_cols) ] # Time Feature Encoding time_cols = [ # SNAPSHOT # day of week ("snapshot_day_of_week_sin", sin_transformer(7), [snapshot_day_of_week_col]), ("snapshot_day_of_week_cos", cos_transformer(7), [snapshot_day_of_week_col]), ("snapshot_day_of_week_spline", periodic_spline_transformer(7, n_splines=3), [snapshot_day_of_week_col]), # year ("snapshot_year_sin", sin_transformer(12), [snapshot_year]), ("snapshot_year_cos", cos_transformer(12), [snapshot_year]), ("snapshot_year_spline", periodic_spline_transformer(12, n_splines=6), [snapshot_year]), # day of year ("snapshot_day_of_year_sin", sin_transformer(366), [snapshot_day_of_year_col]), ("snapshot_day_of_year_cos", cos_transformer(366), [snapshot_day_of_year_col]), ("snapshot_day_of_year_spline", periodic_spline_transformer(366, n_splines=150), [snapshot_day_of_year_col]), # month ("snapshot_month_sin", sin_transformer(30), [snapshot_month_col]), ("snapshot_month_cos", cos_transformer(30), [snapshot_month_col]), ("snapshot_month_spline", periodic_spline_transformer(30, n_splines=15), [snapshot_month_col]), ## RELEASE # year ("release_year_sin", sin_transformer(12), [release_year]), ("release_year_cos", cos_transformer(12), [release_year]), ("release_year_spline", periodic_spline_transformer(12, n_splines=6), [release_year]), # month ("release_month_sin", sin_transformer(30), [release_month_col]), ("release_month_cos", cos_transformer(30), [release_month_col]), ("release_month_spline", periodic_spline_transformer(12, n_splines=6), [release_month_col]), # week ("release_week_sin", sin_transformer(56), [release_week_col]), ("release_week_cos", cos_transformer(56), [release_week_col]), ("release_week_spline", periodic_spline_transformer(56, n_splines=25), [release_week_col]), # day of week ("release_day_of_week_sin", sin_transformer(7), [release_day_of_week_col]), ("release_day_of_week_cos", cos_transformer(7), [release_day_of_week_col]), ("release_day_of_week_spline", periodic_spline_transformer(7, n_splines=3), [release_day_of_week_col]), ] # numercial floats feature pipeline numerical_floats_features_pipeline = [ ("float_features", StandardScaler(), numerical_float_cols) ] # bool features bool_features_pipeline = [ ("boolean_features", OneHotEncoder(), bool_cols), ] feature_selection_pipeline = [] if use_feature_selection: feature_selection_pipeline.append(("feature_selection", SelectPercentile(percentile=90, score_func="mutual_info_regression"))) # column transformers column_transform_fns = ColumnTransformer(transformers=categorical_features_pipeline + \ time_cols + \ numerical_floats_features_pipeline +\ bool_features_pipeline +\ feature_selection_pipeline, remainder = 'drop' ) steps = [("feature_processor", column_transform_fns)] + feature_selection_pipeline return Pipeline(steps=steps)