互联网产品用户流失预防与召回 - 基于商业分析方法论的完整体系

# 基于商业分析框架的用户画像维度设计
user_profile_dimensions = {
 # 趋势分析维度
    'activity_trend': ['登录频次变化', '使用时长变化', '功能使用深度变化'],

 # 生命周期维度
    'lifecycle_stage': ['新手期', '成长期', '成熟期', '衰退期'],

 # 自然周期维度
    'natural_cycle': ['工作日活跃度', '周末活跃度', '节假日活跃度'],

 # 主动行为响应维度
    'campaign_response': ['推送响应率', '活动参与率', '新功能采用率']
}

-- 用户基础信息表
CREATE TABLE game_users (
    user_id VARCHAR(32) PRIMARY KEY,
    install_time DATETIME,
    device_type ENUM('Android', 'iOS'),
    channel VARCHAR(50), -- 渠道来源
    country VARCHAR(50),
    first_pay_time DATETIME,
    total_pay_amount DECIMAL(10,2) DEFAULT 0,
    last_login DATETIME,
    user_level INT DEFAULT 1,
    INDEX idx_install_time (install_time),
    INDEX idx_last_login (last_login)
);

-- 游戏行为事件表
CREATE TABLE game_events (
    event_id VARCHAR(32) PRIMARY KEY,
    user_id VARCHAR(32),
    event_type ENUM('login', 'level_start', 'level_complete', 'level_fail',
                   'purchase', 'ad_view', 'social_share', 'logout'),
    event_time DATETIME,
    level_id INT,
    session_id VARCHAR(32),
    session_length INT, -- 会话时长(秒)
    game_mode ENUM('normal', 'challenge', 'endless'),
    score INT,
    props_used JSON, -- 使用的道具
    INDEX idx_user_time (user_id, event_time),
    INDEX idx_session (session_id),
    INDEX idx_event_type (event_type, event_time)
);

-- 用户留存快照表（每日计算）
CREATE TABLE user_retention_daily (
    date DATE,
    user_id VARCHAR(32),
    install_date DATE,
    days_since_install INT,
    is_active BOOLEAN,
    session_count INT,
    playtime_minutes INT,
    levels_completed INT,
    PRIMARY KEY (date, user_id),
    INDEX idx_install_cohort (install_date, days_since_install)
);

-- 定义流失标准：连续7天未登录的用户
WITH user_last_activity AS (
    SELECT
        user_id,
        MAX(event_time) as last_active_time,
        DATEDIFF(CURRENT_DATE, DATE(MAX(event_time))) as days_inactive
    FROM game_events
    WHERE event_type = 'login'
    GROUP BY user_id
),
user_status AS (
    SELECT
        u.user_id,
        u.install_time,
        u.user_level,
        u.total_pay_amount,
        ula.last_active_time,
        ula.days_inactive,
        CASE
            WHEN ula.days_inactive <= 1 THEN '活跃用户'
            WHEN ula.days_inactive <= 3 THEN '风险用户'
            WHEN ula.days_inactive <= 7 THEN '沉睡用户'
            ELSE '流失用户'
        END as user_status
    FROM game_users u
    LEFT JOIN user_last_activity ula ON u.user_id = ula.user_id
)
SELECT
    user_status,
    COUNT(*) as user_count,
    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER(), 2) as percentage,
    ROUND(AVG(CASE WHEN total_pay_amount > 0 THEN total_pay_amount END), 2) as avg_ltv
FROM user_status
GROUP BY user_status
ORDER BY
    CASE user_status
        WHEN '活跃用户' THEN 1
        WHEN '风险用户' THEN 2
        WHEN '沉睡用户' THEN 3
        WHEN '流失用户' THEN 4
    END;

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, roc_auc_score

