import os, requests, time, json
from collections import Counter, defaultdict
AMP_KEY = os.environ["AMPLITUDE_API_KEY"]
AMP_SECRET = os.environ["AMPLITUDE_SECRET_KEY"]
MV = os.environ["MAVERA_API_KEY"]
amp_auth = (AMP_KEY, AMP_SECRET)
CONVERTER_IDS = os.environ.get("CONVERTER_USER_IDS", "user_101,user_102,user_103,user_104,user_105").split(",")
NON_CONVERTER_IDS = os.environ.get("NON_CONVERTER_USER_IDS", "user_201,user_202,user_203,user_204,user_205").split(",")
def fetch_activity(user_id):
r = requests.get(
"https://amplitude.com/api/2/useractivity",
auth=amp_auth,
params={"user": user_id},
)
if r.status_code == 429:
time.sleep(int(r.headers.get("Retry-After", 10)))
return fetch_activity(user_id)
if r.status_code != 200:
return []
data = r.json()
return data.get("userData", {}).get("events", [])
def extract_path(events, max_events=30):
path = []
for e in events[:max_events]:
name = e.get("event_type", "unknown")
if name.startswith("$"):
continue
path.append(name)
return path
converter_paths = []
for uid in CONVERTER_IDS:
events = fetch_activity(uid)
path = extract_path(events)
if path:
converter_paths.append(path)
time.sleep(1)
non_converter_paths = []
for uid in NON_CONVERTER_IDS:
events = fetch_activity(uid)
path = extract_path(events)
if path:
non_converter_paths.append(path)
time.sleep(1)
def aggregate_paths(paths):
step_freq = Counter()
bigrams = Counter()
path_lengths = []
last_events = Counter()
for path in paths:
path_lengths.append(len(path))
for event in path:
step_freq[event] += 1
for i in range(len(path) - 1):
bigrams[f"{path[i]} → {path[i+1]}"] += 1
if path:
last_events[path[-1]] += 1
return {
"count": len(paths),
"avg_length": sum(path_lengths) / max(len(path_lengths), 1),
"top_events": step_freq.most_common(10),
"top_transitions": bigrams.most_common(10),
"last_events": last_events.most_common(5),
}
conv_agg = aggregate_paths(converter_paths)
non_conv_agg = aggregate_paths(non_converter_paths)
def format_agg(agg, label):
events = "\n".join(f" {e}: {c} occurrences" for e, c in agg["top_events"])
transitions = "\n".join(f" {t}: {c} times" for t, c in agg["top_transitions"])
last = "\n".join(f" {e}: {c}" for e, c in agg["last_events"])
return (
f" **{label}** ({agg['count']} users, avg {agg['avg_length']:.0f} events/journey)\n"
f" Top events:\n{events}\n"
f" Top transitions:\n{transitions}\n"
f" Final events before exit:\n{last}"
)
journey_data = f"""CONVERTING USERS:
{format_agg(conv_agg, "Converters")}
NON-CONVERTING USERS:
{format_agg(non_conv_agg, "Non-Converters")}"""
mave = requests.post(
"https://app.mavera.io/api/v1/mave/chat",
headers={"Authorization": f"Bearer {MV}", "Content-Type": "application/json"},
json={"message": f"""Analyze these user journey patterns from Amplitude. Common paths to conversion? Where do users get stuck?
{journey_data}
Provide:
1. The "golden path" — the most common event sequence leading to conversion
2. Divergence points — where converters and non-converters split
3. Friction signals — events that appear in non-converter paths but not converter paths
4. Drop-off cliffs — transitions where non-converters commonly exit
5. Recommendations for reducing friction at each identified stuck point
6. A simplified journey map with 4-6 key stages"""},
).json()
print("--- Journey Analysis ---")
print(mave.get("content", "")[:3000])
