Script 1421: Weekly Campaign Anomaly

Purpose

The Python script identifies and reports anomalies in weekly campaign performance metrics using configurable thresholds for outlier detection.
To Elaborate

The script is designed to analyze weekly campaign data to identify anomalies in performance metrics such as conversions, conversion rate, click-through rate, and impressions. It uses statistical methods like Interquartile Range (IQR) and deviation thresholds to detect outliers. The script is configured to look back over the last eight weeks by default, but this can be adjusted. It focuses on high-traffic campaigns by filtering based on a specified metric and includes only the top campaigns that contribute significantly to the overall metric. The script calculates forecasts using exponential smoothing and compares actual performance against these forecasts to determine anomalies. It also considers conversion lag when scheduling reports. The output includes a summary of campaigns with anomalous metrics, categorized by campaign type and other optional categories, providing insights into trends and performance issues.
Walking Through the Code

Initialization and Configuration
- The script begins by defining constants for column names and user-configurable parameters, such as metrics to report, thresholds for outlier detection, and the lookback period for forecasts.
- User parameters include the mapping of required columns, metrics to include in the report, and thresholds for filtering top campaigns.
Data Preparation
- The script loads data, either from a server or locally using a pickle file, and prepares it by resetting indices and filling missing values.
- It calculates additional metrics like conversion rate, cost per click, and click-through rate.
Filtering and Aggregation
- The script filters data to focus on top campaigns based on a specified metric and minimum threshold.
- It aggregates data by campaign and calculates forecast and actual values for each metric using defined functions.
Anomaly Detection
- Anomaly scores are calculated for each metric using deviation ratios and outlier scores.
- The script flags metrics as anomalous if they exceed both deviation and outlier thresholds.
Trend Analysis and Reporting
- The script analyzes trends for each campaign and category, determining whether they are positive, negative, or mixed.
- It prepares data for output, including a summary of anomalous campaigns and trends, formatted for reporting.
Output and Debugging
- The script generates a prompt for summarizing the report and prepares data for output, including deep links for further analysis.
- In local development mode, it writes the prompt and output data to files for debugging purposes.
Vitals

Script ID : 1421
Client ID / Customer ID: 1306914662 / 3614566
Action Type: Email Report
Item Changed: None
Output Columns:
Linked Datasource: M1 Report
Reference Datasource: None
Owner: Anton Antonov (aantonov@marinsoftware.com)
Created by Anton Antonov on 2024-10-02 14:37
Last Updated by Anton Antonov on 2024-10-02 14:42
> See it in Action

Python Code

##
## name: Weekly Campaign Anomaly Report with Summary
## description:
##  * identify outliers for configured metrics
##  * calculate anomaly via adjustable IQR and Deviation Thresholds
##  * by default, expects data from last 8 weeks. dial down MIN_FORECAST_LOOKBACK_WEEKS if need shorter lookback.
##  * report should be scheduled to accomodate Conversion Lag
##
##
## author: Michael S. Huang and Anton Antonov
## created: 2024-09-27
## 

RPT_COL_CAMPAIGN = 'Campaign'
RPT_COL_DATE = 'Date'
RPT_COL_PUBLISHER = 'Publisher'
RPT_COL_ACCOUNT = 'Account'
RPT_COL_CAMPAIGN_STATUS = 'Campaign Status'
RPT_COL_CAMPAIGN_TYPE = 'Campaign Type'
RPT_COL_IMPR = 'Impr.'
RPT_COL_CLICKS = 'Clicks'
RPT_COL_PUB_COST = 'Pub. Cost €'
RPT_COL_CONV = 'Conv.'
RPT_COL_REVENUE = 'Revenue €'
RPT_COL_ROAS = 'ROAS'
RPT_COL_MSFT_PURCHASE_CONV = 'Purchase Conv.'
RPT_COL_MSFT_PURCHASE_REVENUE = 'Purchase Revenue €'
RPT_COL_MSFT_PURCHASE_ROAS = 'Purchase ROAS'
RPT_COL_BIG_EVENT = 'Event'

COL_CONV_RATE = 'CVR'
COL_COST_PER_CLICK = 'CPC'
COL_CTR = 'CTR %'
COL_COST_PER_ACQUISITION = 'CPA'

# 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_ANOMALY_FLAG = 'z_is_anomaly'

COL_OUTLIER_DEVIATION_FLAG_COUNT = 'zz_outlier_deviation_flagged'
COL_OVERALL_WEIGHTED_OUTLIER_SCORE = 'zz_overall_outlier_score'

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

COL_TREND = 'Trend'

## NB. Row key columns must be 'mscripts_row_key' for backend to recognize
COL_ROW_KEY = 'mscripts_row_key'

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

