Integrating Marketing and Attribution Data

In plain terms

A customer clicks a search ad, later follows an email link, and buys a subscription. The order database records one purchase, while several marketing reports may assign credit to it. Integrating marketing attribution data means connecting spend, observed interactions, identities, and conversions, then making the credit rule explicit. Assigned credit alone does not establish that an advertisement caused a purchase.

This article builds on the event identity, incremental loading, and managed ingestion articles. The lab uses synthetic data and Python’s standard library to examine campaign mapping, paths, five attribution rules, revisions, calendars, and reconciliation. It does not connect to advertising services or estimate incremental advertising effects.

The five tables and their grains

The grain states what one row represents. Spend is an aggregate: its key includes vendor, account, campaign, reporting date, currency, and any requested breakdown such as device or country. Store the reporting time zone, extraction time, and report configuration too. Campaign IDs are stable within their vendor/account scope; names are descriptive attributes. Mixing different breakdowns can count the same spend twice.

A touchpoint is one observed click or visit, keyed by an event ID, with event time, campaign identifiers, and an anonymous or account ID. Deduplicate retries using that event ID; a vendor click ID is supporting evidence, not a universal event key. Identity links connect identifiers with their provenance, validity interval, and applicable measurement permission. A conversion is one defined business event, keyed by order or conversion ID, with account, time, status, and amount. An account can have many conversions. Refunds, cancellations, and duplicate events require an explicit conversion definition.

Vendor attribution reports are a fifth dataset. Their grain depends on the API and selected dimensions: they may be aggregates with fractional or modeled conversions, rather than individually identifiable purchases. Keep their model, window, metric definition, and report date basis. Business records establish the chosen conversion population; observed touches only describe what the platform captured. Vendor figures remain useful for their reporting and optimization purposes, but cannot be assumed to be a deduplicated cross-channel total.

Campaign taxonomy and safe joins

Maintain a campaign dimension mapping scoped vendor IDs to channel, objective, market, launch, and owner. Preserve names as reported and version the mapping when historical classification matters. A rename must not create a new campaign. A new, unmapped campaign must remain in an unknown bucket and in total spend; report it to the owner for classification. An inner join that drops it makes the dashboard silently incomplete.

Aggregate spend and attributed conversion credit separately to the same reporting grain before joining them. Joining one campaign-day spend row directly to ten touches repeats that spend ten times. For example, 100 dollars of spend and 200 dollars of credited revenue produce a reported ROAS of 2, not 0.2 after a fan-out join. Align currency and its conversion rule, taxes, refunds, date basis, and channel scope first. ROAS uses revenue credit; cost per acquisition uses a defined acquisition count. Neither ratio proves incremental return.

From a click to a conversion path

A landing event carries campaign parameters or a click identifier. Login can link its anonymous ID to an account, but a shared device or later account switch makes a permanent anonymous-to-person mapping unsafe. Resolve links under a documented validity rule and retain the identity snapshot used for a report. Respect applicable collection and linkage permissions; an identifier match alone is not permission to join.

For each conversion, collect eligible touches on its resolved identities and filter to the attribution lookback window. This lab includes the lower boundary and excludes touches at or after conversion time, then sorts by timestamp and event ID for deterministic ties. Its identity links are static and one-to-one; it does not implement a production identity graph. A conversion without an eligible touch is unattributed under this policy. Missing tracking, missing links, or an expired window can produce such a path in production, so unattributed does not establish a direct visit.

A path is evaluated separately for each conversion. If one customer makes two eligible purchases after one click, that click can appear in both paths under an every-purchase policy; each purchase still contributes only one total credit. Counting only the first acquisition, or resetting the path after a purchase, produces a different metric and needs an explicit rule. Repeated delivery of the same conversion ID is a duplicate, not a second purchase. The generator includes at most one conversion per linked account, so it does not exercise repeat-purchase behavior.

Attribution rules and what their totals mean

Last touch assigns all credit to the final eligible touch; first touch to the first; linear divides it equally. Time decay weights recent touches more heavily. The lab uses a seven-day half-life and normalizes the weights to sum to one. Position-based attribution gives 40% each to the first and last touches and shares 20% among the middle touches; one-touch and two-touch paths receive 100% and 50/50 respectively. For search, email, social, a linear path gives each one third; position-based gives 0.4, 0.2, 0.4.