def create_user_features(events_df, users_df, current_date):
    """创建用户流失预测特征"""

 # 定义时间窗口
    window_7d = current_date - timedelta(days=7)
    window_14d = current_date - timedelta(days=14)
    window_30d = current_date - timedelta(days=30)

 # 基础活跃特征
    activity_features = events_df
        events_df['event_time'] >= window_30d
    ].groupby('user_id').agg({
        'event_time': ['count', 'nunique'],  # 总事件数、活跃天数
        'session_id': 'nunique',            # 会话数
        'session_length': ['mean', 'sum'],  # 平均&总会话时长
        'level_id': ['max', 'nunique'],     # 最高等级、玩过关卡数
        'score': ['mean', 'max']            # 平均&最高分数
    }).round(2)

 # 平铺列名
    activity_features.columns =
        'total_events_30d', 'active_days_30d', 'total_sessions_30d',
        'avg_session_length', 'total_playtime', 'max_level', 'levels_played',
        'avg_score', 'max_score'
    ]

 # 近期活跃趋势特征
    recent_7d = events_df[events_df['event_time'] >= window_7d].groupby('user_id').agg({
        'event_time': 'count',
        'session_length': 'sum'
    }).rename(columns={'event_time': 'events_7d', 'session_length': 'playtime_7d'})

    recent_14d = events_df[events_df['event_time'] >= window_14d].groupby('user_id').agg({
        'event_time': 'count',
        'session_length': 'sum'
    }).rename(columns={'event_time': 'events_14d', 'session_length': 'playtime_14d'})

 # 合并所有特征
    features = activity_features.join([recent_7d, recent_14d], how='left').fillna(0)

 # 计算趋势指标
    features['activity_trend'] = (
        features['events_7d'] / (features['events_14d'] - features['events_7d'] + 1)
    ).round(3)

    features['playtime_trend'] = (
        features['playtime_7d'] / (features['playtime_14d'] - features['playtime_7d'] + 1)
    ).round(3)

 # 用户生命周期特征
    users_df['install_days'] = (current_date - users_df['install_time']).dt.days

 # 游戏进度特征
    level_events = events_df[events_df['event_type'].isin(['level_complete', 'level_fail'])]
    progress_features = level_events.groupby('user_id').agg({
        'level_id': 'max',
        'event_type': lambda x: (x == 'level_complete').sum() / len(x)  # 通关率
    }).rename(columns={'level_id': 'current_level', 'event_type': 'success_rate'})

 # 最终特征集合
    final_features = features.join(
        users_df.set_index('user_id')'install_time', 'user_level', 'total_pay_amount', 'install_days',
        progress_features
    ], how='left').fillna(0)

    return final_features

def calculate_churn_labels(events_df, current_date, churn_days=7):
    """计算用户流失标签"""

    churn_threshold = current_date - timedelta(days=churn_days)

 # 计算每个用户最后活跃时间
    last_activity = events_df
        events_df['event_type'] == 'login'
    ].groupby('user_id')['event_time'].max()

 # 标记流失用户
    churn_labels = (last_activity < churn_threshold).astype(int)

    return churn_labels

# 加载数据和特征工程
current_date = datetime.now()
events_df = load_game_events_data()  # 从数据库加载
users_df = load_game_users_data()

# 创建特征和标签
user_features = create_user_features(events_df, users_df, current_date)
churn_labels = calculate_churn_labels(events_df, current_date)

# 合并特征和标签
modeling_data = user_features.join(churn_labels.rename('is_churn'), how='inner')

print(f"建模数据集大小: {modeling_data.shape}")
print(f"流失率: {modeling_data['is_churn'].mean():.2%}")

from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import cross_val_score, GridSearchCV
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import precision_recall_curve, roc_curve, confusion_matrix
import plotly.graph_objects as go
from plotly.subplots import make_subplots

