Machine Learning Guru

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# Загружаем данные
df = pd.read_csv('web_analytics.csv', parse_dates=['HitDateTime', 'VisitDateTime'])

# Агрегируем данные по пользователям
def aggregate_user_behavior(df, days_window=30):
    now = df['HitDateTime'].max()
    window_start = now - timedelta(days=days_window)
    
    user_metrics = df[df['HitDateTime'] >= window_start].groupby('ClientID').agg({
        'VisitID': 'nunique',  # количество сессий
        'WatchID': 'count',    # количество просмотров страниц
        'VisitDuration': ['mean', 'median', 'sum'],  # метрики времени на сайте
        'PageGoals': lambda x: np.mean([len(goals) for goals in x]),  # среднее количество целей
        'isNewUser': 'max'  # новый ли пользователь
    }).reset_index()
    
    # Добавляем признаки источников трафика
    traffic_sources = df.groupby('ClientID')['Source'].value_counts().unstack(fill_value=0)
    user_metrics = user_metrics.merge(traffic_sources, on='ClientID', how='left')
    
    return user_metrics

user_behavior = aggregate_user_behavior(df)

def add_transaction_features(behavior_df, transactions_df):
    """
    Добавляет транзакционные метрики к поведенческим данным
    """
    recent_transactions = transactions_df[transactions_df['date'] >= 
                                       transactions_df['date'].max() - timedelta(days=90)]
    
    tx_metrics = recent_transactions.groupby('ClientID').agg({
        'amount': ['sum', 'mean', 'count', 'std'],
        'date': lambda x: (x.max() - x.min()).days,  # период активности
        'product_id': 'nunique'  # разнообразие покупок
    }).reset_index()
    
    # Добавляем RFM метрики
    tx_metrics['recency'] = (transactions_df['date'].max() - 
                            recent_transactions.groupby('ClientID')['date'].max())
    
    return behavior_df.merge(tx_metrics, on='ClientID', how='left')

# Добавляем транзакционные метрики
user_features = add_transaction_features(user_behavior, transactions_df)

def define_subscription_churn(df, subscription_data):
    """
    Определяет отток для подписочной модели
    """
    today = subscription_data['end_date'].max()
    horizon = today + timedelta(days=30)  # прогнозируем на месяц вперед
    
    churn_labels = subscription_data.groupby('ClientID').agg({
        'end_date': 'max',
        'is_renewed': 'last',
        'subscription_type': 'last'
    }).reset_index()
    
    # Клиент считается ушедшим, если:
    # 1. Явно отменил подписку
    # 2. Не продлил подписку в течение 7 дней после окончания
    churn_labels['is_churned'] = (
        (~churn_labels['is_renewed']) & 
        (churn_labels['end_date'] <= today - timedelta(days=7))
    )
    
    return df.merge(churn_labels[['ClientID', 'is_churned']], on='ClientID', how='left')

def define_activity_churn(df, activity_threshold=30, purchase_threshold=60):
    """
    Определяет отток на основе активности пользователя
    """
    last_date = df['HitDateTime'].max()
    
    # Считаем дни с последнего визита и покупки
    user_recency = df.groupby('ClientID').agg({
        'HitDateTime': lambda x: (last_date - x.max()).days,
        'last_purchase_date': lambda x: (last_date - x.max()).days if not x.empty else np.inf
    }).reset_index()
    
    # Клиент считается ушедшим, если:
    # 1. Не было активности дольше activity_threshold дней
    # 2. Не было покупок дольше purchase_threshold дней
    user_recency['is_churned'] = (
        (user_recency['HitDateTime'] > activity_threshold) |
        (user_recency['last_purchase_date'] > purchase_threshold)
    )
    
    return df.merge(user_recency[['ClientID', 'is_churned']], on='ClientID', how='left')

def create_engagement_features(df):
    """
    Создает признаки вовлеченности пользователя
    """
    engagement = df.groupby('ClientID').agg({
        # Активность на сайте
        'Pageviews': ['sum', 'mean', 'std'],
        'VisitDuration': ['sum', 'mean', 'std'],
        
        # Достижение целей
        'PageGoals': lambda x: np.mean([len(goals) if isinstance(goals, list) else 0 for goals in x]),
        
        # Разнообразие поведения
        'URL': lambda x: x.nunique() / x.count(),  # доля уникальных страниц
        
        # Время между визитами
        'VisitDateTime': lambda x: np.mean(np.diff(sorted(x))) if len(x) > 1 else np.nan
    }).reset_index()
    
    # Добавляем тренды
    engagement['pageview_trend'] = calculate_trend(df, 'ClientID', 'Pageviews')
    engagement['duration_trend'] = calculate_trend(df, 'ClientID', 'VisitDuration')
    
    return engagement

