Script 703: Scripts Campaign Anomaly Detection by Reporting Category V5

Purpose

Python script solves the problem of generating a performance anomaly report for pay-per-click marketing data.
To Elaborate

The Python script takes input data on various metrics such as conversions, revenue, publisher cost, and clicks, and calculates anomaly scores for each metric based on forecasted and actual values. It identifies outliers that may require attention and highlights the metrics with the highest anomaly scores. The script also provides summaries of notable anomalies for reporting categories and campaigns.
Walking Through the Code

The script begins by defining column constants and user-changeable parameters.
It then initializes the data source dictionary and sets up the timezone.
The input data is reduced and filtered based on the conversion lag and trailing 30-day spend.
Forecasts, interquartile ranges, and actual values are calculated for each metric using historical data.
Anomaly scores are calculated for each metric based on the deviation from the forecast and the interquartile range.
Outliers are flagged based on anomaly scores and deviation ratios.
The script identifies notable anomalies for reporting categories and campaigns based on the flagged outliers.
The results are formatted into a prompt string for generating a performance anomaly report.
The prompt string is printed or saved to a file for further use.
Note: The code related to column constants, client timezone, today, change comparison, and output has been excluded from the summary.
Vitals

Script ID : 703
Client ID / Customer ID: 1306912103 / 60267913
Action Type: Email Report
Item Changed: None
Output Columns:
Linked Datasource: M1 Report
Reference Datasource: None
Owner: Grégory Pantaine (gpantaine@marinsoftware.com)
Created by Grégory Pantaine on 2024-02-19 15:22
Last Updated by Grégory Pantaine on 2024-02-19 15:22
> See it in Action

Python Code

##
## name: Campaign Performance Anomaly Report with Summary
## description:
##  * Identify anomalies based on day-of-week forecasts
##  * Limited to top 80% campaigns by spend
##  * Adjustable sensitivity via IQR Weight
## 
## author: Michael S. Huang
## created: 2024-01-17
## copied from Yotel by G Pantaine
## copied on: 2024-02-06


RPT_COL_DATE = 'Date'
RPT_COL_REPCATEGORY = 'Reporting Category'
RPT_COL_ACCOUNT = 'Account'
RPT_COL_CAMPAIGN = 'Campaign'
RPT_COL_CAMPAIGN_TYPE = 'Campaign Type'
RPT_COL_CAMPAIGN_STATUS = 'Campaign Status'

RPT_COL_IMPR = 'Impr.'
RPT_COL_CLICKS = 'Clicks'
RPT_COL_PUB_COST = 'Pub. Cost £'

RPT_COL_CONV = 'NBO Purchases+ GA4'
RPT_COL_REVENUE = 'Revenues+ GA4 £'

RPT_COL_IMPRESSION_SHARE = 'Impr. share %'
RPT_COL_IMPRESSION_SHARE_TOP = 'Impr. Share (Top) %'
RPT_COL_LOST_IMPRESSION_SHARE_BUDGET = 'Lost Impr. Share (Budget) %'

COL_CONV_RATE = 'CVR'
COL_COST_PER_CLICK = 'CPC'
COL_CTR = 'CTR %'
COL_COST_PER_LEAD = 'CPL'
COL_SEARCHES = 'Searches'
COL_IMPRESSIONS_TOP = 'IMPRESSIONS_TOP'
COL_IMPRESSIONS_LOST_BUDGET = 'IMPRESSIONS_LOST_BUDGET'

# column names
COL_FORECAST = 'forecast'
COL_ACTUAL = 'actual'
COL_TRAILING = 'trailing'
COL_IQR = 'z_iqr'
COL_DEVIATION = 'z_deviation'
COL_DEVIATION_PCT = 'z_deviation_pct'

COL_DEVIATION_RATIO = 'z_deviation_ratio'
COL_DEVIATION_RATIO_FLAG_COUNT = COL_DEVIATION_RATIO + '_flagged'

COL_OUTLIER_SCORE = 'z_outlier_score'
COL_OUTLIER_SCORE_FLAG_COUNT = COL_OUTLIER_SCORE + '_flagged'

