# imports import pandas as pd import numpy as np pd.options.mode.chained_assignment = None pd.set_option('display.float_format', lambda x: '%.5f' % x) import datetime as dt import warnings warnings.simplefilter(action='ignore', category=FutureWarning) def prep_timeseries(timeseries_df, small_geos, smaller_geos): timeseries_df['report_date'] = pd.to_datetime(timeseries_df['report_date']) timeseries_df.sort_values(by=['pfn_geo', 'report_date'], inplace=True) timeseries_df = timeseries_df.reset_index(drop=True) timeseries_df['day_of_week'] = timeseries_df['report_date'].dt.day_name() timeseries_df['month_of_year'] = pd.DatetimeIndex(timeseries_df['report_date']).month timeseries_df['date_rank'] = timeseries_df.groupby(['pfn_geo'])['report_date'].rank().astype(int) timeseries_df['prev_7'] = timeseries_df['streams'].shift(periods = 7) timeseries_df['prev_14'] = timeseries_df['streams'].shift(periods = 14) timeseries_df['prev_21'] = timeseries_df['streams'].shift(periods = 21) timeseries_df['prev_28'] = timeseries_df['streams'].shift(periods = 28) timeseries_df['prev4_avg'] = (4 * timeseries_df['prev_7'] + 3 * timeseries_df['prev_14'] + 2 * timeseries_df['prev_21'] + timeseries_df['prev_28']) / 10 columns_to_calculate = ['prev_7', 'prev_14', 'prev_21', 'prev_28'] timeseries_df['prev4_stdev'] = timeseries_df[columns_to_calculate].std(axis=1) timeseries_df['upper_bound'] = pd.NA timeseries_df['upper_bound'][~timeseries_df['geo_country'].isin(small_geos)] = timeseries_df['prev4_avg'] + 7500 timeseries_df['upper_bound'][timeseries_df['geo_country'].isin(list(set(small_geos) - set(smaller_geos)))] = timeseries_df['prev4_avg'] + 2500 timeseries_df['upper_bound'][timeseries_df['geo_country'].isin(smaller_geos)] = timeseries_df['prev4_avg'] + 1000 timeseries_df['upper_bound'][timeseries_df['month_of_year'] == 1] = timeseries_df['upper_bound'] * 1.15 timeseries_df['upper_bound'][timeseries_df['month_of_year'] == 12] = timeseries_df['upper_bound'] * 1.15 timeseries_df['combined_forecast'] = timeseries_df['upper_bound'] timeseries_df['combined_forecast'][timeseries_df['date_rank'] < 28.0] = np.nan return timeseries_df def calculate_consecutive(timeseries_df, consecutive_days = 0, failed_days = 0): trending_data = list(timeseries_df['is_trending']) consecutive_list = [] fail_list = [] for x in trending_data: if x == 1: consecutive_days += 1 consecutive_list.append(consecutive_days) failed_days = 0 fail_list.append(0) else: if failed_days == 0: consecutive_list.append(consecutive_days) else: consecutive_list.append(0) consecutive_days = 0 failed_days += 1 fail_list.append(failed_days) timeseries_df['consecutive_trend'] = consecutive_list timeseries_df['consecutive_fails'] = fail_list return timeseries_df def iron_out_trends(timeseries_df, verbose=False): # set the basic version of these with bubbles in the contact paper. timeseries_df['is_trending'] = 0 timeseries_df['is_trending'][timeseries_df['streams'].astype(float) > timeseries_df['combined_forecast']] = 1 ironed_out_df = pd.DataFrame() trend_birth_length = 2 trend_death_length = 2 trend_too_long_length = 60 tracks = list(timeseries_df['pfn_geo'].unique()) print('--- --- number of unique pfn geos:', len(tracks)) #we split the df into tracks that dont need to loop (no trend) and tracks that do need to loop grouped_timeseries_df = timeseries_df.groupby('pfn_geo')['is_trending'].sum().reset_index() timeseries_df_no_trend = grouped_timeseries_df.loc[grouped_timeseries_df['is_trending'] == 0] timeseries_df_trend = grouped_timeseries_df.loc[grouped_timeseries_df['is_trending'] != 0] # Get lists of pfn_geo values pfn_geo_trend = timeseries_df_trend['pfn_geo'].tolist() pfn_geo_no_trend = timeseries_df_no_trend['pfn_geo'].tolist() # Filter the original timeseries_df to only include rows with pfn_geo in pfn_geo_trend timeseries_df_with_trend = timeseries_df[timeseries_df['pfn_geo'].isin(pfn_geo_trend)] timeseries_df_with_trend.sort_values(by=['pfn_geo', 'report_date'], inplace=True) timeseries_df_with_trend = timeseries_df_with_trend.reset_index(drop=True) # Filter the original timeseries_df to only include rows with pfn_geo in pfn_geo_no_trend timeseries_df_without_trend = timeseries_df[timeseries_df['pfn_geo'].isin(pfn_geo_no_trend)] print('--- --- number of unique pfn geos with trend:', len(pfn_geo_trend)) print('--- ---', dt.datetime.now().strftime("%H:%M:%S"), ': starting loops') for x in range(len(pfn_geo_trend)): if x % 1000 == 0: if x != 0: ironed_out_df = pd.concat([ironed_out_df, mini_agg_df], ignore_index=False) mini_agg_df = pd.DataFrame() print('--- --- --- ', dt.datetime.now().strftime("%H:%M:%S"), ': finished', x) #we would change this timeseries df to just one geo df one_track_df = timeseries_df_with_trend[timeseries_df_with_trend['pfn_geo'] == pfn_geo_trend[x]] one_track_df['locked_daily'] = 0 one_track_df['in_trend'] = 0 in_trend = 0 loop_position = one_track_df.index.min() loop_end = one_track_df.index.max() + 1 if verbose == True: print('range:', loop_position, loop_end) # so, we'll say while len(one_track_df) > 0, keep looping. while loop_position < loop_end: if verbose == True: print('this loop:', loop_position) # first we'll write the section for if we are outside of a trend: if in_trend == 0: if verbose == True: print('not trending') one_track_df['is_trending'][(one_track_df['streams'] < one_track_df['combined_forecast']) & (one_track_df.index >= loop_position)] = 0 one_track_df['is_trending'][(one_track_df['streams'] >= one_track_df['combined_forecast']) & (one_track_df.index >= loop_position)] = 1 one_track_df = calculate_consecutive(one_track_df) trend_born = one_track_df.index[(one_track_df['consecutive_trend'] == trend_birth_length) & (one_track_df.index > loop_position)].min() if trend_born > 0: if verbose == True: print('trend_born :', trend_born) # first lets make the locked daily. start_idx = max(0, trend_born - 7) upper_bound_data = one_track_df.loc[start_idx:trend_born - 1, ['day_of_week', 'upper_bound']] result_dict = upper_bound_data.groupby('day_of_week')['upper_bound'].mean().to_dict() one_track_df['locked_daily'][one_track_df.index >= trend_born - trend_birth_length] = one_track_df['day_of_week'].map(result_dict) one_track_df['combined_forecast'][one_track_df.index >= trend_born - trend_birth_length] = one_track_df['locked_daily'] one_track_df['in_trend'][(one_track_df.index > trend_born - trend_birth_length)] = 1 in_trend = 1 loop_position = trend_born else: if verbose == True: print('currently not trending') one_track_df['in_trend'][one_track_df.index >= loop_position] = 0 mini_agg_df = pd.concat([mini_agg_df, one_track_df], ignore_index=False) one_track_df = pd.DataFrame() loop_position = loop_end else: if verbose == True: print('in trend') # now we need to set the is_trending to assess against locked_daily. one_track_df['is_trending'][(one_track_df['streams'] < one_track_df['combined_forecast']) & (one_track_df.index >= loop_position)] = 0 one_track_df['is_trending'][(one_track_df['streams'] >= one_track_df['combined_forecast']) & (one_track_df.index >= loop_position)] = 1 one_track_df = calculate_consecutive(one_track_df) # then we look for the first 3 trend_death = one_track_df.index[(one_track_df['consecutive_fails'] == trend_death_length) & (one_track_df.index > loop_position)].min() trend_too_long = one_track_df.index[(one_track_df['consecutive_trend'] >= trend_too_long_length) & (one_track_df.index > loop_position)].min() if verbose == True: if trend_too_long > 0: print('found trend_too_long:', trend_too_long) if trend_too_long < trend_death: trend_death = trend_too_long * 1 if np.isnan(trend_death): if trend_too_long > 0: trend_death = trend_too_long * 1 if trend_death > 0: if verbose == True: print('found trend death:', trend_death) one_track_df['in_trend'][one_track_df.index >= trend_death - trend_death_length] = 0 one_track_df['combined_forecast'][one_track_df.index > trend_death] = one_track_df['upper_bound'] # then we set in_trend = 1 in_trend = 0 loop_position = trend_death else: if verbose == True: print('still trending') one_track_df['in_trend'][one_track_df.index >= loop_position] = 1 mini_agg_df = pd.concat([mini_agg_df, one_track_df], ignore_index=False) one_track_df = pd.DataFrame() loop_position = loop_end # last one, for the unfinshed 1000. ironed_out_df = pd.concat([ironed_out_df, mini_agg_df], ignore_index=False) #here we would also append final_ironed_out_df = pd.concat([ironed_out_df, timeseries_df_without_trend], ignore_index=False) #fillna with zeros after for ironed out df final_ironed_out_df.fillna(0, inplace=True) return final_ironed_out_df