长期效应评估：如何测量A/B测试的持续影响

I. 为什么长期效应评估至关重要

短期指标的局限性

短期A/B测试通常关注立即的转化率提升、点击率变化或收入增长。然而，这些指标往往无法捕捉变化的长期影响。例如：

• 新奇效应：用户可能因新鲜感而暂时改变行为，但效果会随时间衰减
• 学习曲线：用户需要时间适应新功能，真正价值可能延迟显现
• 生态系统影响：局部优化可能对产品其他部分产生负面连锁反应

实例分析：社交媒体的"优化"陷阱

某社交媒体平台通过A/B测试发现，增加推送通知频率可以显著提升短期活跃度。基于显著的短期结果，团队全面实施了新策略。然而，六个月后的数据显示：

• 用户月流失率增加了25%
• 用户满意度评分下降18%
• 长期用户价值(LTV)降低12%

进一步分析发现，频繁的推送虽然提高了即时打开率，但导致了用户疲劳和厌烦，最终损害了长期关系。

长期评估的商业价值

II. 长期效应评估的理论框架

时间维度分析

长期效应评估需要在多个时间尺度上进行分析：

因果推断挑战

长期评估面临独特的因果推断挑战：

• 时间依赖性：效应可能随时间变化
• 竞争风险：用户可能因多种原因流失
• 选择性偏差：长期留存的用户可能本身就有不同特征

评估指标体系

建立全面的长期评估指标体系：

class LongTermMetricsFramework:
    """长期评估指标框架"""
    
    def __init__(self):
        self.metrics_categories = {
            'engagement': ['DAU', 'WAU', 'session_duration', 'feature_adoption'],
            'retention': ['D1_retention', 'D7_retention', 'D30_retention', 'churn_rate'],
            'monetization': ['LTV', 'ARPU', 'purchase_frequency', 'average_order_value'],
            'quality': ['user_satisfaction', 'nps', 'complaint_rate', 'support_tickets']
        }
    
    def calculate_retention_metrics(self, user_data, treatment_col, date_col, user_col):
        """计算留存指标"""
        retention_results = {}
        
        for cohort_period in ['D1', 'D7', 'D30']:
            # 这里实现具体的留存计算逻辑
            retention_rate = self._compute_cohort_retention(
                user_data, treatment_col, date_col, user_col, cohort_period
            )
            retention_results[f'{cohort_period}_retention'] = retention_rate
        
        return retention_results
    
    def calculate_ltv(self, user_data, treatment_col, revenue_col, periods=365):
        """计算用户终身价值"""
        ltv_results = {}
        
        treatment_groups = user_data[treatment_col].unique()
        for group in treatment_groups:
            group_data = user_data[user_data[treatment_col] == group]
            # 简化版的LTV计算 - 实际中可能使用更复杂的模型
            avg_daily_revenue = group_data[revenue_col].mean()
            predicted_ltv = avg_daily_revenue * periods * 0.7  # 应用折扣因子
            
            ltv_results[group] = predicted_ltv
        
        return ltv_results

III. 长期效应评估的统计方法

生存分析应用

生存分析是评估长期效应的强大工具，特别适用于分析用户留存和流失模式：

import pandas as pd
import numpy as np
from lifelines import KaplanMeierFitter, CoxPHFitter
import matplotlib.pyplot as plt
import seaborn as sns