For every conversion, credits must be finite, nonnegative, and sum to one, including an unattributed bucket for an empty path. Summing these shares yields conversion credit, not additional purchases. Multiplying each share by that conversion’s eligible revenue gives revenue credit, with a separate refund policy and rounding rule. The lab reports counts and does not use its generated amount field to calculate ROAS.

These are educational warehouse rules, not a list of every vendor’s current options. Google Ads documents data-driven attribution and last click, and has retired first click, linear, time decay, and position-based models. See Google Ads attribution models. Vendor models and windows vary, and reports can include view-through or modeled conversions. The lab’s overlapping claims simply give each channel one credit whenever it has an eligible click in its illustrative window. That yields 773 claims for 499 purchases; it does not reproduce a real vendor algorithm.

Version the window, model, eligible touch types, identity snapshot, conversion definition, and effective date. Preserve permitted source snapshots and calculation versions so past reports remain explainable within retention limits. Comparing rules shows sensitivity to a reporting choice. To estimate additional purchases caused by advertising, use an appropriate incrementality design, such as randomized holdouts, with its own assumptions and analysis.

Reporting dates, conversion lag, and restatements

An event has an instant; a daily report has a calendar and a date basis. Store event timestamps in UTC and convert them with a named time zone when constructing reporting days. The lab uses America/Los_Angeles, whose offset changes during March 2026. A fixed minus-eight-hour offset would misclassify some dates after daylight saving starts. A vendor account need not use Pacific time. Attribution reports may group by interaction date or conversion date; choosing the same time zone does not resolve that difference. See Analytics attribution settings and Python zoneinfo.

A daily spend aggregate cannot generally be converted into one UTC date: a vendor day spans parts of two UTC days. Preserve its original reporting date and time zone. To compare, aggregate your events on that same calendar; to split spend across UTC days, obtain finer-grained data or explicitly label an allocation estimate.

Keep three horizons separate. The attribution window defines which past touches may receive credit. Cohort maturation determines how long to observe a set of campaign interactions before assessing later conversions. The ingestion reread horizon captures late source corrections. Conversion lag informs maturation and window selection, but does not by itself set every reprocessing horizon. Late identities, refunds, source delays, and policy changes can require wider or targeted backfills.

For revised spend, fetch complete snapshots for selected reporting partitions, stage and validate them, then atomically replace those partitions, including rows that vanished or an empty snapshot. Retain extraction versions for as-of reporting. Choose routine reread periods from source behavior and monitoring, with deeper reconciliation and explicit backfills for older corrections. The lab only replays revisions to existing campaign-day keys; it does not implement a complete snapshot loader.

Reconciliation and privacy in the pipeline

Compare like-for-like metrics before explaining a discrepancy: account, campaign, date basis, time zone, currency, extraction version, conversion definition, and attribution settings. The lab compares observed clicks with initially reported clicks on the same UTC calendar. These are not necessarily billable clicks in a real API. Its 0.80–1.05 alert band and 88% capture rate are chosen simulator parameters, not industry baselines. A changing ratio suggests investigation of tags, redirects, consent, deduplication, reporting revisions, or scope; it does not identify the cause. Handle zero denominators and low-volume groups separately.

Reconcile connector spend against invoices with an explicit bridge for taxes, credits, fees, exchange rates, and reporting cutoffs. Compare platform and vendor conversions with their definitions visible; neither is required to be larger. Report unmapped spend, source revisions, missing partitions, and volume anomalies. Alert tolerances should reflect volume and expected behavior, rather than a universal percentage.

For this example’s proposed policy, collect measurement touches only where measurement consent permits it. This is a design choice under a reviewed policy, not a universal legal rule. Minimize identifiers and restrict access. Hashed emails and pseudonymous IDs can remain linkable and must not be described as anonymous merely because their representation changed. Record applicable consent state and policy version; govern retention, downstream deletion, and any affected derived reports under the relevant requirements. The synthetic lab does not implement these workflows.

Keep observed and modeled conversion metrics distinguishable where the API exposes that distinction, and otherwise record that they are mixed or unknown. A modeled conversion aggregate may be compared at an aggregate level with aligned scope and uncertainty, but cannot necessarily be matched to an individual order. This metadata belongs to conversion reporting, not to an invented observed-versus-modeled spend flag. Google modeled online conversions explains how observed and modeled conversions can appear together in reporting.

