User Composition → Persona Calibration

import os, requests, time
from collections import defaultdict

AMP_KEY = os.environ["AMPLITUDE_API_KEY"]
AMP_SECRET = os.environ["AMPLITUDE_SECRET_KEY"]
MV = os.environ["MAVERA_API_KEY"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
amp_auth = (AMP_KEY, AMP_SECRET)

r = requests.get(
    "https://amplitude.com/api/2/composition",
    auth=amp_auth,
    params={"start": "20260215", "end": "20260317"},
)
if r.status_code == 429:
    retry = int(r.headers.get("Retry-After", 60))
    time.sleep(retry)
    r = requests.get(
        "https://amplitude.com/api/2/composition",
        auth=amp_auth,
        params={"start": "20260215", "end": "20260317"},
    )
r.raise_for_status()
composition = r.json()

series = composition.get("data", {}).get("series", [])
xvals = composition.get("data", {}).get("xValues", [])

segments = defaultdict(lambda: {"total": 0, "countries": defaultdict(int),
    "platforms": defaultdict(int), "devices": defaultdict(int)})

for i, date_label in enumerate(xvals):
    for seg in series:
        name = seg.get("name", "all")
        val = seg.get("value", [0] * len(xvals))
        count = val[i] if i < len(val) else 0
        segments[name]["total"] += count

existing = requests.get(f"{MB}/personas", headers=MH).json()
existing_names = {p.get("name", ""): p for p in (existing if isinstance(existing, list) else [])}

PERSONA_SEGMENTS = {
    "Mobile Power Users": {
        "filter": lambda s: "mobile" in s.lower() or "ios" in s.lower() or "android" in s.lower(),
        "description": "Users primarily accessing via mobile devices. Shorter sessions, gesture-driven navigation, on-the-go usage patterns.",
    },
    "Desktop Evaluators": {
        "filter": lambda s: "desktop" in s.lower() or "windows" in s.lower() or "mac" in s.lower(),
        "description": "Desktop-first users. Longer sessions, deeper feature exploration, likely evaluating the product for team adoption.",
    },
    "International Users": {
        "filter": lambda s: any(c in s.lower() for c in ["uk", "de", "fr", "in", "br", "jp", "au"]),
        "description": "Non-US users who may need localized content, timezone-aware engagement, and region-specific messaging.",
    },
}

created = []
for persona_name, config in PERSONA_SEGMENTS.items():
    matching = {k: v for k, v in segments.items() if config["filter"](k)}
    if not matching:
        continue

    total = sum(v["total"] for v in matching.values())
    segment_names = list(matching.keys())[:5]

    full_name = f"Amplitude: {persona_name}"
    payload = {
        "name": full_name,
        "description": (
            f"{config['description']} "
            f"Amplitude composition data (30d). Total: {total:,} users. "
            f"Segments: {', '.join(segment_names)}."
        ),
        "demographic": {
            "source": "amplitude_composition",
            "segments": segment_names,
            "total_users": total,
        },
    }

    if full_name in existing_names:
        pid = existing_names[full_name]["id"]
        requests.patch(f"{MB}/personas/{pid}", headers=MH, json=payload).raise_for_status()
        created.append({"name": full_name, "id": pid, "action": "updated", "users": total})
        print(f"  Updated: {full_name} ({pid}) — {total:,} users")
    else:
        p = requests.post(f"{MB}/personas", headers=MH, json=payload).json()
        created.append({"name": full_name, "id": p["id"], "action": "created", "users": total})
        print(f"  Created: {full_name} ({p['id']}) — {total:,} users")
    time.sleep(0.3)

overall_total = sum(s["total"] for s in segments.values())
print(f"\nCalibrated {len(created)} personas from {overall_total:,} total composition data points")