COL_OUTLIER_DEVIATION_FLAG_COUNT = 'z_outlier_deviation_flagged'
COL_OUTLIER_DEVIATION_FLAG_COUNT_SCALED = COL_OUTLIER_DEVIATION_FLAG_COUNT + '_scaled'

COL_MOST_UPWARD_OUTLIER_METRIC = 'z_most_upward_outlier_metric'
COL_MOST_DOWNWARD_OUTLIER_METRIC = 'z_most_downward_outlier_metric'

COL_TOTAL_FLAG_COUNT_SCALED = 'z_total_flag_count_scaled'
COL_TRAILING_COST = RPT_COL_PUB_COST + ' (Trailing)'

########### START - User Params ###########

FRACTION_OF_TRAILING_30_DAY_SPEND_TO_INCLUDE = 0.80
CONVERSION_LAG_DAYS = 1
MIN_FORECAST_LOOKBACK_WEEKS = 7

# Metrics to include in Report
# Format: (Metric, Outlier Threshold, Deviation Threshold)
# Metric = metrics to analyze
# Outlier Threshold: IQR multiplier; 1.5 is equivalent to 97.5% percentile
# Deviation Threshold: deviation threshold (in decimal; 0.20 = 20%)
REPORT_METRICS = [
    (RPT_COL_CLICKS,        1.5,    0.20),
    (RPT_COL_PUB_COST,      1.5,    0.20),
    (RPT_COL_CONV,          1.5,    0.20),
    (RPT_COL_REVENUE,       1.5,    0.20),
]

########### END - User Params ###########

########### START - Local Mode Config ###########
# Step 1: Uncomment download_preview_input flag and run Preview successfully with the Datasources you want
download_preview_input=False
# Step 2: In MarinOne, go to Scripts -> Preview -> Logs, download 'dataSourceDict' pickle file, and update pickle_path below
# pickle_path = ''
pickle_path = '/Users/mhuang/Downloads/pickle/yotel_summary_hotel_20240206_datasource_dict.pkl'
# Step 3: Copy this script into local IDE with Python virtual env loaded with pandas and numpy.
# Step 4: Run locally with below code to init dataSourceDict

# determine if code is running on server or locally
def is_executing_on_server():
    try:
        # Attempt to access a known restricted builtin
        dict_items = dataSourceDict.items()
        return True
    except NameError:
        # NameError: dataSourceDict object is missing (indicating not on server)
        return False

local_dev = False

if is_executing_on_server():
    print("Code is executing on server. Skip init.")
elif len(pickle_path) > 3:
    print("Code is NOT executing on server. Doing init.")
    local_dev = True
    # load dataSourceDict via pickled file
    import pickle
    dataSourceDict = pickle.load(open(pickle_path, 'rb'))

    # print shape and first 5 rows for each entry in dataSourceDict
    for key, value in dataSourceDict.items():
        print(f"Shape of dataSourceDict[{key}]: {value.shape}")
        # print(f"First 5 rows of dataSourceDict[{key}]:\n{value.head(5)}")

    # set outputDf same as inputDf
    inputDf = dataSourceDict["1"]
    outputDf = inputDf.copy()

    # setup timezone
    import datetime
    # Chicago Timezone is GMT-5. Adjust as needed.
    CLIENT_TIMEZONE = datetime.timezone(datetime.timedelta(hours=-5))

    # import pandas
    import pandas as pd
    import numpy as np

    # other imports
    import re
    import urllib

    # import Marin util functions
    # from marin_scripts_utils import tableize, select_changed

    # pandas settings
    pd.set_option('display.max_columns', None)  # Display all columns
    pd.set_option('display.max_colwidth', None)  # Display full content of each column

else:
    print("Running locally but no pickle path defined. dataSourceDict not loaded.")
    exit(1)
########### END - Local Mode Setup ###########


########### Anomaly Detection Libray Functions #############

### Forecast and Anomaly functions

# get forecast via exponential smoothing of previous weeks
def get_forecasts(data):
    if len(data) >= MIN_FORECAST_LOOKBACK_WEEKS * 7:
        # print("data len: ", len(data))
        # print("data.index", data.index)
        # print("data", data)
        index_previous_weeks = list(range(-8, -len(data)-1, -7))
        index_previous_weeks_ordered = index_previous_weeks[::-1]
        # print(index_previous_weeks_ordered)
        hist_data = data.iloc[index_previous_weeks_ordered]
        # forecasts = np.mean(hist_data, axis=0)

        # exponential smoothing
        alpha = 0.5 # smoothing factor
        forecasts = hist_data.ewm(alpha=alpha).mean().iloc[-1]

        return forecasts
    else:
        print("not enough data. skipping: ", data.index)
    
    return None

