Stripe Payment Analytics Exporter

Stripe Payment Analytics Exporter

Problem Statement

Finance and DevOps teams often need to reconcile Stripe payment data with internal reporting systems, spreadsheets, or BI dashboards. The Stripe Dashboard provides interactive views, but extracting historical metrics like charges, refunds, and monthly recurring revenue (MRR) in a scriptable, repeatable way requires custom code. Manually exporting CSVs from the dashboard is error-prone, doesn't scale across multiple accounts, and can't be scheduled.

This script solves that by querying the Stripe API for a configurable time window, aggregating the requested metrics, and writing a clean CSV report. It's designed to run headless (cron, CI, container job) and to fail loudly with actionable error messages when Stripe returns rate limits, auth failures, or malformed data.

How It Works

The script uses the official stripe Python SDK. On startup it reads the API key and options from environment variables and CLI flags, then constructs a time window (--since 30d, --since 7d, or an ISO date). For each requested metric it dispatches to a small collector function that pages through the Stripe API using auto_paging_iter(), which transparently handles cursor pagination and rate-limit retries built into the SDK.

Results are accumulated into a list of flat dicts and written with csv.DictWriter. The design uses a registry pattern (METRIC_COLLECTORS) so adding a new metric is a one-function change. Monetary values are normalized to decimal strings (Stripe returns integer cents), and all timestamps are emitted as ISO-8601 UTC. Errors from stripe.error.StripeError subclasses are caught and re-raised as RuntimeError with context, so cron logs show exactly what failed.

Full Script

#!/usr/bin/env python3
"""
Stripe Payment Analytics Exporter

Queries the Stripe API for a rolling or absolute time window and exports
selected metrics (charges, refunds, mrr, customers) to a CSV file.

Usage:
    python3 stripe_export.py --since 30d --metrics charges,refunds,mrr \
        --output stripe_report.csv
"""

from __future__ import annotations

import argparse
import csv
import os
import re
import sys
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Any, Callable, Iterable

import stripe

# ---------------------------------------------------------------------------
# Configuration (environment variables with sensible defaults)
# ---------------------------------------------------------------------------
STRIPE_API_KEY: str | None = os.environ.get("STRIPE_API_KEY")
STRIPE_API_VERSION: str = os.environ.get("STRIPE_API_VERSION", "2024-06-20")
DEFAULT_OUTPUT: str = os.environ.get("STRIPE_EXPORT_OUTPUT", "stripe_report.csv")
DEFAULT_METRICS: str = os.environ.get("STRIPE_EXPORT_METRICS", "charges,refunds")
MAX_PAGES: int = int(os.environ.get("STRIPE_EXPORT_MAX_PAGES", "100"))

# Stripe caps page size at 100 for most list endpoints.
PAGE_SIZE = 100


# ---------------------------------------------------------------------------
# Time window parsing
# ---------------------------------------------------------------------------
_WINDOW_RE = re.compile(r"^(\d+)([dhm])$")


def parse_since(value: str) -> datetime:
    """Parse '30d', '12h', '90m', or an ISO-8601 date into a UTC datetime."""
    match = _WINDOW_RE.match(value.strip().lower())
    if match:
        amount, unit = int(match.group(1)), match.group(2)
        delta = {
            "d": timedelta(days=amount),
            "h": timedelta(hours=amount),
            "m": timedelta(minutes=amount),
        }[unit]
        return datetime.now(timezone.utc) - delta

    # Fall back to ISO-8601 date/datetime
    try:
        dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            f"Invalid --since value {value!r}. Use 30d/12h/90m or ISO-8601."
        ) from exc
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def to_unix(dt: datetime) -> int:
    """Stripe expects integer Unix timestamps (seconds)."""
    return int(dt.timestamp())


def cents_to_decimal(amount: int | None) -> Decimal:
    """Convert Stripe's integer cents into a Decimal with 2 places."""
    return Decimal(amount or 0) / Decimal(100)


# ---------------------------------------------------------------------------
# Metric collectors
# ---------------------------------------------------------------------------
def collect_charges(since: datetime, until: datetime) -> Iterable[dict[str, Any]]:
    """Yield one row per successful charge."""
    params = {
        "created": {"gte": to_unix(since), "lte": to_unix(until)},
        "limit": PAGE_SIZE,
    }
    for i, charge in enumerate(stripe.Charge.list(**params).auto_paging_iter()):
        if i >= MAX_PAGES * PAGE_SIZE:
            break
        yield {
            "metric": "charge",
            "id": charge["id"],
            "created": datetime.fromtimestamp(
                charge["created"], tz=timezone.utc
            ).isoformat(),
            "amount": str(cents_to_decimal(charge["amount"])),
            "currency": charge["currency"].upper(),
            "status": charge["status"],
            "customer": charge.get("customer") or "",
            "description": (charge.get("description") or "")[:200],
        }