Assembling the subscription pipeline

Use managed connectors or source-specific adapters for ad reports and email click events. Land raw report versions with extraction manifests. Validate keys, requested dimensions, currencies, date ranges, and completeness before replacing reporting partitions. Merge email clicks by event ID while keeping message ID as a separate attribute; one message can have several clicks. Ingest business conversions and versioned identity links from the database and event pipelines.

Build eligible paths from those inputs, persist conversion-by-touch credit with a rule version, and represent empty paths explicitly as unattributed. Aggregate credit and spend separately before the reporting join. Rebuild affected conversions when an input or policy changes; a nightly rolling window is an optimization supplemented by targeted historical backfills. Keep rule configuration, input versions, run ID, and reconciliation results together. A dashboard user should be able to tell whether a number changed because of new business activity, a source revision, or a different attribution rule.

Lab

The generator assumes one currency, one account per channel, globally distinct campaign IDs, and UTC report days. It captures roughly 88% of generated paid clicks except for a planted tag outage, links about 85% of observed visitors, then samples converters only from those linked visitors. Only conversions before the month-end cutoff are retained. These choices create selection and observation-cutoff bias: this sample cannot estimate real tracking loss or the full conversion-lag distribution. Run both setup blocks before any exercise. Python 3.12 with IANA time-zone data is used.

import random
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo
import math

rng = random.Random(0)
start = datetime(2026, 3, 1, tzinfo=timezone.utc)
CAMPAIGNS = {"A-101": ("search", "brand"), "A-102": ("search", "generic"), "B-7": ("social", "spring_sale"),
             "B-8": ("social", "retargeting"), "C-3": ("affiliate", "coupon_site"), "E-1": ("email", "newsletter")}
WINDOWS = {"search": 30, "social": 28, "affiliate": 7, "email": 7}

def vendor_name(campaign, day):
    """Vendors report names, and names change; B-7 was renamed on day 15 and B-9 was created on day 20 by the marketing team."""
    if campaign == "B-7":
        return "Spring Sale" if day < 15 else "Spring Sale - final week"
    return {"A-101": "Brand terms", "A-102": "Generic terms", "B-8": "Retargeting", "C-3": "Coupon partners", "B-9": "Influencer test"}[campaign]

spend_reports, touches, conversions, links = [], [], [], {}
for day in range(30):
    for campaign, (channel, _) in list(CAMPAIGNS.items()) + ([("B-9", ("social", None))] if day >= 20 else []):
        if channel == "email":
            continue
        clicks = rng.randint(30, 90)
        spend_reports.append({"reported_on": day + 1, "day": day, "campaign": campaign, "name": vendor_name(campaign, day),
                              "spend_cents": clicks * rng.randint(40, 120), "clicks": clicks})
        observed = 0 if (campaign == "A-102" and day == 12) else int(clicks * 0.88)
        for _ in range(observed):
            touches.append({"anonymous_id": f"A{rng.randint(1, 4000)}", "campaign": campaign, "channel": channel,
                            "at": start + timedelta(days=day, seconds=rng.randrange(86400)), "click_id": f"{campaign}-{rng.randrange(10**6)}"})
    for _ in range(40):
        touches.append({"anonymous_id": f"A{rng.randint(1, 4000)}", "campaign": "E-1", "channel": "email",
                        "at": start + timedelta(days=day, hours=9), "click_id": None})
for event_id, touch in enumerate(touches):
    touch["event_id"] = event_id
touches.sort(key=lambda t: (t["at"], t["event_id"]))
by_anonymous = defaultdict(list)
for t in touches:
    by_anonymous[t["anonymous_id"]].append(t)
for anonymous_id in by_anonymous:
    if rng.random() < 0.85:
        links[anonymous_id] = "U" + anonymous_id[1:]
for anonymous_id in rng.sample(sorted(links), 700):
    last = by_anonymous[anonymous_id][-1]["at"]
    at = last + timedelta(days=min(rng.expovariate(1 / 4), 25), seconds=rng.randrange(86400))
    if at < start + timedelta(days=30):
        conversions.append({"user_id": links[anonymous_id], "at": at, "amount_cents": rng.randint(2000, 9000)})
