[数据分析]长期影响评估：如何测量A-B测试的长期效应？

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

# 配置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DataCollectionSystem:
    def __init__(self, db_connection):
        self.db_connection = db_connection
        self.events_to_track = [
            'user_signup', 'user_login', 'purchase', 'content_view',
            'feature_usage', 'session_start', 'invite_sent'
        ]
    
    def track_event(self, user_id, event_type, properties=None):
        """
        跟踪用户事件
        """
        try:
            event_data = {
                'user_id': user_id,
                'event_type': event_type,
                'event_timestamp': datetime.now(),
                'properties': properties or {}
            }
            
            # 这里简化处理，实际应写入数据库或数据管道
            self._store_event(event_data)
            logger.info(f"Tracked event: {event_type} for user: {user_id}")
            
        except Exception as e:
            logger.error(f"Error tracking event: {str(e)}")
    
    def assign_experiment_group(self, user_id, experiment_name, group_name):
        """
        分配用户到实验组
        """
        assignment_data = {
            'user_id': user_id,
            'experiment_name': experiment_name,
            'group_name': group_name,
            'assignment_timestamp': datetime.now()
        }
        
        self._store_assignment(assignment_data)
        logger.info(f"Assigned user {user_id} to {experiment_name}/{group_name}")
    
    def _store_event(self, event_data):
        """存储事件数据到数据库"""
        # 实际实现中这里会有数据库写入逻辑
        pass
    
    def _store_assignment(self, assignment_data):
        """存储实验分配数据到数据库"""
        # 实际实现中这里会有数据库写入逻辑
        pass

# 使用示例
data_collector = DataCollectionSystem(db_connection="your_db_connection")

# 新用户注册时分配实验组
user_id = "user_123"
experiment_name = "new_ui_redesign"
group_name = "treatment" if np.random.random() > 0.5 else "control"

data_collector.assign_experiment_group(user_id, experiment_name, group_name)

# 跟踪用户事件
data_collector.track_event(user_id, "user_signup", {"signup_method": "email"})

def calculate_sample_size_long_term(alpha=0.05, power=0.8, baseline_value=0.1, 
                                  mde=0.02, attrition_rate=0.1, duration_weeks=8):
    """
    计算长期实验所需的样本量，考虑用户流失率
    
    参数:
    alpha: 显著性水平 (I类错误概率)
    power: 统计功效 (1 - II类错误概率)
    baseline_value: 基线指标值
    mde: 最小可检测效应 (Minimum Detectable Effect)
    attrition_rate: 每周用户流失率
    duration_weeks: 实验持续时间（周）
    
    返回:
    每组所需的初始样本量
    """
    from statsmodels.stats.power import NormalIndPower
    import math
    
    # 计算有效样本量（考虑流失）
    effective_sample_ratio = 0
    for week in range(duration_weeks):
        effective_sample_ratio += (1 - attrition_rate) ** week
    
    effective_sample_ratio /= duration_weeks
    
    # 计算不考虑流失的样本量
    power_analysis = NormalIndPower()
    base_sample_size = power_analysis.solve_power(
        effect_size=2 * math.asin(math.sqrt(baseline_value + mde)) - 2 * math.asin(math.sqrt(baseline_value)),
        power=power,
        alpha=alpha,
        ratio=1
    )
    
    # 调整样本量考虑流失
    adjusted_sample_size = base_sample_size / effective_sample_ratio
    
    return math.ceil(adjusted_sample_size)

# 示例计算
sample_size = calculate_sample_size_long_term(
    baseline_value=0.15,  # 15% 基线留存率
    mde=0.02,            # 检测2%的绝对提升
    attrition_rate=0.05,  # 每周5%流失率
    duration_weeks=6      # 6周实验
)

print(f"每组所需样本量: {sample_size}")
print(f"总样本量 (两组): {sample_size * 2}")

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