# map required columns to actual report columns
COL_CONVERSIONS = RPT_COL_CONV
COL_CATEGORY_1 = RPT_COL_CAMPAIGN_TYPE
# COL_CATEGORY_2 = RPT_COL_COUNTRY
COL_CATEGORY_2 = None

# Metrics to include in Report.
# Order: first metric is the main focus of the report, and metrics that follow contribute exponentially less.
# Format: 
#   Metric: {
#     'alias': short name for use in Summary
#     'outlier_threshold': IQR multiplier; 1.5 is equivalent to 97.5% percentile,
#     'deviation_threshold': deviation threshold (in decimal; 0.20 = 20%),
#     'forecast_precision': number of decimal places; 0 for integer
#   }
REPORT_METRICS = {
    RPT_COL_CONV: {
        'alias': 'Conversions',
        'outlier_threshold': 1.5,
        'deviation_threshold': 0.20,
        'forecast_precision': 0,
    },
    COL_CONV_RATE: {
        'alias': 'Conv Rate',
        'outlier_threshold': 1.5,
        'deviation_threshold': 0.20,
        'forecast_precision': 1,
    },
    COL_CTR: {
        'alias': 'CTR',
        'outlier_threshold': 1.5,
        'deviation_threshold': 0.20,
        'forecast_precision': 1,
    },
    RPT_COL_IMPR: {
        'alias': 'Impressions',
        'outlier_threshold': 1.5,
        'deviation_threshold': 0.20,
        'forecast_precision': 0,
    }
}

# Campaign Grid View to deep link to
CAMPAIGN_GRID_VIEW_ID = 14971665

# Only focus on high traffic campaigns
# caveat: can't use Pub Cost/Revenue for Cross Client reports due to lack of currency conversion
TOP_CAMPAIGN_METRIC = RPT_COL_PUB_COST
FRACTION_OF_TOP_CAMPAIGNS_TO_INCLUDE = 0.90
MIN_THRESHOLD_METRIC = COL_CONVERSIONS
MIN_THRESHOLD = 10

# lookback window for forecast
MIN_FORECAST_LOOKBACK_WEEKS = 7

########### 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/mass_general_weekly_anomaly_20240926.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 simple exponential smoothing
# note: changed to handle weekly not daily data
def get_forecasts(data, decimals=0):
    if len(data) >= MIN_FORECAST_LOOKBACK_WEEKS:
        # print("data len: ", len(data))
        # print("data.index", data.index)
        # print("data", data)

        # exclude most recent data point
        hist_data = data.iloc[:-1]

        # exponential smoothing
        alpha = 0.5 # smoothing factor
        forecasts = hist_data.ewm(alpha=alpha).mean().iloc[-1]
        if decimals == 0:
            forecasts = forecasts.round(0).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:
        # print("data.index", data.index)
        # print("data", data)

        # exclude most recent data point
        hist_data = data.iloc[:-1]

        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=4, decimals=0):
    if len(data) >= window:
        index_previous_periods = list(range(-1, -(window+1), -1))
        hist_data = data.iloc[index_previous_periods]

        if decimals == 0:
            total = hist_data.sum().round(0).astype(int)
        else:
            total = hist_data.sum().round(decimals)

        return total 
    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_dict):
    df = df.copy()

    deviation_ratios = {}
    outlier_scores = {}

    for metric, params in metrics_dict.items():
        forecast = df.loc[:, (metric, COL_FORECAST)].fillna(0)
        actual = df.loc[:, (metric, COL_ACTUAL)].fillna(0)
        iqr = df.loc[:, (metric, COL_IQR)].fillna(0)


        # 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))
        
        # save deviation ratio for each metric for post-loop processing
        deviation_ratios[metric] = 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)

        # save outlier_scores for each metric for post-loop processing
        outlier_scores[metric] = score
        
        df.loc[:, (metric, COL_OUTLIER_SCORE)] = score

        # flag metric if both deviation and outlier score exceed threshold
        outlier_threshold = params['outlier_threshold']
        deviation_ratio_threshold = params['deviation_threshold']
        anomaly = (np.abs(score) > outlier_threshold) & (np.abs(deviation_ratio) > deviation_ratio_threshold)
        df.loc[anomaly, (metric, COL_ANOMALY_FLAG)] = True


    # get count of metrics that are anomalous (outlier deviations)
    df[COL_OUTLIER_DEVIATION_FLAG_COUNT] = df.xs(COL_ANOMALY_FLAG, level=1, axis=1).sum(axis=1)

    ## overall weighted anomaly score:
    #  - heavily weights primary metric (first position in REPORT_METRICS)
    #  - scaled by larger spenders using clicks

    # Calculate the sum of weighted deviations for each metric using natural log for decay
    weighted_deviations = []
    for i, metric in enumerate(metrics_dict.keys()):
        # don't allow cancellation, so take abs
        deviation = df.loc[:, (metric, COL_DEVIATION_RATIO)].abs()
        decay_weight = np.exp(-i)  # Decay weight using natural log directly
        weighted_deviations.append(deviation * decay_weight)

    # Sum the weighted deviations
    total_weighted_deviations = sum(weighted_deviations)

    # use trailing clicks as proxy metrics, since this script runs across many currencies and don't want to currency conversion here
    trailing_clicks = df.loc[:, (RPT_COL_CLICKS, COL_TRAILING)]

    # Calculate the overall weighted outlier score
    df[COL_OVERALL_WEIGHTED_OUTLIER_SCORE] = total_weighted_deviations * trailing_clicks
    

    ## 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_scores = {}
    for metric in metrics_dict.keys():
        deviation_ratio = np.nan_to_num(deviation_ratios[metric], copy=False)
        outlier_score = np.nan_to_num(outlier_scores[metric], copy=False)
        change_scores[metric] = np.multiply(deviation_ratio, np.abs(outlier_score))

    # Initialize arrays to store max and min scores and their corresponding metrics
    max_scores = np.full(df.shape[0], -np.inf)
    min_scores = np.full(df.shape[0], np.inf)
    max_score_metrics = np.full(df.shape[0], '', dtype=str)
    min_score_metrics = np.full(df.shape[0], '', dtype=str)

    # Iterate over each metric and update the max and min scores
    for metric, scores in change_scores.items():
        max_mask = scores > max_scores
        min_mask = scores < min_scores

        max_scores = np.where(max_mask, scores, max_scores)
        min_scores = np.where(min_mask, scores, min_scores)

        max_score_metrics = np.where(max_mask, metric, max_score_metrics)
        min_score_metrics = np.where(min_mask, metric, min_score_metrics)

    # Assign the most upward and downward outlier metrics
    df[COL_MOST_UPWARD_OUTLIER_METRIC] = [metric if score > 0 else np.nan for score, metric in zip(max_scores, max_score_metrics)]
    df[COL_MOST_DOWNWARD_OUTLIER_METRIC] = [metric if score < 0 else np.nan for score, metric in zip(min_scores, min_score_metrics)]

    ## 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)

    # sort results with largest spender with outlier being first
    return df.sort_values(by=COL_OVERALL_WEIGHTED_OUTLIER_SCORE, 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 Code 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()}")

