import subprocess subprocess.check_call('pip install -r /opt/ml/processing/input/dependencies/requirements-maze.txt', shell=True) import os import pandas as pd import numpy as np import time import pickle import datetime from datetime import date from imblearn.over_sampling import SMOTE 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 from sklearn.metrics import classification_report, confusion_matrix, accuracy_score, log_loss from sklearn.model_selection import GridSearchCV from sklearn.preprocessing import MinMaxScaler, StandardScaler from sklearn.linear_model import LogisticRegression from sklearn.inspection import permutation_importance from sklearn.utils import class_weight from abc import ABC, abstractmethod import warnings warnings.filterwarnings("ignore") import snowflake.connector from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives.asymmetric import dsa from cryptography.hazmat.primitives import serialization import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s.%(msecs)03d %(levelname)s %(module)s - %(funcName)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S', ) import pyecharts as echarts import boto3 import re from getpass import getpass from absl import logging log_level = "DEBUG" ticket_code = "EXP_1" logging.set_verbosity(log_level) logging.debug("READY!!!") sec_id = 'dev/sagemaker-notebook-instance/SNOWFLAKE_PASSWORD' def get_secret_value(name, version=None): """Gets the value of a secret. Version (if defined) is used to retrieve a particular version of the secret. """ secrets_client = boto3.client("secretsmanager", region_name='us-east-1') kwargs = {'SecretId': name} if version is not None: kwargs['VersionStage'] = version response = secrets_client.get_secret_value(**kwargs) return response def get_snowflake_creds(username="SAGEMAKER", account="orchard", warehouse="DEV_OWS_ENGINEERING"): """ Fetches and returns snowflake creds for connecting to snowflake Please use this within the scope of a function if using this on a shared instance This is so that the password is in memory only when its needed and gets dropped once its no longer required. returns: - creds (dict) - a dictionary containing user creds """ creds = { "user": username, "password": get_secret_value(sec_id)['SecretString'], "account": "orchard", "warehouse": warehouse, "protocol": 'https' } return creds def snowflake_connector_factory(creds=None): """ A Factory for creating snowflake connectors. This returns the cursor after opening a session with snowflake. params: - creds - snowflake credentials returns: - cursor - snowflake session cursor """ try: if creds: _creds = creds else: _creds = get_snowflake_creds() return snowflake.connector.connect(**_creds).cursor() except Exception as e: logging.error(f"Something went wrong - {str(e)}") def _is_version_number(s): "Check and returns true if its a version number" return re.search("^[0-9][.0-9]*[0-9]$", s) is not None def test_connection(): """ tests connection to snowflake """ with snowflake_connector_factory() as cs: try: cs.execute("SELECT current_version()") one_row = cs.fetchone() assert len(one_row) == 1 assert _is_version_number(one_row[0]) logging.info(f"Your snowflake version - {one_row[0]} PASSED!") except Exception as e: logging.error(f"Something went wrong - {str(e)}") ######################################################################################################################################################################## Data prep ####################################### ################################################################################################################################# 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'], format='%Y-%m-%d') df['LATEST_HIT_DATE'] = pd.to_datetime(df['LATEST_HIT_DATE'], format='%Y-%m-%d') df['Month_release'] = df['FIXED_RELEASE_DATE'].dt.strftime('%b') df['Month_chart'] = df['MIN_CHART_DATE'].dt.strftime('%b') df['Month_latest_chart'] = df['LATEST_HIT_DATE'].dt.strftime('%b') return df def prepare_sample(df): prefixes = ['USS1Z', 'QMS1Z', 'ISRC', 'QZS1Z', 'JMK40', 'AB'] mask = df['ISRC'].str.startswith(tuple(prefixes)) filtered = df[~mask] filtered['FIXED_RELEASE_DATE'] = filtered['FIXED_RELEASE_DATE'].dt.date sub = filtered[filtered['FIXED_RELEASE_DATE'] <= date(2024,2, 13)] country_sample = sub[(sub['COUNTRY_CODE'].isin(['GB', 'US', 'IN', 'CA', 'ZA', 'ΝL', 'MZ', 'NA', 'BW', 'ZW'])) | (sub['CHART_COUNTRY']=='ZA')] return country_sample def artist_score_calculation(df): conditions = [(df['PREVIOUS_HIT']==0), (df['PREVIOUS_HIT']==1) & (df['DAYS_SINCE_LAST_HIT'] > 365), (df['PREVIOUS_HIT']==1) & (df['DAYS_SINCE_LAST_HIT'] <= 365)] choices = ['no_previous_hit', 'non_recent_hit', 'recent_hit'] df['artist_score'] = np.select(conditions, choices) return df def fix_dataframe(df): df = fix_time_variables(df) #df = extract_country(df) df = artist_score_calculation(df) sub = prepare_sample(df) return sub def prepare_data(data): data = fix_dataframe(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']] data['MODE'] = data['MODE'].apply(str) numeric_features= data[['ACOUSTICNESS', 'DANCEABILITY', 'DURATION_MS', 'ENERGY', 'INSTRUMENTALNESS', 'KEY', 'LIVENESS', 'LOUDNESS', 'SPEECHINESS', 'TEMPO', 'TIME_SIGNATURE', 'VALENCE']] categorical_features = data[['Month_release', 'artist_score']] binary_features = data[['MODE']] scaler = StandardScaler() ordinal_encoder = OrdinalEncoder() label_encoder = LabelEncoder() X_processed = pd.DataFrame(scaler.fit_transform(numeric_features), columns = numeric_features.columns) encoded = ordinal_encoder.fit_transform(categorical_features) X_categorical1 = pd.DataFrame(encoded, columns = categorical_features.columns) X_categorical2 = pd.DataFrame(label_encoder.fit_transform(binary_features), columns = binary_features.columns) new_X = pd.concat([X_processed.reset_index(), X_categorical1.reset_index(), X_categorical2.reset_index()], axis=1) new_X = new_X.drop(columns=['index'], axis=1) return (new_X , y) def balancing_minor_class(new_X): (X_prepared, y_prepared) = prepare_data(new_X) X_resampled, y_resampled = SMOTE(random_state=2024, sampling_strategy='minority').fit_resample(X_prepared, y_prepared) return X_resampled, y_resampled def split_train_test_data(new_X): """ Cross validation with balanced classes """ X_resampled, y_resampled = balancing_minor_class(new_X) X_train, X_test, y_train, y_test = train_test_split(X_resampled, y_resampled, test_size=0.3, random_state=2023, stratify=y_resampled) # 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) ######################################################################################################################################################################## Regression ####################################### ################################################################################################################################# class Model(ABC): def __init__(self): super().__init__() logging.info('Initialising Model') @abstractmethod def train(self): logging.info('Training Model') @abstractmethod def predict(self): logging.info('Predicting...') ############################################################################################################# ####################### Prdictors ################################################################ ############################################################################################################# class LogisticRegressionModel(Model): def __init__(self, prepared_data, outcome, cutoff = 0.50): super().__init__() self.X = prepared_data self.y = outcome self.model = None self.cutoff = cutoff self.class_weights = None def train(self): best_params = self.search_best_parameters(self.X, self.y) self.class_weights = class_weight.compute_class_weight('balanced', classes=np.unique(self.y), y=self.y.values.reshape(-1)) self.model = LogisticRegression(solver = best_params['solver'], C = best_params['C'], max_iter = best_params['max_iter'], penalty = best_params['penalty'], class_weight = 'balanced', n_jobs=os.cpu_count()).fit(self.X, self.y.values.ravel()) def predicted_probabilities(self, X_to_predict): return self.model.predict_proba(X_to_predict) def predict(self, X_to_predict): pred = self.predicted_probabilities(X_to_predict) prob_1 = [x[1] for x in pred] return np.where(np.array(prob_1) > self.cutoff, 1, 0) def model_intercept(self): #print(self.model.estimator_.intercept_) print(self.model.intercept_) def model_coefficients(self): for coefficients, feature in zip(self.model.coef_[0], self.X.columns): print(f"{feature}: {coefficients}") def search_best_parameters(self, X_train, y_train): """ Using grid search to identify a good set of parameters """ logging.info('Searching params for Logistic Regression ... \n') param_grid = { #'solver': ['liblinear', 'newton-cg', 'lbfgs', 'saga'], # solver and penalty params can't be on at the same time. l1 and l2 support different solvers 'solver': ['liblinear', 'saga'], 'C': [0.001, 0.01, 0.1, 1, 10, 100, 1000], 'max_iter': [1000], 'penalty' : ['l1', 'l2']} # CV_logreg = GridSearchCV(estimator=LogisticRegression(random_state=2023), param_grid=param_grid, cv=10, error_score='raise', n_jobs=os.cpu_count()) CV_logreg.fit(X_train, y_train.values.ravel()) result = CV_logreg.best_params_ logging.info('Best parameters found: %s', result) return result def important_features(self): model_fi = permutation_importance(self.model, self.X, self.y, scoring = 'f1') importances = model_fi.importances_mean for feature, importance in zip(self.X.columns, importances): print(f"{feature}: {importance}") def ROC_plot(self): return sklearn.metrics.plot_roc_curve(CV_logreg, X_train, y_train.values.ravel()) def estimate_predictor(model, X_test, y_test): """ Estimates accuracy of model """ pred_y = model.predict(X_test) observed_y = y_test return classification_report(observed_y, pred_y, output_dict=True) def confusion_matrix_calculation(model, X_test, y_test): pred_y = model.predict(X_test) observed_y = y_test return confusion_matrix(observed_y, pred_y) def accuracy_calculator(model, X_test, y_test): pred_y = model.predict(X_test) observed_y = y_test return accuracy_score(observed_y, pred_y) def loss_evaluation(model, X_test, y_test): pred_y = model.predict(X_test) observed_y = y_test return log_loss(observed_y, pred_y) ################################################################################################################################# if __name__ == '__main__': with snowflake_connector_factory(get_snowflake_creds()) as cs: try: cs.execute("USE WAREHOUSE DEV_PERFORMANCE_WAREHOUSE;") cs.execute(""" select * from dev_engineering.eimpara.maze_country_table; """) rows = cs.fetchall() except Exception as e: logging.error(f"Something went wrong - {str(e)}") data_df = pd.DataFrame(rows, columns=list(map(lambda meta: meta[0], cs.description))) df = data_df.drop_duplicates().copy() print("{} rows loaded".format(df.shape[0])) (X_train, y_train, X_test, y_test) = split_train_test_data(df) log_model = LogisticRegressionModel(X_train, y_train) log_model.train() logging.info(f"South Africa") log_performances = estimate_predictor(log_model, X_test, y_test) logging.info(f"Perfromance metric for Logistic Model {log_performances}") matrix_log = confusion_matrix_calculation(log_model, X_test, y_test) logging.info(f"Perfromance metric for Logistic Model {matrix_log}") # with open('/Users/impr001/Documents/Jupyter_Notebooks/Maze/log_model_GB.pkl', 'wb') as file: # pickle.dump(log_model, file) # Save the model to disk pkl_filename = 'log_model_ZA_v2.pkl' with open(pkl_filename, 'wb') as file: pickle.dump(log_model, file) s3_resource = boto3.resource('s3') bucket = 'dev-cucumbers' key = 'eimpara/MAZE/log_model_ZA_v2.pkl' # Read the pickled file as bytes with open(pkl_filename, 'rb') as f: pickle_bytes_obj = f.read() # Upload the pickled file to S3 s3_resource.Bucket(bucket).put_object(Key=key, Body=pickle_bytes_obj)