class SurvivalAnalysis:
    """生存分析用于长期效应评估"""
    
    def __init__(self):
        self.kmf = KaplanMeierFitter()
    
    def prepare_survival_data(self, user_data, start_date_col, end_date_col, 
                            event_col, treatment_col):
        """准备生存分析数据"""
        
        # 计算生存时间
        user_data['survival_time'] = (
            user_data[end_date_col] - user_data[start_date_col]
        ).dt.days
        
        # 标记事件（如流失）
        user_data['event_occurred'] = user_data[event_col]
        
        return user_data[['survival_time', 'event_occurred', treatment_col]]
    
    def kaplan_meier_analysis(self, survival_data, treatment_col):
        """执行Kaplan-Meier分析"""
        
        plt.figure(figsize(12, 8))
        
        treatment_groups = survival_data[treatment_col].unique()
        
        for group in treatment_groups:
            group_data = survival_data[survival_data[treatment_col] == group]
            self.kmf.fit(
                group_data['survival_time'],
                group_data['event_occurred'],
                label=f'Group {group}'
            )
            self.kmf.plot_survival_function()
        
        plt.title('Kaplan-Meier生存曲线 - 长期留存比较')
        plt.xlabel('时间 (天)')
        plt.ylabel('留存概率')
        plt.grid(True, alpha=0.3)
        plt.legend()
        plt.show()
        
        return self.kmf
    
    def cox_proportional_hazards(self, survival_data, treatment_col, covariates):
        """Cox比例风险模型"""
        
        cox_data = survival_data.copy()
        
        # 添加协变量
        for covariate in covariates:
            if covariate not in cox_data.columns:
                # 在实际应用中，这里应该从原始数据中获取协变量
                cox_data[covariate] = np.random.normal(0, 1, len(cox_data))
        
        cph = CoxPHFitter()
        cph.fit(cox_data, duration_col='survival_time', event_col='event_occurred')
        
        # 显示结果
        cph.print_summary()
        
        # 可视化风险比
        plt.figure(figsize(10, 6))
        cph.plot()
        plt.title('Cox模型变量重要性')
        plt.tight_layout()
        plt.show()
        
        return cph

# 示例使用
def generate_survival_data(n_users=2000, treatment_effect=0.3):
    """生成模拟生存数据"""
    
    np.random.seed(42)
    
    # 创建基础数据
    user_data = pd.DataFrame({
        'user_id': range(n_users),
        'treatment': np.random.choice([0, 1], n_users, p=[0.5, 0.5]),
        'signup_date': pd.date_range('2024-01-01', periods=n_users, freq='H'),
        'age': np.random.normal(35, 10, n_users),
        'activity_level': np.random.exponential(2, n_users)
    })
    
    # 模拟生存时间 - 处理组有更好的留存
    base_survival_time = np.random.exponential(180, n_users)  # 基础生存时间
    
    # 添加处理效应
    user_data['survival_time'] = base_survival_time * (
        1 + user_data['treatment'] * treatment_effect
    )
    
    # 添加随机审查
    user_data['censored'] = np.random.binomial(1, 0.8, n_users)
    user_data['event_occurred'] = 1 - user_data['censored']
    
    # 计算结束日期
    user_data['end_date'] = user_data['signup_date'] + pd.to_timedelta(
        user_data['survival_time'], unit='D'
    )
    
    return user_data

# 执行生存分析
survival_analyzer = SurvivalAnalysis()
simulated_data = generate_survival_data(2000, 0.25)

survival_data = survival_analyzer.prepare_survival_data(
    simulated_data, 'signup_date', 'end_date', 'event_occurred', 'treatment'
)

# Kaplan-Meier分析
km_results = survival_analyzer.kaplan_meier_analysis(survival_data, 'treatment')

# Cox比例风险模型
covariates = ['age', 'activity_level']
cox_model = survival_analyzer.cox_proportional_hazards(
    survival_data, 'treatment', covariates
)

差异中的差异(DID)方法

对于长期评估，DID方法可以控制时间趋势和组间固有差异：

