import math import random import scipy import pandas as pd import matplotlib.pyplot as plt import numpy as np import seaborn as sns import statsmodels.formula.api as smf import statsmodels.api as sm import pmdarima as pm import time from statsmodels.tsa.statespace.sarimax import SARIMAX from statsmodels.tsa.arima.model import ARIMA from statsmodels.graphics.tsaplots import plot_acf, plot_pacf from statsmodels.tsa.seasonal import seasonal_decompose from statsmodels.tools.eval_measures import mse,rmse, meanabs from statsmodels.tsa.stattools import adfuller from statsmodels.tsa.statespace.tools import diff from sklearn.metrics import mean_squared_error from datetime import datetime, date from sklearn.preprocessing import StandardScaler import snowflake.connector import os 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 warnings from password import PRIVATE_KEY_PASSPHRASE from multiprocessing import Pool from snowflake.connector.pandas_tools import write_pandas warnings.filterwarnings("ignore") pd.options.display.float_format = '{:.2f}'.format # Param to change if you want to run it key_path = "/Users/impr001/Keys_snowflake/rsa_key.p8" user='eimpara' account='orchard' role= 'PROD_DATALYTICS_ROLE' warehouse = "ANALYTICS_AD_HOC" database="DEV_ENGINEERING" schema="EIMPARA" table_name = '!!!_DO_NOT_USE_!!!' # Uncomment to allow writing to snowflake # table_name = 'MOMENTS_PRED_TABLE_140_DAYS' with open(key_path, "rb") as key: p_key= serialization.load_pem_private_key( key.read(), password=PRIVATE_KEY_PASSPHRASE.encode(), backend=default_backend() ) pkb = p_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption()) def timing_val(func): def wrapper(*arg, **kw): t1 = time.time() res = func(*arg, **kw) t2 = time.time() return (t2 - t1), res, func.__name__ return wrapper # returns full df @timing_val def load_full_data(): ctx = snowflake.connector.connect( user=user, account=account, private_key=pkb, role= role, warehouse = warehouse ) try: cs = ctx.cursor() sql = """ WITH invalid AS (select ISRC, count(*) as count from intelligence.dbt_prod_project_moments.project_moments_spotify_streams_120_day group by ISRC having count(*) <> 140), infelction_data as(select distinct ACTIVITY_DATE, Inflection_Point, ISRC from dev_engineering.eimpara.MOMENTS_FOURIERTABLE_APR_SEPT_22_V2 where Inflection_Point =1 and ACTIVITY_DATE > '2022-09-01'), data_to_analyse AS(select distinct ACTIVITY_DATE, ISRC, TRACK_NAME, ARTIST_ID, ARTIST_NAME, STREAMS from intelligence.dbt_prod_project_moments.project_moments_spotify_streams_120_day where ISRC not in (select ISRC from invalid) and ISRC in (select ISRC from infelction_data)) select * from data_to_analyse order by ISRC asc, ACTIVITY_DATE desc """ cs.execute(sql) df = cs.fetch_pandas_all() return df finally: cs.close() ctx.close() def split_universe(df, n): def chunks(l, n): for i in range(0, len(l), n): yield l[i:i + n] isrcs = df['ISRC'].unique().tolist() isrcs.sort() isrc_chunks = chunks(isrcs, (len(isrcs) // n) + 1) return list(isrc_chunks) @timing_val def split_work(full_df, n, f): isrc_chunks = split_universe(full_df, n) splitted_df = [] for chunk in isrc_chunks: df_chunk = full_df[full_df['ISRC'].isin(chunk)] splitted_df.append(df_chunk) run_id = str(time.time() * 1000000) print("ISRCs have been splitted in {} dataframes for run id={}".format(n, run_id)) with Pool(n) as p: splitted_df_with_index = list(enumerate(splitted_df)) splitted_df_with_args = map(lambda x: (run_id, x[0], x[1]), splitted_df_with_index) # parallel computation happens here p.map(f, splitted_df_with_args) print("Moments calculation finished.") # Save one chunk of work # def save_chunk(df, run_id, chunk_id, path="~/Documents/Moments", prefix="df_moments_"): # filepath = "{}/{}_{}_{}.csv".format(path, prefix, run_id, chunk_id) # df.to_csv(filepath) def save_chunk(df, run_id, chunk_id, table_name = table_name): def create_table_if_not_exists(cs): cs.execute("USE WAREHOUSE ANALYTICS_AD_HOC") cs.execute("USE DATABASE DEV_ENGINEERING") cs.execute("USE SCHEMA EIMPARA") cs.execute( "CREATE TABLE IF NOT EXISTS " + table_name + "(ISRC string, PRED_TYPE string, MSE float, RMSE float, " + "NRMSE float, AVG_RMSE float, MAE float, SUM_FORECAST_ERRORS float, AVG_FE float, LEN_DF integer)") def save_to_snowflake(ctx, df): return write_pandas( conn=ctx, df=df, table_name=table_name, database=database, schema=schema, quote_identifiers=False) ctx = snowflake.connector.connect( user=user, account=account, private_key=pkb, role= role, warehouse = warehouse) try: cs = ctx.cursor() create_table_if_not_exists(cs) success, num_chunks, num_rows, output = save_to_snowflake(ctx, d) print("Chunk {} saved: success={},rows={},output={}".format(chunk_id,success, num_rows, output)) finally: cs.close() def auto_arima(df_isrc): cutoff_train_test_sets = 133 (train, test) = (df_isrc.iloc[:cutoff_train_test_sets], df_isrc.iloc[cutoff_train_test_sets:]) train.index = pd.to_datetime(train['ACTIVITY_DATE']) train = train.sort_index(axis = 0) auto_df = train[['STREAMS']].copy() auto_model = pm.auto_arima(auto_df, start_p=1, start_q=1, test='adf', max_p=3, max_q=3, m=7, start_P=0, seasonal=True, d=None, D=1, trace=False, error_action='ignore', suppress_warnings=True, stepwise=True) return (auto_model.seasonal_order, auto_model.order) def iterate_group_by_key(df, col_for_key, sort_data=True): if sort_data: df = df.sort_values(by=col_for_key) def key_at(i): return np.array(df[col_for_key].iloc[i]) index, size = (0, df.shape[0]) while index < size: current_key = key_at(index) res = [] while index < size and list(current_key) == list(key_at(index)): res.append(df.iloc[index]) index = index + 1 resdf = pd.DataFrame(res, columns=df.columns) yield resdf @timing_val def arima_iscrs(df): new_df = pd.DataFrame(columns=['ISRC','pred_type', 'mse', 'rmse','nrmse', 'avg_rmse', 'mae', 'avg_mae', 'sum_forecast_errors', 'avg_fe','len_df']) count = 1 for subdf in iterate_group_by_key(df, ['ISRC']): df_isrc = subdf isrc = df_isrc['ISRC'].iloc[0] df_isrc = df_isrc.sort_values(by=['ACTIVITY_DATE']) try: cutoff_train_test_sets = 133 start_pred = 132 end_pred = 139 (train, test) = (df_isrc.iloc[:cutoff_train_test_sets], df_isrc.iloc[cutoff_train_test_sets:]) (end_test, end_train) = (len(test), len(train)) (seasonal_order, order) = auto_arima(train) arima_model = ARIMA(train['STREAMS'], order=order, seasonal_order=seasonal_order,enforce_stationarity=False).fit() pred = arima_model.get_prediction(start= 1, end = (end_train + end_test-1), dynamic=False) forecast_errors = np.subtract(np.array(test['STREAMS']), np.array(pred.predicted_mean[start_pred:end_pred])) max_min = test['STREAMS'].max() - test['STREAMS'].min() mse = np.square(forecast_errors).mean() rmse = np.sqrt(mse) nrmse = rmse/max_min avg_rmse = rmse/test.shape[0] mae = np.abs(np.subtract(np.array(test['STREAMS']),np.array(pred.predicted_mean[start_pred:end_pred])).mean()) avg_mae = mae/test.shape[0] sum_forecast_errors = round(forecast_errors.sum(), 3) avg_fe = sum_forecast_errors/test.shape[0] all_positives = all(map(lambda x: x > 0, forecast_errors)) all_negatives = all(map(lambda x: x < 0, forecast_errors)) pred_categorical = None if all_positives: pred_categorical = 'actuals_above_predicted' elif all_negatives: pred_categorical = 'actuals_below_predicted' else: pred_categorical = 'actuals_crossing_predicted' new_df = new_df.append(pd.DataFrame([{ 'ISRC': isrc, 'pred_type': pred_categorical, 'mse': mse, 'rmse': rmse, 'nrmse': nrmse, 'avg_rmse':avg_rmse, 'mae': mae, 'avg_mae': avg_mae, 'sum_forecast_errors': sum_forecast_errors, 'avg_fe': avg_fe, 'len_df': df_isrc.shape[0] }], columns=new_df.columns)) if count % 100 ==0: print("Worker processed {} ISRCs.".format(count)) count = count+1 except Exception as e: print("Error while processing ISRC {}: '{}'".format(isrc, e)) return new_df def work_load(arg): (run_id, chunk_id, df) = arg print("Arguments {},{},{}".format(run_id, chunk_id, type(df))) timing, res, _ = arima_iscrs(df) save_chunk(res, run_id, chunk_id) if __name__ == '__main__': #### Running the code parallelism = 10 timing, full_df, _ = load_full_data() print("{} rows loaded in {} seconds".format(full_df.shape[0], timing)) print("Splitting work into {} chunks".format(parallelism)) split_work(full_df, parallelism, work_load)