# calculate report coverage date

report_date_start = (pd.to_datetime(max_input_date))
report_date_end = report_date_start + pd.Timedelta(days=6)

print(f"Most recent input date is {max_input_date.date()}. Report Week set to: {report_date_start.date()} to {report_date_end.date()}.")

inputDf_reduced = inputDf \
                         .reset_index() \
                         .set_index([RPT_COL_ACCOUNT, RPT_COL_CAMPAIGN])

if COL_CATEGORY_1 is not None and len(COL_CATEGORY_1) > 1:
    # turn nulls into Unspecified 
    inputDf_reduced[COL_CATEGORY_1] = inputDf_reduced[COL_CATEGORY_1].fillna('Unspecified')
else:
    # create placeholder column with empty string
    # don't use nan since would be filled with zeros later
    COL_CATEGORY_1 = 'Category1'
    inputDf_reduced[COL_CATEGORY_1] = ''

if COL_CATEGORY_2 is not None and len(COL_CATEGORY_2) > 1:
    # turn nulls into Unspecified 
    inputDf_reduced[COL_CATEGORY_2] = inputDf_reduced[COL_CATEGORY_2].fillna('Unspecified')
else:
    # create placeholder column with empty string
    # don't use nan since would be filled with zeros later
    COL_CATEGORY_2 = 'Category2'
    inputDf_reduced[COL_CATEGORY_2] = ''

### Keep only Top Campaigns via configured metric TOP_CAMPAIGN_METRIC (30-day lookback)

# get trailing 30-day total by campaign

