Functory
functory.
All Articles
5 min read
Functory

Compute unsubscribe rate from Customer.io subscriber lists in Python with Functory for quarterly marketing reviews

This article shows how to write a single-file Python function that computes an actionable unsubscribe rate metric from Customer.io subscriber lists and email open events. It targets small marketing teams that need an on-demand, reproducible script (or hosted API) for quarterly reviews without hiring a data engineer. You will get a precise definition of the metric, concrete input schemas, a runnable pandas-based implementation, and instructions for publishing the function on Functory so non-technical teammates can trigger it via a web UI or API.

What this function does and why it matters

We compute the "unsubscribe rate after an open": the proportion of unique subscribers who opened at least one email in the reporting window and then unsubscribed within N days after their last open. Inputs can be CSV/JSON exports from Customer.io (subscriber list and event logs). Output is a small report (CSV and a JSON summary) with per-campaign and overall unsubscribe rates, plus raw counts: unique_opens, unsubscribes_within_N_days, unsubscribe_rate_pct.

Input data expectations

  • Subscribers CSV (subscribers.csv): columns id (string), email (string), created_at (ISO8601), unsubscribed_at (ISO8601 or empty).
  • Events CSV/JSONL (events.csv): columns event_id, customer_id (matches subscribers.id), event_name (e.g., "email_open"), event_timestamp (ISO8601), campaign_id (string).
  • Time window: start_date and end_date (YYYY-MM-DD) applied to opens. lookback_days: integer window to consider unsubscribes after an open (default 7).

What transformation the script performs

In detail: read subscribers and events, filter events for event_name == "email_open" in the date range, compute unique opens per subscriber and campaign, find subscribers whose unsubscribed_at exists and is within N days after their last open, aggregate by campaign and overall, and save a CSV report. The report contains columns: campaign_id, unique_openers, unsubscribed_within_N_days, unsubscribe_rate_pct.

Real-world scenario (concrete)

Files: subscribers_2024Q2.csv (12,432 rows) and events_2024Q2.csv (45,287 rows). Problem: the Head of Growth wants to know whether recent re-engagement emails are causing increased unsubscribes within 7 days of opens. Desired output: per-campaign rows like:

campaign_id,unique_openers,unsubscribed_within_7_days,unsubscribe_rate_pct
welcome_series_v2,3,1,33.33
promo_may2024,1,0,0.00
quarterly_newsletter_q2,4523,12,0.27

Example dataset and specific problem

Example dataset: 3 files totalling ~60k rows (12k subscribers, 45k events, ~100MB CSVs compressed). Problem solved: turns manual CSV merges and ad-hoc spreadsheet formulas into a reproducible function you can run on demand. The function calculates whether certain campaigns have unusually high unsubscribe rates after opens, which informs subject line or frequency tests.

When to use this metric

  • Quarterly marketing review where you need a consistent denominator (openers) rather than sends.
  • Post-campaign safety checks: if unsubscribe_rate_pct > 1% in a low-volume campaign, flag for review.
  • Automated guardrails for A/B subject-line experiments.

Step-by-step mini workflow

  1. Export subscribers.csv and events.csv from Customer.io for the quarter (or upload to Functory UI).
  2. Call the Python function with start_date=2024-04-01, end_date=2024-06-30, lookback_days=7.
  3. The function reads CSVs, computes per-campaign unique openers and unsubscribes within 7 days, writes unsubscribe_report.csv, and returns a JSON summary.
  4. Download the CSV from the Functory result or fetch the JSON via the API, include in your quarterly slides.

Algorithm (high-level)

  1. Filter events to event_name == "email_open" and event_timestamp within [start_date, end_date].
  2. For each customer_id and campaign_id, get the last open timestamp.
  3. Join to subscribers on customer_id; compute delta = unsubscribed_at - last_open; mark unsubscribed_within_N_days if 0 <= delta.days <= N.
  4. Aggregate counts per campaign: unique_openers, unsubscribed_within_N_days, compute percentage.

Python implementation (runnable example)

import pandas as pd
from datetime import datetime, timedelta