def collect_refunds(since: datetime, until: datetime) -> Iterable[dict[str, Any]]:
    """Yield one row per refund."""
    params = {
        "created": {"gte": to_unix(since), "lte": to_unix(until)},
        "limit": PAGE_SIZE,
    }
    for i, refund in enumerate(stripe.Refund.list(**params).auto_paging_iter()):
        if i >= MAX_PAGES * PAGE_SIZE:
            break
        yield {
            "metric": "refund",
            "id": refund["id"],
            "created": datetime.fromtimestamp(
                refund["created"], tz=timezone.utc
            ).isoformat(),
            "amount": str(cents_to_decimal(refund["amount"])),
            "currency": refund["currency"].upper(),
            "status": refund["status"],
            "customer": refund.get("charge") or "",
            "description": (refund.get("reason") or "")[:200],
        }


def collect_mrr(_since: datetime, _until: datetime) -> Iterable[dict[str, Any]]:
    """
    Emit a single MRR row computed from active subscriptions.

    MRR = sum(monthly-normalized price * quantity) for all active subs.
    Yearly prices are divided by 12, weekly multiplied by 52/12.
    """
    total = Decimal(0)
    currency = "USD"
    count = 0

    for sub in stripe.Subscription.list(
        status="active", limit=PAGE_SIZE, expand=["data.items.data.price"]
    ).auto_paging_iter():
        count += 1
        for item in sub["items"]["data"]:
            price = item["price"]
            unit_amount = Decimal(price.get("unit_amount") or 0) / Decimal(100)
            qty = Decimal(item.get("quantity") or 1)
            interval = price.get("recurring", {}).get("interval", "month")
            multiplier = {"month": Decimal(1), "year": Decimal(1) / 12,
                          "week": Decimal(52) / 12, "day": Decimal(365) / 12}[interval]
            total += unit_amount * qty * multiplier
            currency = (price.get("currency") or currency).upper()

    yield {
        "metric": "mrr",
        "id": "aggregate",
        "created": datetime.now(timezone.utc).isoformat(),
        "amount": str(total.quantize(Decimal("0.01"))),
        "currency": currency,
        "status": "active",
        "customer": f"{count} subscriptions",
        "description": "Monthly recurring revenue",
    }


def collect_customers(since: datetime, until: datetime) -> Iterable[dict[str, Any]]:
    """Yield one row per customer created in the window."""
    params = {
        "created": {"gte": to_unix(since), "lte": to_unix(until)},
        "limit": PAGE_SIZE,
    }
    for i, cust in enumerate(stripe.Customer.list(**params).auto_paging_iter()):
        if i >= MAX_PAGES * PAGE_SIZE:
            break
        yield {
            "metric": "customer",
            "id": cust["id"],
            "created": datetime.fromtimestamp(
                cust["created"], tz=timezone.utc
            ).isoformat(),
            "amount": "0",
            "currency": "",
            "status": "created",
            "customer": cust["id"],
            "description": (cust.get("email") or "")[:200],
        }


METRIC_COLLECTORS: dict[str, Callable[[datetime, datetime], Iterable[dict[str, Any]]]] = {
    "charges": collect_charges,
    "refunds": collect_refunds,
    "mrr": collect_mrr,
    "customers": collect_customers,
}

