Script 249: Title Strategy Assignment
Purpose:
The Python script assigns a specific campaign strategy and publisher strategy based on the current date matching the “CPA Date” in the data.
To Elaborate
The Python script is designed to automate the assignment of campaign strategies for marketing campaigns. Specifically, it checks if the “CPA Date” in the campaign data matches the current date. If a match is found, the script updates the campaign strategy to “TITLE | AUTOMATION | CPA” and sets the publisher strategy to “ManualCpc” while clearing the “Bidding Movement” field. This ensures that campaigns are automatically adjusted to a predefined strategy when they reach a specific date, facilitating structured budget allocation and strategic planning. The script also includes functionality to detect changes in the strategy assignments and outputs the modified data for further processing or reporting.
Walking Through the Code
- Initialization and Setup:
- The script begins by defining constants for column names used in the input and output dataframes.
- It sets up temporary columns in the input dataframe to hold potential changes in strategy and bidding movement.
- Logic Application:
- The script checks each row in the input dataframe to see if the “CPA Date” matches the current date.
- If a match is found, it assigns the predefined values for strategy, publisher bid strategy, and bidding movement to the temporary columns.
- Change Detection:
- The script identifies rows where the temporary columns differ from the original columns, indicating a change in strategy.
- It constructs an output dataframe containing only the changed rows, renaming the temporary columns to match the output column names.
- Testing and Execution:
- A unit test function is included to verify the script’s functionality using sample data.
- The script processes the input dataframe and prints the final output, showcasing the updated campaign strategies.
Vitals
- Script ID : 249
- Client ID / Customer ID: 1306913736 / 50395
- Action Type: Bulk Upload (Preview)
- Item Changed: Campaign
- Output Columns: Account, Campaign, Strategy, Publisher Bid Strategy, Bidding Movement
- Linked Datasource: M1 Report
- Reference Datasource: None
- Owner: Michael Huang (mhuang@marinsoftware.com)
- Created by Michael Huang on 2023-07-23 17:17
- Last Updated by Jeremy Brown on 2023-12-06 04:01
> See it in Action
Python Code
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
#
# Auto Strategy Assignment
# When "CPA Date" is today:
# - Assign campaign strategy to "TITLE | CPA $"
# - set Publisher Strategy to manualCpc and "Bidding Movement" to blank
#
# Author: Michael S. Huang
# Created: 2023-07-24
#
RPT_COL_CAMPAIGN = 'Campaign'
RPT_COL_ACCOUNT = 'Account'
RPT_COL_PUBLISHER_BIDSTRATEGY = 'Publisher Bid Strategy'
RPT_COL_BIDDING_MOVEMENT = 'Bidding Movement'
RPT_COL_CPA_DATE = 'CPA Date'
RPT_COL_STRATEGY = 'Strategy'
BULK_COL_ACCOUNT = 'Account'
BULK_COL_CAMPAIGN = 'Campaign'
BULK_COL_STRATEGY = 'Strategy'
BULK_COL_PUBLISHER_BIDSTRATEGY = 'Publisher Bid Strategy'
BULK_COL_BIDDING_MOVEMENT = 'Bidding Movement'
outputDf[BULK_COL_STRATEGY] = "<<YOUR VALUE>>"
outputDf[BULK_COL_PUBLISHER_BIDSTRATEGY] = "<<YOUR VALUE>>"
outputDf[BULK_COL_BIDDING_MOVEMENT] = "<<YOUR VALUE>>"
today = datetime.datetime.now(CLIENT_TIMEZONE).date()
print("today: ", today)
VAL_PUB_STRATEGY_MANUALCPC = 'ManualCpc'
VAL_MARIN_STRATEGY_TITLE_CPA = 'TITLE | AUTOMATION | CPA'
VAL_BIDDING_MOVEMENT_BLANK = ''
def process(inputDf):
# setup tmp columns
TMP_COL_PUBLISHER_BIDSTRATEGY = RPT_COL_PUBLISHER_BIDSTRATEGY + '_'
TMP_COL_STRATEGY = RPT_COL_STRATEGY + '_'
TMP_COL_BIDDING_MOVEMENT = RPT_COL_BIDDING_MOVEMENT + '_'
inputDf[TMP_COL_PUBLISHER_BIDSTRATEGY] = np.nan
inputDf[TMP_COL_STRATEGY] = np.nan
inputDf[TMP_COL_BIDDING_MOVEMENT] = np.nan
print("inputDf", inputDf.to_string())
# apply logic
inputDf.loc[ inputDf[RPT_COL_CPA_DATE] == today, \
[TMP_COL_PUBLISHER_BIDSTRATEGY, TMP_COL_STRATEGY, TMP_COL_BIDDING_MOVEMENT] \
] = [VAL_PUB_STRATEGY_MANUALCPC, VAL_MARIN_STRATEGY_TITLE_CPA, VAL_BIDDING_MOVEMENT_BLANK]
# Check for changes
changed = (inputDf[TMP_COL_PUBLISHER_BIDSTRATEGY].notnull() & (inputDf[TMP_COL_PUBLISHER_BIDSTRATEGY] != inputDf[RPT_COL_PUBLISHER_BIDSTRATEGY])) | \
(inputDf[TMP_COL_STRATEGY].notnull() & (inputDf[TMP_COL_STRATEGY] != inputDf[RPT_COL_STRATEGY])) | \
(inputDf[TMP_COL_BIDDING_MOVEMENT].notnull() & (inputDf[TMP_COL_BIDDING_MOVEMENT] != inputDf[RPT_COL_BIDDING_MOVEMENT]))
if changed.sum() > 0:
# print changed rows
print("changed", inputDf[changed].to_string())
# construct outputDf
outputDf = inputDf[changed][[RPT_COL_ACCOUNT, RPT_COL_CAMPAIGN, TMP_COL_PUBLISHER_BIDSTRATEGY, TMP_COL_STRATEGY, TMP_COL_BIDDING_MOVEMENT]].copy()
outputDf.rename(columns={ \
TMP_COL_PUBLISHER_BIDSTRATEGY: BULK_COL_PUBLISHER_BIDSTRATEGY, \
TMP_COL_STRATEGY: BULK_COL_STRATEGY,
TMP_COL_BIDDING_MOVEMENT: BULK_COL_BIDDING_MOVEMENT
}, inplace=True)
else:
print("No changes detected, returning an empty dataframe")
outputDf = pd.DataFrame(columns=[BULK_COL_ACCOUNT, BULK_COL_CAMPAIGN, BULK_COL_PUBLISHER_BIDSTRATEGY, BULK_COL_STRATEGY, BULK_COL_BIDDING_MOVEMENT])
return outputDf
def test_process():
print("#### UNIT TEST START ####")
try:
# Input data with additional columns
input_data = [
[
'Campaign 1',
'Account 1',
'Max ROAS',
'Increase',
(datetime.datetime.today() - datetime.timedelta(days=1)).date(),
'Max Conv'
],
[
'Campaign 2',
'Account 1',
'Max ROAS',
'Increase',
(datetime.datetime.today() + datetime.timedelta(days=1)).date(),
'Max Conv'
],
[
'Campaign 3',
'Account 1',
'Max ROAS',
'Increase',
datetime.datetime.today().date(),
'Max Conv'
],
]
# Build input dataframe
inputDf = pd.DataFrame(input_data, columns=[
RPT_COL_CAMPAIGN,
RPT_COL_ACCOUNT,
RPT_COL_PUBLISHER_BIDSTRATEGY,
RPT_COL_BIDDING_MOVEMENT,
RPT_COL_CPA_DATE,
RPT_COL_STRATEGY
])
# Expected output DataFrame
expected_output_data = {
BULK_COL_ACCOUNT: ["Account 1"],
BULK_COL_CAMPAIGN: ["Campaign 3"],
BULK_COL_PUBLISHER_BIDSTRATEGY: [VAL_PUB_STRATEGY_MANUALCPC],
BULK_COL_STRATEGY: [VAL_MARIN_STRATEGY_TITLE_CPA],
BULK_COL_BIDDING_MOVEMENT: [VAL_BIDDING_MOVEMENT_BLANK],
}
expected_outputDf = pd.DataFrame(expected_output_data)
outputDf = process(inputDf)
# Assertions
assert len(outputDf) == 1
# if outputDf.equals(expected_outputDf):
assert np.array_equal(expected_outputDf.values, outputDf.values)
print("#### PASSED ####")
print("#### UNIT TEST END ####")
except AssertionError:
print("#### FAILED ####")
print("expected", tableize(expected_outputDf), expected_outputDf.dtypes, expected_outputDf.values)
print("actual", tableize(outputDf), outputDf.dtypes, outputDf.values)
# run unit test
test_process()
# Process the inputDf
outputDf = process(inputDf)
print("final output", outputDf.to_string())
Post generated on 2025-03-11 01:25:51 GMT