Продуктовая аналитика с AI — PostHog, Mixpanel и умные инсайты без дата-аналитика

claude mcp add --transport http posthog https://mcp.posthog.com/mcp

import requests

def get_mixpanel_funnel(project_id, funnel_id, api_secret):
    url = f"https://mixpanel.com/api/2.0/funnels"
    params = {
        "funnel_id": funnel_id,
        "from_date": "2026-09-01",   # пример: подставь свои даты
        "to_date": "2026-09-30",
    }
    response = requests.get(
        url,
        params=params,
        auth=(api_secret, "")
    )
    return response.json()

import anthropic
import json

def predict_churn(users_activity_data: list[dict]) -> dict:
    """
    users_activity_data — список словарей с активностью каждого пользователя:
    [{"user_id": "u123", "last_login_days_ago": 8, "sessions_last_30d": 2,
      "feature_usage": {"core": 5, "advanced": 0}, "plan": "free"}]
    """
    client = anthropic.Anthropic()

    prompt = f"""
Ты аналитик продукта. Вот данные об активности {len(users_activity_data)} пользователей за последние 30 дней:

{json.dumps(users_activity_data, ensure_ascii=False, indent=2)}

Проанализируй паттерны и:
1. Выдели HIGH RISK пользователей (вероятно уйдут в ближайшие 2 недели)
2. Объясни признаки по которым определил риск
3. Предложи конкретное действие для каждой risk-группы (email, звонок, in-app уведомление)

Ответь в формате JSON:
{{
  "high_risk": [список user_id],
  "medium_risk": [список user_id],
  "risk_signals": "объяснение",
  "actions": {{"high_risk": "действие", "medium_risk": "действие"}}
}}
"""

    message = client.messages.create(
        model="claude-sonnet-5-5",   # актуальные модели: страница «Актуальное сейчас»
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )

    return json.loads("".join(b.text for b in message.content if b.type == "text"))

def interpret_ab_test(test_data: dict) -> str:
    client = anthropic.Anthropic()

    message = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"""
Проанализируй результаты A/B теста:
{json.dumps(test_data, ensure_ascii=False)}

Ответь:
1. Какой вариант победил и насколько уверенно?
2. Статистически значимый ли результат (выборка {test_data.get('total_users', 'N/A')} пользователей)?
3. Рекомендуй: раскатывать победителя или ждать больше данных?
Ответ на русском, кратко — 3-4 предложения.
"""
        }]
    )

    return "".join(b.text for b in message.content if b.type == "text")

def analyze_session_patterns(sessions_metadata: list[dict]) -> str:
    rage_clicks = [s for s in sessions_metadata if s.get("rage_clicks", 0) > 2]
    high_exit_pages = {}

    for session in sessions_metadata:
        exit_page = session.get("exit_page", "")
        high_exit_pages[exit_page] = high_exit_pages.get(exit_page, 0) + 1

    analysis_data = {
        "total_sessions": len(sessions_metadata),
        "frustration_sessions": len(rage_clicks),
        "top_exit_pages": sorted(
            high_exit_pages.items(), key=lambda x: x[1], reverse=True
        )[:5],
        "avg_session_duration_minutes": sum(
            s.get("duration_seconds", 0) for s in sessions_metadata
        ) / len(sessions_metadata) / 60
    }

    client = anthropic.Anthropic()
    message = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""
Данные о сессиях пользователей:
{json.dumps(analysis_data, ensure_ascii=False, indent=2)}

Найди главные точки боли пользователей. Что нужно исправить в первую очередь?
Ответ: три конкретные проблемы с предложениями по исправлению.
"""
        }]
    )

    return "".join(b.text for b in message.content if b.type == "text")

import os
import json
import anthropic
import requests
from datetime import datetime, timedelta
from dotenv import load_dotenv

load_dotenv()

# Ключи берём из .env, а не пишем в коде
POSTHOG_PERSONAL_API_KEY = os.getenv("POSTHOG_PERSONAL_API_KEY")  # персональный ключ (phx_...) для чтения данных
POSTHOG_PROJECT_ID = os.getenv("POSTHOG_PROJECT_ID")
POSTHOG_HOST = os.getenv("POSTHOG_HOST", "https://us.posthog.com")  # для Европы: https://eu.posthog.com
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")

def run_hogql(query: str) -> list:
    """Выполняет запрос на HogQL (язык запросов PostHog) через API."""
    response = requests.post(
        f"{POSTHOG_HOST}/api/projects/{POSTHOG_PROJECT_ID}/query/",
        headers={"Authorization": f"Bearer {POSTHOG_PERSONAL_API_KEY}"},
        json={"query": {"kind": "HogQLQuery", "query": query}, "name": "weekly report"},
    )
    return response.json().get("results", []) if response.ok else []

def fetch_posthog_insights() -> dict:
    """Забираем ключевые метрики за прошлую неделю"""
    end_date = datetime.now()
    start_date = end_date - timedelta(days=7)

    # Просмотры страниц по дням
    pageviews_by_day = run_hogql(
        "select toDate(timestamp) as day, count() from events "
        "where event = '$pageview' and timestamp >= now() - interval 7 day "
        "group by day order by day"
    )

    # Активные пользователи за неделю
    active_users = run_hogql(
        "select count(distinct person_id) from events "
        "where timestamp >= now() - interval 7 day"
    )

    # Новые регистрации
    signups = run_hogql(
        "select count() from events "
        "where event = 'signed_up' and timestamp >= now() - interval 7 day"
    )

    return {
        "period": f"{start_date.strftime('%d.%m')} — {end_date.strftime('%d.%m.%Y')}",
        "pageviews_by_day": pageviews_by_day,
        "active_users": active_users,
        "signups": signups,
    }

def generate_weekly_report(insights: dict) -> str:
    """Claude генерирует человеческий отчёт"""
    client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)

    message = client.messages.create(
        model="claude-sonnet-5-5",   # актуальные модели: страница «Актуальное сейчас»
        max_tokens=1500,
        messages=[{
            "role": "user",
            "content": f"""
