Predictive Analytics → Persona Enrichment

Pull Klaviyo profiles with predictive CLV and churn risk, group by value tier, and create Mavera Custom Personas enriched with predictive data

Scenario

Klaviyo computes predictive analytics for each profile — predicted customer lifetime value (CLV), churn risk probability, expected next order date, and historic CLV. You pull profiles with these predictions, group them by CLV tier (high/medium/low), and create Mavera Custom Personas enriched with predictive data. The result is personas that know not just who your customers are, but what they’re worth and how likely they are to leave.

Architecture

Code

Python

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")

const KL_KEY = process.env.KLAVIYO_API_KEY;
const MV = process.env.MAVERA_API_KEY;
const MB = "https://app.mavera.io/api/v1";
const MH = { Authorization: `Bearer ${MV}`, "Content-Type": "application/json" };
const KH = {
  Authorization: `Klaviyo-API-Key ${KL_KEY}`,
  "Content-Type": "application/json",
  revision: "2024-10-15",
};

const profiles = [];
let url = "https://a.klaviyo.com/api/profiles?fields[profile]=email,properties,predictive_analytics,location&page[size]=100";

for (let page = 0; page < 5; page++) {
  const res = await fetch(url, { headers: KH });
  if (res.status === 429) {
    await new Promise((r) => setTimeout(r, parseInt(res.headers.get("Retry-After") || "10") * 1000));
    continue;
  }
  const data = await res.json();
  profiles.push(...(data.data || []));
  url = data.links?.next;
  if (!url) break;
  await new Promise((r) => setTimeout(r, 500));
}

console.log(`Fetched ${profiles.length} profiles\n`);

const tiers = { high_value: [], growth: [], at_risk: [], low_value: [] };
for (const p of profiles) {
  const attrs = p.attributes || {};
  const pred = attrs.predictive_analytics || {};
  const clv = parseFloat(pred.predicted_customer_lifetime_value || pred.historic_clv || 0);
  const churn = parseFloat(pred.predicted_churn_risk || pred.churn_probability || 0.5);

  const data = { clv, churn, country: attrs.location?.country || "", ...Object.fromEntries(Object.entries(attrs.properties || {}).filter(([, v]) => typeof v !== "object")) };

  if (clv > 500 && churn < 0.2) tiers.high_value.push(data);
  else if (clv > 150) tiers.growth.push(data);
  else if (churn > 0.5) tiers.at_risk.push(data);
  else tiers.low_value.push(data);
}

const created = [];
for (const [tierName, users] of Object.entries(tiers)) {
  if (users.length < 3) continue;
  const label = tierName.replace(/_/g, " ").replace(/\b\w/g, (c) => c.toUpperCase());
  const avgClv = users.reduce((s, u) => s + u.clv, 0) / users.length;
  const avgChurn = users.reduce((s, u) => s + u.churn, 0) / users.length;

  const countries = {};
  users.forEach((u) => { if (u.country) countries[u.country] = (countries[u.country] || 0) + 1; });
  const topCountries = Object.keys(countries).sort((a, b) => countries[b] - countries[a]).slice(0, 3);

  const p = await fetch(`${MB}/personas`, { method: "POST", headers: MH,
    body: JSON.stringify({
      name: `Klaviyo: ${label}`,
      description: `${label} from Klaviyo predictive. N=${users.length}. CLV: $${avgClv.toFixed(0)}. Churn: ${(avgChurn * 100).toFixed(0)}%. Countries: ${topCountries.join(", ") || "N/A"}.`,
      demographic: { source: "klaviyo_predictive", countries: topCountries },
      psychographic: { clv_tier: tierName, avg_predicted_clv: avgClv, avg_churn_risk: avgChurn },
    }),
  }).then((r) => r.json());

  created.push({ tier: label, id: p.id, n: users.length, clv: avgClv });
  console.log(`  ${label}: ${p.id} (N=${users.length}, CLV=$${avgClv.toFixed(0)}, churn=${(avgChurn * 100).toFixed(0)}%)`);
  await new Promise((r) => setTimeout(r, 300));
}

console.log(`\nCreated ${created.length} predictive personas`);

Example Output

{
  "profiles_fetched": 500,
  "personas": [
    { "tier": "High Value", "id": "per_kl_hv_1", "n": 48, "avg_clv": 1240, "avg_churn": "8%" },
    { "tier": "Growth", "id": "per_kl_gr_2", "n": 156, "avg_clv": 310, "avg_churn": "25%" },
    { "tier": "At Risk", "id": "per_kl_ar_3", "n": 89, "avg_clv": 120, "avg_churn": "68%" },
    { "tier": "Low Value", "id": "per_kl_lv_4", "n": 207, "avg_clv": 45, "avg_churn": "35%" }
  ]
}

Error Handling

Predictive analytics availability

Predictive CLV and churn risk require Klaviyo Growth+ plan and sufficient data (500+ customers, 180+ days of orders). On lower plans, predictive_analytics fields are empty. Fall back to historic_clv from order history.

Revision header required

Every Klaviyo API request must include a revision header. Without it, requests fail with 400. Use the latest stable revision from Klaviyo’s API changelog.

Tiered rate limits

Klaviyo’s rate limits vary: Profiles GET allows 10-75 req/sec depending on plan. On 429, read Retry-After header. The burst limit (up to 350/sec) applies to Tracking/Events endpoints, not Profiles.

---

What’s Next

Klaviyo Integration

Back to Klaviyo integration overview

Flow Content Refresh

Rewrite underperforming flow content

Personas API

Full reference for POST /api/v1/personas

Mave Agent

Full reference for POST /api/v1/mave/chat