class SurvivalAnalysis:
    def __init__(self):
        self.kmf = KaplanMeierFitter()
    
    def prepare_survival_data(self, user_data, event_data, end_date):
        """
        准备生存分析数据
        
        参数:
        user_data: 包含用户ID和实验组别的DataFrame
        event_data: 包含用户事件和时间的DataFrame
        end_date: 分析截止日期
        
        返回:
        包含每个用户生存时间和是否发生事件的DataFrame
        """
        # 获取每个用户的首次活跃日期
        first_activity = event_data.groupby('user_id')['event_timestamp'].min().reset_index()
        first_activity.columns = ['user_id', 'first_date']
        
        # 获取每个用户的末次活跃日期
        last_activity = event_data.groupby('user_id')['event_timestamp'].max().reset_index()
        last_activity.columns = ['user_id', 'last_date']
        
        # 合并数据
        survival_data = user_data.merge(first_activity, on='user_id')
        survival_data = survival_data.merge(last_activity, on='user_id')
        
        # 计算生存时间（天）
        survival_data['signup_to_end'] = (end_date - survival_data['first_date']).dt.days
        survival_data['last_to_end'] = (end_date - survival_data['last_date']).dt.days
        
        # 定义是否流失（假设30天无活动为流失）
        survival_data['churned'] = (survival_data['last_to_end'] > 30).astype(int)
        
        # 计算生存时间（从注册到流失或截尾）
        survival_data['T'] = np.where(
            survival_data['churned'] == 1,
            (survival_data['last_date'] - survival_data['first_date']).dt.days,
            survival_data['signup_to_end']
        )
        
        return survival_data[['user_id', 'group', 'T', 'churned']]
    
    def plot_survival_curve(self, survival_data):
        """
        绘制生存曲线
        """
        plt.figure(figsize=(10, 6))
        
        groups = survival_data['group'].unique()
        for group in groups:
            group_data = survival_data[survival_data['group'] == group]
            self.kmf.fit(durations=group_data['T'], event_observed=group_data['churned'], label=group)
            self.kmf.plot_survival_function()
        
        plt.title('Survival Function by Experiment Group')
        plt.xlabel('Days since signup')
        plt.ylabel('Survival Probability')
        plt.legend()
        plt.grid(True)
        plt.show()
    
    def run_cox_regression(self, survival_data):
        """
        运行Cox比例风险模型
        """
        # 准备数据
        cox_data = survival_data.copy()
        
        # 可以添加其他协变量
        cox_data['group_code'] = (cox_data['group'] == 'treatment').astype(int)
        
        # 拟合Cox模型
        cph = CoxPHFitter()
        cph.fit(cox_data[['T', 'churned', 'group_code']], duration_col='T', event_col='churned')
        
        # 显示结果
        cph.print_summary()
        return cph

# 示例使用
def example_survival_analysis():
    # 生成模拟数据
    np.random.seed(42)
    n_users = 1000
    
    # 创建用户数据
    user_data = pd.DataFrame({
        'user_id': range(n_users),
        'group': np.random.choice(['control', 'treatment'], n_users, p=[0.5, 0.5])
    })
    
    # 创建事件数据（简化）
    event_data = []
    for user_id in range(n_users):
        signup_date = pd.Timestamp('2023-01-01') + pd.Timedelta(days=np.random.randint(0, 30))
        n_events = np.random.poisson(lam=10 if user_id % 3 != 0 else 2)  # 部分用户不活跃
        
        for i in range(n_events):
            event_date = signup_date + pd.Timedelta(days=np.random.randint(0, 90))
            event_data.append({
                'user_id': user_id,
                'event_timestamp': event_date,
                'event_type': np.random.choice(['login', 'purchase', 'view'], p=[0.6, 0.2, 0.2])
            })
    
    event_data = pd.DataFrame(event_data)
    end_date = pd.Timestamp('2023-04-01')
    
    # 进行生存分析
    sa = SurvivalAnalysis()
    survival_data = sa.prepare_survival_data(user_data, event_data, end_date)
    sa.plot_survival_curve(survival_data)
    cph_model = sa.run_cox_regression(survival_data)
    
    return survival_data, cph_model

# 运行示例
survival_data, cph_model = example_survival_analysis()

class DifferenceInDifferences:
    def __init__(self):
        pass
    
    def prepare_did_data(self, metric_data, experiment_start_date):
        """
        准备DID分析数据
        
        参数:
        metric_data: 包含日期、组别和指标的DataFrame
        experiment_start_date: 实验开始日期
        
        返回:
        适合DID分析的数据格式
        """
        did_data = metric_data.copy()
        did_data['post_treatment'] = (did_data['date'] >= experiment_start_date).astype(int)
        did_data['treatment_group'] = (did_data['group'] == 'treatment').astype(int)
        did_data['did_term'] = did_data['post_treatment'] * did_data['treatment_group']
        
        return did_data
    
    def run_did_analysis(self, did_data, metric_column):
        """
        运行DID分析
        """
        import statsmodels.formula.api as smf
        
        # 拟合DID模型
        model = smf.ols(f'{metric_column} ~ post_treatment + treatment_group + did_term', 
                       data=did_data).fit()
        
        # 输出结果
        print(model.summary())
        
        # 计算平均处理效应
        ate = model.params['did_term']
        print(f"\nAverage Treatment Effect (ATE): {ate:.4f}")
        
        return model

