ActivationPal

Funnel analysis Python: onboarding counts

Python can rank differences between mobile onboarding stage counts retrieved from ActivationPal's stats API. This funnel analysis Python example fetches those counts, calculates adjacent ratios and cohort shares, and produces a ranked drop-off report. Run it offline first, then switch to your app's response.

What the onboarding counts measure

Use funnel.cohort as the installed count. It counts distinct app/device identities with first_open inside the selected reporting window. Each stage counts identities from that cohort whose matching event also falls inside that window.

funnel.steps contains observed onboarding steps ordered by index. Each step has index, id, count, answers and pct; count is distinct devices, while answer totals count responses. funnel.completed counts distinct devices that sent onboarding_completed.

These are independent stage-reach totals, not timestamp-ordered intersections of the same people. Dividing one stage by its predecessor compares counts; it does not establish conditional conversion or prove individual abandonment. The event definitions explain the fixed onboarding and paywall events that feed these stages.

Keep all denominators in that cohort. Headline installs, raw event totals, answer totals and paywallPlacements are different inputs. Missing step indexes mean unobserved steps, not measured zeroes, so this script connects only the observed steps.

Fetch ActivationPal funnel counts with Python

Create a read-only account key in dashboard settings under the mcp tab. Keys read only the apps their owner can access, narrowed by the key's scope. Set AP_KEY in your process environment, then use authenticated GET /api/v1/list_apps to find an app's id.

The live request is GET /api/v1/get_stats?app=APP_ID&range=7d&tz=UTC with an Authorization: Bearer header. Choose one app: portfolio mode omits individual onboarding steps. Explicit UTC makes the reporting boundaries consistent between runs; use that timezone when comparing against the dashboard.

The response is { "ok": true, ...snapshot }: read response["funnel"], without a data wrapper. The API also supplies funnel.transitions with fromKey, toKey, rate, lost and pct. This example calculates from raw counts so its denominator checks and treatment of increasing stages are visible.

Run the Python example on sample data

Save this complete script as funnel_analysis.py. It uses Python's standard library and defaults to an offline fixture. The fixture is an illustrative response excerpt using repository-verified fields, based on the repository's onboarding-transition test: 10 installed, 8 at welcome, 5 completed, 5 shown a paywall, 3 selected a plan and 1 purchased. These are synthetic counts, not customer results or a captured live response.

import argparse
import json
import os
import urllib.parse
import urllib.request


SAMPLE = {
    "ok": True,
    "error": None,
    "funnel": {
        "cohort": 10,
        "steps": [
            {"index": 0, "id": "welcome", "count": 8,
             "answers": [], "pct": 80}
        ],
        "completed": 5,
        "paywall": {"shown": 5, "planSelected": 3, "purchased": 1},
    },
}


def count(value):
    if type(value) is not int or value < 0:
        raise ValueError("Counts and indexes must be nonnegative integers")
    return value


def percent(numerator, denominator):
    return None if denominator == 0 else numerator / denominator * 100


def stages(payload):
    if not isinstance(payload, dict):
        raise ValueError("Expected a response object")
    if payload.get("ok") is not True or payload.get("error"):
        raise ValueError("API did not return a successful snapshot")
    try:
        funnel = payload["funnel"]
        cohort = count(funnel["cohort"])
        steps = funnel["steps"]
        if not isinstance(steps, list):
            raise ValueError("Expected a steps array")
        observed = []
        seen = set()
        for step in steps:
            index = count(step["index"])
            if index in seen:
                raise ValueError("Duplicate step index")
            seen.add(index)
            label = step["id"]
            if label is not None and not isinstance(label, str):
                raise ValueError("Expected a string or null step id")
            label = " ".join((label or f"step-{index}").split())
            observed.append((index, label, count(step["count"])))
        result = [("installed", cohort)]
        result += [(label, n) for _, label, n in sorted(observed)]
        result.append(("completed", count(funnel["completed"])))
        paywall = funnel["paywall"]
        if paywall is not None:
            if not isinstance(paywall, dict):
                raise ValueError("Expected a paywall object or null")
            for field in ("shown", "planSelected", "purchased"):
                result.append((field, count(paywall[field])))
        if any(n > cohort for _, n in result):
            raise ValueError("Stage count exceeds install cohort")
        return cohort, result
    except (KeyError, TypeError) as exc:
        raise ValueError("Missing or malformed required funnel fields") from exc


def analyze(payload):
    cohort, sequence = stages(payload)
    ranked, increases = [], []
    for order, ((left, source), (right, destination)) in enumerate(
        zip(sequence, sequence[1:])
    ):
        row = {
            "transition": f"{left} -> {right}",
            "source": source,
            "destination": destination,
            "ratio": percent(destination, source),
            "gap": source - destination,
            "share": percent(destination, cohort),
            "order": order,
        }
        (increases if destination > source else ranked).append(row)
    ranked.sort(key=lambda row: (-row["gap"], row["order"]))
    return ranked, increases


def display(rows):
    width = max([32] + [len(row["transition"]) for row in rows])
    print(f"{'Transition':<{width}} {'From':>5} {'To':>5} "
          f"{'Ratio':>8} {'Gap':>5} {'Cohort':>8}")
    def fmt(value):
        return "N/A" if value is None else f"{value:.1f}%"
    for row in rows:
        print(f"{row['transition']:<{width}} {row['source']:>5} "
              f"{row['destination']:>5} {fmt(row['ratio']):>8} "
              f"{row['gap']:>5} {fmt(row['share']):>8}")


