import datetime as dt import logging import warnings import numpy as np import pandas as pd from pandas.errors import SettingWithCopyWarning from tadas.domain import constants logger = logging.getLogger(__name__) warnings.simplefilter(action='ignore', category=FutureWarning) warnings.simplefilter(action='ignore', category=RuntimeWarning) warnings.simplefilter(action='ignore', category=SettingWithCopyWarning) def prep_timeseries(timeseries_df): 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(constants.small_geos)] = timeseries_df[ 'prev4_avg'] + 5000 # for global (XX) we want upper_bound to be 15_000. As we just added 5_000, we add 10_000 more timeseries_df['upper_bound'][timeseries_df['geo_country'].isin(constants.global_geos)] = timeseries_df[ 'prev4_avg'] + 10000 # total 15_000 timeseries_df['upper_bound'][ timeseries_df['geo_country'].isin(list(set(constants.small_geos) - set(constants.smaller_geos)))] = timeseries_df[ 'prev4_avg'] + 2500 timeseries_df['upper_bound'][timeseries_df['geo_country'].isin(constants.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()) logger.info(f'--- --- 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)] logger.info(f'--- --- number of unique pfn geos with trend: {len(pfn_geo_trend)}') logger.info(f"--- --- {dt.datetime.now().strftime('%H:%M:%S')} : starting loops") mini_agg_df = pd.DataFrame() 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() logger.info(f"--- --- --- {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.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: logger.info(f'range: {loop_position} {loop_end}') # so, we'll say while len(one_track_df) > 0, keep looping. while loop_position < loop_end: if verbose: logger.info(f'this loop: {loop_position}') # first we'll write the section for if we are outside of a trend: if in_trend == 0: if verbose: logger.info('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: logger.info(f'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: logger.info('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: logger.info('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: if trend_too_long > 0: logger.info(f'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: logger.info(f'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: logger.info('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