Script 1105: Campaign Anomaly Detection Daxko

Purpose

The Python script detects anomalies in campaign performance metrics by comparing actual values against forecasted values using statistical methods.

To Elaborate

The script is designed to identify anomalies in campaign performance metrics by analyzing historical data and comparing it with current performance. It uses statistical methods such as exponential smoothing to forecast expected values for various metrics like conversions, clicks, and costs. The script then calculates deviations from these forecasts and determines if they are significant enough to be considered anomalies. Anomalies are flagged based on user-defined thresholds for interquartile range (IQR) and deviation percentages. The script is particularly useful for pay-per-click marketing data analysts who need to monitor campaign performance and identify any unusual patterns or trends that may require attention. The analysis is performed on a day-of-week basis, allowing for more accurate forecasting and anomaly detection.

Walking Through the Code

User Parameters and Setup
- The script begins by defining user-changeable parameters such as CONVERSION_LAG_DAYS and MIN_FORECAST_LOOKBACK_WEEKS, which control the data analysis window and forecast lookback period.
- Metrics to be analyzed are specified in REPORT_METRICS, where each metric is associated with an outlier threshold and a deviation threshold.
Data Loading and Initialization
- The script checks if it is running on a server or locally. If local, it loads data from a specified pickle file.
- It initializes the data by setting up the input DataFrame and configuring necessary libraries like pandas and numpy.
Forecasting and Anomaly Detection Functions
- Functions such as get_forecasts and get_inter_quartile_ranges are defined to calculate forecasted values and IQRs for the metrics.
- The calc_anomaly_scores function computes deviation ratios and anomaly scores for each metric, flagging those that exceed user-defined thresholds.
Data Processing and Aggregation
- The script processes the input data to calculate additional metrics like conversion rate and cost per click.
- It uses groupby and aggregation functions to compute forecasts, IQRs, and actual values for each campaign.
Anomaly Scoring and Reporting
- Anomaly scores are calculated, and campaigns with significant deviations are highlighted.
- The script generates a summary report, which includes notable campaigns and trends, formatted in Markdown for easy readability.
Output and Debugging
- The results are saved to CSV files for local debugging, and a prompt is generated for further analysis or reporting.

Vitals

Script ID : 1105
Client ID / Customer ID: 1306927457 / 60270313
Action Type: Email Report
Item Changed: None
Output Columns:
Linked Datasource: M1 Report
Reference Datasource: None
Owner: Kent Pearce (kpearce@marinsoftware.com)
Created by Kent Pearce on 2024-05-15 22:53
Last Updated by Kent Pearce on 2024-05-21 15:28

> See it in Action

Python Code

##
## name: Campaign Performance Anomaly Report with Summary
## description:
##  * Identify anomalies based on day-of-week forecast
##  * Calculate anomaly via adjustable IQR and Deviation Thresholds
## 
## author: Dana Waidhas
## created: 2024-04-01
## 

RPT_COL_DATE = 'Date'
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 = 'Pub. Conv.'
RPT_COL_REVENUE = 'Pub. Revenue'

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 = '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 = 'zz_outlier_deviation_flagged'
COL_OUTLIER_DEVIATION_FLAG_COUNT_SCALED = COL_OUTLIER_DEVIATION_FLAG_COUNT + '_scaled'

COL_MOST_UPWARD_OUTLIER_METRIC = 'zz_most_upward_outlier_metric'
COL_MOST_DOWNWARD_OUTLIER_METRIC = 'zz_most_downward_outlier_metric'