def build_churn_prediction_model(modeling_data):
    """构建用户流失预测模型"""

 # 特征选择
    feature_cols =
        'total_events_30d', 'active_days_30d', 'total_sessions_30d',
        'avg_session_length', 'total_playtime', 'max_level', 'levels_played',
        'activity_trend', 'playtime_trend', 'install_days', 'current_level',
        'success_rate', 'total_pay_amount'
    ]

    X = modeling_data[feature_cols].fillna(0)
    y = modeling_data['is_churn']

 # 训练测试分离
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

 # 模型对比实验
    models = {
        'RandomForest': RandomForestClassifier(n_estimators=100, random_state=42),
        'GradientBoosting': GradientBoostingClassifier(n_estimators=100, random_state=42)
    }

    results = {}
    for name, model in models.items():
 # 交叉验证评估
        cv_scores = cross_val_score(model, X_train, y_train, cv=5, scoring='roc_auc')

 # 训练模型
        model.fit(X_train, y_train)

 # 预测和评估
        y_pred = model.predict(X_test)
        y_prob = model.predict_proba(X_test)[:, 1]

        auc_score = roc_auc_score(y_test, y_prob)

        results[name] = {
            'model': model,
            'cv_mean': cv_scores.mean(),
            'cv_std': cv_scores.std(),
            'test_auc': auc_score,
            'y_pred': y_pred,
            'y_prob': y_prob
        }

        print(f"{name} - CV AUC: {cv_scores.mean():.3f}(±{cv_scores.std():.3f})")
        print(f"{name} - Test AUC: {auc_score:.3f}")
        print(classification_report(y_test, y_pred))
        print("-" * 50)

 # 选择最佳模型
    best_model_name = max(results.keys(), key=lambda x: results[x]['test_auc'])
    best_model = results[best_model_name]['model']

 # 特征重要性分析
    feature_importance = pd.DataFrame({
        'feature': feature_cols,
        'importance': best_model.feature_importances_
    }).sort_values('importance', ascending=False)

    return best_model, feature_importance, results

# 构建预测模型
best_model, feature_importance, model_results = build_churn_prediction_model(modeling_data)

print("特征重要性排序:")
print(feature_importance.head(10))

def create_user_segments(modeling_data, best_model):
    """基于流失预测进行用户分群"""

    feature_cols =
        'total_events_30d', 'active_days_30d', 'total_sessions_30d',
        'avg_session_length', 'total_playtime', 'max_level', 'levels_played',
        'activity_trend', 'playtime_trend', 'install_days', 'current_level',
        'success_rate', 'total_pay_amount'
    ]

    X = modeling_data[feature_cols].fillna(0)

 # 流失概率预测
    churn_prob = best_model.predict_proba(X)[:, 1]
    modeling_data['churn_probability'] = churn_prob

 # 用户分群逻辑
    def assign_segment(row):
        prob = row['churn_probability']
        is_paying = row['total_pay_amount'] > 0
        activity_level = row['total_events_30d']

        if prob < 0.2:
            return '健康用户'
        elif prob < 0.5:
            if is_paying:
                return '付费风险用户'
            else:
                return '免费风险用户'
        elif prob < 0.8:
            if activity_level > 50:
                return '高活跃流失风险'
            else:
                return '低活跃流失风险'
        else:
            return '极高流失风险'

    modeling_data['user_segment'] = modeling_data.apply(assign_segment, axis=1)

 # 分群统计
    segment_stats = modeling_data.groupby('user_segment').agg({
        'churn_probability': ['count', 'mean'],
        'total_pay_amount': ['mean', 'sum'],
        'total_events_30d': 'mean',
        'is_churn': 'mean'
    }).round(3)

    return modeling_data, segment_stats

# 用户分群
segmented_data, segment_stats = create_user_segments(modeling_data, best_model)
print("用户分群统计:")
print(segment_stats)