# 示例使用
def example_did_analysis():
    # 生成模拟数据
    np.random.seed(42)
    n_days = 90
    dates = pd.date_range('2023-01-01', periods=n_days)
    
    did_data = []
    for date in dates:
        for group in ['control', 'treatment']:
            # 基线值
            base_value = 100 + 0.1 * (date - dates[0]).days
            
            # 实验开始后的处理效应
            treatment_effect = 0
            if date >= pd.Timestamp('2023-02-01'):
                treatment_effect = 15 if group == 'treatment' else 0
            
            # 添加随机噪声
            noise = np.random.normal(0, 5)
            
            value = base_value + treatment_effect + noise
            
            did_data.append({
                'date': date,
                'group': group,
                'metric_value': value
            })
    
    did_data = pd.DataFrame(did_data)
    
    # 运行DID分析
    did_analyzer = DifferenceInDifferences()
    prepared_data = did_analyzer.prepare_did_data(did_data, pd.Timestamp('2023-02-01'))
    model = did_analyzer.run_did_analysis(prepared_data, 'metric_value')
    
    return prepared_data, model

# 运行示例
did_data, did_model = example_did_analysis()

class SyntheticControl:
    def __init__(self):
        pass
    
    def create_synthetic_control(self, pre_period_data, treatment_unit, control_units):
        """
        创建合成控制组
        
        参数:
        pre_period_data: 实验前期的数据
        treatment_unit: 处理单元标识符
        control_units: 控制单元列表
        """
        from sklearn.linear_model import LinearRegression
        
        # 准备数据
        pre_data = pre_period_data.pivot(index='date', columns='unit', values='metric')
        
        # 分离处理单元和控制单元
        y_pre = pre_data[treatment_unit].values
        X_pre = pre_data[control_units].values
        
        # 拟合模型找到最优权重
        model = LinearRegression(fit_intercept=False, positive=True)
        model.fit(X_pre, y_pre)
        
        # 计算权重
        weights = model.coef_
        
        # 创建合成控制组
        synthetic_control = pd.Series(
            np.dot(pre_data[control_units].values, weights),
            index=pre_data.index
        )
        
        return synthetic_control, weights, control_units
    
    def calculate_treatment_effect(self, full_data, treatment_unit, synthetic_control, intervention_date):
        """
        计算处理效应
        """
        # 提取处理组数据
        treatment_data = full_data[full_data['unit'] == treatment_unit].set_index('date')['metric']
        
        # 合并数据
        comparison = pd.DataFrame({
            'treatment': treatment_data,
            'synthetic_control': synthetic_control
        })
        
        # 分割实验前后数据
        pre_period = comparison[comparison.index < intervention_date]
        post_period = comparison[comparison.index >= intervention_date]
        
        # 计算处理效应
        pre_diff = (pre_period['treatment'] - pre_period['synthetic_control']).mean()
        post_diff = (post_period['treatment'] - post_period['synthetic_control']).mean()
        
        treatment_effect = post_diff - pre_diff
        
        return treatment_effect, comparison

# 示例使用（简化）
def example_synthetic_control():
    # 注意：这里需要实际的时间序列数据
    # 以下为概念性代码框架
    pass

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