def self_check():
    ranked, increases = analyze(SAMPLE)
    assert len(ranked) == 5 and not increases
    assert ranked[0]["transition"] == "welcome -> completed"
    assert (ranked[0]["gap"], ranked[0]["ratio"],
            ranked[0]["share"]) == (3, 62.5, 50.0)
    assert [row["order"] for row in ranked] == [1, 0, 3, 4, 2]
    def fixture(cohort=0, step=0, completed=0):
        return {"ok": True, "funnel": {
            "cohort": cohort,
            "steps": [{"index": 0, "id": "welcome", "count": step}],
            "completed": completed, "paywall": None}}
    zero, _ = analyze(fixture())
    assert all(row["ratio"] is None and row["share"] is None
               for row in zero)
    _, up = analyze(fixture(10, 0, 2))
    assert len(up) == 1 and up[0]["ratio"] is None
    assert up[0]["gap"] == -2
    _, up = analyze(fixture(10, 2, 3))
    assert up[0]["ratio"] == 150.0
    assert len(analyze(fixture(10, 8, 5))[0]) == 2
    invalid = [fixture(10, value) for value in (-1, 1.5, True, 11)]
    missing = fixture()
    del missing["funnel"]["completed"]
    duplicate = fixture()
    duplicate["funnel"]["steps"] *= 2
    invalid += [missing, duplicate, {"ok": True},
                {"ok": False}, dict(SAMPLE, error="unavailable")]
    for payload in invalid:
        try:
            analyze(payload)
        except ValueError:
            continue
        raise AssertionError("Invalid input was accepted")


def fetch(app):
    key = os.environ.get("AP_KEY", "").strip()
    if not key:
        raise ValueError("Set AP_KEY in the process environment")
    query = urllib.parse.urlencode({"app": app, "range": "7d", "tz": "UTC"})
    request = urllib.request.Request(
        "https://activationpal.com/api/v1/get_stats?" + query,
        headers={"Authorization": "Bearer " + key},
    )
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return json.load(response)
    except (OSError, ValueError) as exc:
        raise ValueError("Stats request failed: HTTP, network or JSON error") from exc


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--live", action="store_true")
    parser.add_argument("--app")
    args = parser.parse_args()
    if args.live and (not args.app or args.app == "all"):
        parser.error("--live requires --app with one app id")
    if args.app and not args.live:
        parser.error("Use --live with --app")
    self_check()
    try:
        ranked, increases = analyze(fetch(args.app) if args.live else SAMPLE)
    except ValueError as exc:
        parser.exit(1, f"{exc}\n")
    print("Ranked count gaps")
    display(ranked)
    if increases:
        print("\nIncreasing counts: excluded from ranking")
        display(increases)


if __name__ == "__main__":
    main()

Run python3 funnel_analysis.py for the sample. After configuring AP_KEY, run python3 funnel_analysis.py --live --app APP_ID to fetch your app. Only --live sends a request; neither mode prints the key. An HTTP failure, invalid JSON, unsuccessful response or invalid required count stops the report.

Sample output:

Ranked count gaps
Transition                        From    To    Ratio   Gap   Cohort
welcome -> completed                 8     5    62.5%     3    50.0%
installed -> welcome                10     8    80.0%     2    80.0%
shown -> planSelected                5     3    60.0%     2    30.0%
planSelected -> purchased            3     1    33.3%     2    10.0%
completed -> shown                   5     5   100.0%     0    50.0%

Rank onboarding drop-offs

The report calculates three values for every adjacent pair: count ratio is destination divided by source times 100; count gap is source minus destination; cohort share is destination divided by installed times 100. Sorting by descending positive gap surfaces the largest count differences first. Ties retain stage order, and zero-gap rows follow positive gaps.

In this fixture, welcome to completed ranks first: 8 devices reached welcome and 5 completed. The ratio is 62.5%, the gap is 3, and completion represents 50% of the install cohort. That makes this segment the first place to inspect instrumentation and the flow, rather than evidence of three proven abandonments.

Use ActivationPal to inspect the corresponding onboarding stages and answer splits. Check that step indexes and completion events match the intended screens before changing a screen based on this ranking. Rerun the same app, range and timezone after investigating, keeping the denominator consistent.

Handle zero counts and increasing stages

A zero denominator prints N/A. Zero installs therefore produces unavailable cohort shares; a zero source produces an unavailable adjacent ratio. A destination count of zero with a positive source produces 0.0%, which is different from an undefined ratio.

A later stage can exceed an earlier one because the counts are independent: a device may emit completion without an observed preceding step. The script preserves the negative gap and any defined ratio, marks that transition as increasing, and excludes it from the ranking. It never clamps the gap into a convincing-looking loss.

Negative, fractional or boolean counts, duplicate step indexes, missing required fields and counts above cohort size are rejected. A null paywall ends the sequence at completed; it does not create three zero-valued stages. The executable assertions cover these cases as well as the sample ranking, so the offline run checks the calculation before printing results.

Frequently asked questions

How do you calculate funnel conversion in Python?

Divide the destination count by the source count and multiply by 100, returning an unavailable value when the source is zero. For this API, that is an adjacent reach-count ratio; conditional conversion requires checking that the same devices passed through both stages in order.

Why can a later funnel stage have more devices?

Each stage counts its own distinct devices from the install cohort, without requiring an earlier stage event. Missing instrumentation or paths that skip a step can therefore produce an increase. Inspect those transitions separately before treating them as drop-offs.

What happens when a denominator is zero?

The script returns None internally and prints N/A, rather than dividing by zero. A zero numerator with a positive denominator still produces a valid 0.0%.

Can this API build a funnel from arbitrary custom events?

The stats snapshot's onboarding funnel uses fixed onboarding and paywall event names, not an arbitrary custom-event sequence. Custom events are available through get_events, but constructing an ordered funnel from them requires separate event-level analysis.