# get interquartile range from previous weeks
def get_inter_quartile_ranges(data):
    if len(data) >= MIN_FORECAST_LOOKBACK_WEEKS * 7:
        # print("data.index", data.index)
        # print("data", data)

        index_previous_weeks = list(range(-8, -len(data), -7))
        index_previous_weeks_ordered = index_previous_weeks[::-1]
        hist_data = data.iloc[index_previous_weeks_ordered]

        if np.isnan(hist_data).any():
            print("fixing nan in hist_data")
            hist_data = hist_data.fillna(0)
        

        # iqrs = np.std(hist_data, axis=0)

        # calculate interquartile range (IQR)
        Q1 = hist_data.quantile(0.25)
        Q3 = hist_data.quantile(0.75)
        IQR = Q3 - Q1

        return IQR
    else:
        print("not enough data. skipping: ", data.index)
    
    return None


# most recent data point is the last item
get_actuals = lambda x: x.iloc[[-1]]

# get trailing total
def get_trailing_total(data, window=7):
    if len(data) >= window:
        index_previous_days = list(range(-1, -(window+1), -1))
        hist_data = data.iloc[index_previous_days]
        return hist_data.sum()
    else:
        print("not enough data. skipping: ", data.index)
    
    return None

# ### Calculate anomaly score