class EcommerceDataGenerator:
    def __init__(self, n_users=10000, start_date='2023-01-01'):
        self.n_users = n_users
        self.start_date = pd.Timestamp(start_date)
    
    def generate_user_data(self):
        """生成用户基本数据"""
        user_ids = range(self.n_users)
        
        # 随机分配实验组别
        groups = np.random.choice(['control', 'treatment'], self.n_users, p=[0.5, 0.5])
        
        # 生成注册日期（实验前30天内）
        signup_dates = [self.start_date - timedelta(days=np.random.randint(1, 30)) 
                       for _ in range(self.n_users)]
        
        user_data = pd.DataFrame({
            'user_id': user_ids,
            'group': groups,
            'signup_date': signup_dates
        })
        
        return user_data
    
    def generate_behavior_data(self, user_data, end_date='2023-04-30'):
        """生成用户行为数据"""
        end_date = pd.Timestamp(end_date)
        behavior_data = []
        
        for _, user in user_data.iterrows():
            user_id = user['user_id']
            group = user['group']
            signup_date = user['signup_date']
            
            # 确定用户活跃期
            if group == 'treatment':
                # 处理组用户更可能长期活跃
                churn_prob = 0.005
                purchase_prob = 0.1
            else:
                churn_prob = 0.008
                purchase_prob = 0.08
            
            current_date = signup_date
            while current_date <= end_date and np.random.random() > churn_prob:
                # 生成每日行为
                n_sessions = np.random.poisson(1.5 if group == 'treatment' else 1.2)
                
                for _ in range(n_sessions):
                    session_data = {
                        'user_id': user_id,
                        'date': current_date,
                        'sessions': 1,
                        'page_views': np.random.poisson(8 if group == 'treatment' else 6),
                        'time_on_site': np.random.normal(300 if group == 'treatment' else 240, 60),
                        'made_purchase': 1 if np.random.random() < purchase_prob else 0
                    }
                    
                    if session_data['made_purchase']:
                        session_data['purchase_value'] = np.random.lognormal(3.5, 0.8)
                    else:
                        session_data['purchase_value'] = 0
                    
                    behavior_data.append(session_data)
                
                # 移动到下一天
                current_date += timedelta(days=1)
                
                # 随时间降低活跃度（自然衰减）
                churn_prob += 0.0001
        
        return pd.DataFrame(behavior_data)

# 生成数据
data_gen = EcommerceDataGenerator(n_users=5000)
user_data = data_gen.generate_user_data()
behavior_data = data_gen.generate_behavior_data(user_data)

# 保存数据
user_data.to_csv('ecommerce_user_data.csv', index=False)
behavior_data.to_csv('ecommerce_behavior_data.csv', index=False)

class EcommerceLongTermAnalysis:
    def __init__(self, user_data, behavior_data):
        self.user_data = user_data
        self.behavior_data = behavior_data
        self.merged_data = pd.merge(behavior_data, user_data, on='user_id')
    
    def calculate_daily_metrics(self):
        """计算每日指标"""
        daily_metrics = self.merged_data.groupby(['date', 'group']).agg({
            'user_id': 'nunique',
            'sessions': 'sum',
            'page_views': 'sum',
            'time_on_site': 'mean',
            'made_purchase': 'sum',
            'purchase_value': 'sum'
        }).reset_index()
        
        daily_metrics.columns = [
            'date', 'group', 'active_users', 'total_sessions', 
            'total_page_views', 'avg_time_on_site', 
            'purchase_count', 'total_revenue'
        ]
        
        # 计算转化率
        daily_metrics['conversion_rate'] = daily_metrics['purchase_count'] / daily_metrics['active_users']
        # 计算平均订单价值
        daily_metrics['aov'] = daily_metrics['total_revenue'] / daily_metrics['purchase_count']
        daily_metrics['aov'] = daily_metrics['aov'].fillna(0)
        
        return daily_metrics
    
    def analyze_cohort_retention(self, cohort_period='W'):
        """分析队列留存率"""
        # 创建注册队列
        self.user_data['cohort'] = self.user_data['signup_date'].dt.to_period(cohort_period)
        
        # 合并行为数据
        user_cohort_data = pd.merge(
            self.merged_data, 
            self.user_data[['user_id', 'cohort']], 
            on='user_id'
        )
        
        # 计算队列周期
        user_cohort_data['period'] = (
            (user_cohort_data['date'] - user_cohort_data['signup_date']).dt.days // 7
        )
        
        # 计算队列留存
        cohort_pivot = user_cohort_data.groupby(['cohort', 'group', 'period']).agg({
            'user_id': 'nunique'
        }).reset_index()
        
        # 计算留存率
        cohort_size = cohort_pivot.groupby(['cohort', 'group'])['user_id'].transform('first')
        cohort_pivot['retention_rate'] = cohort_pivot['user_id'] / cohort_size
        
        return cohort_pivot
    
    def calculate_ltv(self, prediction_horizon=90):
        """计算用户生命周期价值"""
        # 按组别计算每日ARPU
        daily_arpu = self.merged_data.groupby(['group', 'date']).agg({
            'user_id': 'nunique',
            'purchase_value': 'sum'
        }).reset_index()
        
        daily_arpu['arpu'] = daily_arpu['purchase_value'] / daily_arpu['user_id']
        
        # 计算留存曲线
        from scipy.optimize import curve_fit
        
        def exponential_decay(t, a, b, c):
            return a * np.exp(-b * t) + c
        
        # 按组别拟合留存曲线
        ltv_results = {}
        
        for group in ['control', 'treatment']:
            group_data = daily_arpu[daily_arpu['group'] == group].copy()
            group_data['day'] = (group_data['date'] - group_data['date'].min()).dt.days
            
            # 拟合留存曲线
            popt, _ = curve_fit(
                exponential_decay, 
                group_data['day'], 
                group_data['arpu'],
                p0=[1, 0.01, 0.1],
                maxfev=5000
            )
            
            # 预测未来ARPU
            future_days = np.arange(0, prediction_horizon)
            predicted_arpu = exponential_decay(future_days, *popt)
            
            # 计算LTV（累计预测ARPU）
            ltv = np.sum(predicted_arpu)
            ltv_results[group] = ltv
        
        return ltv_results