class DifferenceInDifferences:
    """差异中的差异分析"""
    
    def __init__(self):
        self.results = {}
    
    def prepare_did_data(self, data, time_periods, treatment_group, outcome_var):
        """准备DID分析数据"""
        
        did_data = data.copy()
        
        # 确保数据包含所需的时间周期和处理组信息
        required_cols = [time_periods, treatment_group, outcome_var]
        assert all(col in did_data.columns for col in required_cols)
        
        return did_data
    
    def calculate_did_estimate(self, data, time_periods, treatment_group, outcome_var):
        """计算DID估计值"""
        
        # 分组计算平均值
        summary = data.groupby([time_periods, treatment_group])[outcome_var].agg([
            'mean', 'count', 'std'
        ]).reset_index()
        
        # 计算DID估计值
        pre_treatment_control = summary[
            (summary[time_periods] == 'pre') & (summary[treatment_group] == 0)
        ]['mean'].values[0]
        
        pre_treatment_treatment = summary[
            (summary[time_periods] == 'pre') & (summary[treatment_group] == 1)
        ]['mean'].values[0]
        
        post_treatment_control = summary[
            (summary[time_periods] == 'post') & (summary[treatment_group] == 0)
        ]['mean'].values[0]
        
        post_treatment_treatment = summary[
            (summary[time_periods] == 'post') & (summary[treatment_group] == 1)
        ]['mean'].values[0]
        
        # DID计算
        did_estimate = (
            (post_treatment_treatment - pre_treatment_treatment) -
            (post_treatment_control - pre_treatment_control)
        )
        
        # 计算标准误（简化版本）
        n_treatment_pre = summary[
            (summary[time_periods] == 'pre') & (summary[treatment_group] == 1)
        ]['count'].values[0]
        
        n_treatment_post = summary[
            (summary[time_periods] == 'post') & (summary[treatment_group] == 1)
        ]['count'].values[0]
        
        n_control_pre = summary[
            (summary[time_periods] == 'pre') & (summary[treatment_group] == 0)
        ]['count'].values[0]
        
        n_control_post = summary[
            (summary[time_periods] == 'post') & (summary[treatment_group] == 0)
        ]['count'].values[0]
        
        # 合并方差估计
        pooled_variance = (
            summary[summary[treatment_group] == 1]['std'].var() +
            summary[summary[treatment_group] == 0]['std'].var()
        ) / 2
        
        se = np.sqrt(pooled_variance * (1/n_treatment_pre + 1/n_treatment_post + 
                                      1/n_control_pre + 1/n_control_post))
        
        # 置信区间
        confidence_interval = (
            did_estimate - 1.96 * se,
            did_estimate + 1.96 * se
        )
        
        self.results = {
            'did_estimate': did_estimate,
            'standard_error': se,
            'confidence_interval': confidence_interval,
            't_statistic': did_estimate / se,
            'significant': abs(did_estimate / se) > 1.96
        }
        
        return self.results
    
    def plot_did_analysis(self, data, time_periods, treatment_group, outcome_var):
        """可视化DID分析"""
        
        plt.figure(figsize(10, 6))
        
        # 准备绘图数据
        plot_data = data.groupby([time_periods, treatment_group])[outcome_var].mean().reset_index()
        
        # 分别绘制处理组和对照组
        for group in [0, 1]:
            group_data = plot_data[plot_data[treatment_group] == group]
            plt.plot([0, 1], group_data[outcome_var].values, 
                    marker='o', linewidth=2, label=f'Group {group}')
        
        plt.xticks([0, 1], ['Pre-Treatment', 'Post-Treatment'])
        plt.ylabel(outcome_var)
        plt.title('Difference-in-Differences Analysis')
        plt.legend()
        plt.grid(True, alpha=0.3)
        
        # 添加DID估计标注
        did_result = self.results.get('did_estimate', 0)
        plt.annotate(f'DID Estimate: {did_result:.3f}', 
                    xy=(0.5, 0.5), xycoords='axes fraction',
                    bbox=dict(boxstyle="round,pad=0.3", fc="white", ec="gray", alpha=0.8))
        
        plt.tight_layout()
        plt.show()