def compute_unsubscribe_report(sub_csv, events_csv, start_date, end_date, lookback_days=7):
    subs = pd.read_csv(sub_csv, parse_dates=["created_at", "unsubscribed_at"])  # id,email,created_at,unsubscribed_at
    events = pd.read_csv(events_csv, parse_dates=["event_timestamp"])  # event_id,customer_id,event_name,event_timestamp,campaign_id

    start = pd.to_datetime(start_date)
    end = pd.to_datetime(end_date) + pd.Timedelta(days=1) - pd.Timedelta(seconds=1)

    opens = events[(events["event_name"]=="email_open") & (events["event_timestamp"]>=start) & (events["event_timestamp"]<=end)].copy()

    # last open per customer per campaign
    last_open = opens.sort_values("event_timestamp").groupby(["customer_id","campaign_id"]).tail(1)
    last_open = last_open[["customer_id","campaign_id","event_timestamp"]].rename(columns={"event_timestamp":"last_open_at"})

    joined = last_open.merge(subs, left_on="customer_id", right_on="id", how="left")

    joined["unsubscribed_within"] = (joined["unsubscribed_at"] - joined["last_open_at"]).dt.days
    joined["unsub_within_N"] = joined["unsubscribed_within"].between(0, lookback_days)

    agg = joined.groupby("campaign_id").agg(
        unique_openers=("customer_id","nunique"),
        unsubscribed_within_N_days=("unsub_within_N","sum")
    ).reset_index()

    agg["unsubscribe_rate_pct"] = (agg["unsubscribed_within_N_days"] / agg["unique_openers"]) * 100
    agg = agg.sort_values("unsubscribe_rate_pct", ascending=False)

    out_csv = f"unsubscribe_report_{start_date}_to_{end_date}.csv"
    agg.to_csv(out_csv, index=False)
    return out_csv, agg.head(10).to_dict(orient="records")

# Example call
if __name__ == '__main__':
    path, summary = compute_unsubscribe_report(
        "subscribers_2024Q2.csv",
        "events_2024Q2.csv",
        "2024-04-01",
        "2024-06-30",
        lookback_days=7
    )
    print("Wrote:", path)
    print(summary)

How Functory Makes It Easy

To publish this as a Functory function, wrap the core logic into a single main(...) entrypoint that accepts typed parameters: subscribers_csv (FilePath), events_csv (FilePath), start_date (str), end_date (str), lookback_days (int). On Functory the developer chooses an exact Python runtime (for example 3.11.11) and provides a requirements.txt with pinned versions like pandas==2.2.3 and python-dateutil==2.8.2. Functory will present each parameter as an input field in the web UI and expose them on the API as JSON or multipart uploads for files. The function should return a path-like value (e.g., "unsubscribe_report_2024-04-01_to_2024-06-30.csv") so Functory exposes it as a downloadable result. Benefits: no servers to manage, automatic CPU autoscaling for large CSVs, built-in logging via print(), and pay-per-use billing handled by the platform. You can chain this function with a pre-processing Functory function that fetches Customer.io exports (API key input) and a downstream reporting function that publishes the CSV to a shared Google Drive or Slack notification.

Alternative approaches and why this function-based approach is better

Teams often solve this with (a) manual CSV merges in spreadsheets, (b) Jupyter notebooks run locally, or (c) scheduled ETL pipelines requiring a data engineer. Spreadsheets are error-prone for joins on 12k rows; notebooks are difficult to share and schedule. Packaging the logic as a small function (or a hosted Functory API) provides repeatability, correct joins, and an easy UI for non-technical marketers. It also reduces drift: the same code runs each quarter with pinned library versions and documented inputs.

Business impact

Quantified benefit: converting a manual spreadsheet process to this script reduces reporting time from ~3 hours per quarter to ~20 minutes for exports and verification — a ~89% reduction in manual effort. It also reduces human errors in joins that could otherwise skew unsubscribe rates by several percentage points.

Key takeaways

Conclusion: You now have a clear definition, algorithm, and a runnable Python example to compute unsubscribe rates from Customer.io exports. Next steps: wrap the example in a main(...) function for Functory, pin dependencies (e.g., pandas==2.2.3) and a Python patch version (e.g., 3.11.11), and test with one quarter of data. If you want to automate fetching exports from Customer.io, add a small preprocessor that calls the Customer.io API and uploads CSVs to the Functory function. Publish and iterate — then add a Slack notification step to speed up decision-making in the next marketing review.

Thanks for reading.