for conversion_id, conversion in enumerate(conversions):
    conversion["conversion_id"] = conversion_id
print(len(spend_reports), "spend rows,", len(touches), "touches,", len(conversions), "conversions,", len(links), "identity links")
# 160 spend rows, 9612 touches, 499 conversions, 3072 identity links
print("touches per converting person:", sorted(Counter(min(len(by_anonymous["A" + c["user_id"][1:]]), 4) for c in conversions).items()))
# touches per converting person: [(1, 141), (2, 147), (3, 109), (4, 102)]

The resolver below is the join from a conversion back to the touches that may deserve credit: every anonymous id linked to the converting account, every touch on those ids, filtered to the lookback window before the conversion and ordered in time. Exercises 2, 3, and 5 use it.

anonymous_for = defaultdict(list)
for anonymous_id, user_id in links.items():
    anonymous_for[user_id].append(anonymous_id)

def eligible_touches(conversion, lookback_days=7):
    if isinstance(lookback_days, bool) or not isinstance(lookback_days, (int, float)) or not math.isfinite(lookback_days) or not 0 <= lookback_days <= 36500:
        raise ValueError("lookback_days must be finite and between 0 and 36500")
    touched = [t for a in anonymous_for.get(conversion["user_id"], []) for t in by_anonymous[a]]
    return sorted((t for t in touched if conversion["at"] - timedelta(days=lookback_days) <= t["at"] < conversion["at"]), key=lambda t: (t["at"], t["event_id"]))

1. Map campaigns. Total spend by taxonomy campaign joined on the vendor’s id, then by the vendor’s name, and list the spend that maps to nothing.

by_id, by_name, unmapped = Counter(), Counter(), Counter()
for r in spend_reports:
    if r["campaign"] in CAMPAIGNS:
        by_id[CAMPAIGNS[r["campaign"]][1]] += r["spend_cents"]
    else:
        unmapped[(r["campaign"], r["name"])] += r["spend_cents"]
    by_name[r["name"]] += r["spend_cents"]
print("spend by taxonomy campaign (joined on vendor id):", dict(by_id))
# spend by taxonomy campaign (joined on vendor id): {'brand': 146055, 'generic': 162515, 'spring_sale': 122540, 'retargeting': 142748, 'coupon_site': 137978}
print("spend by vendor name:", {k: v for k, v in by_name.items() if "Spring" in k})
# spend by vendor name: {'Spring Sale': 65832, 'Spring Sale - final week': 56708}
print("unmapped spend:", dict(unmapped), "=", f"{sum(unmapped.values()) / sum(r['spend_cents'] for r in spend_reports):.1%} of the month")
# unmapped spend: {('B-9', 'Influencer test'): 41467} = 5.5% of the month
assert sum(by_id.values()) + sum(unmapped.values()) == sum(r["spend_cents"] for r in spend_reports)

2. Build the paths. For every conversion, resolve the touchpoints in a 7-day window and count how many each path has; repeat with 30 days; list the most common channel paths.

paths = [eligible_touches(c) for c in conversions]
print("touches in the 7-day window before conversion:", sorted(Counter(min(len(p), 3) for p in paths).items()), "(3 means 3 or more)")
# touches in the 7-day window before conversion: [(0, 48), (1, 314), (2, 110), (3, 27)] (3 means 3 or more)
print("with a 30-day window:", sorted(Counter(min(len(eligible_touches(c, 30)), 3) for c in conversions).items()))
# with a 30-day window: [(1, 141), (2, 147), (3, 211)]
print("most common channel paths:", Counter(" > ".join(t["channel"] for t in p) or "(none)" for p in paths).most_common(4))
# most common channel paths: [('social', 133), ('search', 91), ('affiliate', 49), ('(none)', 48)]

3. Five models and overlapping claims. Credit each conversion’s path under last touch, first touch, linear, time decay, and position-based rules, and total the credit by channel. Check that every model’s total equals the number of conversions. Then compute the illustrative channel claims under each channel’s window.