def calculate_trend(df, group_col, metric_col, windows=[7, 14, 30]):
    """
    Рассчитывает тренды метрик за различные периоды
    """
    trends = []
    for window in windows:
        # Разбиваем данные на периоды
        df['period'] = df['HitDateTime'].dt.to_period('D')
        window_data = df.groupby([group_col, 'period'])[metric_col].mean()
        
        # Считаем наклон линии тренда
        from scipy import stats
        def calc_slope(series):
            if len(series) < 2:
                return 0
            x = np.arange(len(series))
            slope, _, _, _, _ = stats.linregress(x, series)
            return slope
        
        trend = window_data.groupby(group_col).rolling(window=window).apply(calc_slope)
        trends.append(trend.rename(f'{metric_col}_trend_{window}d'))
    
    return pd.concat(trends, axis=1)

def create_journey_features(df):
    """
    Создает признаки на основе пути пользователя по сайту
    """
    # Создаем последовательности переходов
    df['next_url'] = df.groupby('VisitID')['URL'].shift(-1)
    
    journey_features = df.groupby('ClientID').agg({
        # Паттерны навигации
        'URL': lambda x: analyze_navigation_patterns(x),
        
        # Точки входа и выхода
        'StartURL': lambda x: x.value_counts().index[0],  # самая частая точка входа
        'EndURL': lambda x: x.value_counts().index[0],    # самая частая точка выхода
        
        # Конверсионный путь
        'PageGoals': lambda x: analyze_conversion_path(x)
    }).reset_index()
    
    return journey_features

def analyze_navigation_patterns(urls):
    """
    Анализирует паттерны навигации пользователя
    """
    from collections import defaultdict
    patterns = defaultdict(int)
    
    # Создаем n-граммы из последовательности URL
    for i in range(len(urls) - 1):
        pattern = (urls[i], urls[i + 1])
        patterns[pattern] += 1
    
    return dict(sorted(patterns.items(), key=lambda x: x[1], reverse=True)[:5])

def train_base_models(X_train, y_train, X_val, y_val):
    """
    Обучает набор базовых моделей
    """
    from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
    from lightgbm import LGBMClassifier
    from xgboost import XGBClassifier
    
    models = {
        'rf': RandomForestClassifier(
            n_estimators=100,
            max_depth=10,
            class_weight='balanced',
            random_state=42
        ),
        'gb': GradientBoostingClassifier(
            n_estimators=100,
            learning_rate=0.1,
            max_depth=5,
            random_state=42
        ),
        'lgb': LGBMClassifier(
            n_estimators=100,
            learning_rate=0.1,
            max_depth=5,
            class_weight='balanced',
            random_state=42
        ),
        'xgb': XGBClassifier(
            n_estimators=100,
            learning_rate=0.1,
            max_depth=5,
            scale_pos_weight=sum(y_train==0)/sum(y_train==1),  # балансировка классов
            random_state=42
        )
    }
    
    # Обучаем модели и собираем метрики
    from sklearn.metrics import roc_auc_score, precision_recall_curve
    results = {}
    
    for name, model in models.items():
        # Обучение
        model.fit(X_train, y_train)
        
        # Предсказания
        train_preds = model.predict_proba(X_train)[:, 1]
        val_preds = model.predict_proba(X_val)[:, 1]
        
        # Метрики
        results[name] = {
            'model': model,
            'train_auc': roc_auc_score(y_train, train_preds),
            'val_auc': roc_auc_score(y_val, val_preds),
            'val_preds': val_preds
        }
        
    return results

# Обучаем модели
model_results = train_base_models(X_train, y_train, X_val, y_val)

def build_stacking_ensemble(base_models_results, X_train, y_train, X_val, y_val):
    """
    Создает ансамбль моделей методом стекинга
    """
    from sklearn.linear_model import LogisticRegression
    
    # Собираем предсказания базовых моделей
    base_train_preds = np.column_stack([
        model['model'].predict_proba(X_train)[:, 1]
        for model in base_models_results.values()
    ])
    
    base_val_preds = np.column_stack([
        model['model'].predict_proba(X_val)[:, 1]
        for model in base_models_results.values()
    ])
    
    # Обучаем мета-модель
    meta_model = LogisticRegression(C=1.0, class_weight='balanced')
    meta_model.fit(base_train_preds, y_train)
    
    # Финальные предсказания
    final_preds = meta_model.predict_proba(base_val_preds)[:, 1]
    
    return {
        'meta_model': meta_model,
        'ensemble_auc': roc_auc_score(y_val, final_preds),
        'final_preds': final_preds
    }