# 进行分析
analysis = EcommerceLongTermAnalysis(user_data, behavior_data)
daily_metrics = analysis.calculate_daily_metrics()
cohort_retention = analysis.analyze_cohort_retention()
ltv_results = analysis.calculate_ltv()

print("LTV Results:")
for group, ltv in ltv_results.items():
    print(f"{group}: ${ltv:.2f}")

# 计算相对提升
ltv_lift = (ltv_results['treatment'] - ltv_results['control']) / ltv_results['control'] * 100
print(f"\nLTV Lift: {ltv_lift:.2f}%")

import matplotlib.pyplot as plt
import seaborn as sns

def create_long_term_visualizations(daily_metrics, cohort_retention, ltv_results):
    """创建长期效应可视化"""
    plt.figure(figsize=(15, 10))
    
    # 1. 每日活跃用户趋势
    plt.subplot(2, 2, 1)
    for group in ['control', 'treatment']:
        group_data = daily_metrics[daily_metrics['group'] == group]
        plt.plot(group_data['date'], group_data['active_users'], label=group)
    
    plt.title('Daily Active Users Over Time')
    plt.xlabel('Date')
    plt.ylabel('Active Users')
    plt.legend()
    plt.xticks(rotation=45)
    
    # 2. 转化率趋势
    plt.subplot(2, 2, 2)
    for group in ['control', 'treatment']:
        group_data = daily_metrics[daily_metrics['group'] == group]
        plt.plot(group_data['date'], group_data['conversion_rate'], label=group)
    
    plt.title('Conversion Rate Over Time')
    plt.xlabel('Date')
    plt.ylabel('Conversion Rate')
    plt.legend()
    plt.xticks(rotation=45)
    
    # 3. 队列留存热图
    plt.subplot(2, 2, 3)
    treatment_retention = cohort_retention[cohort_retention['group'] == 'treatment']
    retention_pivot = treatment_retention.pivot_table(
        values='retention_rate', index='cohort', columns='period', fill_value=0
    )
    
    sns.heatmap(retention_pivot, cmap='YlGnBu', annot=True, fmt='.2%')
    plt.title('Treatment Group Cohort Retention')
    plt.xlabel('Weeks Since Signup')
    plt.ylabel('Cohort')
    
    # 4. LTV比较
    plt.subplot(2, 2, 4)
    groups = list(ltv_results.keys())
    ltv_values = list(ltv_results.values())
    
    plt.bar(groups, ltv_values, color=['blue', 'orange'])
    plt.title('Lifetime Value Comparison')
    plt.xlabel('Experiment Group')
    plt.ylabel('LTV ($)')
    
    for i, v in enumerate(ltv_values):
        plt.text(i, v + 0.5, f'${v:.2f}', ha='center')
    
    plt.tight_layout()
    plt.savefig('long_term_impact_analysis.png', dpi=300, bbox_inches='tight')
    plt.show()

# 创建可视化
create_long_term_visualizations(daily_metrics, cohort_retention, ltv_results)

# requirements.txt
pandas==1.5.3
numpy==1.24.3
matplotlib==3.7.1
seaborn==0.12.2
lifelines==0.27.7
scipy==1.10.1
statsmodels==0.14.0
scikit-learn==1.2.2
python-dotenv==1.0.0
sqlalchemy==2.0.19

