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