def credit(path, model):
    if model not in {"last", "first", "linear", "time decay", "position"}:
        raise ValueError("unknown attribution model")
    path = sorted(path, key=lambda t: (t["at"], t["event_id"]))
    if not path:
        return {"unattributed": 1.0}
    n = len(path)
    if model == "last":
        return {path[-1]["channel"]: 1.0}
    if model == "first":
        return {path[0]["channel"]: 1.0}
    if model == "linear":
        weights = [1 / n] * n
    elif model == "time decay":
        raw = [0.5 ** ((path[-1]["at"] - t["at"]).total_seconds() / 86400 / 7) for t in path]
        weights = [w / sum(raw) for w in raw]
    elif model == "position":
        weights = [1.0] if n == 1 else [0.5, 0.5] if n == 2 else [0.4] + [0.2 / (n - 2)] * (n - 2) + [0.4]
    out = defaultdict(float)
    for t, w in zip(path, weights):
        out[t["channel"]] += w
    return out

paths = [eligible_touches(c) for c in conversions]
print(f"{'model':11}" + "".join(f"{c:>13}" for c in ["search", "social", "affiliate", "email", "unattributed"]))
# model             search       social    affiliate        email unattributed
for model in ("last", "first", "linear", "time decay", "position"):
    total = defaultdict(float)
    for path in paths:
        shares = credit(path, model)
        assert all(math.isfinite(v) and v >= 0 for v in shares.values())
        assert math.isclose(sum(shares.values()), 1.0)
        for channel, share in shares.items():
            total[channel] += share
    print(f"{model:11}" + "".join(f"{total:13.1f}" for c in ["search", "social", "affiliate", "email", "unattributed"]), f"  sum {sum(total.values()):.0f}")
# last               138.0        187.0         70.0         56.0         48.0   sum 499
# first              137.0        183.0         72.0         59.0         48.0   sum 499
# linear             136.3        185.6         70.8         58.2         48.0   sum 499
# time decay         136.8        185.8         70.2         58.1         48.0   sum 499
# position           136.8        185.3         71.0         57.9         48.0   sum 499

claims = Counter()
for c in conversions:
    for channel, window in WINDOWS.items():
        if any(t["channel"] == channel for t in eligible_touches(c, window)):
            claims[channel] += 1
print("claims under the illustrative channel windows:", dict(claims), "=", sum(claims.values()), "claimed for", len(conversions), "real conversions")
# claims under the illustrative channel windows: {'search': 288, 'social': 315, 'email': 78, 'affiliate': 92} = 773 claimed for 499 real conversions

4. Restatements. Add the vendors’ revisions: social spend revised down 10% three days after the day, search spend revised up 2% ten days after. Compute spend for the first six days as a nightly job with a 14-day lookback held it on the evening of day 6 and as it stands after the month, count the changed campaign-day keys, and compare lookbacks of 0, 3, and 14 days.

restatements = []
for r in spend_reports:
    channel = CAMPAIGNS.get(r["campaign"], ("social",))[0]
    if channel == "social":
        restatements.append(dict(r, reported_on=r["day"] + 3, spend_cents=int(r["spend_cents"] * 0.9), clicks=int(r["clicks"] * 0.9)))
    if channel == "search":
        restatements.append(dict(r, reported_on=r["day"] + 10, spend_cents=int(r["spend_cents"] * 1.02)))

def spend_as_of(day, lookback):
    """Replay available row revisions within a reread horizon; not a full snapshot loader."""
    if any(type(v) is not int or v < 0 for v in (day, lookback)):
        raise ValueError("day and lookback must be nonnegative integers")
    table = {}
    for report in sorted(spend_reports + restatements, key=lambda r: r["reported_on"]):
        if report["reported_on"] <= day and report["day"] >= report["reported_on"] - 1 - lookback:
            table[(report["day"], report["campaign"])] = report["spend_cents"]
    return table

early, final = spend_as_of(6, 14), spend_as_of(40, 14)
changed = [k for k in early if early[k] != final.get(k)]
print("spend for days 0-5 as known on day 6:", sum(v for (d, _), v in early.items() if d < 6), "cents; after all simulated revisions:", sum(v for (d, _), v in final.items() if d < 6))
# spend for days 0-5 as known on day 6: 133574 cents; after all simulated revisions: 133433
print(len(changed), "of", len(early), "day-campaign keys changed after day 6")
# 16 of 30 day-campaign keys changed after day 6
frozen = spend_as_of(40, 0)
print("a job with no lookback would still hold the first report for", sum(frozen[k] != final[k] for k in frozen), "keys;",
      "a 3-day lookback misses", sum(spend_as_of(40, 3)[k] != final[k] for k in frozen), "of them, a 14-day lookback misses", sum(spend_as_of(40, 14)[k] != final[k] for k in frozen))
