import logging import pandas as pd import numpy as np from sklearn.pipeline import Pipeline from sklearn.compose import ColumnTransformer from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler, OneHotEncoder, LabelEncoder, OrdinalEncoder def timing_val(func): ''' Utility function that calculates the execution time of other functions. When @timing_val is present, the function will return a tuple containing result and execution time. From source: http://www.daniweb.com/code/snippet368.html ''' def wrapper(*arg, **kw): t1 = time.time() res = func(*arg, **kw) t2 = time.time() return (t2 - t1), res, func.__name__ return wrapper def extract_country(df): df['ISRC_first_two_characters'] = df['ISRC'].str[:2] conditions = [(df['ISRC_first_two_characters']=='AR'), (df['ISRC_first_two_characters']=='AU'), (df['ISRC_first_two_characters']=='AT'), (df['ISRC_first_two_characters']=='BY'), (df['ISRC_first_two_characters']=='BE'), (df['ISRC_first_two_characters']=='BO'), (df['ISRC_first_two_characters'].isin(['BR', 'BP', 'BX', 'BC', 'BK'])), (df['ISRC_first_two_characters'].isin(['CA', 'CB'])), (df['ISRC_first_two_characters']=='CL'), (df['ISRC_first_two_characters']=='CO'), (df['ISRC_first_two_characters']=='CY'), (df['ISRC_first_two_characters']=='CY'), (df['ISRC_first_two_characters']=='CZ'), (df['ISRC_first_two_characters'].isin(['DK', 'GL', 'FO'])), (df['ISRC_first_two_characters']=='DO'), (df['ISRC_first_two_characters']=='EC'), (df['ISRC_first_two_characters']=='EG'), (df['ISRC_first_two_characters']=='SV'), (df['ISRC_first_two_characters']=='EE'), (df['ISRC_first_two_characters']=='FI'), (df['ISRC_first_two_characters'].isin(['FR', 'FX'])), (df['ISRC_first_two_characters']=='DE'), (df['ISRC_first_two_characters']=='GR'), (df['ISRC_first_two_characters']=='GT'), (df['ISRC_first_two_characters']=='HN'), (df['ISRC_first_two_characters']=='HK'), (df['ISRC_first_two_characters']=='HU'), (df['ISRC_first_two_characters']=='IS'), (df['ISRC_first_two_characters']=='IN'), (df['ISRC_first_two_characters']=='ID'), (df['ISRC_first_two_characters']=='IE'), (df['ISRC_first_two_characters']=='IL'), (df['ISRC_first_two_characters']=='IT'), (df['ISRC_first_two_characters']=='JP'), (df['ISRC_first_two_characters']=='KZ'), (df['ISRC_first_two_characters']=='LV'), (df['ISRC_first_two_characters']=='LT'), (df['ISRC_first_two_characters']=='LU'), (df['ISRC_first_two_characters']=='MY'), (df['ISRC_first_two_characters']=='MX'), (df['ISRC_first_two_characters']=='MA'), (df['ISRC_first_two_characters']=='NL'), (df['ISRC_first_two_characters']=='NZ'), (df['ISRC_first_two_characters']=='NG'), (df['ISRC_first_two_characters']=='NO'), (df['ISRC_first_two_characters']=='PK'), (df['ISRC_first_two_characters']=='PA'), (df['ISRC_first_two_characters']=='PY'), (df['ISRC_first_two_characters']=='PE'), (df['ISRC_first_two_characters']=='PH'), (df['ISRC_first_two_characters']=='PL'), (df['ISRC_first_two_characters']=='PT'), (df['ISRC_first_two_characters']=='RO'), (df['ISRC_first_two_characters']=='SA'), (df['ISRC_first_two_characters']=='SG'), (df['ISRC_first_two_characters']=='SK'), (df['ISRC_first_two_characters'].isin(['ZA', 'ZB'])), (df['ISRC_first_two_characters']=='ES'), (df['ISRC_first_two_characters']=='SE'), (df['ISRC_first_two_characters']=='CH'), (df['ISRC_first_two_characters']=='TH'), (df['ISRC_first_two_characters']=='TR'), (df['ISRC_first_two_characters']=='AE'), (df['ISRC_first_two_characters']=='UA'), (df['ISRC_first_two_characters'].isin(['UK', 'GX', 'GB'])), (df['ISRC_first_two_characters']=='UY'), (df['ISRC_first_two_characters'].isin(['US', 'QM', 'QZ'])), (df['ISRC_first_two_characters']=='VE'), (df['ISRC_first_two_characters']=='VN'), (df['ISRC_first_two_characters'].isin(['TC', 'DG', 'ZZ', 'CP', 'CS', 'YU', 'YE', 'WZ', 'SF'])), (df['ISRC_first_two_characters']=='CN'), (df['ISRC_first_two_characters']=='TW'), (df['ISRC_first_two_characters']=='BB'), (df['ISRC_first_two_characters']=='BG'), (df['ISRC_first_two_characters']=='RU'), (df['ISRC_first_two_characters'].isin(['KR','KS'])), (df['ISRC_first_two_characters']=='LK'), (df['ISRC_first_two_characters']=='SI'), (df['ISRC_first_two_characters']=='BH'), (df['ISRC_first_two_characters']=='JM'), (df['ISRC_first_two_characters']=='GD'), (df['ISRC_first_two_characters']=='NP'), (df['ISRC_first_two_characters']=='AI'), (df['ISRC_first_two_characters']=='CU')] choices = ['Argentina','Australia','Austria', 'Belarus', 'Belgium', 'Bolivia', 'Brazil', 'Bulgaria', 'Canada', 'Chile', 'Colombia', 'Cyprus', 'Czech_Republic', 'Denmark','Dominican_Republic', 'Ecuador', 'Egypt', 'El_Salvador', 'Estonia', 'Finland', 'France', 'Germany', 'Greece', 'Guatemala', 'Honduras', 'Hong_Kong', 'Hungary','Iceland', 'India', 'Indonesia', 'Ireland', 'Israel', 'Italy', 'Japan', 'Kazakhstan', 'Latvia', 'Lithuania', 'Luxembourg', 'Malaysia', 'Mexico', 'Morocco','Netherlands', 'New_Zealand', 'Nigeria', 'Norway', 'Pakistan', 'Panama', 'Paraguay', 'Peru', 'Philippines', 'Poland', 'Portugal', 'Romania', 'Saudi_Arabia', 'Singapore', 'Slovakia', 'South_Africa', 'Spain', 'Sweden', 'Switzerland', 'Thailand', 'Turkey', 'UAE', 'Ukraine', 'United_Kingdom', 'Uruguay', 'USA', 'Venezuela', 'Vietnam', 'Worldwide', 'China', 'Chinese_Taipei', 'Barbados', 'Bulgaria', 'Russia', 'South_Korea', 'Sri_Lanka', 'Slovenia', 'Bahrain', 'Jamaica','Grenada', 'Nepal', 'Anguilla', 'Cuba'] df['Market_manually_derived'] = np.select(conditions, choices) return df def fix_time_variables(df): df['FIXED_RELEASE_DATE'] = pd.to_datetime(df['FIXED_RELEASE_DATE']) df['MIN_CHART_DATE'] = pd.to_datetime(df['MIN_CHART_DATE']) df['LATEST_HIT_DATE'] = pd.to_datetime(df['LATEST_HIT_DATE']) df['Month_release'] = df['FIXED_RELEASE_DATE'].dt.month_name() df['Month_chart'] = df['MIN_CHART_DATE'].dt.month_name() df['Month_latest_chart'] = df['LATEST_HIT_DATE'].dt.month_name() return df def fix_dataframe(df): df = extract_country(df) df = fix_time_variables(new_df) return df def prepare_data(data): # data = data.drop_duplicates(subset='ISRC', keep='first').reset_index(drop=True) #data = data.sort_values(by=['ISRC']) X = data.drop(columns = ['ISRC','MIN_RELEASE_DATE', 'LATEST_HIT_DATE', 'DAYS_SINCE_LAST_HIT', 'DAYS_TO_CHART', 'FIXED_RELEASE_DATE', 'ARTIST_NAME']) y = data[['HIT']] numeric_features= data[['ACOUSTICNESS', 'DANCEABILITY', 'DURATION_MS', 'ENERGY', 'INSTRUMENTALNESS', 'KEY', 'LIVENESS', 'LOUDNESS', 'MODE', 'SPEECHINESS', 'TEMPO', 'TIME_SIGNATURE', 'VALENCE']] categorical_features1 = data[['Month_release', 'artist_score']] scaler = StandardScaler() label_encoder1 = OrdinalEncoder() scaler.fit(numeric_features) X_processed = pd.DataFrame(scaler.transform(numeric_features), columns = numeric_features.columns) label_encoder1.fit(categorical_features1) encoded1 = label_encoder1.transform(categorical_features1) X_categorical1 = pd.DataFrame(encoded1, columns = categorical_features1.columns) new_X = pd.concat([X_processed.reset_index(), X_categorical1.reset_index()], axis=1) new_X = new_X.drop(columns=['index'], axis=1) scalers = [scaler, label_encoder1] return (new_X , y, scalers) def split_train_test_data(X, y): """ Cross validation with balanced classes """ X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=2023, stratify=y) # checking if y is balanced train_0, train_1 = len(y_train[y_train==0]), len(y_train[y_train==1]) test_0, test_1 = len(y_test[y_test==0]), len(y_test[y_test==1]) logging.info('> Train: 0=%d, 1=%d, Test: 0=%d, 1=%d' % (train_0, train_1, test_0, test_1)) return (X_train, y_train, X_test, y_test)