# calc anomaly score across list of metrics
def calc_anomaly_scores(df, metrics_and_weights):
    df = df.copy()

    deviation_ratio_list = []
    outlier_score_list = []

    for (metric, _, _) in metrics_and_weights:
        forecast = df.loc[:, (metric, COL_FORECAST)]
        actual = df.loc[:, (metric, COL_ACTUAL)]
        iqr = df.loc[:, (metric, COL_IQR)]

        if np.isnan(forecast).any():
            print("nan in forecast")
            forecast = np.nan_to_num(forecast)
            
        if np.isnan(actual).any():
            print("nan in actual")
            actual = np.nan_to_num(actual)

        if np.isnan(iqr).any():
            print("nan in iqr")
            iqr = np.nan_to_num(iqr)

        # negative deviation when less than forecasted
        deviation = np.subtract(actual, forecast)
        
        df.loc[:, (metric, COL_DEVIATION)] = deviation

        # when both forecasted and actual values are ZERO, deviation should be ZERO
        # when forecasted is ZERO but actual is not, set to 50% deviation
        deviation_ratio = np.where(forecast > 0, \
                            deviation/forecast, \
                            np.where(actual > 0, 0.5, 0.0))
        deviation_ratio_list.append(deviation_ratio)
        df.loc[:, (metric, COL_DEVIATION_RATIO)] = deviation_ratio
        df.loc[:, (metric, COL_DEVIATION_PCT)] = np.char.add(np.char.mod('%0.0f', deviation_ratio * 100), '%')

        # anomaly score is ratio of deviation with Inter Quartile Range; score of 1.5 would be 97.5% percentile
        # positive score means exceeding forecast
        score = deviation / iqr
        outlier_score_list.append(score)
        df.loc[:, (metric, COL_OUTLIER_SCORE)] = score

    # flag scores that exceed the anomaly threshold
    anomaly_thresholds = np.array([threshold for (_, threshold, _) in metrics_and_weights])
    scores_stack = np.stack(outlier_score_list, axis=0)
    flagged_outlier_score_list = np.where(np.abs(scores_stack) > anomaly_thresholds[:, None], 1, 0)
    # sum across metrics and save for output
    df[COL_OUTLIER_SCORE_FLAG_COUNT] = np.sum(flagged_outlier_score_list, axis=0)

    # flag deviation ratios that exceed the deviation threshold
    deviation_thresholds = np.array([threshold for (_, _, threshold) in metrics_and_weights])
    deviation_ratios_stack = np.stack(deviation_ratio_list, axis=0)
    flagged_deviation_ratio_list = np.where(np.abs(deviation_ratios_stack) > deviation_thresholds[:, None], 1, 0)
    # sum across metrics and save for output
    df[COL_DEVIATION_RATIO_FLAG_COUNT] = np.sum(flagged_deviation_ratio_list, axis=0)

    # flag anomalous deviations by combining both flags above
    # AND flags in flagged_outlier_score_list and flagged_deviation_ratio_list
    # get the count of metrics where both are 1
    combined_flags = np.logical_and(flagged_outlier_score_list, flagged_deviation_ratio_list)
    df[COL_OUTLIER_DEVIATION_FLAG_COUNT] = np.sum(combined_flags, axis=0)

    # scaled score highlights larger spenders with more anomaly or deviation flags
    forecast_cost = df.loc[:, (RPT_COL_PUB_COST, COL_FORECAST)]
    actual_cost = df.loc[:, (RPT_COL_PUB_COST, COL_ACTUAL)]
    nominal_cost = np.maximum(forecast_cost, actual_cost)
    df[COL_TOTAL_FLAG_COUNT_SCALED] = np.round((df[COL_OUTLIER_SCORE_FLAG_COUNT] + df[COL_DEVIATION_RATIO_FLAG_COUNT]) * nominal_cost, 0)
    
    # another version that highlights larger spenders with anomalous deviation flags
    df[COL_OUTLIER_DEVIATION_FLAG_COUNT_SCALED] = np.round(df[COL_OUTLIER_DEVIATION_FLAG_COUNT] * nominal_cost, 0)


    # calc best & worst changes; scale change by outlier score
    change_score = np.multiply(deviation_ratio_list, np.abs(outlier_score_list))
    scores_stack = np.stack(change_score, axis=0)
    max_scores = np.maximum.reduce(scores_stack, axis=0)
    min_scores = np.minimum.reduce(scores_stack, axis=0)
    max_score_indices = np.argmax(scores_stack, axis=0)
    min_score_indices = np.argmin(scores_stack, axis=0)

    # fill in corresponding metric names
    metric_names = [metric for (metric, weight, _) in metrics_and_weights]
    df[COL_MOST_UPWARD_OUTLIER_METRIC] = [metric_names[idx] if score > 0 else '' for (score, idx) in zip(max_scores, max_score_indices)]
    df[COL_MOST_DOWNWARD_OUTLIER_METRIC] = [metric_names[idx] if score < 0 else '' for (score, idx) in zip(min_scores, min_score_indices)]

    # resort columns
    df.columns = df.columns.swaplevel(0, 1)
    df.sort_index(axis=1, inplace=True)
    df.columns = df.columns.swaplevel(1, 0)
    df.sort_index(axis=1, inplace=True)

    return df.sort_values(by=COL_OUTLIER_SCORE_FLAG_COUNT, axis=0, ascending=False)


def safe_ratio(numerator, denominator):
    
    return np.where(denominator > 0, \
                    numerator / denominator, \
                    0)


########### END Functions ###########


#### User Starts Here

print('inputDf.info\n',inputDf.info())

min_input_date = min(inputDf[RPT_COL_DATE])
max_input_date = max(inputDf[RPT_COL_DATE])
print(f"Input date range: {min_input_date.date()} to {max_input_date.date()}")

### Reduce columns and drop latest N days due to converion lag

report_date = (pd.to_datetime(max_input_date - datetime.timedelta(days=CONVERSION_LAG_DAYS)))
print(f"Most recent input date is {max_input_date.date()}. Using conversion lag of {CONVERSION_LAG_DAYS} days, set Report Date to {report_date.date()} and discard more recent dates.")

min_hist_date = min_input_date
max_hist_date = report_date - datetime.timedelta(days=1)
hist_days = (max_hist_date - min_hist_date).days + 1

print(f"Input has {hist_days} days of historical data from {min_hist_date.date()} to {max_hist_date.date()}")

inputDf_reduced = inputDf.loc[ inputDf[RPT_COL_DATE] <= report_date ] \
                         .drop([RPT_COL_CAMPAIGN_TYPE, RPT_COL_CAMPAIGN_STATUS], axis=1) \
                         .reset_index() \
                         .set_index([RPT_COL_REPCATEGORY, RPT_COL_CAMPAIGN])