def design_retention_strategies():
    """设计个性化用户召回策略"""

    strategies = {
        '健康用户': {
            'strategy': '价值延续',
            'actions':
                '新内容推荐：推送新关卡和游戏模式',
                '社交功能：邀请好友一起游戏',
                '成就系统：设置长期目标和奖励'
            ],
            'channels': ['App内推送', '游戏内消息'],
            'frequency': '周2-3次',
            'kpi': '提升用户LTV和活跃时长'
        },

        '付费风险用户': {
            'strategy': '价值保护',
            'actions':
                '专属客服：一对一游戏指导',
                '付费专享：VIP专属内容和特权',
                '个性化推荐：基于游戏偏好推荐'
            ],
            'channels': ['短信', 'App推送', '客服电话'],
            'frequency': '每日1次',
            'kpi': '降低流失率到5%以下'
        },

        '免费风险用户': {
            'strategy': '激活转化',
            'actions':
                '限时免费：解锁付费道具试用',
                '新手引导：优化游戏教程',
                '激励机制：签到奖励和任务系统'
            ],
            'channels': ['App推送', '游戏内引导'],
            'frequency': '每日2-3次',
            'kpi': '提升次日留存率到60%'
        },

        '高活跃流失风险': {
            'strategy': '体验优化',
            'actions':
                '难度调优：个性化关卡难度',
                '内容更新：加速新内容发布',
                '社区互动：参与玩家社区活动'
            ],
            'channels': ['App推送', '社区消息'],
            'frequency': '每日1-2次',
            'kpi': '减少游戏挫败感，提升完成率'
        },

        '低活跃流失风险': {
            'strategy': '重新激活',
            'actions':
                '回归奖励：大量金币和道具补偿',
                '简化操作：降低游戏门槛',
                '情感召回：怀旧元素和回忆杀'
            ],
            'channels': ['短信', '邮件', 'App推送'],
            'frequency': '隔日1次',
            'kpi': '7日召回率达到25%'
        },

        '极高流失风险': {
            'strategy': '最后挽救',
            'actions':
                '终极奖励：稀有道具和角色',
                '人工外呼：了解流失原因',
                '产品改进：收集反馈优化体验'
            ],
            'channels': ['短信', '邮件', '推送'],
            'frequency': '立即执行，后续每周1次',
            'kpi': '挽回5%高价值用户'
        }
    }

    return strategies

# 策略制定
retention_strategies = design_retention_strategies()

for segment, strategy in retention_strategies.items():
    print(f"\n【{segment}】召回策略:")
    print(f"核心策略: {strategy['strategy']}")
    print(f"具体行动: {', '.join(strategy['actions'])}")
    print(f"触达渠道: {', '.join(strategy['channels'])}")
    print(f"执行频率: {strategy['frequency']}")
    print(f"目标KPI: {strategy['kpi']}")

-- A/B测试分组与效果追踪
CREATE TABLE retention_experiment (
    experiment_id VARCHAR(32),
    user_id VARCHAR(32),
    user_segment VARCHAR(50),
    test_group ENUM('control', 'treatment_a', 'treatment_b'),
    strategy_type VARCHAR(100),
    start_date DATE,
    end_date DATE,
    PRIMARY KEY (experiment_id, user_id)
);

-- 实验效果监控查询
WITH experiment_results AS (
    SELECT
        re.experiment_id,
        re.test_group,
        re.user_segment,
        COUNT(DISTINCT re.user_id) as total_users,
        COUNT(DISTINCT CASE
            WHEN ge.event_time BETWEEN re.start_date AND re.end_date
            THEN re.user_id
        END) as returned_users,
        COUNT(DISTINCT CASE
            WHEN ge.event_time >= re.end_date
            THEN re.user_id
        END) as active_after_experiment
    FROM retention_experiment re
    LEFT JOIN game_events ge ON re.user_id = ge.user_id
        AND ge.event_type = 'login'
    WHERE re.start_date >= '2024-01-01'
    GROUP BY re.experiment_id, re.test_group, re.user_segment
)
SELECT
    experiment_id,
    user_segment,
    test_group,
    total_users,
    returned_users,
    ROUND(returned_users * 100.0 / total_users, 2) as return_rate,
    active_after_experiment,
    ROUND(active_after_experiment * 100.0 / total_users, 2) as retention_rate
FROM experiment_results
ORDER BY experiment_id, user_segment, test_group;

import plotly.graph_objects as go
from plotly.subplots import make_subplots
import plotly.express as px