def optimize_threshold(y_true, y_pred_proba):
    """
    Оптимизирует порог для максимизации бизнес-метрики
    """
    # Рассчитываем стоимость ошибок
    cost_matrix = {
        'false_positive': 1,    # стоимость ложного срабатывания
        'false_negative': 5,    # стоимость пропуска оттока
        'true_positive': -10,   # выгода от предотвращения оттока
        'true_negative': 0      # без затрат
    }
    
    def calculate_business_metric(threshold):
        y_pred = (y_pred_proba >= threshold).astype(int)
        
        # Считаем метрики
        tp = np.sum((y_true == 1) & (y_pred == 1))
        tn = np.sum((y_true == 0) & (y_pred == 0))
        fp = np.sum((y_true == 0) & (y_pred == 1))
        fn = np.sum((y_true == 1) & (y_pred == 0))
        
        # Считаем общую стоимость
        total_cost = (
            tp * cost_matrix['true_positive'] +
            tn * cost_matrix['true_negative'] +
            fp * cost_matrix['false_positive'] +
            fn * cost_matrix['false_negative']
        )
        
        return -total_cost  # минимизируем затраты
    
    # Ищем оптимальный порог
    from scipy.optimize import minimize_scalar
    
    result = minimize_scalar(
        calculate_business_metric,
        bounds=(0, 1),
        method='bounded'
    )
    
    return result.x

def interpret_predictions(model, X, feature_names):
    """
    Интерпретирует предсказания модели с помощью SHAP
    """
    import shap
    
    # Создаем SHAP explainer
    explainer = shap.TreeExplainer(model)
    shap_values = explainer.shap_values(X)
    
    # Визуализация важности признаков
    plt.figure(figsize=(12, 8))
    shap.summary_plot(shap_values, X, feature_names=feature_names)
    
    # Анализ конкретных предсказаний
    def explain_prediction(client_index):
        plt.figure(figsize=(10, 6))
        shap.force_plot(
            explainer.expected_value,
            shap_values[client_index,:],
            X.iloc[client_index,:],
            feature_names=feature_names,
            matplotlib=True
        )
    
    return shap_values, explain_prediction

class ChurnPredictor:
    def __init__(self, model, feature_processor, threshold=0.5):
        self.model = model
        self.feature_processor = feature_processor
        self.threshold = threshold
        
    def predict_customer_risk(self, customer_data):
        """
        Оценивает риск оттока для конкретного клиента
        """
        # Подготовка данных
        features = self.feature_processor.transform(customer_data)
        
        # Получение предсказания
        churn_probability = self.model.predict_proba(features)[:, 1][0]
        
        # Определение уровня риска
        risk_level = self._categorize_risk(churn_probability)
        
        return {
            'client_id': customer_data['ClientID'],
            'churn_probability': float(churn_probability),
            'risk_level': risk_level,
            'main_factors': self._get_risk_factors(features)
        }
    
    def _categorize_risk(self, probability):
        """
        Категоризация уровня риска
        """
        if probability >= self.threshold * 1.5:
            return 'CRITICAL'
        elif probability >= self.threshold:
            return 'HIGH'
        elif probability >= self.threshold * 0.5:
            return 'MEDIUM'
        return 'LOW'
    
    def _get_risk_factors(self, features):
        """
        Определяет основные факторы риска
        """
        import shap
        
        explainer = shap.TreeExplainer(self.model)
        shap_values = explainer.shap_values(features)[1][0]  # для positive класса
        
        # Сортируем факторы по важности
        feature_importance = pd.DataFrame({
            'feature': features.columns,
            'importance': np.abs(shap_values)
        }).sort_values('importance', ascending=False)
        
        return feature_importance.head(3).to_dict('records')