CSV_FIELDS = ["metric", "id", "created", "amount", "currency",
              "status", "customer", "description"]


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Export Stripe payment analytics to CSV."
    )
    p.add_argument("--since", default="30d",
                   help="Time window: 30d, 12h, 90m, or ISO-8601 (default: 30d)")
    p.add_argument("--until", default="now",
                   help="End of window: 'now' or ISO-8601 (default: now)")
    p.add_argument("--metrics", default=DEFAULT_METRICS,
                   help=f"Comma-separated: {','.join(METRIC_COLLECTORS)}")
    p.add_argument("--output", default=DEFAULT_OUTPUT,
                   help="CSV output path (default: %(default)s)")
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    if not STRIPE_API_KEY:
        print("ERROR: STRIPE_API_KEY environment variable is not set.",
              file=sys.stderr)
        return 2

    stripe.api_key = STRIPE_API_KEY
    stripe.api_version = STRIPE_API_VERSION
    # SDK-level retries for transient network errors.
    stripe.max_network_retries = 3

    since = parse_since(args.since)
    until = (datetime.now(timezone.utc) if args.until == "now"
             else parse_since(args.until))

    requested = [m.strip().lower() for m in args.metrics.split(",") if m.strip()]
    unknown = [m for m in requested if m not in METRIC_COLLECTORS]
    if unknown:
        print(f"ERROR: unknown metric(s): {', '.join(unknown)}", file=sys.stderr)
        print(f"Available: {', '.join(METRIC_COLLECTORS)}", file=sys.stderr)
        return 2

    rows: list[dict[str, Any]] = []
    for metric in requested:
        collector = METRIC_COLLECTORS[metric]
        try:
            collected = list(collector(since, until))
        except stripe.error.AuthenticationError as exc:
            print(f"ERROR: Stripe authentication failed: {exc}", file=sys.stderr)
            return 3
        except stripe.error.RateLimitError as exc:
            print(f"ERROR: Stripe rate limit hit for {metric}: {exc}",
                  file=sys.stderr)
            return 4
        except stripe.error.StripeError as exc:
            print(f"ERROR: Stripe API error while collecting {metric}: {exc}",
                  file=sys.stderr)
            return 5
        print(f"  {metric}: {len(collected)} rows", file=sys.stderr)
        rows.extend(collected)

    try:
        with open(args.output, "w", newline="", encoding="utf-8") as fh:
            writer = csv.DictWriter(fh, fieldnames=CSV_FIELDS)
            writer.writeheader()
            writer.writerows(rows)
    except OSError as exc:
        print(f"ERROR: could not write {args.output}: {exc}", file=sys.stderr)
        return 6

    print(f"Wrote {len(rows)} rows to {args.output}", file=sys.stderr)
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

CLI flags --since, --until, --metrics, and --output override the corresponding defaults. --since accepts 30d, 12h, 90m, or any ISO-8601 date/datetime.

Usage Examples

1. Last 30 days of charges and refunds:

$ export STRIPE_API_KEY=sk_test_xxx
$ python3 stripe_export.py --since 30d --metrics charges,refunds --output stripe_report.csv
  charges: 412 rows
  refunds: 18 rows
Wrote 430 rows to stripe_report.csv

2. Current MRR snapshot plus year-to-date customers:

$ python3 stripe_export.py --since 2024-01-01 --metrics mrr,customers --output ytd.csv
  mrr: 1 rows
  customers: 1_284 rows
Wrote 1285 rows to ytd.csv

3. Cron job — daily 7-day rolling export:

0 6 * * * STRIPE_API_KEY=sk_live_xxx /usr/bin/python3 /opt/scripts/stripe_export.py \
    --since 7d --metrics charges,refunds --output /var/reports/stripe_$(date +\%F).csv

Sample CSV output:

metric,id,created,amount,currency,status,customer,description
charge,ch_3Pq...,2024-07-01T09:14:22+00:00,49.00,USD,succeeded,cus_Nx...,Pro plan
refund,re_3Pq...,2024-07-02T11:03:10+00:00,49.00,USD,succeeded,ch_3Pq...,requested_by_customer
mrr,aggregate,2024-07-08T06:00:00+00:00,12840.50,USD,active,312 subscriptions,Monthly recurring revenue

Error Handling

• Missing STRIPE_API_KEY → exits with code 2 and a clear message before any network call.
• Invalid --since → argparse.ArgumentTypeError gives a usage hint rather than a stack trace.
• Unknown metric → lists valid metric names and exits code 2.
• stripe.error.AuthenticationError → exit code 3 (bad/revoked key).
• stripe.error.RateLimitError → exit code 4. The SDK already retries internally; if it still fails, the job is safe to re-run.
• Other stripe.error.StripeError → exit code 5 with the metric name in the message.
• Disk/permission errors on the output file → exit code 6.
• Transient network failures → stripe.max_network_retries = 3 handles most before surfacing.
• Runaway pagination → MAX_PAGES caps rows per metric so a misconfigured window can't exhaust memory.

Because all errors return non-zero exit codes, cron, systemd, or CI pipelines can alert on failure without parsing stderr.

Extending Ideas

• Add more metrics: implement a function with signature (since, until) -> Iterable[dict], register it in METRIC_COLLECTORS, and it's instantly available via --metrics. Good candidates: payouts, disputes, invoices, balance_transactions.
• Multi-account support: read a JSON file of {name: api_key} pairs and loop main() per account, prefixing each CSV with the account name (useful for platforms with connected accounts via stripe_account=).
• Direct-to-warehouse: replace the CSV writer with a boto3 S3 upload or a psycopg COPY into Postgres for scheduled ingestion into a BI tool.
• Aggregated summaries: add a --group-by day|week|month flag that buckets rows and emits totals instead of per-object rows, which is much faster to chart in a spreadsheet.
• Structured logging: swap print(..., file=sys.stderr) for the logging module with JSON formatting so log aggregators (Loki, CloudWatch) can index failures by metric and error type.