# a job with no lookback would still hold the first report for 130 keys; a 3-day lookback misses 60 of them, a 14-day lookback misses 0

5. Two calendars and the lag. Count conversions whose calendar day differs between the platform’s UTC and a vendor reporting in Pacific time, and compare one day’s count. Then measure the lag from last touch to conversion and the share whose last touch lies outside a 7-day attribution window.

pacific = ZoneInfo("America/Los_Angeles")
utc_days = Counter(c["at"].date() for c in conversions)
vendor_days = Counter(c["at"].astimezone(pacific).date() for c in conversions)
moved = sum(c["at"].date() != c["at"].astimezone(pacific).date() for c in conversions)
print(moved, "of", len(conversions), "conversions fall on a different calendar day in the vendor's Pacific-time report than in the platform's UTC day")
# 140 of 499 conversions fall on a different calendar day in the vendor's Pacific-time report than in the platform's UTC day
print("March 10 conversions, platform:", utc_days[datetime(2026, 3, 10).date()], " vendor:", vendor_days[datetime(2026, 3, 10).date()])
# March 10 conversions, platform: 5  vendor: 7
lags = sorted((c["at"] - eligible_touches(c, 30)[-1]["at"]).total_seconds() / 86400 for c in conversions if eligible_touches(c, 30))
print(f"days from last touch to conversion: median {lags[len(lags) // 2]:.1f}, 90th percentile {lags[int(len(lags) * 0.9)]:.1f}, max {lags[-1]:.1f}")
# days from last touch to conversion: median 2.4, 90th percentile 6.9, max 16.5
print("a 7-day attribution window excludes the last touch for", f"{sum(l > 7 for l in lags) / len(lags):.0%}", "of retained simulated conversions")
# a 7-day attribution window excludes the last touch for 10% of retained simulated conversions

6. Reconcile clicks. For every campaign and day, divide the clicks the platform observed by the clicks the vendor initially reported, report the median, and list the campaign-days outside a band of 0.80 to 1.05.

def click_comparison(observed_count, reported_count, minimum_reported=20):
    if any(type(v) is not int or v < 0 for v in (observed_count, reported_count)):
        raise ValueError("click counts must be nonnegative integers")
    if type(minimum_reported) is not int or minimum_reported < 1:
        raise ValueError("minimum_reported must be a positive integer")
    if reported_count == 0:
        return None, "zero denominator"
    ratio = observed_count / reported_count
    if reported_count < minimum_reported:
        return ratio, "low volume"
    return ratio, "compare"

observed = Counter((t["campaign"], (t["at"] - start).days) for t in touches if t["click_id"])
alerts, ratios, exceptions = [], [], []
for r in spend_reports:
    count = observed[(r["campaign"], r["day"])]
    ratio, status = click_comparison(count, r["clicks"])
    if status != "compare":
        exceptions.append((r["day"], r["campaign"], r["clicks"], count, status))
        continue
    ratios.append(ratio)
    if not 0.8 <= ratio <= 1.05:
        alerts.append((r["day"], r["campaign"], r["clicks"], count, round(ratio, 2)))
if ratios:
    print(f"observed clicks over initially reported clicks: median {sorted(ratios)[len(ratios) // 2]:.2f} across {len(ratios)} campaign-days")
else:
    print("no campaign-days meet the comparison rule")
# observed clicks over initially reported clicks: median 0.87 across 160 campaign-days
print("outside the 0.80-1.05 band:", alerts)
# outside the 0.80-1.05 band: [(12, 'A-102', 83, 0, 0.0)]
print("zero-denominator or low-volume groups:", exceptions)
# zero-denominator or low-volume groups: []

7. Write the attribution policy. Produce the one-page policy the attribution job implements: the lookback window, the model, which touch types count, how unattributed conversions are reported, the reprocessing windows for conversions and spend, the reconciliation checks and their bands, the consent rule, and the rule version and effective date.