agg_func = {
		TOP_CAMPAIGN_METRIC: ['sum'],
        MIN_THRESHOLD_METRIC: ['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_ACCOUNT, RPT_COL_CAMPAIGN]) \
                              .agg(agg_func) \
                              .droplevel(1, axis=1) \
                              .sort_values([TOP_CAMPAIGN_METRIC], ascending=False)

metric_subtotal = df_camp_agg[TOP_CAMPAIGN_METRIC].sum()

COL_TOP_CAMPAIGN_METRIC_CUMULATIVE = TOP_CAMPAIGN_METRIC+'_cumulative'
COL_TOP_CAMPAIGN_METRIC_CUMULATIVE_PCT = TOP_CAMPAIGN_METRIC+'_cumulative_pct'

df_camp_agg[COL_TOP_CAMPAIGN_METRIC_CUMULATIVE] = df_camp_agg[TOP_CAMPAIGN_METRIC].cumsum()
df_camp_agg[COL_TOP_CAMPAIGN_METRIC_CUMULATIVE_PCT] = df_camp_agg[COL_TOP_CAMPAIGN_METRIC_CUMULATIVE] / metric_subtotal

top_campaign_metric_cutoff = metric_subtotal * FRACTION_OF_TOP_CAMPAIGNS_TO_INCLUDE
df_top_campaigns = df_camp_agg.loc[ df_camp_agg[COL_TOP_CAMPAIGN_METRIC_CUMULATIVE] <= top_campaign_metric_cutoff ] \
                                       .sort_values([TOP_CAMPAIGN_METRIC], ascending=False)

print(f"For metric '{TOP_CAMPAIGN_METRIC}', trailing 30-day sub-total across all {df_camp_agg.shape[0]:,} campaigns is {round(metric_subtotal):,}. {FRACTION_OF_TOP_CAMPAIGNS_TO_INCLUDE*100}% of it ({round(top_campaign_metric_cutoff):,}) comes from just {df_top_campaigns.shape[0]} campaigns.")

# apply minimum threshold
before_count = df_top_campaigns.shape[0]
df_top_campaigns = df_top_campaigns.loc[df_top_campaigns[MIN_THRESHOLD_METRIC] > MIN_THRESHOLD]
after_count = df_top_campaigns.shape[0]
print(f"Applying min thres of {MIN_THRESHOLD} to {MIN_THRESHOLD_METRIC} would trim off {before_count-after_count} campaigns")


# actually filter by top campaigns
before_count = inputDf_reduced.shape[0]
inputDf_reduced_filtered = inputDf_reduced.loc[df_top_campaigns.index]
after_count = inputDf_reduced_filtered.shape[0]
# only apply filter if there is still data left
if after_count > 0:
    inputDf_reduced = inputDf_reduced_filtered
    print(f"Applying top campaign criteria reduced input row count from {before_count} to {after_count}")
else:
    print(f"Warning: Filter criteria would remove ALL campaigns. Skipped filter. Still using {inputDf_reduced.shape[0]} rows.")


# compute ratio metrics for each date

inputDf_reduced[COL_CONV_RATE] = safe_percentage(inputDf_reduced[COL_CONVERSIONS], 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_COST_PER_ACQUISITION] = inputDf_reduced[RPT_COL_PUB_COST] / inputDf_reduced[COL_CONVERSIONS]
inputDf_reduced[COL_CTR] = safe_percentage(inputDf_reduced[RPT_COL_CLICKS], inputDf_reduced[RPT_COL_IMPR])


### Aggregate across Dates for Campaigns; calculate Forecast & Actual values
agg_spec = {
    metric: [
        (COL_FORECAST, lambda x, d=params['forecast_precision']: get_forecasts(x, d)),
        (COL_IQR, get_inter_quartile_ranges),
        (COL_ACTUAL, get_actuals)
    ] 
    for metric, params in REPORT_METRICS.items()
}

# Now, add the additional keys for RPT_COL_PUB_COST and RPT_COL_CLICKS
agg_spec[RPT_COL_PUB_COST] = agg_spec.get(RPT_COL_PUB_COST, []) + [(COL_TRAILING, get_trailing_total)]
agg_spec[RPT_COL_CLICKS] = agg_spec.get(RPT_COL_CLICKS, []) + [(COL_TRAILING, get_trailing_total)]

# safe to always groupby both Category columns, since placeholder columns are created if they are not available
group_by_columns = [RPT_COL_PUBLISHER, RPT_COL_ACCOUNT, RPT_COL_CAMPAIGN, COL_CATEGORY_1, COL_CATEGORY_2]

df_campaign = inputDf_reduced \
                        .fillna(0) \
                        .replace([np.inf, -np.inf], 0) \
                        .groupby(group_by_columns) \
                        .agg(agg_spec)

### Compute Anomaly Scores

df_campaign_anomaly = calc_anomaly_scores(df_campaign, REPORT_METRICS)


### Find Outlier Trafficking Accounts and Campaigns

# Get the first metric key from REPORT_METRICS
first_metric_key = list(REPORT_METRICS.keys())[0]

# Only highlight campaigns where the primary metric is anomalous
highlight_campaigns = df_campaign_anomaly.loc[ \
                                              df_campaign_anomaly[(first_metric_key, COL_ANOMALY_FLAG)] == True
                                             ] \
                                         .sort_values(by=[COL_OVERALL_WEIGHTED_OUTLIER_SCORE], ascending=[False]) \
                                         .reset_index()
print("highlight_campaigns: ", highlight_campaigns.shape[0])

### Add some metadata

# add row key for Deep Links
highlight_campaigns[COL_ROW_KEY] = ['row_' + str(i) + '_key' for i in range(1, len(highlight_campaigns) + 1)]

# get list of Category1 and Category2 for building prompt
categories_1 = []
categories_2 = []

# get count of non-empty values for Category 1
has_category1 = highlight_campaigns[COL_CATEGORY_1].replace('', np.nan).dropna()
if len(has_category1) > 0:
    categories_1 = has_category1.value_counts().index.tolist()
else:
    categories_1 = []
print(f"categories_1={COL_CATEGORY_1}, values:", categories_1)

# get count of non-empty values for Category 2
has_category2 = highlight_campaigns[COL_CATEGORY_2].replace('', np.nan).dropna()
if (len(has_category2) > 0):
    categories_2 = has_category2.value_counts().index.tolist()
else:
    categories_2 = []
print(f"categories_2={COL_CATEGORY_2}, values:", categories_2)

# Deterine whether deviation for given metric is net Positive or Negative
# Return value is a signed decimal where sign indicates Positive/Negative and value indicates magnitude
def get_trend_direction(metric, deviation_ratio):
    # default is Positive trend
    dir = 1
    
    # except for Cost and CPA
    if metric in [RPT_COL_PUB_COST, COL_COST_PER_ACQUISITION]:
        dir = -1

    # flip the sign if deviation is negative
    if deviation_ratio < 0:
        dir *= -1
    
    # use square(deviation) to weight so larger deviation trump smaller ones
    return dir * np.abs(deviation_ratio)

def get_trend(row):
    trend_sum = 0

    metrics_list = list(REPORT_METRICS.keys())
    for i, metric in enumerate(metrics_list):
        deviation_ratio = row[(metric, COL_DEVIATION_RATIO)]
        # Exponential decay factor, heavier weight for earlier metrics
        weight = np.exp(-i)
        trend_sum += get_trend_direction(metric, deviation_ratio) * weight

    threshold = list(REPORT_METRICS.values())[0]['deviation_threshold']

    if trend_sum > threshold:
        return 'Positive'
    elif trend_sum < -1*threshold:
        return 'Negative'
    
    return 'Mixed'

# set Trend for each campaign
highlight_campaigns[COL_TREND] = highlight_campaigns.apply(get_trend, axis=1)

# get aggregate count
category1_trend_count = highlight_campaigns[[COL_CATEGORY_1, COL_TREND]]
category1_trend_count.columns = category1_trend_count.columns.get_level_values(0)
category1_trend_count = category1_trend_count.value_counts().reset_index(name='count')
category1_trend_count = category1_trend_count.sort_values(by=[COL_CATEGORY_1, COL_TREND], ascending=False)

category2_trend_count = highlight_campaigns[[COL_CATEGORY_2, COL_TREND]]
category2_trend_count.columns = category2_trend_count.columns.get_level_values(0)
category2_trend_count = category2_trend_count.value_counts().reset_index(name='count')
category2_trend_count = category2_trend_count.sort_values(by=[COL_CATEGORY_2, COL_TREND], ascending=False)


### Determine the Metric Trends for each section
# - for each campaign, flag each metric that is anomalous (exceed both thresholds)
# - pivot dataframe so each anomalous metric has its own row
# - call get_trend as usual to get trend direction
# - then do value_count

def find_outlier_trend_for_cohort(highlight_campaigns, section):

    if highlight_campaigns.empty:
        return pd.DataFrame(columns=[section, 'metric', 'count', 'trend'])

    # Filter campaigns that have outliers
    outlier_campaigns = highlight_campaigns[highlight_campaigns[COL_OUTLIER_DEVIATION_FLAG_COUNT] > 0]

    ## Pivot dataframe so each metric has its own row
    outlier_campaigns_melted = outlier_campaigns.melt(
        id_vars = [(section,''), (RPT_COL_CAMPAIGN, '')],
        value_vars = [(col, COL_DEVIATION_RATIO) for col in REPORT_METRICS.keys()] + [(col, COL_ANOMALY_FLAG) for col in REPORT_METRICS.keys()],
    )

    outlier_campaigns_pivoted = outlier_campaigns_melted.pivot_table(
        index = [(section,''), (RPT_COL_CAMPAIGN,''), 'variable_0'],
        columns='variable_1',
        values='value',
        aggfunc='first'
    ).reset_index()
    outlier_campaigns_pivoted.columns = [section, RPT_COL_CAMPAIGN, 'metric', COL_DEVIATION_RATIO, COL_ANOMALY_FLAG]

    # Filter out rows where COL_ANOMALY_FLAG is NaN (i.e., non-anomalous metrics)
    outlier_campaigns_pivoted = outlier_campaigns_pivoted.dropna(subset=[COL_ANOMALY_FLAG])

    ## Determine trend direction for each row

    # use direction with magnitude to break ties when there are equal counts in same direction
    outlier_campaigns_pivoted['trend_direction_with_magnitude'] = outlier_campaigns_pivoted.apply(
        lambda row: get_trend_direction(row['metric'], row[COL_DEVIATION_RATIO]), axis=1
    )

    # for directional count
    outlier_campaigns_pivoted['trend_direction'] = outlier_campaigns_pivoted['trend_direction_with_magnitude'].apply(lambda x: 1 if x > 0 else -1 if x < 0 else 0)

    # Group by section, metric and count the occurrence of trend_direction values while summing trend_direction_with_magnitude
    trend_count = outlier_campaigns_pivoted.groupby([section, 'metric', 'trend_direction']).agg(
        count=('trend_direction', 'count'),
        sum_magnitude_abs=('trend_direction_with_magnitude', lambda x: x.abs().sum())
    ).reset_index()

    # Map trend direction to human-readable format
    trend_count['trend'] = trend_count['trend_direction'].map({1: 'Positive', -1: 'Negative', 0: 'Mixed'})

    # sort by absolute value and only take top trend
    trend_count = trend_count.sort_values(by=[section, 'count', 'sum_magnitude_abs'], ascending=[True, False, False])

    print(f"trend count for section: {section}", trend_count) 
   
    return trend_count.drop(columns=['trend_direction','sum_magnitude_abs'])


metric_trend_for_category1 = find_outlier_trend_for_cohort(highlight_campaigns, COL_CATEGORY_1)
print("category1 trends\n", metric_trend_for_category1)

metric_trend_for_category2 = find_outlier_trend_for_cohort(highlight_campaigns, COL_CATEGORY_2)
print("category2 trends\n", metric_trend_for_category2)



### Construct Complete Prompt

def convert_dataframe_to_formatted_string(df):

    if df.empty or len(df) < 1:
        return ""

    # add quotes around column names and values
    quoted_column_names = ['"{}"'.format(col) for col in df.columns]
    output_str = df.to_string(index=False, header=quoted_column_names, formatters={col: lambda x: f'"{x}"' for col in df.columns})

    return output_str

def prepare_dataframe_for_output(df):
    if df.empty:
        return pd.DataFrame()
    else:
        primary_data = df[[COL_CATEGORY_1, COL_TREND, RPT_COL_CAMPAIGN]]

        metric_cols = [(metric, sub_metric) for metric in REPORT_METRICS.keys() for sub_metric in [COL_DEVIATION_PCT, COL_FORECAST]]
        metric_data = df.loc[:, metric_cols]
       
        trailing_data = df.loc[:, [(RPT_COL_PUB_COST, COL_TRAILING), (RPT_COL_CLICKS, COL_TRAILING)]]
        
        other_data = df[[RPT_COL_PUBLISHER, RPT_COL_ACCOUNT, COL_CATEGORY_2, COL_ROW_KEY]]
       
        output_df = pd.concat([primary_data, metric_data, trailing_data, other_data], axis=1)

        output_df.columns = ['{}_{}'.format(col[0], col[1]) if col[1] else col[0] for col in output_df.columns]

        output_df = output_df.sort_values(by=[COL_CATEGORY_1, COL_TREND]) \
                             .reset_index(drop=True)

        return output_df
    
## Build Prompt

# blank out prompt in case there is no actual output
emailSummaryPrompt = ''
if highlight_campaigns.empty:
    # do nothing
    print("No output. Skip building prompt")
else:

    prompt_category_1 = "\n".join(
    f'''
## write a succint headline ending with an action verb that highlights the metrics and trends with high counts in "Metric Trend for {category_1} category" DataFrame. Write headline only without name of Dataframe.

EXAMPLE: ## {category_1} {list(REPORT_METRICS.values())[0]['alias']} Lifts with {list(REPORT_METRICS.values())[3]['alias']} Soaring

use "Trend Count for {category_1} category" DataFrame to determine the number of paragraphs for each trend.
for each trend paragraph, provide the trend count.
for each trend paragraph, using bullet points, summarize the performance anomaly for at most 3 campaigns with that trend in the "{category_1} category" dataframe.
each bullet point should start with the campaign name ('{RPT_COL_CAMPAIGN}') as link to row key ('{COL_ROW_KEY}') and bolded trailing cost ('{COL_TRAILING_COST}') with currency symbol).
include actual values when explaining percentage deviation ('METRIC_{COL_DEVIATION_PCT}') from baseline value ('METRIC_{COL_FORECAST}').

EXAMPLE:

3 campaigns doing better. Examples:
* [My Campaign 1](COL_ROW_KEY) (__£1,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial increase of __+60%__ from baseline of 1,345, __{list(REPORT_METRICS.values())[1]['alias']}__ lifted __+17%__ from projection of 3.1, __{list(REPORT_METRICS.values())[2]['alias']}__ dipped slightly __-2%__ from projection of 5, and __{list(REPORT_METRICS.values())[3]['alias']}__ was on par with forecast of 1,000.
* [My Campaign 2](COL_ROW_KEY) (__£2,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial lift of __+30%__ from baseline of 2,345, and __{list(REPORT_METRICS.values())[3]['alias']}__ improved __+7%__ from projection of 5.1.

1 campaign facing declines:
* [My Campaign 3](COL_ROW_KEY) (__£3,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial increase of __+40%__ from baseline of 3,345, __{list(REPORT_METRICS.values())[1]['alias']}__ lifted __+37%__ from projection of 4.1, __{list(REPORT_METRICS.values())[2]['alias']}__ dipped slightly __-3%__ from projection of 6, and __{list(REPORT_METRICS.values())[3]['alias']}__ was on par with forecast of 3,500.

2 campaigns with mixed trend. Example:
* [My Campaign 4](COL_ROW_KEY) (__£4,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial increase of __+20%__ from baseline of 4,000, but __{list(REPORT_METRICS.values())[1]['alias']}__ dropped __-15%__ from projection of 3.7.
    '''
        for category_1 in categories_1
    )


    dataframe_category_1 = "\n\n".join(
    f'''
DataFrame of "Trend Count for {category_1} category":
[[[
{convert_dataframe_to_formatted_string(category1_trend_count.loc[category1_trend_count[COL_CATEGORY_1] == category_1])}
]]]

DataFrame of "Metric Trend for {category_1} category":
[[[
{convert_dataframe_to_formatted_string(metric_trend_for_category1.loc[metric_trend_for_category1[COL_CATEGORY_1] == category_1])}
]]]

DataFrame of "{category_1} category" campaigns:
[[[
{convert_dataframe_to_formatted_string(prepare_dataframe_for_output(highlight_campaigns.loc[highlight_campaigns[COL_CATEGORY_1] == category_1].head(20)))}
]]]
    '''
        for category_1 in categories_1
    )



    if len(categories_2) > 0:
        prompt_category_2 = "\n".join(
    f'''
## write a succint headline ending with an action verb that highlights the metrics and trends with high counts in "Metric Trend for {category_2} category2" DataFrame.  Write headline only without name of Dataframe.

EXAMPLE: ## {category_2} {list(REPORT_METRICS.values())[0]['alias']} Lifts with {list(REPORT_METRICS.values())[3]['alias']} Jumping

use "Trend Count for {category_2} category" DataFrame to determine the number of paragraphs for each trend.
use the full category2 name instead of ISO category2 code. example: use Japan instead of JP, France instead of FR, Germany instead of DE.
for each trend paragraph, provide the trend count.
for each trend paragraph, using bullet points, summarize the performance anomaly for at most 3 campaigns with that trend in the "{category_2} category" dataframe.
each bullet point should start with the campaign name ('{RPT_COL_CAMPAIGN}') as link to row key ('{COL_ROW_KEY}') and bolded trailing cost ('{COL_TRAILING_COST}') with currency symbol).
include actual values when explaining percentage deviation ('METRIC_{COL_DEVIATION_PCT}') from baseline value ('METRIC_{COL_FORECAST}').

EXAMPLE:

3 campaigns doing really well. Examples:
* [My Campaign 1](COL_ROW_KEY) (__£1,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial increase of __+60%__ from baseline of 1,345, __{list(REPORT_METRICS.values())[1]['alias']}__ lifted __+17%__ from projection of 3.1, __{list(REPORT_METRICS.values())[2]['alias']}__ dipped slightly __-2%__ from projection of 5, and __{list(REPORT_METRICS.values())[3]['alias']}__ matched forecast of 1,000.
* [My Campaign 2](COL_ROW_KEY) (__£2,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial lift of __+30%__ from baseline of 2,345, and __{list(REPORT_METRICS.values())[3]['alias']}__ improved __+7%__ from projection of 5.1.

5 campaigns worst than before. Examples:
* [My Campaign 3](COL_ROW_KEY) (__£1,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial drop of __-30%__ from baseline of 3,345, and __{list(REPORT_METRICS.values())[2]['alias']}__ tanked __-27%__ from projection of 4.1.
* [My Campaign 4](COL_ROW_KEY) (__£5,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial increase of __+40%__ from baseline of 4,345, __{list(REPORT_METRICS.values())[1]['alias']}__ lifted __+37%__ from projection of 5.1, __{list(REPORT_METRICS.values())[2]['alias']}__ dipped slightly __-7%__ from projection of 12, and __{list(REPORT_METRICS.values())[3]['alias']}__ tracked forecast of 5,500.

1 campaign with mixed trend:
* [My Campaign 5](COL_ROW_KEY) (__£4,439__): __{list(REPORT_METRICS.values())[0]['alias']}__ experienced a substantial increase of __+20%__ from baseline of 1,545, __{list(REPORT_METRICS.values())[1]['alias']}__ lifted __+17%__ from projection of 3.1, __{list(REPORT_METRICS.values())[2]['alias']}__ dipped slightly __-2%__ from projection of 5, and __{list(REPORT_METRICS.values())[3]['alias']}__ was on par with forecast of 1,000.
    '''
            for category_2 in categories_2
        )

    dataframe_category_2 = "\n\n".join(
    f'''
DataFrame of "Trend Count for {category_2} category2":
[[[
{convert_dataframe_to_formatted_string(category2_trend_count.loc[category2_trend_count[COL_CATEGORY_2] == category_2])}
]]]

DataFrame of "Metric Trend for {category_2} category2":
[[[
{convert_dataframe_to_formatted_string(metric_trend_for_category2.loc[metric_trend_for_category2[COL_CATEGORY_2] == category_2])}
]]]

DataFrame of "{category_2} category2" campaigns:
[[[
{convert_dataframe_to_formatted_string(prepare_dataframe_for_output(highlight_campaigns.loc[highlight_campaigns[COL_CATEGORY_2] == category_2].head(20)))}
]]]
    '''
        for category_2 in categories_2
    )

    alias_instruction = "\n".join(
    f'''* Refer to '{metric}' as '{params['alias']}', except when referenced within backticks (`)'''
        for metric, params in REPORT_METRICS.items()
    )


    forecast_instruction = "\n".join(
    f"""* '{metric + '_' + COL_FORECAST}' is the baseline value for '{metric}' """
        for metric in REPORT_METRICS.keys()
    )

    deviation_instruction = "\n".join(
    f"""* '{metric + '_' + COL_DEVIATION_PCT}' is the percent deviation from baseline value for '{metric}' """
        for metric in REPORT_METRICS.keys()
    )

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

{dataframe_category_1 if len(categories_1) > 0 else ''}
{dataframe_category_2 if len(categories_2) > 0 else ''}

Interpret above data using these guidelines:
{alias_instruction}
{forecast_instruction}
{deviation_instruction}
* present results grouped by '{COL_CATEGORY_1}', with values: {categories_1 if len(has_category1) > 0 else ''}.
* if available, also present results by '{COL_CATEGORY_2}', with values: {categories_2 if len(has_category2) > 0 else ''}.
* include all listed metrics, in the order provided, when explaining performance: {', '.join(REPORT_METRICS.keys())}


Please summarize the weekly anomaly report using a professional tone.
Generate output in Markdown, using this format:

# Campaigns with Anomalous {list(REPORT_METRICS.values())[0]['alias']} for Week of {report_date_start.strftime('%b %d, %Y')}

Note: Weekly metrics are from {report_date_start.strftime('%b %d, %Y')} to {report_date_end.strftime('%b %d, %Y')}, inclusive. Changes are compared to a recent 4-week baseline. Metrics scanned: ___{', '.join([metric['alias'] for metric in REPORT_METRICS.values()])}___.

{('# By ' + COL_CATEGORY_1) if len(has_category1) > 0 else ''}
{prompt_category_1 if len(has_category1) > 0 else ''}

{('# By ' + COL_CATEGORY_2) if len(has_category2) > 0 else ''}
{prompt_category_2 if len(has_category2) > 0 else ''}

    '''

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

#### email output

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

### debug output

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]


#### config deep links

config = {
    "email_marinone_links": [
        {
            "index": index + 1,
            "view_id": CAMPAIGN_GRID_VIEW_ID,
            "filters": [f"campaign_name:{row[RPT_COL_CAMPAIGN]}", f"pca_alias:{row[RPT_COL_ACCOUNT]}"]
        } for index, row in outputDf.iterrows()
    ]
} if not outputDf.empty else {}

# Create the final JSON string by combining all rows
import json
mscripts_output_config = json.dumps(config)

print(f"mscript_output_config: {mscripts_output_config}")

### 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("===========")
Post generated on 2024-11-27 06:58:46 GMT
02 Oct 2024
« Script 1417: Weekly Campaign Anomaly Report with Summary Delvaie Sciences Script 1423: Weekly Campaign Anomaly Sky Betting »
MarinOne Scripts Creator's Corner