def create_churn_monitoring_dashboard(segmented_data, daily_stats):
    """创建用户流失监控仪表板"""

 # 创建多子图布局
    fig = make_subplots(
        rows=3, cols=2,
        subplot_titles=
            '用户流失风险分布', '各分群流失预测',
            '召回策略效果追踪', '关键指标趋势',
            '流失原因分析', '预警用户列表'
        ],
        specs=
            [{"type": "pie"}, {"type": "bar"}],
            [{"type": "scatter"}, {"type": "scatter"}],
            [{"type": "heatmap"}, {"type": "table"}]
        ]
    )

 # 1. 流失风险分布饼图
    risk_distribution = segmented_data['user_segment'].value_counts()
    fig.add_trace(
        go.Pie(
            labels=risk_distribution.index,
            values=risk_distribution.values,
            textinfo='label+percent',
            name="风险分布"
        ),
        row=1, col=1
    )

 # 2. 各分群流失预测对比
    segment_risk = segmented_data.groupby('user_segment')['churn_probability'].mean()
    fig.add_trace(
        go.Bar(
            x=segment_risk.index,
            y=segment_risk.values,
            text=[f'{v:.1%}' for v in segment_risk.values],
            textposition='auto',
            name="平均流失概率"
        ),
        row=1, col=2
    )

 # 3. 召回效果趋势
    if 'recall_effect' in daily_stats.columns:
        fig.add_trace(
            go.Scatter(
                x=daily_stats['date'],
                y=daily_stats['recall_rate'],
                mode='lines+markers',
                name='召回成功率',
                line=dict(color='green')
            ),
            row=2, col=1
        )

 # 4. 关键指标趋势
    fig.add_trace(
        go.Scatter(
            x=daily_stats['date'],
            y=daily_stats['churn_rate'],
            mode='lines+markers',
            name='流失率',
            line=dict(color='red')
        ),
        row=2, col=2
    )

    fig.add_trace(
        go.Scatter(
            x=daily_stats['date'],
            y=daily_stats['retention_rate'],
            mode='lines+markers',
            name='留存率',
            line=dict(color='blue'),
            yaxis='y2'
        ),
        row=2, col=2
    )

 # 更新布局
    fig.update_layout(
        title_text="用户流失预防监控仪表板",
        showlegend=True,
        height=1200
    )

    return fig

# 生成仪表板
dashboard = create_churn_monitoring_dashboard(segmented_data, daily_stats)
dashboard.show()

# 分层召回策略设计
churn_prevention_strategies = {
 # 预防性策略（针对风险用户）
    'prevention': {
        'target': '活跃度下降但未流失的用户',
        'strategies':
            '个性化内容推荐',
            '学习路径优化',
            '社群互动增强'
        ],
        'timing': '活跃度下降30%时触发',
        'cost_per_user': 5  # 元
    },

 # 挽回性策略（针对流失用户）
    'retention': {
        'target': '已经7天未登录的付费用户',
        'strategies':
            '专属客服联系',
            '课程延期补偿',
            '限时优惠券'
        ],
        'timing': '流失后48小时内执行',
        'cost_per_user': 25  # 元
    },

 # 召回策略（针对长期流失用户）
    'winback': {
        'target': '流失超过30天的高价值用户',
        'strategies':
            '重大优惠活动',
            '新课程免费体验',
            '老朋友回归礼包'
        ],
        'timing': '每月1次',
        'cost_per_user': 50  # 元
    }
}

# 完整的流失分析指标体系
kpi_framework = {
 # 结果指标（滞后指标）
    'result_metrics': {
        'churn_rate_7d': '7日流失率',
        'churn_rate_30d': '30日流失率',
        'retention_rate': '用户留存率',
        'winback_success_rate': '召回成功率'
    },

 # 过程指标（领先指标）
    'process_metrics': {
        'engagement_score': '用户参与度评分',
        'feature_adoption_rate': '核心功能使用率',
        'support_ticket_rate': '客服咨询率',
        'nps_score': '净推荐值'
    },

 # 投入指标
    'input_metrics': {
        'retention_campaign_cost': '召回活动成本',
        'customer_success_cost': '客户成功团队成本',
        'product_improvement_cost': '产品优化投入'
    }
}

# 一周内各天的用户流失率
weekly_churn_pattern = {
    'Monday': 8.2,      # 周一流失率最低（新周开始动力强）
    'Tuesday': 9.1,     # 逐渐上升
    'Wednesday': 10.5,  # 周中疲惫期
    'Thursday': 11.8,   # 继续上升
    'Friday': 15.2,     # 周五最高（周末诱惑大）
    'Saturday': 12.3,   # 周末有所下降
    'Sunday': 13.7      # 周日较高（明日恐惧症）
}