class ChurnPreventionSystem:
    def __init__(self, predictor, communication_service):
        self.predictor = predictor
        self.communication_service = communication_service
        
    async def process_customer_batch(self, customers_data):
        """
        Обрабатывает пакет клиентских данных и запускает превентивные действия
        """
        results = []
        for customer in customers_data:
            # Оценка риска
            risk_assessment = self.predictor.predict_customer_risk(customer)
            
            # Определение действий
            actions = self._determine_actions(risk_assessment, customer)
            
            # Запуск коммуникаций
            if actions:
                await self._execute_actions(actions, customer)
            
            results.append({
                'client_id': customer['ClientID'],
                'risk_assessment': risk_assessment,
                'actions_taken': actions
            })
            
        return results
    
    def _determine_actions(self, risk_assessment, customer_data):
        """
        Определяет необходимые действия на основе оценки риска
        """
        actions = []
        
        if risk_assessment['risk_level'] == 'CRITICAL':
            actions.extend([
                {
                    'type': 'PERSONAL_CALL',
                    'priority': 'HIGH',
                    'message_template': 'urgent_retention_call'
                },
                {
                    'type': 'SPECIAL_OFFER',
                    'offer_type': 'PREMIUM_DISCOUNT',
                    'validity_days': 7
                }
            ])
        elif risk_assessment['risk_level'] == 'HIGH':
            actions.append({
                'type': 'EMAIL',
                'template': 'retention_email',
                'personalization': {
                    'risk_factors': risk_assessment['main_factors']
                }
            })
        elif risk_assessment['risk_level'] == 'MEDIUM':
            actions.append({
                'type': 'PUSH_NOTIFICATION',
                'template': 'engagement_reminder'
            })
            
        return actions
    
    async def _execute_actions(self, actions, customer):
        """
        Выполняет запланированные действия
        """
        for action in actions:
            try:
                await self.communication_service.send_message(
                    customer['ClientID'],
                    action
                )
            except Exception as e:
                logger.error(f"Failed to execute action {action['type']} "
                           f"for customer {customer['ClientID']}: {str(e)}")

class ModelMonitor:
    def __init__(self, model_version):
        self.model_version = model_version
        self.metrics_store = MetricsDatabase()
        
    def log_prediction(self, prediction_data, actual_outcome=None):
        """
        Логирует предсказание и фактический результат
        """
        log_entry = {
            'timestamp': datetime.now(),
            'model_version': self.model_version,
            'prediction': prediction_data,
            'actual_outcome': actual_outcome,
            'features_snapshot': prediction_data['features']
        }
        
        self.metrics_store.save_prediction_log(log_entry)
        
    def calculate_metrics(self, time_window=timedelta(days=7)):
        """
        Рассчитывает метрики качества модели
        """
        # Получаем логи за период
        recent_predictions = self.metrics_store.get_predictions(
            from_date=datetime.now() - time_window
        )
        
        # Считаем метрики
        metrics = {
            'prediction_volume': len(recent_predictions),
            'average_probability': np.mean([p['prediction']['churn_probability'] 
                                         for p in recent_predictions]),
            'risk_distribution': self._calculate_risk_distribution(recent_predictions)
        }
        
        # Если есть фактические результаты
        predictions_with_outcomes = [p for p in recent_predictions 
                                  if p['actual_outcome'] is not None]
        
        if predictions_with_outcomes:
            metrics.update(self._calculate_performance_metrics(predictions_with_outcomes))
            
        return metrics
        
    def _calculate_risk_distribution(self, predictions):
        """
        Рассчитывает распределение рисков
        """
        risk_levels = [p['prediction']['risk_level'] for p in predictions]
        return dict(Counter(risk_levels))

