Script 1373: [Trial] Anomaly with Summary and Links Core4 4

Purpose

The Python script identifies and analyzes anomalies in weekly campaign metrics to detect outliers and trends for performance optimization.
To Elaborate

The Python script is designed to identify and analyze anomalies in weekly campaign metrics for a retail enterprise. It focuses on detecting outliers in key performance metrics such as conversions, conversion rate, click-through rate, and impressions over the past eight weeks. The script uses statistical methods like Interquartile Range (IQR) and deviation thresholds to calculate anomalies. It allows for user-configurable parameters to adjust the sensitivity of anomaly detection. The script is intended to be run on weekdays and is designed to accommodate conversion lag in reporting. It filters campaigns based on high traffic and applies a minimum threshold to ensure only significant data is analyzed. The output includes a summary of campaigns with anomalous behavior, categorized by campaign type and vehicle brand, and provides insights into trends and performance changes.
Walking Through the Code

User Parameters Setup
- The script begins by defining user-configurable parameters, such as the metrics to include in the report, thresholds for outlier detection, and the lookback period for forecasts.
- Key parameters include REPORT_METRICS, TOP_CAMPAIGN_METRIC, FRACTION_OF_TOP_CAMPAIGNS_TO_INCLUDE, and MIN_THRESHOLD.
Data Initialization and Filtering
- The script checks if it is running on a server or locally, loading data accordingly.
- It filters the input data to focus on top campaigns based on a specified metric and applies a minimum threshold to ensure significant data is analyzed.
Metric Calculation and Aggregation
- The script calculates ratio metrics such as conversion rate and cost per acquisition.
- It aggregates data across dates for each campaign, computing forecast and actual values using exponential smoothing and interquartile ranges.
Anomaly Detection
- Anomaly scores are calculated for each metric using deviation ratios and outlier scores.
- The script flags metrics that exceed both deviation and outlier thresholds, identifying campaigns with significant anomalies.
Trend Analysis and Reporting
- The script analyzes trends for each campaign, categorizing them as positive, negative, or mixed based on deviation ratios.
- It prepares a summary report, highlighting campaigns with anomalous behavior and providing insights into performance trends.
Output Preparation
- The script prepares the output data for reporting, including deep links for further analysis.
- It generates a prompt for summarizing the weekly anomaly report, formatted in Markdown for easy interpretation.
Vitals

Script ID : 1373
Client ID / Customer ID: 1306928255 / 60270463
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-09-05 11:09
Last Updated by Grégory Pantaine on 2024-09-05 11:09
> See it in Action

Python Code

##
## Script name: [Demo] Weekly Campaign Anomaly with Summary and Links - Powpow Enterprise Retail
## Report name: [Trial] Scripts - Weekly Campaign Anomaly - last 8 weeks
## Runs: 11am on weekdays / include dates without metrics / show : weekly / past 8 weeks.
## 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
## created: 2024-05-31
## copied by Grég Pantaine on 5th September 2024
## 

RPT_COL_CAMPAIGN = 'Campaign'
RPT_COL_DATE = 'Date'
RPT_COL_PUBLISHER = 'Publisher'
RPT_COL_ACCOUNT = 'Account'
RPT_COL_VEHICLEBRAND = 'Vehicle Brand'
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 = 'Conv.'
RPT_COL_REVENUE = 'Revenue $'

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_VEHICLEBRAND
# COL_CATEGORY_2 = ""

# 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/powpow_weekly_anomaly_20240603.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(REPORT_METRICS.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:
    # set null to Unspecified Category 1
    inputDf_reduced[COL_CATEGORY_1] = inputDf_reduced[COL_CATEGORY_1].fillna('Unspecified')
else:
    # if column undefined, set to blank so section still appears but with no name
    inputDf_reduced[COL_CATEGORY_1] = ''

if COL_CATEGORY_2:
    # set null to Unspecified Category 2
    inputDf_reduced[COL_CATEGORY_2] = inputDf_reduced[COL_CATEGORY_2].fillna('Unspecified')
else:
    # if column undefined, set to blank to be dropped later. can't use np.nan since it gets replaced with zero
    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 = inputDf_reduced.loc[df_top_campaigns.index]
after_count = inputDf_reduced.shape[0]
# but if everything is filtered out, then remove filter
if after_count == 0:
    inputDf_reduced = inputDf.reset_index().set_index([RPT_COL_ACCOUNT, RPT_COL_CAMPAIGN])
    print(f"Warning: Filter criteria would remove ALL campaigns. Skipped filter. Using {inputDf_reduced.shape[0]} rows.")
else:
    print(f"Applying top campaign criteria reduced input row count from {before_count} to {after_count}")


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

df_campaign = inputDf_reduced \
                        .fillna(0) \
                        .replace([np.inf, -np.inf], 0) \
                        .groupby([RPT_COL_PUBLISHER, RPT_COL_ACCOUNT, RPT_COL_CAMPAIGN, COL_CATEGORY_1, COL_CATEGORY_2]) \
                        .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 = []

has_category1 = highlight_campaigns[COL_CATEGORY_1].dropna()
if len(has_category1) > 0:
    categories_1 = has_category1.value_counts().index.tolist()
    print(f"categories_1={COL_CATEGORY_1}, values:", categories_1)

has_category2 = highlight_campaigns[COL_CATEGORY_2].replace('', np.nan).dropna()
if (len(has_category2) > 0):
    categories_2 = has_category2.value_counts().index.tolist()
    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()],
        var_name=['metric' , 'variable'],
        value_name='value'
    )
    outlier_campaigns_pivoted = outlier_campaigns_melted.pivot_table(
        index = [(section,''), (RPT_COL_CAMPAIGN,''), 'metric'],
        columns='variable',
        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", metric_trend_for_category1)

metric_trend_for_category2 = find_outlier_trend_for_cohort(highlight_campaigns, COL_CATEGORY_2)
print("category2 trends", 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.

    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.

    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}
    {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: {', '.join(categories_1)}.
    * if available, also present results by '{COL_CATEGORY_2}', with values: {', '.join(categories_2)}.
    * 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}

    {prompt_category_1}

    # By {COL_CATEGORY_2}

    {prompt_category_2 if len(categories_2) > 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
mscripts_output_config  = str(config).replace("'", '"')

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
05 Sep 2024
« Script 1371: Top Kw's Clear Script 1377: Target CPA Cap $1000 »
MarinOne Scripts Creator's Corner