print(f"Reduced from {inputDf.shape[0]} to {inputDf_reduced.shape[0]} rows")
print(f"Reduced input date range: {min(inputDf_reduced[RPT_COL_DATE]).date()} to {max(inputDf_reduced[RPT_COL_DATE]).date()}")

# calculate impression counts since impression share cannot be directly aggregated
inputDf_reduced[COL_SEARCHES] = inputDf_reduced[RPT_COL_IMPR] / inputDf_reduced[RPT_COL_IMPRESSION_SHARE]
inputDf_reduced[COL_IMPRESSIONS_TOP] = inputDf_reduced[COL_SEARCHES] * inputDf_reduced[RPT_COL_IMPRESSION_SHARE_TOP]
inputDf_reduced[COL_IMPRESSIONS_LOST_BUDGET] = inputDf_reduced[COL_SEARCHES] * inputDf_reduced[RPT_COL_LOST_IMPRESSION_SHARE_BUDGET]

# ### Keep only Top 80% Spenders (30-day lookback)

# # get trailing 30-day total spend by campaign

# agg_func = {
#       RPT_COL_PUB_COST: ['sum'],
# }

# thirty_days_ago = pd.to_datetime(max_input_date - datetime.timedelta(days=30))

# df_camp_agg = inputDf_reduced.loc[inputDf_reduced[RPT_COL_DATE] >= thirty_days_ago] \
#                               .groupby([RPT_COL_REPCATEGORY, RPT_COL_CAMPAIGN]) \
#                               .agg(agg_func) \
#                               .droplevel(1, axis=1) \
#                               .sort_values([RPT_COL_PUB_COST], ascending=False)

# total_spend = df_camp_agg[RPT_COL_PUB_COST].sum()

# COL_SPEND_CUMULATIVE = RPT_COL_PUB_COST+'_cumulative'
# COL_SPEND_CUMULATIVE_PCT = RPT_COL_PUB_COST+'_cumulative_pct'

# df_camp_agg[COL_SPEND_CUMULATIVE] = df_camp_agg[RPT_COL_PUB_COST].cumsum()
# df_camp_agg[COL_SPEND_CUMULATIVE_PCT] = df_camp_agg[COL_SPEND_CUMULATIVE] / total_spend

# top_cumulative_spend = total_spend * FRACTION_OF_TRAILING_30_DAY_SPEND_TO_INCLUDE
# df_top_spend_campaigns = df_camp_agg.loc[ df_camp_agg[COL_SPEND_CUMULATIVE] <= top_cumulative_spend ] \
#                                        .sort_values([RPT_COL_PUB_COST], ascending=False)

# print(f"Trailing 30-day Spend across all {df_camp_agg.shape[0]:,} campaigns is ${round(total_spend):,}. {FRACTION_OF_TRAILING_30_DAY_SPEND_TO_INCLUDE*100}% of it (${round(top_cumulative_spend):,}) comes from just {df_top_spend_campaigns.shape[0]} campaigns.")


# # actually filter by top campaigns
# before_count = inputDf_reduced.shape[0]
# inputDf_reduced = inputDf_reduced.loc[df_top_spend_campaigns.index]
# after_count = inputDf_reduced.shape[0]
# print(f"Taking top {FRACTION_OF_TRAILING_30_DAY_SPEND_TO_INCLUDE*100}% spend, further reduced from {before_count} to {after_count} rows")


### Prepare FORECAST and ACTUAL metrics

agg_func_selection = {
    RPT_COL_IMPR: ['sum'],
    RPT_COL_CLICKS: ['sum'],
    RPT_COL_PUB_COST: ['sum'],
    RPT_COL_CONV: ['sum'],
    RPT_COL_REVENUE: ['sum'],
    COL_SEARCHES: ['sum'],
    COL_IMPRESSIONS_TOP: ['sum'],
    COL_IMPRESSIONS_LOST_BUDGET: ['sum'],
}

df_agg_reporting_category_date = inputDf_reduced.groupby([RPT_COL_REPCATEGORY,RPT_COL_DATE]) \
                                            .agg(agg_func_selection) \
                                            .droplevel(1, axis=1)