# 生成DID模拟数据
def generate_did_data(n_samples=4000):
    """生成DID分析模拟数据"""
    
    np.random.seed(42)
    
    data = pd.DataFrame({
        'user_id': range(n_samples),
        'time_period': np.random.choice(['pre', 'post'], n_samples),
        'treatment_group': np.random.choice([0, 1], n_samples, p=[0.7, 0.3]),
        'age': np.random.normal(35, 10, n_samples),
        'pre_existing_trend': np.random.normal(0, 1, n_samples)
    })
    
    # 生成结果变量 - 包含时间趋势和处理效应
    base_outcome = 100
    time_trend = 5  # 自然增长趋势
    treatment_effect = 8  # 真实处理效应
    
    data['outcome'] = (
        base_outcome +
        (data['time_period'] == 'post') * time_trend +
        (data['treatment_group'] == 1) * (data['time_period'] == 'post') * treatment_effect +
        data['pre_existing_trend'] +
        np.random.normal(0, 10, n_samples)  # 随机噪声
    )
    
    return data

# 执行DID分析
did_analyzer = DifferenceInDifferences()
did_data = generate_did_data()

did_results = did_analyzer.calculate_did_estimate(
    did_data, 'time_period', 'treatment_group', 'outcome'
)

print("DID分析结果:")
for key, value in did_results.items():
    print(f"{key}: {value}")

did_analyzer.plot_did_analysis(did_data, 'time_period', 'treatment_group', 'outcome')

长期评估方法总结

IV. 实例分析：电商平台搜索算法更新的长期影响

背景介绍

某大型电商平台对搜索算法进行了重大更新，旨在提高搜索结果的准确性。短期A/B测试（两周）显示：

• 搜索点击率提升12%（p < 0.01）
• 添加到购物车率提升8%（p < 0.05）
• 无负面指标显著变化

基于这些积极结果，团队决定全量发布新算法。

长期跟踪与发现

团队建立了系统的长期评估框架，跟踪以下指标六个月：