Ты аналитик продукта. Вот данные за неделю {insights['period']}:

{json.dumps(insights, ensure_ascii=False, indent=2, default=str)}

Напиши еженедельный отчёт для основателя стартапа. Структура:
📊 НЕДЕЛЯ {insights['period']}

🟢 Что хорошо (2-3 пункта)
🔴 На что обратить внимание (1-2 пункта)
🎯 Приоритет на эту неделю (1 конкретное действие)

Тон: дружелюбный, конкретный, без воды. Пиши по-русски.
Если данных мало, так и скажи, не выдумывай причины.
"""
        }]
    )

    return "".join(b.text for b in message.content if b.type == "text")

def send_to_telegram(text: str):
    """Отправляем отчёт в Telegram"""
    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
        json={
            "chat_id": TELEGRAM_CHAT_ID,
            "text": text,
        }
    )

def run_weekly_report():
    print("Собираем данные из PostHog...")
    insights = fetch_posthog_insights()

    print("Claude анализирует...")
    report = generate_weekly_report(insights)

    print("Отправляем в Telegram...")
    send_to_telegram(report)
    print("Готово!")

if __name__ == "__main__":
    run_weekly_report()

<script>
  !function(t,e){var o,n,p,r;e.__SV||(window.posthog=e,e._i=[],e.init=function(i,s,a){...}
  posthog.init('phc_ваш_ключ', {api_host: 'https://us.i.posthog.com'})   // для Европы: https://eu.i.posthog.com
</script>

pip install posthog

from posthog import Posthog

posthog = Posthog('phc_ваш_ключ', host='https://us.i.posthog.com')
# Именованные аргументы работают в разных версиях библиотеки
posthog.capture(event='signed_up', distinct_id='user_123', properties={'plan': 'free', 'source': 'google'})

mkdir analytics_bot
cd analytics_bot
pip install anthropic requests python-dotenv

POSTHOG_PERSONAL_API_KEY=phx_ваш_персональный_ключ
POSTHOG_PROJECT_ID=ваш_id_проекта
POSTHOG_HOST=https://us.posthog.com
ANTHROPIC_API_KEY=sk-ant-ваш_ключ
TELEGRAM_BOT_TOKEN=ваш_токен
TELEGRAM_CHAT_ID=ваш_chat_id

python weekly_report.py

from dotenv import load_dotenv
import os
import anthropic
import requests
import json

load_dotenv()

POSTHOG_HOST = os.getenv("POSTHOG_HOST", "https://us.posthog.com")

def get_at_risk_users():
    """Получаем пользователей не заходивших 7+ дней (упрощённо: фильтр и поля сверь с документацией PostHog)"""
    headers = {"Authorization": f"Bearer {os.getenv('POSTHOG_PERSONAL_API_KEY')}"}

    # Упрощённый запрос — пользователи с низкой активностью
    response = requests.get(
        f"{POSTHOG_HOST}/api/projects/{os.getenv('POSTHOG_PROJECT_ID')}/persons/",
        headers=headers,
        params={"properties": '[{"key":"$last_seen","value":"-7d","operator":"is_before_date"}]'}
    )

    if not response.ok:
        return []

    persons = response.json().get("results", [])
    # В модель отправляем только внутренние ID и счётчики, без имён и email
    return [
        {
            "user_id": p.get("distinct_ids", ["unknown"])[0],
            "last_seen": p.get("properties", {}).get("$last_seen"),
            "total_sessions": p.get("properties", {}).get("$session_count", 0),
        }
        for p in persons[:50]  # Берём первые 50
    ]

def run_churn_check():
    users = get_at_risk_users()
    if not users:
        print("Нет пользователей в зоне риска")
        return

    client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
    message = client.messages.create(
        model="claude-haiku-4-5",  # Haiku достаточно для этой задачи (проверь, что модель ещё доступна в API)
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""
Вот пользователи которые давно не заходили ({len(users)} человек):
{json.dumps(users[:10], ensure_ascii=False)}

Что им написать чтобы вернуть? Одно короткое письмо (3-4 предложения) для реактивации.
"""
        }]
    )

    print("Рекомендованный текст для реактивации:")
    print("".join(b.text for b in message.content if b.type == "text"))
    print(f"\nПользователей в зоне риска: {len(users)}")

if __name__ == "__main__":
    run_churn_check()

# Открыть crontab
crontab -e

# Добавить строки:
# Понедельник 9:00 — недельный отчёт
0 9 * * 1 cd /path/to/analytics_bot && python weekly_report.py

# Пятница 17:00 — проверка оттока
0 17 * * 5 cd /path/to/analytics_bot && python churn_detector.py >> churn_log.txt