df_agg_reporting_category_date[COL_CONV_RATE] = safe_ratio(df_agg_reporting_category_date[RPT_COL_CONV], df_agg_reporting_category_date[RPT_COL_CLICKS])
df_agg_reporting_category_date[COL_COST_PER_CLICK] = safe_ratio(df_agg_reporting_category_date[RPT_COL_PUB_COST], df_agg_reporting_category_date[RPT_COL_CLICKS])
df_agg_reporting_category_date[COL_CTR] = safe_ratio(df_agg_reporting_category_date[RPT_COL_CLICKS], df_agg_reporting_category_date[RPT_COL_IMPR])
df_agg_reporting_category_date[RPT_COL_IMPRESSION_SHARE] = safe_ratio(df_agg_reporting_category_date[RPT_COL_IMPR], df_agg_reporting_category_date[COL_SEARCHES])
df_agg_reporting_category_date[RPT_COL_IMPRESSION_SHARE_TOP] = safe_ratio(df_agg_reporting_category_date[COL_IMPRESSIONS_TOP], df_agg_reporting_category_date[COL_SEARCHES])
df_agg_reporting_category_date[RPT_COL_LOST_IMPRESSION_SHARE_BUDGET] = safe_ratio(df_agg_reporting_category_date[COL_IMPRESSIONS_LOST_BUDGET], df_agg_reporting_category_date[COL_SEARCHES])



### Use gropuby and agg to calculate forecast, iqr, and actual

df_reporting_category = df_agg_reporting_category_date \
     .fillna(0) \
     .replace([np.inf, -np.inf], 0) \
     .groupby([RPT_COL_REPCATEGORY]) \
     .agg({
        RPT_COL_REVENUE:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        RPT_COL_CONV:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        RPT_COL_CLICKS:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        COL_CONV_RATE:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        RPT_COL_PUB_COST:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals), (COL_TRAILING, get_trailing_total)],
        COL_COST_PER_CLICK:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        COL_CTR:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        RPT_COL_IMPR:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        COL_SEARCHES:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        RPT_COL_IMPRESSION_SHARE:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        RPT_COL_IMPRESSION_SHARE_TOP:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
        RPT_COL_LOST_IMPRESSION_SHARE_BUDGET:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
      })

# ### Do the same for Campaigns

# inputDf_reduced[COL_CONV_RATE] = safe_ratio(inputDf_reduced[RPT_COL_CONV], inputDf_reduced[RPT_COL_CLICKS])
# inputDf_reduced[COL_COST_PER_CLICK] = safe_ratio(inputDf_reduced[RPT_COL_PUB_COST], inputDf_reduced[RPT_COL_CLICKS])
# inputDf_reduced[COL_CTR] = safe_ratio(inputDf_reduced[RPT_COL_CLICKS], inputDf_reduced[RPT_COL_IMPR])


# df_campaign = inputDf_reduced \
#                         .fillna(0) \
#                         .replace([np.inf, -np.inf], 0) \
#                         .groupby([RPT_COL_REPCATEGORY,RPT_COL_ACCOUNT,RPT_COL_CAMPAIGN]) \
#                         .agg({ \
#                             RPT_COL_REVENUE:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             RPT_COL_CONV:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             RPT_COL_CLICKS:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             COL_CONV_RATE:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             RPT_COL_PUB_COST:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals), (COL_TRAILING, get_trailing_total)],
#                             COL_COST_PER_CLICK:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             COL_CTR:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             RPT_COL_IMPR:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             COL_SEARCHES:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             RPT_COL_IMPRESSION_SHARE:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             RPT_COL_IMPRESSION_SHARE_TOP:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                             RPT_COL_LOST_IMPRESSION_SHARE_BUDGET:[(COL_FORECAST, get_forecasts), (COL_IQR, get_inter_quartile_ranges), (COL_ACTUAL, get_actuals)],
#                         })

### Compute Anomaly Scores

df_reporting_category_anomaly = calc_anomaly_scores(df_reporting_category, REPORT_METRICS)

# df_campaign_anomaly = calc_anomaly_scores(df_campaign, REPORT_METRICS)


### Find Outlier Trafficking Accounts and Campaigns