# database.py
import sqlalchemy as db
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
from dotenv import load_dotenv

load_dotenv()

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    user_id = Column(Integer, primary_key=True)
    group = Column(String(50))
    signup_date = Column(DateTime)
    country = Column(String(100))
    device_type = Column(String(100))

class UserEvent(Base):
    __tablename__ = 'user_events'
    
    event_id = Column(Integer, primary_key=True, autoincrement=True)
    user_id = Column(Integer)
    event_type = Column(String(100))
    event_timestamp = Column(DateTime)
    session_id = Column(String(100))
    page_url = Column(String(500))
    duration = Column(Float)

class DatabaseManager:
    def __init__(self, connection_string=None):
        if connection_string is None:
            connection_string = os.getenv('DB_CONNECTION_STRING', 
                                         'sqlite:///ab_testing.db')
        
        self.engine = create_engine(connection_string)
        self.Session = sessionmaker(bind=self.engine)
        
    def init_database(self):
        """初始化数据库"""
        Base.metadata.create_all(self.engine)
        
    def get_session(self):
        """获取数据库会话"""
        return self.Session()
    
    def store_experiment_assignment(self, user_id, experiment_name, group_name):
        """存储实验分配结果"""
        session = self.get_session()
        try:
            assignment = ExperimentAssignment(
                user_id=user_id,
                experiment_name=experiment_name,
                group_name=group_name,
                assignment_timestamp=db.func.now()
            )
            session.add(assignment)
            session.commit()
        except Exception as e:
            session.rollback()
            raise e
        finally:
            session.close()

# 使用示例
db_manager = DatabaseManager()
db_manager.init_database()

# analysis_pipeline.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from database import DatabaseManager
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AnalysisPipeline:
    def __init__(self, db_manager):
        self.db_manager = db_manager
    
    def extract_data(self, experiment_name, start_date, end_date):
        """从数据库提取实验数据"""
        session = self.db_manager.get_session()
        
        try:
            # 获取实验用户
            query = """
            SELECT u.user_id, u.group, u.signup_date, u.country, u.device_type,
                   e.event_type, e.event_timestamp, e.session_id, e.duration
            FROM users u
            LEFT JOIN user_events e ON u.user_id = e.user_id
            WHERE u.signup_date BETWEEN :start_date AND :end_date
            AND EXISTS (
                SELECT 1 FROM experiment_assignments ea
                WHERE ea.user_id = u.user_id 
                AND ea.experiment_name = :experiment_name
            )
            """
            
            data = pd.read_sql_query(
                query, 
                self.db_manager.engine,
                params={'start_date': start_date, 'end_date': end_date, 
                       'experiment_name': experiment_name}
            )
            
            return data
            
        except Exception as e:
            logger.error(f"Error extracting data: {str(e)}")
            raise e
        finally:
            session.close()
    
    def calculate_metrics(self, raw_data):
        """计算关键指标"""
        # 确保时间格式正确
        raw_data['event_timestamp'] = pd.to_datetime(raw_data['event_timestamp'])
        raw_data['signup_date'] = pd.to_datetime(raw_data['signup_date'])
        
        # 按日期和组别聚合
        daily_metrics = raw_data.groupby([
            pd.Grouper(key='event_timestamp', freq='D'), 
            'group'
        ]).agg({
            'user_id': 'nunique',
            'session_id': 'nunique',
            'duration': 'sum'
        }).reset_index()
        
        daily_metrics.columns = ['date', 'group', 'active_users', 'sessions', 'total_duration']
        
        # 计算衍生指标
        daily_metrics['avg_session_duration'] = daily_metrics['total_duration'] / daily_metrics['sessions']
        daily_metrics['sessions_per_user'] = daily_metrics['sessions'] / daily_metrics['active_users']
        
        return daily_metrics
    
    def run_survival_analysis(self, raw_data, end_date):
        """运行生存分析"""
        from lifelines import KaplanMeierFitter
        
        # 准备生存分析数据
        user_last_activity = raw_data.groupby('user_id')['event_timestamp'].max().reset_index()
        user_first_activity = raw_data.groupby('user_id')['event_timestamp'].min().reset_index()
        
        user_data = raw_data[['user_id', 'group', 'signup_date']].drop_duplicates()
        
        survival_data = user_data.merge(user_first_activity, on='user_id')
        survival_data = survival_data.merge(user_last_activity, on='user_id', 
                                           suffixes=('_first', '_last'))
        
        # 计算生存时间
        end_date = pd.to_datetime(end_date)
        survival_data['T'] = (survival_data['event_timestamp_last'] - 
                             survival_data['event_timestamp_first']).dt.days
        
        # 标记删失（在分析期结束时仍活跃）
        survival_data['E'] = (survival_data['event_timestamp_last'] < end_date).astype(int)
        
        # 分组进行生存分析
        kmf = KaplanMeierFitter()
        
        results = {}
        for group in survival_data['group'].unique():
            group_data = survival_data[survival_data['group'] == group]
            kmf.fit(group_data['T'], event_observed=group_data['E'], label=group)
            results[group] = kmf
            
        return results, survival_data
    
    def calculate_retention_curves(self, raw_data, periods=[1, 7, 30]):
        """计算留存曲线"""
        retention_results = {}
        
        for period in periods:
            # 计算period日留存
            retention_data = raw_data.copy()
            retention_data['signup_period'] = retention_data['signup_date'].dt.to_period('D')
            retention_data['event_period'] = retention_data['event_timestamp'].dt.to_period('D')
            
            # 计算每个用户是否在注册后period天活跃
            user_retention = retention_data.groupby(['user_id', 'group', 'signup_period']).agg({
                'event_period': lambda x: (x.max() - x.min()).days >= period
            }).reset_index()
            
            user_retention.columns = ['user_id', 'group', 'signup_period', f'retained_{period}d']
            
            # 计算留存率
            retention_rate = user_retention.groupby(['group', 'signup_period'])[f'retained_{period}d'].mean()
            retention_results[period] = retention_rate.reset_index()
        
        return retention_results