# 生命周期各阶段的流失原因与应对策略
lifecycle_churn_analysis = {
    'newbie_stage': {
        'duration': '0-7天',
        'churn_rate': 0.85,
        'main_reasons':
            '产品功能不匹配用户需求（40%）',
            '操作门槛过高，学习成本大（30%）',
            '期望值过高，体验不达预期（30%）'
        ],
        'interventions':
            '优化新手引导流程',
            '个性化推荐首次学习内容',
            '48小时内客服主动联系'
        ],
        'success_rate': 0.25  # 干预成功率
    },

    'growth_stage': {
        'duration': '8-30天',
        'churn_rate': 0.45,
        'main_reasons':
            '学习习惯未建立，坚持困难（50%）',
            '学习效果不明显，信心不足（30%）',
            '时间冲突，优先级下降（20%）'
        ],
        'interventions':
            '学习计划个性化定制',
            '阶段性成果展示与激励',
            '学习社群组建与互动'
        ],
        'success_rate': 0.40
    },

    'mature_stage': {
        'duration': '31-180天',
        'churn_rate': 0.15,
        'main_reasons':
            '竞品吸引，功能更优（40%）',
            '学习目标达成，需求满足（35%）',
            '价格敏感，寻找替代品（25%）'
        ],
        'interventions':
            '高级功能逐步开放',
            '学习路径延伸与拓展',
            '价值感知提升策略'
        ],
        'success_rate': 0.60
    },

    'decline_stage': {
        'duration': '180天+',
        'churn_rate': 0.35,
        'main_reasons':
            '用户需求变化，产品价值下降（45%）',
            '使用习惯固化，缺乏新鲜感（35%）',
            '生活状态变化，时间精力不足（20%）'
        ],
        'interventions':
            '产品功能升级与创新',
            '新的学习目标设定',
            '忠诚用户专属权益'
        ],
        'success_rate': 0.20
    }
}

# 主动行为分析的五阶段模型
> active_behavior_analysis = {
>     '上线前期': {
>         'duration': '4月1日-4月14日',
>         'baseline_churn_rate': 0.175,
>         'characteristics': '正常运营状态，流失率相对稳定'
>     },
>
>     '功能发布期': {
>         'duration': '4月15日-4月21日',
>         'churn_rate': 0.142,
>         'change': -18.9,  # 相比基准期变化百分比
>         'analysis': '新功能吸引用户关注，流失率明显下降'
>     },
>
>     '用户适应期': {
>         'duration': '4月22日-5月5日',
>         'churn_rate': 0.158,
>         'change': -9.7,
>         'analysis': '部分用户适应新功能，效果有所收敛'
>     },
>
>     '效果稳定期': {
>         'duration': '5月6日-5月19日',
>         'churn_rate': 0.165,
>         'change': -5.7,
>         'analysis': '新功能效果稳定，但仍有改善作用'
>     },
>
>     '长期影响期': {
>         'duration': '5月20日至今',
>         'churn_rate': 0.169,
>         'change': -3.4,
>         'analysis': '长期来看有轻微改善，需要继续观察'
>     }
> }
> ```

**主动行为分析的商业价值判断**：

> [!tip] AI智能答疑功能的ROI分析
>
> **投入成本**：
> - 开发成本：120万元
> - 运营成本：15万元/月
> - 总投入（6个月）：210万元
>
> **收益估算**：
> - 流失率平均下降：6.2%
> - 挽回用户数：约8500人
> - 每用户价值：800元LTV
> - 总收益：680万元
>
> **ROI计算**：(680-210)/210 = 223.8%
> **结论**：该功能投入产出比优秀，建议继续优化并推广

## 核心洞察与经验总结

### 关键发现

1. **预警时机是关键**：流失预测要在用户真正流失前7-14天进行干预
2. **分群精准度决定效果**：不同价值用户需要完全不同的召回策略
3. **多渠道协同效果**：单一渠道召回率15%，多渠道组合可达40%+

### 数据驱动的核心指标

```python
# 核心指标监控代码
def calculate_key_metrics(events_df, users_df):
    """计算流失预防的核心指标"""

    metrics = {
 # 预测准确性指标
        'prediction_precision': 0.73,  # 预测流失的准确率
        'prediction_recall': 0.68,     # 流失用户的识别率
        'prediction_f1': 0.70,        # 综合指标

 # 召回效果指标
        'overall_recall_rate': 0.28,   # 整体召回成功率
        'high_value_recall_rate': 0.45, # 高价值用户召回率
        'recall_cost_per_user': 12.50,  # 单用户召回成本

 # 业务影响指标
        'churn_rate_reduction': 0.15,   # 流失率下降幅度
        'ltv_protection': 0.82,         # 用户价值保护率
        'roi_retention_campaign': 3.2   # 召回活动ROI
    }

    return metrics

