AI Email Marketing Automation System Development

AI-система автоматизации email-маркетинга

Email-автоматизация без AI — статичные последовательности: «через 3 дня после регистрации отправить письмо N». AI-автоматизация адаптирует содержимое, время отправки и путь каждого получателя индивидуально. Разница: 25-40% прирост click-through rate и 2-3x улучшение конверсии.

Предсказание оптимального времени отправки

import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from anthropic import Anthropic
import json

class SendTimeOptimizer:
    """Оптимальное время отправки для каждого получателя"""

    def __init__(self):
        self.model = GradientBoostingRegressor(
            n_estimators=100, learning_rate=0.1, random_state=42
        )

    def train(self, email_history: pd.DataFrame):
        """
        email_history: user_id, sent_hour, sent_weekday, opened (bool), clicked (bool)
        """
        features = self._extract_features(email_history)
        target = email_history['clicked'].astype(float)
        self.model.fit(features, target)

    def _extract_features(self, df: pd.DataFrame) -> pd.DataFrame:
        return pd.DataFrame({
            'sent_hour': df['sent_hour'],
            'sent_weekday': df['sent_weekday'],
            'is_weekend': (df['sent_weekday'] >= 5).astype(int),
            'is_morning': ((df['sent_hour'] >= 7) & (df['sent_hour'] <= 10)).astype(int),
            'is_lunch': ((df['sent_hour'] >= 12) & (df['sent_hour'] <= 14)).astype(int),
            'is_evening': ((df['sent_hour'] >= 19) & (df['sent_hour'] <= 22)).astype(int),
        })

    def predict_best_time(self, user_id: str,
                           user_open_history: list[dict]) -> dict:
        """Лучший час и день для пользователя"""
        if not user_open_history:
            return {'best_hour': 10, 'best_weekday': 1, 'confidence': 0.3}

        # Анализ паттернов открытий конкретного пользователя
        opens_df = pd.DataFrame(user_open_history)
        opens_df = opens_df[opens_df['opened'] == True]

        if len(opens_df) < 5:
            # Недостаточно данных → общий паттерн аудитории
            return {'best_hour': 10, 'best_weekday': 1, 'confidence': 0.4}

        # Перебираем все слоты и предсказываем CTR
        best_score, best_hour, best_weekday = -1, 10, 1

        for weekday in range(5):  # Пн-Пт
            for hour in [8, 10, 12, 14, 17, 19]:
                features = self._extract_features(pd.DataFrame([{
                    'sent_hour': hour, 'sent_weekday': weekday
                }]))
                score = self.model.predict(features)[0]
                if score > best_score:
                    best_score, best_hour, best_weekday = score, hour, weekday

        # Уверенность растёт с количеством данных
        confidence = min(0.95, 0.4 + len(opens_df) * 0.02)

        return {
            'best_hour': best_hour,
            'best_weekday': best_weekday,
            'predicted_ctr': round(best_score, 3),
            'confidence': round(confidence, 2)
        }


class EmailContentPersonalizer:
    """Персонализация содержимого писем"""

    def __init__(self):
        self.llm = Anthropic()

    def generate_personalized_email(self, template: dict,
                                     recipient: dict,
                                     campaign_type: str) -> dict:
        """Генерация персонализированного email"""
        context = {
            'name': recipient.get('first_name', 'Пользователь'),
            'company': recipient.get('company', ''),
            'industry': recipient.get('industry', ''),
            'last_product_used': recipient.get('last_feature', ''),
            'days_inactive': recipient.get('days_since_last_login', 0),
            'plan': recipient.get('subscription_plan', 'free'),
        }

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=400,
            messages=[{
                "role": "user",
                "content": f"""Write a personalized email for this recipient.

Campaign type: {campaign_type}
Template theme: {template.get('theme', '')}
Recipient context: {json.dumps(context, ensure_ascii=False)}

Requirements:
- Subject line: compelling, 40-60 chars, personalized
- Body: 3-4 short paragraphs
- CTA: single clear action
- Tone: professional but warm
- Language: Russian
- NO generic phrases like "we hope this email finds you well"

Return JSON: {{"subject": "...", "body": "...", "cta_text": "...", "cta_url_param": "..."}}"""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return {
                'subject': f"{context['name']}, специальное предложение",
                'body': template.get('default_body', ''),
                'cta_text': 'Открыть',
                'cta_url_param': ''
            }

    def generate_subject_line_variants(self, base_subject: str,
                                        audience_segment: str,
                                        n_variants: int = 5) -> list[str]:
        """A/B тестирование: несколько вариантов темы письма"""
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=200,
            messages=[{
                "role": "user",
                "content": f"""Generate {n_variants} subject line variants for A/B testing.
Base: "{base_subject}"
Audience: {audience_segment}
Language: Russian
Mix strategies: curiosity, urgency, social proof, benefit, question.
Return JSON array: ["variant1", "variant2", ...]"""
            }]
        )

        try:
            return json.loads(response.content[0].text)[:n_variants]
        except Exception:
            return [base_subject]


class EmailSequenceOrchestrator:
    """Оркестрация email-последовательностей"""

    def __init__(self):
        self.send_time_optimizer = SendTimeOptimizer()
        self.content_personalizer = EmailContentPersonalizer()

    def process_trigger_event(self, event: dict,
                               recipient: dict,
                               sequences: dict) -> dict:
        """Обработка триггерного события → выбор и запуск последовательности"""
        event_type = event.get('type')

        # Маппинг события на последовательность
        sequence_map = {
            'signup': 'onboarding',
            'trial_start': 'trial_nurture',
            'trial_expire_soon': 'conversion_push',
            'purchase': 'post_purchase',
            'inactivity_7d': 'reactivation',
            'feature_not_used': 'feature_adoption',
        }

        sequence_name = sequence_map.get(event_type)
        if not sequence_name or sequence_name not in sequences:
            return {'action': 'skip', 'reason': f'No sequence for event: {event_type}'}

        sequence = sequences[sequence_name]
        first_email = sequence[0]

        # Персонализация первого письма
        email_content = self.content_personalizer.generate_personalized_email(
            first_email, recipient, sequence_name
        )

        # Оптимизация времени отправки
        send_time = self.send_time_optimizer.predict_best_time(
            recipient['id'],
            recipient.get('email_history', [])
        )

        return {
            'sequence': sequence_name,
            'email': email_content,
            'send_at': {
                'hour': send_time['best_hour'],
                'weekday': send_time['best_weekday']
            },
            'remaining_steps': len(sequence) - 1,
            'tracking_params': {
                'campaign': sequence_name,
                'user_id': recipient['id']
            }
        }

    def evaluate_sequence_performance(self,
                                       sequence_logs: pd.DataFrame) -> pd.DataFrame:
        """Метрики эффективности последовательности"""
        return sequence_logs.groupby(['sequence_name', 'step_number']).agg(
            sent=('email_id', 'count'),
            open_rate=('opened', 'mean'),
            ctr=('clicked', 'mean'),
            conversion_rate=('converted', 'mean'),
            unsubscribe_rate=('unsubscribed', 'mean')
        ).round(3)

Персонализированные email-кампании с AI-оптимизацией достигают open rate 35-45% (против 20-25% для массовых рассылок) и CTR 8-15% (против 2-3%). Ключевой антипаттерн: гиперперсонализация без достаточного объёма данных создаёт странные письма, снижающие доверие.