# Dependencies import os import subprocess subprocess.check_call('pip install -r /opt/ml/processing/input/dependencies/requirements.txt', shell=True) # subprocess.check_call('pip -q install snowflake-connector-python', shell=True) # subprocess.check_call('pip -q install pytest', shell=True) # subprocess.check_call('pip -q install pytest-sugar', shell=True) # subprocess.check_call('pip -q install pyecharts', shell=True) # subprocess.check_call('pip -q install absl-py', shell=True) # subprocess.check_call('pip -q install statsmodels', shell=True) # subprocess.check_call('pip -q install pmdarima', shell=True) # subprocess.check_call('pip -q install s3fs', shell=True) # subprocess.check_call('pip -q install boto3', shell=True) # subprocess.check_call('pip -q install datetime', shell=True) 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 pandas as pd import boto3 import math import random import scipy import numpy as np import statsmodels.formula.api as smf import statsmodels.api as sm import pmdarima as pm import time from datetime import datetime, date, timedelta from statsmodels.tsa.arima.model import ARIMA from statsmodels.tools.eval_measures import mse,rmse from statsmodels.tsa.statespace.tools import diff from scipy import fftpack 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 ################################################################################################################## ############################################# FOURIER ############################################################ ################################################################################################################## def fourier_table(one_track_df, loop=60): ''' Apply Fast Fourier Transform (FFT) to individual time series. FFT is a method for expressing a function as a sum of its periodic components. This is used to clean the noise from the data using low pass filter, which cuts off high frequencies. The function also identifies the inflection points of the FFT time series. one_track_df -- DataFrame containing the time series data loop -- the number of loops for the Fourier Transform. ''' # Computing FFT and applying low pass filter y = np.array(one_track_df['STREAMS']) y_fft_filtered = fftpack.fft(y).copy() freq = fftpack.fftfreq(len(y), d = 1/len(y)) # Filtering the frequencies cut_off = len(y)/loop y_fft_filtered[np.abs(freq) > cut_off] = 0 Fourier = fftpack.ifft(y_fft_filtered) # Computing second derivative to identify inflection points y_diff = pd.Series(Fourier).diff() y_diff_diff = y_diff.diff() locs2 = abs(np.diff(np.sign(y_diff_diff))) zero_diff_diff = np.where(np.logical_and(locs2 !=0, np.isnan(locs2)==False)) inflection_point = one_track_df.iloc[zero_diff_diff[0]] # Compiling table containing FFT columns Fourier = pd.DataFrame(Fourier, columns = ['Fourier']) Fourier['ACTIVITY_DATE'] = np.array(one_track_df['ACTIVITY_DATE']) streams = pd.DataFrame(one_track_df[['STREAMS','ACTIVITY_DATE']]) df_fourier = streams.merge(Fourier, on = ['ACTIVITY_DATE']) df_fourier['Inflection_Point'] = np.where(df_fourier['ACTIVITY_DATE'] .isin(inflection_point['ACTIVITY_DATE']), 1,0) df_fourier['Fourier_real_part'] = np.array(df_fourier['Fourier']).real df_fourier['ISRC'] = np.array(one_track_df['ISRC']) return df_fourier @timing_val def compile_fourier_table(df): '''Assembling the dataframe containing FFT and Inflection Point columns''' res = pd.DataFrame(columns=['ACTIVITY_DATE', 'STREAMS', 'Fourier', 'Inflection_Point', 'ISRC']) unique_isrcs = df['ISRC'].unique() for isrc in unique_isrcs: one_track_df = unique_isrc_df(isrc, df) if check_one_track_df(one_track_df): one_track_fourier = fourier_table(one_track_df, loop=60) res = pd.concat([res, one_track_fourier]) else: logging.error(f"Error on {isrc}: data has length {len(one_track_df)}") return res def compute_fourier_for_chunk(chunk_df, chunk_id, dry_run): '''Compute FastFourierTransform for a the given chunk of data, and save the data to S3''' logging.info(f"Starting to compute FFT on data chunk {chunk_id} ...") fourier_table = compile_fourier_table(chunk_df) logging.info(f"FFT on data chunk {chunk_id} finished.") fourier_df = pd.DataFrame(fourier_table[1]) fourier_df['ACTIVITY_DATE'] = pd.to_datetime(fourier_df['ACTIVITY_DATE']).dt.date logging.info(f"Saving FFT data to S3 ...") save_dataframe_s3(fourier_df, chunk_id, dry_run) logging.info(f"FFT data saved.") return fourier_df @timing_val def take_isrcs_with_inflection_last_week(fourier_df): ''' Function that selects only ISRCs that recorded an inflection point in the last 7 days of their lifecycle ''' first_day_pred = fourier_df['ACTIVITY_DATE'].max() - timedelta(days=7) list_for_pred = fourier_df[(fourier_df['Inflection_Point']==1) & (fourier_df['ACTIVITY_DATE'] > first_day_pred)]['ISRC'].unique().tolist() subset_for_pred = fourier_df[fourier_df['ISRC'].isin(list_for_pred)].copy() return subset_for_pred def save_dataframe_s3(df, chunk_id, dry_run): '''Saving FFT calculated data to S3''' if dry_run: logging.warning(f"DRY_RUN flag is True, no FFT data will be saved.") else: s3 = boto3.client('s3') bucket_name = 'dev-cucumbers' today = datetime.today().strftime('%Y%m%d-%H%M%S') filepath = "eimpara/Moments_2023_batches/Fourier_TESTING_{}_{}.csv".format(chunk_id, today) csv_buffer = df.to_csv(index=False).encode('utf-8') s3.put_object(Body=csv_buffer, Bucket=bucket_name, Key=filepath) logging.info(f"Table saved to S3 bucket: {bucket_name}, with file name: {filepath}") ################################################################################################################## ############################################# UTILS ############################################################## ################################################################################################################## def check_one_track_df(one_track_df): '''Checking that the length of the individual time series is 140 days''' return len(one_track_df)==140 def unique_isrc_df(isrc, df): ''' As time series analysis is univariate, decomposing the dataframe as to have one-track dataframes at time of analysis. ''' one_track_df = df[df['ISRC']==isrc].copy() one_track_df['ACTIVITY_DATE'] = pd.to_datetime(one_track_df['ACTIVITY_DATE']) one_track_df['ACTIVITY_DATE'] = one_track_df['ACTIVITY_DATE'].dt.date one_track_df = one_track_df[['ISRC', 'ACTIVITY_DATE', 'STREAMS']] return one_track_df.sort_values(by = 'ACTIVITY_DATE') def data_for_chunk(df, chunk_id, chunk_count): ''' Function sorting ISRCs and splitting them into n chuncks and taking only one of them (chunk_id) ''' IDs = df['ISRC'].unique().tolist() IDs.sort() chunk_size = len(IDs) // chunk_count chunked = [IDs[n:n+chunk_size] for n in range(0, len(IDs), chunk_size)] ids = chunked[chunk_id] chunk_df = df[df['ISRC'].isin(ids)] return chunk_df def split_universe(df, n): ''' Function splitting dataframe for parallelism df -- initial dataframe n -- number of chunks ''' # list of sorted ISRCs into n chunks 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, dry_run): ''' full_df -- full dataframe that will be split into n parallel tasks n -- number of parallel tasks f -- function handling one parallel task dry_run -- bolean flag indicating whether the results are saved. ''' isrc_chunks = split_universe(full_df, n) splitted_df = [] # For each chunk, creating an individual dataframe that only includes rows with the same ID for chunk in isrc_chunks: df_chunk = full_df[full_df['ISRC'].isin(chunk)] splitted_df.append(df_chunk) # Unique run ID generator based on timestamp # This is to allow to stitch back files together run_id = str(time.time() * 1000000) logging.info("ISRCs have been splitted in {} dataframes for run id={}".format(n, run_id)) # Using multiprocessing pool to apply the function to each chunk with Pool(n) as p: # Adding indices to the list of chunks splitted_df_with_index = list(enumerate(splitted_df)) # List of arguments for the function splitted_df_with_args = map(lambda x: (dry_run, run_id, x[0], x[1]), splitted_df_with_index) p.map(f, splitted_df_with_args) logging.info("Moments calculation finished.") def work_load(arg): ''' Function that returns the result of arima_iscrs() and saves it in S3''' (dry_run, run_id, chunk_id, df) = arg logging.info(f"Arguments {run_id},{chunk_id},{type(df)}") timing, res, _ = arima_iscrs(df) save_dataframe_s3(res, run_id, chunk_id, dry_run) return res if __name__ == '__main__': # Dry run = True => data will NOT be saved DRY_RUN = False # Splits the universe of ids into CHUNK_COUNT stable chunks CHUNK_COUNT = 100 # We are running the script only for the below CHUNK_ID, out of CHUNK_COUNT chunks CHUNK_ID = 0 # Data CSV File INPUT_FILE = 's3://dev-cucumbers/eimpara/Fourier/2023_data/data_for_2023_analysis_20231107-112155.csv' # Number of parallel processes to process the data PARALLELISM = 1 #max(1, cpu_count() // 2) if DRY_RUN: logging.warning(f"DRY_RUN flag is True, no data will be saved !!!") logging.info(f"Reading data from CSV '{INPUT_FILE}'") logging.info(f"Computing chunk {CHUNK_ID} out of {CHUNK_COUNT} chunk(s)") logging.info(f"Data will be processed with parallelism={PARALLELISM}") # Reading input data file from S3 df = pd.read_csv(INPUT_FILE) # Fixing dates that are not read with the correct type in a CSV file df['ACTIVITY_DATE'] = pd.to_datetime(df['ACTIVITY_DATE']) df['ACTIVITY_DATE'] = df['ACTIVITY_DATE'].dt.date logging.info(f"Min activity date: {df['ACTIVITY_DATE'].min()}") logging.info(f"Max activity date: {df['ACTIVITY_DATE'].max()}") # Using only a CHUNK of the data chunk_table = data_for_chunk(df, CHUNK_ID, CHUNK_COUNT) logging.info(f"Calculation chunk {CHUNK_ID} subset contains {chunk_table.shape[0]} rows.") # Compute FastFourierTransform and save the data to S3 fourier_df = compute_fourier_for_chunk(chunk_table, CHUNK_ID, DRY_RUN) logging.info(f"FFT table contains {fourier_df.shape[0]} rows.") # # Computing ARIMA for all filtered ids from CHUNK_ID # parallelism = PARALLELISM # timing, full_df, _ = take_isrcs_with_inflection_last_week(fourier_df) # logging.info(f"{full_df.shape[0]} rows loaded in {timing} seconds") # logging.info(f"Splitting work into {parallelism} parts for parallel processing") # elapsed_time = split_work(full_df, parallelism, work_load, DRY_RUN) # logging.info(f"Data processed in {elapsed_time} seconds")