key_metrics = calculate_key_metrics(events_df, users_df)
print("流失预防核心指标:")
for metric, value in key_metrics.items():
    print(f"{metric}: {value}")

import redis
from datetime import datetime, timedelta
import json

class RealTimeChurnDetection:
    """实时用户流失预警系统"""

    def __init__(self, redis_client, model):
        self.redis = redis_client
        self.model = model
        self.alert_threshold = 0.7  # 流失概率阈值

    def update_user_realtime_features(self, user_id, event_data):
        """实时更新用户行为特征"""

 # 获取用户历史特征
        key = f"user_features:{user_id}"
        features = self.redis.hgetall(key)

        if not features:
            features = self.initialize_user_features(user_id)

 # 根据事件类型更新特征
        event_type = event_data['event_type']

        if event_type == 'login':
            features['last_login'] = datetime.now().isoformat()
            features['login_count_7d'] = str(int(features.get('login_count_7d', 0)) + 1)

        elif event_type == 'level_complete':
            features['levels_completed'] = str(int(features.get('levels_completed', 0)) + 1)
            features['last_progress'] = datetime.now().isoformat()

        elif event_type == 'purchase':
            features['total_spent'] = str(float(features.get('total_spent', 0)) + event_data['amount'])
            features['last_purchase'] = datetime.now().isoformat()

 # 保存更新的特征
        self.redis.hset(key, mapping=features)
        self.redis.expire(key, 2592000)  # 30天过期

 # 实时流失风险评估
        self.evaluate_churn_risk(user_id, features)

    def evaluate_churn_risk(self, user_id, features):
        """实时评估用户流失风险"""

 # 转换特征格式用于模型预测
        model_features = self.convert_features_for_model(features)

 # 预测流失概率
        churn_prob = self.model.predict_proba([model_features])[1]

 # 如果超过阈值，触发预警
        if churn_prob > self.alert_threshold:
            self.trigger_churn_alert(user_id, churn_prob, features)

    def trigger_churn_alert(self, user_id, churn_prob, features):
        """触发流失预警"""

        alert_data = {
            'user_id': user_id,
            'churn_probability': churn_prob,
            'alert_time': datetime.now().isoformat(),
            'user_segment': self.classify_user_segment(features),
            'suggested_actions': self.get_suggested_actions(features)
        }

 # 发送到预警队列
        self.redis.lpush('churn_alerts', json.dumps(alert_data))

 # 通知运营团队
        self.notify_operations_team(alert_data)

# 部署实时监控
redis_client = redis.Redis(host='localhost', port=6379, db=0)
realtime_detector = RealTimeChurnDetection(redis_client, best_model)

# A/B测试实验设计
ab_test_design = {
    'experiment_name': '付费用户流失召回策略对比',
    'target_audience': '流失7-14天的付费用户',
    'sample_size': 10000,  # 每组5000人

    'control_group': {
        'strategy': '传统邮件+优惠券召回',
        'content': '通用召回邮件 + 20%折扣券',
        'channel': '邮件推送',
        'cost_per_user': 5,
        'expected_recall_rate': 0.15
    },

    'treatment_group': {
        'strategy': '个性化多渠道召回',
        'content': '基于学习轨迹的个性化内容 + 课程延期补偿',
        'channel': '短信+App推送+客服电话',
        'cost_per_user': 25,
        'expected_recall_rate': 0.35
    },

    'test_duration': '14天',
    'key_metrics':
        'recall_success_rate',  # 召回成功率
        'ltv_recovered',        # 挽回的LTV
        'cost_per_recall',      # 单次召回成本
        'roi_14days'           # 14天ROI
    ]
}