class RetentionExperiment:
    def __init__(self, experiment_name, strategies, sample_size_per_group=1000):
        """
        Инициализация эксперимента по тестированию стратегий удержания
        
        Args:
            experiment_name (str): Название эксперимента
            strategies (dict): Словарь стратегий удержания
            sample_size_per_group (int): Размер тестовой группы для каждой стратегии
        """
        self.experiment_name = experiment_name
        self.strategies = strategies
        self.sample_size = sample_size_per_group
        self.start_date = datetime.now()
        self.experiment_groups = {}
        
    def allocate_customers(self, customers_data):
        """
        Распределяет клиентов по группам эксперимента
        """
        # Создаем хэш для стабильного распределения
        def get_customer_hash(customer_id):
            return hashlib.md5(
                f"{self.experiment_name}:{customer_id}".encode()
            ).hexdigest()
        
        allocated_customers = []
        for customer in customers_data:
            customer_hash = get_customer_hash(customer['ClientID'])
            group_index = int(customer_hash, 16) % (len(self.strategies) + 1)  # +1 для контрольной группы
            
            if group_index == 0:
                group = 'control'
                strategy = None
            else:
                group = f'strategy_{group_index}'
                strategy = list(self.strategies.values())[group_index - 1]
            
            allocated_customers.append({
                'customer': customer,
                'group': group,
                'strategy': strategy
            })
            
        return allocated_customers
    
    def track_results(self, customer_id, group, action_taken, outcome):
        """
        Отслеживает результаты применения стратегии
        """
        if group not in self.experiment_groups:
            self.experiment_groups[group] = {
                'total_customers': 0,
                'actions_taken': 0,
                'successful_retentions': 0,
                'failed_retentions': 0,
                'retention_rate': 0.0,
                'cost': 0.0,
                'revenue_saved': 0.0
            }
        
        group_stats = self.experiment_groups[group]
        group_stats['total_customers'] += 1
        
        if action_taken:
            group_stats['actions_taken'] += 1
            if outcome['retained']:
                group_stats['successful_retentions'] += 1
                group_stats['revenue_saved'] += outcome['revenue_saved']
            else:
                group_stats['failed_retentions'] += 1
            
            group_stats['cost'] += outcome['action_cost']
            group_stats['retention_rate'] = (
                group_stats['successful_retentions'] / group_stats['actions_taken']
            )
    
    def get_experiment_results(self):
        """
        Анализирует результаты эксперимента
        """
        results = {
            'experiment_name': self.experiment_name,
            'duration_days': (datetime.now() - self.start_date).days,
            'groups_performance': self.experiment_groups,
            'statistical_significance': self._calculate_significance(),
            'roi_analysis': self._calculate_roi(),
            'recommendations': self._generate_recommendations()
        }
        
        return results
    
    def _calculate_significance(self):
        """
        Рассчитывает статистическую значимость результатов
        """
        from scipy import stats
        
        control_retention = self.experiment_groups['control']['retention_rate']
        significance_results = {}
        
        for group, stats in self.experiment_groups.items():
            if group == 'control':
                continue
                
            # Проводим z-test для сравнения пропорций
            success_treatment = stats['successful_retentions']
            total_treatment = stats['actions_taken']
            success_control = self.experiment_groups['control']['successful_retentions']
            total_control = self.experiment_groups['control']['actions_taken']
            
            z_stat, p_value = stats.proportions_ztest(
                [success_treatment, success_control],
                [total_treatment, total_control]
            )
            
            significance_results[group] = {
                'z_statistic': z_stat,
                'p_value': p_value,
                'is_significant': p_value < 0.05,
                'lift': (stats['retention_rate'] - control_retention) / control_retention * 100
            }
            
        return significance_results

class StrategyOptimizer:
    def __init__(self, initial_strategies):
        """
        Инициализация оптимизатора стратегий удержания
        
        Args:
            initial_strategies (dict): Начальные стратегии удержания
        """
        self.strategies = initial_strategies
        self.strategy_performance = {}
        self.min_confidence_threshold = 0.95
        
    def update_strategies(self, experiment_results):
        """
        Обновляет стратегии на основе результатов экспериментов
        """
        significant_improvements = []
        
        # Анализируем результаты по каждой стратегии
        for group, stats in experiment_results['statistical_significance'].items():
            if not stats['is_significant']:
                continue
                
            if stats['lift'] > 0 and stats['p_value'] < (1 - self.min_confidence_threshold): significant_improvements.append({ 'group': group, 'lift': stats['lift'], 'roi': experiment_results['roi_analysis'][group]['roi'] }) # Обновляем стратегии if significant_improvements: # Сортируем по ROI significant_improvements.sort(key=lambda x: x['roi'], reverse=True) # Обновляем параметры лучших стратегий for improvement in significant_improvements: strategy_name = improvement['group'] if strategy_name in self.strategies: self._optimize_strategy_parameters( strategy_name, experiment_results['groups_performance'][strategy_name] ) def _optimize_strategy_parameters(self, strategy_name, performance_stats): """ Оптимизирует параметры конкретной стратегии """ strategy = self.strategies[strategy_name] # Пример оптимизации для стратегии с скидками if 'discount_percentage' in strategy: # Если стратегия очень прибыльна, можно уменьшить скидку if performance_stats['roi'] > 3.0:  # ROI > 300%
                strategy['discount_percentage'] *= 0.9  # уменьшаем скидку на 10%
            # Если стратегия едва окупается, можно увеличить скидку
            elif performance_stats['roi'] < 1.2:  # ROI < 120%
                strategy['discount_percentage'] *= 1.1  # увеличиваем скидку на 10%
                
        # Оптимизация временных параметров
        if 'offer_duration_days' in strategy:
            acceptance_rate = (
                performance_stats['successful_retentions'] /
                performance_stats['actions_taken']
            )
            
            # Если низкий процент принятия предложения, увеличиваем duration
            if acceptance_rate < 0.3: strategy['offer_duration_days'] += 1 # Если высокий процент принятия, можно уменьшить duration elif acceptance_rate > 0.7:
                strategy['offer_duration_days'] = max(1, strategy['offer_duration_days'] - 1)