COL_TOTAL_FLAG_COUNT_SCALED = 'zz_total_flag_count_scaled'
COL_TRAILING_COST = RPT_COL_PUB_COST + '_' + COL_TRAILING

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

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_CONV,          1.5,    0.20),
    (RPT_COL_PUB_COST,      1.5,    0.20),
    (RPT_COL_CLICKS,        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/universitysouthwest_campaign_anomaly.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, decimals=0):
    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]
        if decimals == 0:
            forecasts = forecasts.astype(int)
        else:
            forecasts = forecasts.round(decimals)

        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 100% deviation
        deviation_ratio = np.where(forecast > 0, \
                            deviation/forecast, \
                            np.where(actual > 0, 1.0, 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
        # if IQR is 0, default anomaly score to 0 so it won't trigger any alerts
        score = np.where(abs(iqr) > 0, deviation / iqr, 0.0)

        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 (use absolute value to avoid change sign of deviation)
    # Using numpy.nan_to_num to fill in NA
    change_score = np.nan_to_num(np.multiply(deviation_ratio_list, np.abs(outlier_score_list)), copy=False)
    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 everything for debugging
    # return (df.sort_values(by=COL_OUTLIER_SCORE_FLAG_COUNT, axis=0, ascending=False), max_scores, min_scores, max_score_indices, min_score_indices)
    return df.sort_values(by=COL_OUTLIER_SCORE_FLAG_COUNT, axis=0, ascending=False)


# convert to percentage units with 2 decimal places
def safe_percentage(numerator, denominator):
    return np.where(denominator > 0, \
                    round(numerator / denominator * 100, 2), \
                    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_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] =  np.where(inputDf_reduced[RPT_COL_IMPRESSION_SHARE] > 0, \
                    np.round(inputDf_reduced[RPT_COL_IMPR] / inputDf_reduced[RPT_COL_IMPRESSION_SHARE], 0), \
                    inputDf_reduced[RPT_COL_IMPR])
inputDf_reduced[COL_IMPRESSIONS_TOP] = np.round(inputDf_reduced[COL_SEARCHES] * inputDf_reduced[RPT_COL_IMPRESSION_SHARE_TOP], 0)
inputDf_reduced[COL_IMPRESSIONS_LOST_BUDGET] = np.round(inputDf_reduced[COL_SEARCHES] * inputDf_reduced[RPT_COL_LOST_IMPRESSION_SHARE_BUDGET], 0)


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

inputDf_reduced[COL_CONV_RATE] = safe_percentage(inputDf_reduced[RPT_COL_CONV], inputDf_reduced[RPT_COL_CLICKS])
inputDf_reduced[COL_COST_PER_CLICK] = inputDf_reduced[RPT_COL_PUB_COST]/inputDf_reduced[RPT_COL_CLICKS]
inputDf_reduced[COL_CTR] = safe_percentage(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_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_campaign_anomaly = calc_anomaly_scores(df_campaign, REPORT_METRICS)


### Find Outlier Trafficking Campaigns

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=[(RPT_COL_PUB_COST, COL_TRAILING)], 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:
        trailing = df.xs(key=COL_TRAILING, level=1, axis=1, drop_level=False).round(0).astype(int)
        forecast = df.xs(key=COL_FORECAST, level=1, axis=1, drop_level=False).round(2)
        deviation_pct = df.xs(key=COL_DEVIATION_PCT, level=1, axis=1, drop_level=False)
        anomaly_score = df.xs(key=COL_OUTLIER_SCORE, level=1, axis=1, drop_level=False).round(2)
        outlier_metrics = df[[COL_MOST_UPWARD_OUTLIER_METRIC, COL_MOST_DOWNWARD_OUTLIER_METRIC]]
        table = pd.concat([trailing, forecast, deviation_pct, anomaly_score, outlier_metrics], 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)
        table = pd.concat([deviation_pct, forecast], axis=1)
        table = table.sort_index(axis=1, level=0)
        trailing = df.xs(key=COL_TRAILING, level=1, axis=1, drop_level=False).round(0).astype(int)
        table = pd.concat([trailing, table], axis=1)
        table = table.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 higher education sector while keeping the tone professional.
Please make sure the report is not alarming while still pointing out the anomalies.

The anomaly report examines this list of METRIC (from most important to least important): {', '.join(["'" + m[0] + "'" for m in REPORT_METRICS])}.
Anomaly score (`METRIC_{COL_OUTLIER_SCORE}`) of a metric is calclated by taking the difference (`METRIC_{COL_DEVIATION}`) between forecast and actual values, and divided by the Inter Quartile Range of historical data.
Forecasted value for each metric is in `METRIC_{COL_FORECAST}`.
For each metric, deviation percentage from forecast is calculated in `METRIC_{COL_DEVIATION_PCT}`.
Metric with anomaly score (`METRIC_{COL_OUTLIER_SCORE}`) greater than 1.5 or less than -1.5 are outliers that may require attention, especially when percentage change `METRIC_{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 `METRIC_{COL_DEVIATION_PCT}` in the response.
Don't include anomaly scores in the response.
State that a metric that increased/decreased certain percentage (`METRIC_{COL_DEVIATION_PCT}`) from the forecasted value (`METRIC_{COL_FORECAST}`) when highlighting an anomalous metric.
Highlight all METRIC names in bold using Markdown `__` notation.
Use the Account and Campaign DataFrame data (blocks surrounded by triple hyphens '---') to highlight issues and summarize trends.
Output at most 5 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 `METRIC_{COL_OUTLIER_SCORE}` greater than 1.5 or less than -1.5 and `METRIC_{COL_DEVIATION_PCT}` greater than 15% or less than -15%.


### 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 Campaigns. Highlight names in bold.
* short summary of negative trends with Campaigns. Highlight names in bold.

===

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


'''

# blank out prompt if there is no actual output
if highlight_campaigns.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_campaigns)

debugDf = highlight_campaigns.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}")

    output_filename = 'outputDf.csv'
    outputDf.to_csv(output_filename, index=False)
    print(f"Local Dev: Output written to: {output_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("===========")
##
## name: Campaigns Anomaly Detection
## description:
##  
## 
## author: 
## created: 2024-05-15
## 

today = datetime.datetime.now(CLIENT_TIMEZONE).date()

# primary data source and columns
inputDf = dataSourceDict["1"]

# output columns and initial values

# user code start here
print(tableize(inputDf.head()))

Post generated on 2024-11-27 06:58:46 GMT

15 May 2024

« Script 1103: Script Group Country Code Script 1111: Campaign Benchmarks General »

MarinOne Scripts Creator's Corner