# 使用示例
def run_complete_analysis():
    db_manager = DatabaseManager()
    pipeline = AnalysisPipeline(db_manager)
    
    # 提取数据
    start_date = '2023-01-01'
    end_date = '2023-04-30'
    experiment_name = 'homepage_redesign'
    
    raw_data = pipeline.extract_data(experiment_name, start_date, end_date)
    
    # 计算指标
    daily_metrics = pipeline.calculate_metrics(raw_data)
    
    # 生存分析
    survival_results, survival_data = pipeline.run_survival_analysis(raw_data, end_date)
    
    # 留存分析
    retention_results = pipeline.calculate_retention_curves(raw_data)
    
    return {
        'daily_metrics': daily_metrics,
        'survival_results': survival_results,
        'survival_data': survival_data,
        'retention_results': retention_results
    }

# 运行分析
results = run_complete_analysis()

# report_generator.py
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.backends.backend_pdf import PdfPages
import seaborn as sns
from datetime import datetime

class ReportGenerator:
    def __init__(self):
        sns.set_style("whitegrid")
        plt.rcParams['figure.figsize'] = [12, 8]
    
    def create_comprehensive_report(self, results, output_path='long_term_analysis_report.pdf'):
        """创建综合报告"""
        with PdfPages(output_path) as pdf:
            # 封面页
            self._create_cover_page(pdf)
            
            # 执行摘要
            self._create_executive_summary(pdf, results)
            
            # 每日指标趋势
            self._create_daily_metrics_plots(pdf, results['daily_metrics'])
            
            # 生存分析结果
            self._create_survival_analysis_plots(pdf, results['survival_results'])
            
            # 留存分析结果
            self._create_retention_analysis_plots(pdf, results['retention_results'])
            
            # 统计检验结果
            self._create_statistical_tests_section(pdf, results)
            
            # 结论和建议
            self._create_conclusions_recommendations(pdf, results)
        
        print(f"Report generated: {output_path}")
    
    def _create_cover_page(self, pdf):
        """创建报告封面"""
        plt.figure(figsize=(11, 8.5))
        plt.text(0.5, 0.7, 'A/B Test Long-Term Impact Analysis', 
                ha='center', va='center', fontsize=20, fontweight='bold')
        plt.text(0.5, 0.6, 'Comprehensive Report', 
                ha='center', va='center', fontsize=16)
        plt.text(0.5, 0.4, f'Generated on: {datetime.now().strftime("%Y-%m-%d %H:%M")}', 
                ha='center', va='center', fontsize=12)
        plt.axis('off')
        pdf.savefig()
        plt.close()
    
    def _create_executive_summary(self, pdf, results):
        """创建执行摘要"""
        plt.figure(figsize=(11, 8.5))
        plt.text(0.1, 0.9, 'Executive Summary', fontsize=16, fontweight='bold')
        
        # 这里添加关键指标摘要
        summary_text = """
        Key Findings:
        - Treatment group showed X% improvement in long-term retention
        - LTV increased by Y% in the treatment group
        - Statistical significance: p < 0.05 for primary metrics
        
        Recommendations:
        - Implement the treatment variant globally
        - Monitor long-term effects for 3 additional months
        - Conduct follow-up analysis on user segments
        """
        
        plt.text(0.1, 0.7, summary_text, fontsize=12, va='top')
        plt.axis('off')
        pdf.savefig()
        plt.close()
    
    def _create_daily_metrics_plots(self, pdf, daily_metrics):
        """创建每日指标图表"""
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        
        metrics_to_plot = ['active_users', 'sessions', 'avg_session_duration', 'sessions_per_user']
        titles = ['Active Users', 'Total Sessions', 'Avg Session Duration', 'Sessions per User']
        
        for i, metric in enumerate(metrics_to_plot):
            ax = axes[i//2, i%2]
            for group in daily_metrics['group'].unique():
                group_data = daily_metrics[daily_metrics['group'] == group]
                ax.plot(group_data['date'], group_data[metric], label=group)
            
            ax.set_title(titles[i])
            ax.legend()
            ax.tick_params(axis='x', rotation=45)
        
        plt.tight_layout()
        pdf.savefig()
        plt.close()
    
    def _create_survival_analysis_plots(self, pdf, survival_results):
        """创建生存分析图表"""
        plt.figure(figsize=(12, 8))
        
        for group, kmf in survival_results.items():
            kmf.plot_survival_function()
        
        plt.title('Survival Analysis by Experiment Group')
        plt.xlabel('Days since first activity')
        plt.ylabel('Survival Probability')
        plt.legend()
        
        pdf.savefig()
        plt.close()
    
    # 其他图表创建方法...
    def _create_retention_analysis_plots(self, pdf, retention_results):
        """创建留存分析图表"""
        # 实现留存图表
        pass
    
    def _create_statistical_tests_section(self, pdf, results):
        """创建统计检验部分"""
        # 实现统计检验结果
        pass
    
    def _create_conclusions_recommendations(self, pdf, results):
        """创建结论和建议部分"""
        # 实现结论和建议
        pass

# 生成报告
report_generator = ReportGenerator()
report_generator.create_comprehensive_report(results)

# scheduler.py
import schedule
import time
from datetime import datetime, timedelta
from analysis_pipeline import AnalysisPipeline
from database import DatabaseManager
from report_generator import ReportGenerator
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class LongTermAnalysisScheduler:
    def __init__(self):
        self.db_manager = DatabaseManager()
        self.pipeline = AnalysisPipeline(self.db_manager)
        self.report_generator = ReportGenerator()
    
    def run_daily_analysis(self):
        """每日分析任务"""
        logger.info("Starting daily long-term analysis...")
        
        try:
            # 获取昨天日期
            yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
            
            # 运行分析（过去90天数据）
            start_date = (datetime.now() - timedelta(days=90)).strftime('%Y-%m-%d')
            end_date = yesterday
            
            results = self.pipeline.run_complete_analysis()
            
            # 生成报告
            report_path = f'reports/long_term_analysis_{yesterday}.pdf'
            self.report_generator.create_comprehensive_report(results, report_path)
            
            logger.info(f"Daily analysis completed. Report saved to: {report_path}")
            
        except Exception as e:
            logger.error(f"Error in daily analysis: {str(e)}")
    
    def run_weekly_summary(self):
        """每周总结任务"""
        logger.info("Starting weekly summary...")
        
        # 实现每周总结逻辑
        # 包括趋势分析、效果评估、建议生成等
        
        logger.info("Weekly summary completed.")
    
    def start_scheduler(self):
        """启动调度器"""
        # 每天凌晨2点运行日常分析
        schedule.every().day.at("02:00").do(self.run_daily_analysis)
        
        # 每周一凌晨3点运行周总结
        schedule.every().monday.at("03:00").do(self.run_weekly_summary)
        
        logger.info("Scheduler started. Press Ctrl+C to exit.")
        
        while True:
            schedule.run_pending()
            time.sleep(60)

# 启动调度器
if __name__ == "__main__":
    scheduler = LongTermAnalysisScheduler()
    scheduler.start_scheduler()