Predictive Analytics → Persona Enrichment

import os, requests, time
from collections import defaultdict

KL_KEY = os.environ["KLAVIYO_API_KEY"]
MV = os.environ["MAVERA_API_KEY"]
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}
KH = {
    "Authorization": f"Klaviyo-API-Key {KL_KEY}",
    "Content-Type": "application/json",
    "revision": "2024-10-15",
}
KB = "https://a.klaviyo.com/api"

profiles = []
url = f"{KB}/profiles"
params = {
    "fields[profile]": "email,properties,predictive_analytics,location",
    "page[size]": 100,
}

for page_num in range(5):
    r = requests.get(url, headers=KH, params=params if page_num == 0 else None)
    if r.status_code == 429:
        retry = int(r.headers.get("Retry-After", 10))
        time.sleep(retry)
        continue
    r.raise_for_status()
    data = r.json()
    profiles.extend(data.get("data", []))

    next_link = data.get("links", {}).get("next")
    if not next_link:
        break
    url = next_link
    params = None
    time.sleep(0.5)

print(f"Fetched {len(profiles)} profiles with predictive data\n")

tiers = {"high_value": [], "growth": [], "at_risk": [], "low_value": []}
for p in profiles:
    attrs = p.get("attributes", {})
    pred = attrs.get("predictive_analytics", {})
    props = attrs.get("properties", {})

    clv = pred.get("predicted_customer_lifetime_value") or pred.get("historic_clv") or 0
    churn = pred.get("predicted_churn_risk") or pred.get("churn_probability") or 0
    next_order = pred.get("expected_date_of_next_order", "")

    profile_data = {
        "clv": float(clv),
        "churn": float(churn) if isinstance(churn, (int, float)) else 0.5,
        "next_order": next_order,
        "city": attrs.get("location", {}).get("city", ""),
        "country": attrs.get("location", {}).get("country", ""),
        **{k: v for k, v in props.items() if isinstance(v, (str, int, float))},
    }

    if profile_data["clv"] > 500 and profile_data["churn"] < 0.2:
        tiers["high_value"].append(profile_data)
    elif profile_data["clv"] > 150:
        tiers["growth"].append(profile_data)
    elif profile_data["churn"] > 0.5:
        tiers["at_risk"].append(profile_data)
    else:
        tiers["low_value"].append(profile_data)

def tier_summary(users, label):
    if not users:
        return None
    avg_clv = sum(u["clv"] for u in users) / len(users)
    avg_churn = sum(u["churn"] for u in users) / len(users)
    countries = defaultdict(int)
    for u in users:
        if u.get("country"):
            countries[u["country"]] += 1
    top_countries = sorted(countries, key=countries.get, reverse=True)[:3]

    return {
        "label": label, "n": len(users),
        "avg_clv": avg_clv, "avg_churn": avg_churn,
        "top_countries": top_countries,
    }

created = []
for tier_name, users in tiers.items():
    summary = tier_summary(users, tier_name)
    if not summary or summary["n"] < 3:
        continue

    label = tier_name.replace("_", " ").title()
    churn_desc = "low" if summary["avg_churn"] < 0.2 else "medium" if summary["avg_churn"] < 0.5 else "high"

    persona = requests.post(f"{MB}/personas", headers=MH, json={
        "name": f"Klaviyo: {label}",
        "description": (
            f"{label} tier from Klaviyo predictive analytics. N={summary['n']}. "
            f"Avg predicted CLV: ${summary['avg_clv']:.0f}. "
            f"Avg churn risk: {summary['avg_churn']:.0%} ({churn_desc}). "
            f"Top countries: {', '.join(summary['top_countries']) if summary['top_countries'] else 'N/A'}."
        ),
        "demographic": {
            "source": "klaviyo_predictive",
            "countries": summary["top_countries"],
        },
        "psychographic": {
            "clv_tier": tier_name,
            "avg_predicted_clv": summary["avg_clv"],
            "avg_churn_risk": summary["avg_churn"],
            "value_segment": churn_desc,
        },
    }).json()

    created.append({"tier": label, "id": persona["id"], "n": summary["n"], "clv": summary["avg_clv"]})
    print(f"  {label}: {persona['id']} (N={summary['n']}, CLV=${summary['avg_clv']:.0f}, churn={summary['avg_churn']:.0%})")
    time.sleep(0.3)

print(f"\nCreated {len(created)} predictive personas")