highlight_reporting_category = df_reporting_category_anomaly.loc[(df_reporting_category_anomaly[(RPT_COL_PUB_COST, COL_ACTUAL)] > 0) & \
                                 (df_reporting_category_anomaly[COL_OUTLIER_DEVIATION_FLAG_COUNT] > 0)
                                ] \
                  .sort_values(by=[COL_OUTLIER_DEVIATION_FLAG_COUNT_SCALED], ascending=[False])

print("highlight_reporting_category: ", highlight_reporting_category.shape[0])

# highlight_campaigns = df_campaign_anomaly.loc[(df_campaign_anomaly[(RPT_COL_PUB_COST, COL_ACTUAL)] > 0) & \
#                                  (df_campaign_anomaly[COL_OUTLIER_DEVIATION_FLAG_COUNT] > 0)
#                                 ] \
#                   .sort_values(by=[COL_OUTLIER_DEVIATION_FLAG_COUNT_SCALED], ascending=[False])

# print("highlight_campaigns: ", highlight_campaigns.shape[0])

### Construct Complete Prompt

def get_anomaly_results_for_prompt_string(df):
    if df.empty:
        return pd.DataFrame()
    else:
        deviation_pct = df.xs(key=COL_DEVIATION_PCT, level=1, axis=1, drop_level=False).round(0)
        anomaly_score = df.xs(key=COL_OUTLIER_SCORE, level=1, axis=1, drop_level=False).round(2)
        trailing = df.xs(key=COL_TRAILING, level=1, axis=1, drop_level=False).round(0).astype(int)
        outlier_metrics = df[[COL_MOST_UPWARD_OUTLIER_METRIC, COL_MOST_DOWNWARD_OUTLIER_METRIC]]
        table = pd.concat([deviation_pct, anomaly_score, outlier_metrics, trailing], axis=1)
        # sorting removes level 1 col name for each column and confuses GPT
        # table = table.sort_index(axis=1, level=0)
        table = table.reset_index()
        # table.columns = ['{}_{}'.format(col[0], col[1]) if col[1] else col[0] for col in table.columns]

        return table.to_string(index=False, formatters={col: lambda x: f'"{x}"' for col in table.columns})

def get_anomaly_results_for_human_dataframe(df):
    if df.empty:
        return pd.DataFrame()
    else:
        deviation_pct = df.xs(key=COL_DEVIATION_PCT, level=1, axis=1, drop_level=False)
        forecast = df.xs(key=COL_FORECAST, level=1, axis=1, drop_level=False).round(2)
        trailing = df.xs(key=COL_TRAILING, level=1, axis=1, drop_level=False).round(0).astype(int)
        table = pd.concat([deviation_pct, forecast, trailing], axis=1)
        table = table.sort_index(axis=1, level=0).reset_index()
        table.columns = ['{}_{}'.format(col[0], col[1]) if col[1] else col[0] for col in table.columns]
        return table

prompt_header = f'''
You are a helpful pay-per-click marketing data analyst with deep understanding of common performance issues.

You are working with the output of a performance anomaly report.
Please summarize the results in a clear, easy to understand, and concise manner.
Make the report useful and insightful to read by using language from the hospitality sector while keeping the tone professional.
Please make sure the report is not alarming while still pointing out the anomalies.

The anomaly report examines these metrics (in order of importance): Conversions, Revenue, Publisher Cost, and Clicks.
Anomaly score (`{COL_OUTLIER_SCORE}`) of a metric is calclated by taking the difference between forecast and actual values (`{COL_DEVIATION}`), and divided by the Inter Quartile Range of historical data (`{COL_IQR}`).
Percentage change for each metric is in `{COL_DEVIATION_PCT}`.
Metric with anomaly score (`{COL_OUTLIER_SCORE}`) greater than 1.5 or less than -1.5 are outliers that may require attention, especially when percentage change `{COL_DEVIATION_PCT}` is also greater than 15% or less than -15%.
For these metrics, positive percentage change is good: Conversion Rate (CVR), Click Through Rate (CTR), Searches, Impression Share, Conversions (Conv), Revenue, Clicks
For these metrics, negative percentage change is good: Cost Per Click (CPC), Publisher Cost (Pub. Cost)
Don't use column names like `{COL_DEVIATION_PCT}` in the response.
Don't include anomaly scores in the response.
Highlight all metric names in bold.
Use the Account and Campaign DataFrame data (blocks surrounded by triple hyphens '---') to highlight issues and summarize trends.
Output at most 3 bullet points for each section, but review all the data given to analyze for trends.

Generate output in Markdown, using this format:

# Performance Anomaly Report for {report_date.date()}

'''