class EcommerceLongTermEvaluation:
    """电商平台长期评估案例"""
    
    def __init__(self):
        self.metrics_timeline = {}
        self.user_cohorts = {}
    
    def simulate_ecommerce_data(self, n_users=10000, observation_period=180):
        """模拟电商平台长期数据"""
        
        np.random.seed(42)
        
        # 创建用户基础数据
        users = pd.DataFrame({
            'user_id': range(n_users),
            'treatment': np.random.choice([0, 1], n_users, p=[0.5, 0.5]),
            'cohort': pd.date_range('2024-01-01', periods=n_users, freq='H'),
            'user_segment': np.random.choice(['new', 'returning', 'vip'], n_users, 
                                           p=[0.6, 0.35, 0.05]),
            'historical_value': np.random.gamma(200, 1, n_users)
        })
        
        # 生成长期行为数据
        long_term_data = []
        
        for user_id, user_row in users.iterrows():
            user_treatment = user_row['treatment']
            user_segment = user_row['user_segment']
            
            # 基础行为模式基于用户细分
            if user_segment == 'new':
                base_activity = 0.3
                retention_prob = 0.6
            elif user_segment == 'returning':
                base_activity = 0.5
                retention_prob = 0.75
            else:  # vip
                base_activity = 0.8
                retention_prob = 0.9
            
            # 处理效应：短期正面，但可能有长期负面效应
            short_term_boost = 0.15 if user_treatment == 1 else 0
            long_term_effect = -0.08 if user_treatment == 1 else 0  # 潜在的负面效应
            
            for day in range(observation_period):
                # 用户留存概率随时间衰减
                daily_retention = retention_prob * np.exp(-0.01 * day)
                
                if np.random.random() < daily_retention:
                    # 用户当天活跃
                    activity_level = base_activity * (
                        1 + short_term_boost * np.exp(-0.1 * day) +  # 短期效应衰减
                        long_term_effect * (1 - np.exp(-0.05 * day))  # 长期效应逐渐显现
                    )
                    
                    # 生成日常指标
                    searches = np.random.poisson(activity_level * 3)
                    clicks = np.random.binomial(searches, 0.4 * (1 + short_term_boost))
                    add_to_cart = np.random.binomial(clicks, 0.2 * (1 + short_term_boost * 0.5))
                    purchases = np.random.binomial(add_to_cart, 0.3)
                    
                    revenue = purchases * np.random.gamma(50, 1)
                    
                    long_term_data.append({
                        'user_id': user_id,
                        'treatment': user_treatment,
                        'user_segment': user_segment,
                        'day': day,
                        'searches': searches,
                        'clicks': clicks,
                        'add_to_cart': add_to_cart,
                        'purchases': purchases,
                        'revenue': revenue,
                        'active': 1
                    })
                else:
                    # 用户流失
                    long_term_data.append({
                        'user_id': user_id,
                        'treatment': user_treatment,
                        'user_segment': user_segment,
                        'day': day,
                        'searches': 0,
                        'clicks': 0,
                        'add_to_cart': 0,
                        'purchases': 0,
                        'revenue': 0,
                        'active': 0
                    })
        
        return pd.DataFrame(long_term_data)
    
    def analyze_long_term_trends(self, long_term_data):
        """分析长期趋势"""
        
        results = {}
        
        # 按周聚合数据
        long_term_data['week'] = long_term_data['day'] // 7
        
        weekly_metrics = long_term_data.groupby(['week', 'treatment']).agg({
            'active': 'mean',
            'searches': 'mean',
            'clicks': 'mean', 
            'add_to_cart': 'mean',
            'purchases': 'mean',
            'revenue': 'mean',
            'user_id': 'nunique'
        }).reset_index()
        
        # 计算每周的相对差异
        control_weekly = weekly_metrics[weekly_metrics['treatment'] == 0]
        treatment_weekly = weekly_metrics[weekly_metrics['treatment'] == 1]
        
        merged_weekly = pd.merge(control_weekly, treatment_weekly, on='week', 
                                suffixes=('_control', '_treatment'))
        
        # 计算效应大小和趋势
        metrics_to_analyze = ['active', 'searches', 'clicks', 'add_to_cart', 'purchases', 'revenue']
        
        for metric in metrics_to_analyze:
            control_col = f'{metric}_control'
            treatment_col = f'{metric}_treatment'
            
            merged_weekly[f'{metric}_effect'] = (
                merged_weekly[treatment_col] - merged_weekly[control_col]
            ) / merged_weekly[control_col]
            
            # 计算效应衰减率（使用简单线性回归）
            from scipy.stats import linregress
            weeks = merged_weekly['week']
            effects = merged_weekly[f'{metric}_effect']
            
            slope, intercept, r_value, p_value, std_err = linregress(weeks, effects)
            
            results[metric] = {
                'initial_effect': effects.iloc[0] if len(effects) > 0 else 0,
                'final_effect': effects.iloc[-1] if len(effects) > 0 else 0,
                'effect_decay_rate': slope,
                'decay_p_value': p_value,
                'mean_effect': effects.mean(),
                'effect_trend': '衰减' if slope < 0 else '稳定' if abs(slope) < 0.001 else '增长'
            }
        
        return results, merged_weekly
    
    def plot_long_term_effects(self, merged_weekly_data):
        """可视化长期效应"""
        
        fig, axes = plt.subplots(2, 2, figsize=(15, 12))
        
        metrics_to_plot = ['active', 'clicks', 'add_to_cart', 'revenue']
        
        for idx, metric in enumerate(metrics_to_plot):
            ax = axes[idx//2, idx%2]
            
            effect_col = f'{metric}_effect'
            
            if effect_col in merged_weekly_data.columns:
                ax.plot(merged_weekly_data['week'], merged_weekly_data[effect_col], 
                       marker='o', linewidth=2, label='相对效应')
                
                # 添加趋势线
                z = np.polyfit(merged_weekly_data['week'], merged_weekly_data[effect_col], 1)
                p = np.poly1d(z)
                ax.plot(merged_weekly_data['week'], p(merged_weekly_data['week']), 
                       "r--", alpha=0.7, label='趋势线')
            
            ax.set_xlabel('周数')
            ax.set_ylabel('相对效应')
            ax.set_title(f'{metric.upper()}指标的长期效应')
            ax.legend()
            ax.grid(True, alpha=0.3)
            
            # 添加零效应参考线
            ax.axhline(y=0, color='black', linestyle='-', alpha=0.5)
        
        plt.tight_layout()
        plt.show()

# 执行电商案例分析
ecommerce_evaluator = EcommerceLongTermEvaluation()
long_term_data = ecommerce_evaluator.simulate_ecommerce_data(5000, 180)

results, weekly_data = ecommerce_evaluator.analyze_long_term_trends(long_term_data)
ecommerce_evaluator.plot_long_term_effects(weekly_data)

print("电商搜索算法长期评估结果:")
for metric, result in results.items():
    print(f"\n{metric}:")
    print(f"  初始效应: {result['initial_effect']:.3f}")
    print(f"  最终效应: {result['final_effect']:.3f}")
    print(f"  效应衰减率: {result['effect_decay_rate']:.5f}")
    print(f"  衰减显著性: {result['decay_p_value']:.3f}")
    print(f"  效应趋势: {result['effect_trend']}")

关键发现与洞见

通过六个月的长期跟踪，团队发现了短期测试未能揭示的重要模式：

V. 长期评估的技术实现框架

数据管道设计

建立稳健的长期评估数据管道：

class LongTermEvaluationPipeline:
    """长期评估数据管道"""
    
    def __init__(self, experiment_id, start_date, evaluation_period=180):
        self.experiment_id = experiment_id
        self.start_date = start_date
        self.evaluation_period = evaluation_period
        self.data_sources = {}
    
    def setup_data_connections(self):
        """设置数据源连接"""
        # 在实际应用中，这里会连接各种数据源
        # 如数据仓库、用户行为日志、业务数据库等
        
        self.data_sources = {
            'user_behavior': 'user_events_table',
            'transactions': 'transactions_table', 
            'user_attributes': 'user_profile_table',
            'experiment_assignments': 'experiment_assignments_table'
        }
        
        return self.data_sources
    
    def extract_long_term_data(self, current_date):
        """提取长期评估数据"""
        
        # 计算应该提取的数据时间范围
        end_date = self.start_date + pd.Timedelta(days=self.evaluation_period)
        
        if current_date < end_date:
            # 如果还在评估期内，提取到当前日期的数据
            extraction_end = current_date
        else:
            extraction_end = end_date
        
        # 模拟数据提取 - 实际中会是SQL查询或API调用
        query = f"""
        SELECT 
            u.user_id,
            u.treatment_group,
            u.assignment_date,
            ub.event_type,
            ub.event_date,
            ub.event_value,
            t.transaction_amount,
            t.transaction_date,
            up.user_segment,
            up.registration_date
        FROM {self.data_sources['user_attributes']} up
        JOIN {self.data_sources['experiment_assignments']} u 
            ON up.user_id = u.user_id
        LEFT JOIN {self.data_sources['user_behavior']} ub 
            ON up.user_id = ub.user_id
        LEFT JOIN {self.data_sources['transactions']} t 
            ON up.user_id = t.user_id
        WHERE u.experiment_id = '{self.experiment_id}'
            AND u.assignment_date >= '{self.start_date}'
            AND COALESCE(ub.event_date, t.transaction_date, u.assignment_date) <= '{extraction_end}'
        """
        
        print(f"执行查询: {query}")
        # 在实际应用中，这里会返回真实的查询结果
        return self._simulate_query_results()
    
    def _simulate_query_results(self):
        """模拟查询结果"""
        # 生成模拟数据代替实际查询
        np.random.seed(42)
        n_users = 2000
        
        simulated_data = pd.DataFrame({
            'user_id': range(n_users),
            'treatment_group': np.random.choice([0, 1], n_users),
            'assignment_date': pd.to_datetime('2024-01-01'),
            'user_segment': np.random.choice(['A', 'B', 'C'], n_users),
            'registration_date': pd.to_datetime('2023-06-01') + 
                               pd.to_timedelta(np.random.randint(0, 210, n_users), unit='D')
        })
        
        # 生成行为事件数据
        events_data = []
        for user_id in range(n_users):
            n_events = np.random.poisson(15)  # 每个用户平均15个事件
            for i in range(n_events):
                event_date = pd.to_datetime('2024-01-01') + pd.to_timedelta(
                    np.random.randint(0, 180), unit='D'
                )
                events_data.append({
                    'user_id': user_id,
                    'event_type': np.random.choice(['search', 'click', 'purchase']),
                    'event_date': event_date,
                    'event_value': np.random.exponential(10)
                })
        
        events_df = pd.DataFrame(events_data)
        
        # 合并数据
        merged_data = pd.merge(simulated_data, events_df, on='user_id', how='left')
        
        return merged_data
    
    def calculate_rolling_metrics(self, data, window_sizes=[7, 30, 90]):
        """计算滚动窗口指标"""
        
        rolling_metrics = {}
        
        for window in window_sizes:
            window_data = data[
                data['event_date'] <= data['assignment_date'] + pd.Timedelta(days=window)
            ]
            
            metrics = window_data.groupby(['user_id', 'treatment_group']).agg({
                'event_type': 'count',
                'event_value': 'sum'
            }).reset_index()
            
            # 计算分组统计
            group_metrics = metrics.groupby('treatment_group').agg({
                'event_type': ['mean', 'std'],
                'event_value': ['mean', 'std']
            })
            
            rolling_metrics[f'{window}_day'] = {
                'data': metrics,
                'summary': group_metrics
            }
        
        return rolling_metrics
    
    def monitor_metric_convergence(self, rolling_metrics, target_metrics):
        """监控指标收敛情况"""
        
        convergence_report = {}
        
        for metric in target_metrics:
            convergence_data = []
            
            for window, window_data in rolling_metrics.items():
                summary = window_data['summary']
                
                # 提取处理效应
                if metric in summary.columns.get_level_values(0):
                    control_mean = summary[metric]['mean'].get(0, 0)
                    treatment_mean = summary[metric]['mean'].get(1, 0)
                    
                    if control_mean > 0:
                        effect_size = (treatment_mean - control_mean) / control_mean
                    else:
                        effect_size = 0
                    
                    convergence_data.append({
                        'window': int(window.split('_')[0]),
                        'effect_size': effect_size
                    })
            
            convergence_df = pd.DataFrame(convergence_data)
            
            # 检查收敛性（效应大小变化小于阈值）
            if len(convergence_df) > 1:
                recent_change = abs(
                    convergence_df['effect_size'].iloc[-1] - 
                    convergence_df['effect_size'].iloc[-2]
                )
                converged = recent_change < 0.01  # 1%变化阈值
            else:
                converged = False
            
            convergence_report[metric] = {
                'converged': converged,
                'convergence_timeline': convergence_df,
                'final_effect_size': convergence_df['effect_size'].iloc[-1] if len(convergence_df) > 0 else 0
            }
        
        return convergence_report

# 使用长期评估管道
pipeline = LongTermEvaluationPipeline(
    experiment_id='search_algorithm_v2',
    start_date=pd.to_datetime('2024-01-01'),
    evaluation_period=180
)

pipeline.setup_data_connections()
current_data = pipeline.extract_long_term_data(pd.to_datetime('2024-04-01'))

rolling_metrics = pipeline.calculate_rolling_metrics(current_data)
convergence_report = pipeline.monitor_metric_convergence(
    rolling_metrics, ['event_type', 'event_value']
)

print("指标收敛报告:")
for metric, report in convergence_report.items():
    print(f"{metric}: 收敛={report['converged']}, 最终效应={report['final_effect_size']:.3f}")

长期评估架构总结

VI. 组织实践与最佳实践

建立长期评估文化

成功实施长期效应评估需要组织文化的支持：

实践路线图

资源分配建议

有效的长期评估需要合理的资源分配：

class LongTermEvaluationROI:
    """长期评估ROI分析"""
    
    def __init__(self):
        self.cost_components = {}
        self.benefit_components = {}
    
    def estimate_costs(self, team_size, project_duration, infrastructure_costs):
        """估算长期评估成本"""
        
        # 人力成本
        monthly_salary = 10000  # 平均月薪
        human_cost = team_size * monthly_salary * project_duration
        
        # 基础设施成本
        infra_cost = infrastructure_costs * project_duration
        
        # 机会成本（简化估计）
        opportunity_cost = human_cost * 0.3
        
        total_cost = human_cost + infra_cost + opportunity_cost
        
        self.cost_components = {
            'human_resources': human_cost,
            'infrastructure': infra_cost,
            'opportunity_cost': opportunity_cost,
            'total_cost': total_cost
        }
        
        return self.cost_components
    
    def estimate_benefits(self, historical_errors, error_cost, prevention_rate=0.7):
        """估算长期评估收益"""
        
        # 基于历史错误避免的收益
        preventable_errors = historical_errors * prevention_rate
        error_avoidance_benefit = preventable_errors * error_cost
        
        # 决策质量提升收益（估计）
        decision_quality_benefit = error_avoidance_benefit * 0.5
        
        # 长期业务健康收益（估计）
        business_health_benefit = error_avoidance_benefit * 0.8
        
        total_benefit = error_avoidance_benefit + decision_quality_benefit + business_health_benefit
        
        self.benefit_components = {
            'error_avoidance': error_avoidance_benefit,
            'decision_quality': decision_quality_benefit,
            'business_health': business_health_benefit,
            'total_benefit': total_benefit
        }
        
        return self.benefit_components
    
    def calculate_roi(self):
        """计算投资回报率"""
        
        if not self.cost_components or not self.benefit_components:
            return "请先估算成本和收益"
        
        total_cost = self.cost_components['total_cost']
        total_benefit = self.benefit_components['total_benefit']
        
        if total_cost > 0:
            roi = (total_benefit - total_cost) / total_cost
            payback_period = total_cost / (total_benefit / 12)  # 月为单位
            
            return {
                'roi': roi,
                'payback_period_months': payback_period,
                'net_benefit': total_benefit - total_cost,
                'benefit_cost_ratio': total_benefit / total_cost
            }
        else:
            return "成本必须大于零"

# ROI分析示例
roi_analyzer = LongTermEvaluationROI()

# 估算成本
costs = roi_analyzer.estimate_costs(
    team_size=2,           # 2人团队
    project_duration=6,    # 6个月项目
    infrastructure_costs=2000  # 每月基础设施成本
)

# 估算收益 - 基于历史数据：过去12个月有5次重大决策错误，平均每次错误成本$50,000
benefits = roi_analyzer.estimate_benefits(
    historical_errors=5,
    error_cost=50000,
    prevention_rate=0.7
)

roi_results = roi_analyzer.calculate_roi()

print("长期评估ROI分析:")
print(f"总成本: ${costs['total_cost']:,.2f}")
print(f"总收益: ${benefits['total_benefit']:,.2f}")
print(f"ROI: {roi_results['roi']:.1%}")
print(f"回收期: {roi_results['payback_period_months']:.1f} 个月")
print(f"净收益: ${roi_results['net_benefit']:,.2f}")