emailSummaryPrompt = f'''
{prompt_header}

## short headline of main trend. focus on metric with `{COL_OUTLIER_SCORE}` greater than 1.5 or less than -1.5 and `{COL_DEVIATION_PCT}` greater than 15% or less than -15%.


### Notable Reporting category

* provide the reporting_category name ('{RPT_COL_REPCATEGORY}') with trailing cost ('{COL_TRAILING_COST}' with currency symbol) in parenthesis and all in bold: __{RPT_COL_REPCATEGORY}__ (__{COL_TRAILING_COST}__
  * summary of metric anomalies, with highlights on metric listed under `{COL_MOST_UPWARD_OUTLIER_METRIC}` and `{COL_MOST_DOWNWARD_OUTLIER_METRIC}`. When highlighting an anomalous metric, use the form: __METRIC__ increased/decreased X% (`METRIC_{COL_DEVIATION_PCT}`) from the forecasted value Y (`METRIC_{COL_FORECAST}`).
  * EXAMPLE: Experienced a substantial increase of __Clicks__ by 60% (forecast: 134), while __Conversions__ decreased by 17% (forecast: 12). __Publisher Cost__ also rose by 53% (forecast: 220).


### Notable Campaigns

* provide the Campaign name ('{RPT_COL_CAMPAIGN}') with trailing cost ('{COL_TRAILING_COST}' with currency symbol) in parenthesis and all in bold: __{RPT_COL_CAMPAIGN}__ (__{COL_TRAILING_COST}__
* summary of metric anomalies, with highlights on metric listed under `{COL_MOST_UPWARD_OUTLIER_METRIC}` and `{COL_MOST_DOWNWARD_OUTLIER_METRIC}`. When highlighting an anomalous metric, use the form: __METRIC__ increased/decreased X% (`METRIC_{COL_DEVIATION_PCT}`) from the forecasted value Y (`METRIC_{COL_FORECAST}`).
  * EXAMPLE: Experienced a substantial increase of __Clicks__ by 60% (forecast: 134), while __Conversions__ decreased by 17% (forecast: 12). __Publisher Cost__ also rose by 53% (forecast: 220).

### Trends 
* short summary of positive trends with reporting_categorys. Highlight names in bold.
* short summary of negative trends with reporting_categorys. Highlight names in bold.
* short summary of positive trends with Campaign name. Highlight names in bold.
* short summary of negative trends with Campaign name. Highlight names in bold.

===

DataFrame of Accounts with at least one large anomalous deviation:
---
{get_anomaly_results_for_prompt_string(highlight_reporting_category.head(20))}
---


'''

# blank out prompt if there is no actual output
if highlight_reporting_category.empty:
    emailSummaryPrompt = ''

print(f"Prompt has ({len(emailSummaryPrompt)} chars)")

#### email output

# TODO: combine account and campaign into output
outputDf = get_anomaly_results_for_human_dataframe(highlight_reporting_category)

debugDf = highlight_reporting_category.reset_index().round(2)
debugDf.columns = ['{}_{}'.format(col[0], col[1]) if col[1] else col[0] for col in debugDf.columns]

print(f"OutputDf has {outputDf.shape[0]} rows")

## local debug
if local_dev:
    with open('prompt.txt', 'w') as file:
        file.write(emailSummaryPrompt)
        print(f"Local Dev: Prompt written to: {file.name}")

    debug_filename = 'outputDf.csv'
    outputDf.to_csv(debug_filename, index=False)
    print(f"Local Dev: Output written to: {debug_filename}")

    debug_filename = 'debugDf.csv'
    debugDf.to_csv(debug_filename, index=False)
    print(f"Local Dev: Debug written to: {debug_filename}")

else:
    print("====== Prompt =====")
    print(emailSummaryPrompt)
    print("===========")
Post generated on 2024-03-10 06:34:12 GMT
19 Feb 2024
« Script 701: SBA Auto Pause Script 707: Campaign Anomaly Detection »
MarinOne Scripts Creator's Corner
