Ab Testing Analyzer

#!/usr/bin/env python3
"""
基础AB测试示例

演示AB测试分析技能的核心功能：
- 数据加载和预处理
- 转化率分析
- 统计显著性检验
- 基础可视化
"""

import sys
import os
import pandas as pd
import numpy as np
from pathlib import Path

# 添加技能路径
skill_path = Path(__file__).parent.parent
sys.path.append(str(skill_path))

from scripts.ab_test_analyzer import ABTestAnalyzer
from scripts.statistical_tests import StatisticalTests
from scripts.visualizer import ABTestVisualizer


def main():
    """主函数：演示基础AB测试分析流程"""
    print("=" * 60)
    print("AB测试分析技能 - 基础示例")
    print("=" * 60)

    # 1. 初始化组件
    print("\n1. 初始化AB测试分析组件...")
    analyzer = ABTestAnalyzer()
    stats_tests = StatisticalTests()
    visualizer = ABTestVisualizer()

    # 2. 加载样本数据
    print("\n2. 加载AB测试样本数据...")
    data_dir = Path(__file__).parent / "sample_data"
    data_file = data_dir / "sample_ab_test_data.csv"

    if not data_file.exists():
        print(f"❌ 数据文件不存在: {data_file}")
        return False

    data = analyzer.load_data(str(data_file))
    if data is None:
        return False

    print(f"✅ 数据加载成功：{len(data)} 条用户记录")
    print(f"   - 实验组别: {data['实验组别'].unique()}")
    print(f"   - 总转化率: {data['转化状态'].apply(lambda x: 1 if str(x).strip() in ['是', 'True', 'true', '1', 'yes'] else 0).mean():.2%}")

    # 3. 数据质量检查
    print("\n3. 数据质量检查...")
    quality_report = analyzer.data_quality_check(
        data,
        required_cols=['用户ID', '实验组别', '转化状态']
    )

    print(f"   - 总行数: {quality_report['total_rows']:,}")
    print(f"   - 总列数: {quality_report['total_columns']}")
    print(f"   - 重复行数: {quality_report['duplicate_rows']}")
    print(f"   - 必需列检查: {quality_report['required_columns_check']}")

    if quality_report['missing_values']:
        print("   - 缺失值统计:")
        for col, info in quality_report['missing_values'].items():
            print(f"     * {col}: {info['count']} ({info['percentage']:.1f}%)")

    # 4. 基础转化率分析
    print("\n4. 基础转化率分析...")
    conversion_results = analyzer.analyze_conversion(
        data=data,
        group_col='实验组别',
        conversion_col='转化状态',
        control_group='对照组'
    )

    print("   - 各组转化率:")
    for group, stats in conversion_results['conversion_analysis'].items():
        print(f"     * {group}: {stats['conversion_rate']:.2%} "
              f"({stats['conversions']}/{stats['sample_size']})")

    # 5. 提升率分析
    print("\n5. 提升率分析...")
    if conversion_results['lift_analysis']:
        for group, lift in conversion_results['lift_analysis'].items():
            print(f"   - {group} vs 对照组:")
            print(f"     * 绝对提升: {lift['absolute_lift']:.2%}")
            print(f"     * 相对提升: {lift['relative_lift']:.2%}")
            print(f"     * 提升百分比: {lift['lift_percentage']:.1f}%")
            ci_lower, ci_upper = lift['lift_confidence_interval']
            print(f"     * 置信区间: [{ci_lower:.2%}, {ci_upper:.2%}]")

    # 6. 统计显著性检验
    print("\n6. 统计显著性检验...")

    # 转化状态数值化
    data_converted = data.copy()
    data_converted['转化状态_数值'] = data_converted['转化状态'].apply(
        lambda x: 1 if str(x).strip() in ['是', 'True', 'true', '1', 'yes'] else 0
    )

    # t检验
    t_test_result = stats_tests.t_test(
        data=data_converted,
        group_col='实验组别',
        metric_col='转化状态_数值',
        test_type='independent'
    )

    print("   - t检验结果:")
    print(f"     * 统计量: {t_test_result['statistic']:.4f}")
    print(f"     * P值: {t_test_result['p_value']:.4f}")
    print(f"     * 效应量 (Cohen's d): {t_test_result['effect_size']:.4f}")
    print(f"     * 效应量解释: {t_test_result['effect_size_interpretation']}")
    print(f"     * 显著性: {'是' if t_test_result['is_significant'] else '否'}")

    # 卡方检验
    chi2_result = stats_tests.chi_square_test(
        data=data,
        group_col='实验组别',
        outcome_col='转化状态',
        test_type='independence'
    )

    print("   - 卡方检验结果:")
    print(f"     * 卡方统计量: {chi2_result['statistic']:.4f}")
    print(f"     * P值: {chi2_result['p_value']:.4f}")
    print(f"     * 效应量 (Cramer's V): {chi2_result['effect_size']:.4f}")
    print(f"     * 效应量解释: {chi2_result['effect_size_interpretation']}")
    print(f"     * 显著性: {'是' if chi2_result['is_significant'] else '否'}")

    # 7. 留存率分析
    print("\n7. 留存率分析...")
    retention_results = analyzer.analyze_retention(
        data=data,
        group_col='实验组别',
        retention_col='留存状态'
    )

    print("   - 各组留存率:")
    for group, stats in retention_results['retention_analysis'].items():
        print(f"     * {group}: {stats['retention_rate']:.2%} "
              f"({stats['retained_users']}/{stats['sample_size']})")

    # 8. 随机分组验证
    print("\n8. 随机分组验证...")
    numeric_features = ['累计消费金额', '年龄']
    randomization_check = analyzer.check_randomization(
        data=data,
        group_col='实验组别',
        feature_cols=numeric_features
    )

    print("   - 分组均衡性检查:")
    for feature, result in randomization_check.items():
        print(f"     * {feature}:")
        print(f"       - 检验方法: {result['test']}")
        print(f"       - P值: {result['p_value']:.4f}")
        print(f"       - 分组均衡: {'是' if result['is_balanced'] else '否'}")

    # 9. 样本量计算
    print("\n9. 样本量计算...")
    baseline_rate = conversion_results['conversion_analysis']['对照组']['conversion_rate']
    expected_lift = 0.10  # 期望提升10%

    required_sample_size = analyzer.calculate_sample_size(
        baseline_rate=baseline_rate,
        expected_lift=expected_lift,
        alpha=0.05,
        power=0.80
    )

    print(f"   - 基准转化率: {baseline_rate:.2%}")
    print(f"   - 期望提升: {expected_lift:.0%}")
    print(f"   - 所需样本量: {required_sample_size:,} (每组)")
    print(f"   - 当前样本量: {len(data[data['实验组别'] == '对照组']):,} (对照组)")

    if len(data[data['实验组别'] == '对照组']) >= required_sample_size:
        print("   - ✅ 样本量充足")
    else:
        print("   - ⚠️ 样本量不足，建议增加样本")

    # 10. 生成可视化
    print("\n10. 生成可视化图表...")
    output_dir = Path(__file__).parent / "basic_output"
    output_dir.mkdir(exist_ok=True)

    # 转化率对比图
    fig1 = visualizer.plot_conversion_comparison(
        conversion_results['conversion_analysis'],
        title='AB测试转化率对比',
        show_confidence_interval=True,
        save_path=str(output_dir / "conversion_comparison.png")
    )
    print("   ✅ 转化率对比图已保存")

    # 留存率曲线图
    fig2 = visualizer.plot_retention_curves(
        retention_results['retention_analysis'],
        title='用户留存率对比',
        save_path=str(output_dir / "retention_curves.png")
    )
    print("   ✅ 留存率曲线图已保存")

    # 置信区间图
    fig3 = visualizer.plot_confidence_intervals(
        conversion_results['conversion_analysis'],
        title='转化率置信区间分析',
        save_path=str(output_dir / "confidence_intervals.png")
    )
    print("   ✅ 置信区间图已保存")

    # 关闭图形以避免内存泄漏
    import matplotlib.pyplot as plt
    plt.close('all')

    # 11. 保存分析结果
    print("\n11. 保存分析结果...")

    # 保存转化率分析结果
    conversion_df = pd.DataFrame([
        {
            '实验组别': group,
            '转化率': stats['conversion_rate'],
            '样本量': stats['sample_size'],
            '转化数': stats['conversions']
        }
        for group, stats in conversion_results['conversion_analysis'].items()
    ])
    conversion_df.to_csv(output_dir / "conversion_results.csv", index=False, encoding='utf-8-sig')

    # 保存统计检验结果
    stats_results = pd.DataFrame([
        {
            '检验方法': 't检验',
            '统计量': t_test_result['statistic'],
            'P值': t_test_result['p_value'],
            '效应量': t_test_result['effect_size'],
            '显著性': t_test_result['is_significant']
        },
        {
            '检验方法': '卡方检验',
            '统计量': chi2_result['statistic'],
            'P值': chi2_result['p_value'],
            '效应量': chi2_result['effect_size'],
            '显著性': chi2_result['is_significant']
        }
    ])
    stats_results.to_csv(output_dir / "statistical_tests.csv", index=False, encoding='utf-8-sig')

    print("✅ 分析结果已保存到 basic_output/ 目录")

    # 12. 总结报告
    print("\n" + "=" * 60)
    print("🎉 AB测试基础分析完成!")
    print("=" * 60)

    print(f"\n📁 生成的文件:")
    print(f"   - 转化率结果: {output_dir}/conversion_results.csv")
    print(f"   - 统计检验结果: {output_dir}/statistical_tests.csv")
    print(f"   - 转化率对比图: {output_dir}/conversion_comparison.png")
    print(f"   - 留存率曲线图: {output_dir}/retention_curves.png")
    print(f"   - 置信区间图: {output_dir}/confidence_intervals.png")

    print(f"\n🎯 关键发现:")

    # 获取测试组和对照组的转化率
    if '测试组' in conversion_results['conversion_analysis'] and '对照组' in conversion_results['conversion_analysis']:
        test_rate = conversion_results['conversion_analysis']['测试组']['conversion_rate']
        control_rate = conversion_results['conversion_analysis']['对照组']['conversion_rate']
        lift = (test_rate - control_rate) / control_rate if control_rate > 0 else 0

        print(f"   - 测试组转化率: {test_rate:.2%}")
        print(f"   - 对照组转化率: {control_rate:.2%}")
        print(f"   - 相对提升: {lift:.1%}")
        print(f"   - 统计显著性: {'显著' if t_test_result['is_significant'] else '不显著'}")

        if t_test_result['is_significant'] and lift > 0:
            recommendation = "建议推广测试组方案"
        elif t_test_result['is_significant'] and lift < 0:
            recommendation = "建议维持对照组方案"
        else:
            recommendation = "需要更多数据或重新设计实验"

        print(f"   - 建议: {recommendation}")

    print(f"\n💡 AB测试分析技能特性:")
    print(f"   ✅ 完整的转化率和留存率分析")
    print(f"   ✅ 多种统计显著性检验方法")
    print(f"   ✅ 效应量计算和解释")
    print(f"   ✅ 随机分组验证")
    print(f"   ✅ 样本量计算和功效分析")
    print(f"   ✅ 丰富的可视化图表")
    print(f"   ✅ 专业的分析报告生成")

    return True


if __name__ == "__main__":
    try:
        success = main()
        if success:
            print("\n🚀 AB测试分析技能验证成功！可以开始使用。")
        sys.exit(0 if success else 1)
    except Exception as e:
        print(f"\n❌ 示例运行失败: {str(e)}")
        import traceback
        traceback.print_exc()
        sys.exit(1)

用户ID,实验组别,转化状态,留存状态,累计消费金额,性别,年龄,价值组别,设备类型,注册时间
U001,测试组,是,TRUE,599.99,男,25,高价值,移动端,2024-01-01
U002,对照组,否,FALSE,89.99,女,32,低价值,PC端,2024-01-02
U003,测试组,是,TRUE,299.99,男,28,中价值,移动端,2024-01-03
U004,对照组,否,FALSE,199.99,女,35,中价值,PC端,2024-01-04
U005,测试组,是,TRUE,799.99,男,22,高价值,移动端,2024-01-05
U006,对照组,否,FALSE,49.99,女,29,低价值,PC端,2024-01-06
U007,测试组,否,FALSE,399.99,男,31,中价值,移动端,2024-01-07
U008,对照组,是,TRUE,699.99,女,27,高价值,PC端,2024-01-08
U009,测试组,是,TRUE,899.99,男,24,高价值,移动端,2024-01-09
U010,对照组,否,FALSE,119.99,女,38,低价值,PC端,2024-01-10
U011,测试组,否,FALSE,259.99,男,33,中价值,移动端,2024-01-11
U012,对照组,是,TRUE,549.99,女,26,高价值,PC端,2024-01-12
U013,测试组,是,TRUE,449.99,男,30,中价值,移动端,2024-01-13
U014,对照组,否,FALSE,79.99,女,36,低价值,PC端,2024-01-14
U015,测试组,是,TRUE,999.99,男,23,高价值,移动端,2024-01-15
U016,对照组,否,FALSE,159.99,女,34,低价值,PC端,2024-01-16
U017,测试组,否,FALSE,349.99,男,29,中价值,移动端,2024-01-17
U018,对照组,是,TRUE,749.99,女,25,高价值,PC端,2024-01-18
U019,测试组,是,TRUE,849.99,男,21,高价值,移动端,2024-01-19
U020,对照组,否,FALSE,139.99,女,37,低价值,PC端,2024-01-20
U021,测试组,否,FALSE,289.99,男,32,中价值,移动端,2024-01-21
U022,对照组,是,TRUE,649.99,女,28,高价值,PC端,2024-01-22
U023,测试组,是,TRUE,929.99,男,26,高价值,移动端,2024-01-23
U024,对照组,否,FALSE,99.99,女,35,低价值,PC端,2024-01-24
U025,测试组,是,TRUE,599.99,男,24,中价值,移动端,2024-01-25
U026,对照组,否,FALSE,179.99,女,31,中价值,PC端,2024-01-26
U027,测试组,否,FALSE,419.99,男,30,中价值,移动端,2024-01-27
U028,对照组,是,TRUE,799.99,女,27,高价值,PC端,2024-01-28
U029,测试组,是,TRUE,879.99,男,22,高价值,移动端,2024-01-29
U030,对照组,否,FALSE,129.99,女,39,低价值,PC端,2024-01-30
U031,测试组,否,FALSE,269.99,男,34,中价值,移动端,2024-01-31
U032,对照组,是,TRUE,699.99,女,29,高价值,PC端,2024-02-01
U033,测试组,是,TRUE,949.99,男,25,高价值,移动端,2024-02-02
U034,对照组,否,FALSE,109.99,女,36,低价值,PC端,2024-02-03
U035,测试组,是,TRUE,579.99,男,23,中价值,移动端,2024-02-04
U036,对照组,否,FALSE,189.99,女,33,中价值,PC端,2024-02-05
U037,测试组,否,FALSE,389.99,男,31,中价值,移动端,2024-02-06
U038,对照组,是,TRUE,729.99,女,28,高价值,PC端,2024-02-07
U039,测试组,是,TRUE,859.99,男,26,高价值,移动端,2024-02-08
U040,对照组,否,FALSE,149.99,女,38,低价值,PC端,2024-02-09
U041,测试组,否,FALSE,319.99,男,35,中价值,移动端,2024-02-10
U042,对照组,是,TRUE,679.99,女,30,高价值,PC端,2024-02-11
U043,测试组,是,TRUE,919.99,男,24,高价值,移动端,2024-02-12
U044,对照组,否,FALSE,119.99,女,37,低价值,PC端,2024-02-13
U045,测试组,是,TRUE,639.99,男,27,中价值,移动端,2024-02-14
U046,对照组,否,FALSE,199.99,女,32,中价值,PC端,2024-02-15
U047,测试组,否,FALSE,429.99,男,29,中价值,移动端,2024-02-16
U048,对照组,是,TRUE,759.99,女,26,高价值,PC端,2024-02-17
U049,测试组,是,TRUE,889.99,男,21,高价值,移动端,2024-02-18
U050,对照组,否,FALSE,139.99,女,36,低价值,PC端,2024-02-19

#!/usr/bin/env python3
"""
快速测试AB测试分析技能的核心功能
"""

import sys
import os
import pandas as pd
import numpy as np
from pathlib import Path

# 添加技能路径
skill_path = Path(__file__).parent
sys.path.append(str(skill_path))

def main():
    """快速测试主要功能"""
    print("🚀 AB测试分析技能快速测试")

    try:
        # 1. 测试模块导入
        print("\n1. 测试模块导入...")
        from scripts.ab_test_analyzer import ABTestAnalyzer
        from scripts.statistical_tests import StatisticalTests
        from scripts.segment_analyzer import SegmentAnalyzer
        from scripts.visualizer import ABTestVisualizer
        print("   ✓ 核心模块导入成功")

        # 2. 测试数据加载和分析
        print("\n2. 测试AB测试分析器...")
        analyzer = ABTestAnalyzer()

        data_dir = skill_path / "examples" / "sample_data"
        data_file = data_dir / "sample_ab_test_data.csv"

        if data_file.exists():
            data = analyzer.load_data(str(data_file))
            if data is not None:
                print(f"   ✓ 数据加载成功: {len(data)} 条记录")

                # 测试转化率分析
                conversion_results = analyzer.analyze_conversion(
                    data, group_col='实验组别', conversion_col='转化状态'
                )
                print(f"   ✓ 转化率分析成功: {len(conversion_results['conversion_analysis'])} 个组")

                # 测试统计检验
                from scripts.statistical_tests import StatisticalTests
                stats_tests = StatisticalTests()

                # 准备数据用于统计检验
                data_converted = data.copy()
                data_converted['转化状态_数值'] = data_converted['转化状态'].apply(
                    lambda x: 1 if str(x).strip() in ['是', 'True', 'true', '1', 'yes'] else 0
                )

                t_test_result = stats_tests.t_test(
                    data=data_converted,
                    group_col='实验组别',
                    metric_col='转化状态_数值'
                )
                print(f"   ✓ t检验成功: p值 = {t_test_result['p_value']:.4f}")

                # 测试分群分析
                segment_analyzer = SegmentAnalyzer()
                value_segments = segment_analyzer.value_based_segmentation(
                    data, value_col='累计消费金额', n_tiers=3
                )
                print(f"   ✓ 价值分群成功: {value_segments['n_segments']} 个分群")

                # 测试可视化
                visualizer = ABTestVisualizer()
                fig = visualizer.plot_conversion_comparison(
                    conversion_results['conversion_analysis'],
                    title='快速测试 - 转化率对比'
                )
                if fig is not None:
                    print("   ✓ 转化率可视化成功")
                    # 关闭图形以避免内存泄漏
                    import matplotlib.pyplot as plt
                    plt.close(fig)

        # 3. 测试样本量计算
        print("\n3. 测试样本量计算...")
        sample_size = analyzer.calculate_sample_size(
            baseline_rate=0.10,
            expected_lift=0.20,
            alpha=0.05,
            power=0.80
        )
        print(f"   ✓ 样本量计算成功: {sample_size:,} (每组)")

        # 4. 测试贝叶斯分析
        print("\n4. 测试贝叶斯AB测试...")
        bayesian_result = analyzer.bayesian_ab_test(
            data,
            group_col='实验组别',
            conversion_col='转化状态',
            n_simulations=1000  # 使用较小的模拟数量以加快测试
        )
        if bayesian_result:
            print(f"   ✓ 贝叶斯分析成功: {len(bayesian_result)} 个结果")

        print("\n🎉 核心功能测试通过！")
        print("\nAB测试分析技能已就绪，可以使用以下命令运行完整示例:")
        print("   python examples/basic_ab_test_example.py")
        print("   python examples/advanced_segmentation_example.py")
        print("   python examples/comprehensive_analysis_example.py")

        return True

    except Exception as e:
        print(f"\n❌ 测试失败: {str(e)}")
        import traceback
        traceback.print_exc()
        return False

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)

"""
AB测试分析技能核心模块

提供专业的AB测试分析功能，包括：
- AB测试设计和分析
- 统计显著性检验
- 用户分群和交互效应分析
- 可视化和报告生成
"""

__version__ = "1.0.0"
__author__ = "Claude Code Skills"

from .ab_test_analyzer import ABTestAnalyzer
from .statistical_tests import StatisticalTests
from .segment_analyzer import SegmentAnalyzer
from .visualizer import ABTestVisualizer

__all__ = [
    'ABTestAnalyzer',
    'StatisticalTests',
    'SegmentAnalyzer',
    'ABTestVisualizer'
]

"""
AB测试核心分析模块

提供完整的AB测试分析功能，包括：
- 基础转化率和留存率分析
- 样本量计算和功效分析
- 提升率计算和置信区间
- 贝叶斯AB测试方法
"""

import pandas as pd
import numpy as np
from scipy import stats
from typing import Dict, List, Tuple, Optional, Union
import warnings
warnings.filterwarnings('ignore')


class ABTestAnalyzer:
    """AB测试分析器主类"""

    def __init__(self):
        self.data = None
        self.significance_level = 0.05
        self.statistical_power = 0.8
        self.multiple_comparison_correction = None

    def load_data(self, file_path: str, **kwargs) -> pd.DataFrame:
        """加载AB测试数据"""
        try:
            self.data = pd.read_csv(file_path, **kwargs)
            print(f"数据加载成功: {len(self.data)} 条记录")
            return self.data
        except Exception as e:
            print(f"数据加载失败: {str(e)}")
            return None

    def set_significance_level(self, alpha: float = 0.05):
        """设置显著性水平"""
        if not 0 < alpha < 1:
            raise ValueError("显著性水平必须在0和1之间")
        self.significance_level = alpha
        print(f"显著性水平设置为: {alpha}")

    def set_statistical_power(self, power: float = 0.8):
        """设置统计功效"""
        if not 0 < power < 1:
            raise ValueError("统计功效必须在0和1之间")
        self.statistical_power = power
        print(f"统计功效设置为: {power}")

    def set_multiple_comparison_correction(self, method: str = None):
        """设置多重比较校正方法"""
        valid_methods = ['bonferroni', 'fdr_bh', 'holm', 'sidak']
        if method and method not in valid_methods:
            raise ValueError(f"校正方法必须是: {valid_methods}")
        self.multiple_comparison_correction = method
        print(f"多重比较校正方法设置为: {method}")

    def calculate_conversion_rates(self,
                                 data: pd.DataFrame,
                                 group_col: str,
                                 conversion_col: str,
                                 value_mapping: Optional[Dict] = None) -> Dict:
        """计算各组转化率"""
        df = data.copy()

        # 处理转化状态映射
        if value_mapping:
            df[conversion_col] = df[conversion_col].map(value_mapping)

        # 确保转化列是数值类型
        if df[conversion_col].dtype == 'object':
            df[conversion_col] = df[conversion_col].apply(lambda x: 1 if str(x).strip() in ['是', 'True', 'true', '1', 'yes'] else 0)

        # 计算转化率
        results = {}
        for group in df[group_col].unique():
            group_data = df[df[group_col] == group]
            conversion_rate = group_data[conversion_col].mean()
            sample_size = len(group_data)
            conversions = group_data[conversion_col].sum()

            # 计算置信区间
            se = np.sqrt(conversion_rate * (1 - conversion_rate) / sample_size)
            ci_lower = max(0, conversion_rate - 1.96 * se)
            ci_upper = min(1, conversion_rate + 1.96 * se)

            results[group] = {
                'conversion_rate': conversion_rate,
                'sample_size': sample_size,
                'conversions': int(conversions),
                'standard_error': se,
                'confidence_interval': (ci_lower, ci_upper)
            }

        return results

    def calculate_lift(self,
                      conversion_results: Dict,
                      control_group: str,
                      test_group: str) -> Dict:
        """计算提升率和相关指标"""
        if control_group not in conversion_results or test_group not in conversion_results:
            raise ValueError("指定的组别在结果中不存在")

        control_rate = conversion_results[control_group]['conversion_rate']
        test_rate = conversion_results[test_group]['conversion_rate']

        # 绝对提升
        absolute_lift = test_rate - control_rate

        # 相对提升
        relative_lift = absolute_lift / control_rate if control_rate > 0 else 0

        # 提升率的置信区间 (使用Delta方法)
        control_se = conversion_results[control_group]['standard_error']
        test_se = conversion_results[test_group]['standard_error']

        lift_se = np.sqrt(test_se**2 + control_se**2)
        lift_ci_lower = absolute_lift - 1.96 * lift_se
        lift_ci_upper = absolute_lift + 1.96 * lift_se

        return {
            'absolute_lift': absolute_lift,
            'relative_lift': relative_lift,
            'lift_percentage': relative_lift * 100,
            'lift_standard_error': lift_se,
            'lift_confidence_interval': (lift_ci_lower, lift_ci_upper)
        }

    def calculate_retention_rates(self,
                                data: pd.DataFrame,
                                group_col: str,
                                retention_col: str) -> Dict:
        """计算留存率"""
        df = data.copy()

        # 确保留存列是布尔类型
        if df[retention_col].dtype == 'object':
            df[retention_col] = df[retention_col].apply(lambda x: True if str(x).strip() in ['TRUE', 'true', '1', '是'] else False)

        results = {}
        for group in df[group_col].unique():
            group_data = df[df[group_col] == group]
            retention_rate = group_data[retention_col].mean()
            sample_size = len(group_data)
            retained_users = group_data[retention_col].sum()

            # 计算置信区间
            se = np.sqrt(retention_rate * (1 - retention_rate) / sample_size)
            ci_lower = max(0, retention_rate - 1.96 * se)
            ci_upper = min(1, retention_rate + 1.96 * se)

            results[group] = {
                'retention_rate': retention_rate,
                'sample_size': sample_size,
                'retained_users': int(retained_users),
                'standard_error': se,
                'confidence_interval': (ci_lower, ci_upper)
            }

        return results

    def calculate_sample_size(self,
                            baseline_rate: float,
                            expected_lift: float,
                            alpha: float = None,
                            power: float = None,
                            two_tailed: bool = True) -> int:
        """计算AB测试所需样本量"""
        alpha = alpha or self.significance_level
        power = power or self.statistical_power

        # 获取Z值
        if two_tailed:
            z_alpha = stats.norm.ppf(1 - alpha/2)
        else:
            z_alpha = stats.norm.ppf(1 - alpha)

        z_beta = stats.norm.ppf(power)

        # 计算效应量
        p1 = baseline_rate
        p2 = p1 * (1 + expected_lift)

        # 使用Cohen's h计算样本量
        h = 2 * np.arcsin(np.sqrt(p2)) - 2 * np.arcsin(np.sqrt(p1))

        if h == 0:
            raise ValueError("效应量为0，无法计算样本量")

        # 计算每组所需样本量
        n_per_group = 2 * (z_alpha + z_beta)**2 / (h**2)

        return int(np.ceil(n_per_group))

    def check_randomization(self,
                           data: pd.DataFrame,
                           group_col: str,
                           feature_cols: List[str]) -> Dict:
        """检查随机分组是否均衡"""
        df = data.copy()
        results = {}

        for feature in feature_cols:
            if df[feature].dtype == 'object':
                # 分类变量：使用卡方检验
                contingency_table = pd.crosstab(df[group_col], df[feature])
                chi2, p_value, dof, expected = stats.chi2_contingency(contingency_table)
                results[feature] = {
                    'test': 'Chi-square Test',
                    'statistic': chi2,
                    'p_value': p_value,
                    'degrees_of_freedom': dof,
                    'is_balanced': p_value > self.significance_level
                }
            else:
                # 数值变量：使用t检验或方差分析
                groups = [df[df[group_col] == group][feature].dropna() for group in df[group_col].unique()]
                if len(groups) == 2:
                    statistic, p_value = stats.ttest_ind(*groups)
                    test_name = 'Independent t-test'
                else:
                    statistic, p_value = stats.f_oneway(*groups)
                    test_name = 'ANOVA'

                results[feature] = {
                    'test': test_name,
                    'statistic': statistic,
                    'p_value': p_value,
                    'is_balanced': p_value > self.significance_level
                }

        return results

    def data_quality_check(self,
                          data: pd.DataFrame,
                          required_cols: List[str]) -> Dict:
        """数据质量检查"""
        df = data.copy()
        results = {
            'total_rows': len(df),
            'total_columns': len(df.columns),
            'missing_values': {},
            'data_types': {},
            'duplicate_rows': df.duplicated().sum(),
            'required_columns_check': {}
        }

        # 检查缺失值
        for col in df.columns:
            missing_count = df[col].isnull().sum()
            if missing_count > 0:
                results['missing_values'][col] = {
                    'count': missing_count,
                    'percentage': missing_count / len(df) * 100
                }

        # 检查数据类型
        for col in df.columns:
            results['data_types'][col] = str(df[col].dtype)

        # 检查必需列
        for col in required_cols:
            results['required_columns_check'][col] = col in df.columns

        return results

    def analyze_conversion(self,
                          data: pd.DataFrame,
                          group_col: str,
                          conversion_col: str,
                          control_group: Optional[str] = None) -> Dict:
        """综合转化率分析"""
        # 计算转化率
        conversion_results = self.calculate_conversion_rates(data, group_col, conversion_col)

        # 如果指定了对照组，计算提升率
        lift_results = {}
        if control_group and len(conversion_results) > 1:
            for group in conversion_results:
                if group != control_group:
                    lift_results[group] = self.calculate_lift(
                        conversion_results, control_group, group
                    )

        return {
            'conversion_analysis': conversion_results,
            'lift_analysis': lift_results,
            'summary': {
                'total_users': len(data),
                'groups': list(conversion_results.keys()),
                'overall_conversion_rate': data[conversion_col].mean() if data[conversion_col].dtype != 'object'
                else data[conversion_col].apply(lambda x: 1 if str(x).strip() in ['是', 'True', 'true', '1', 'yes'] else 0).mean()
            }
        }

    def analyze_retention(self,
                         data: pd.DataFrame,
                         group_col: str,
                         retention_col: str) -> Dict:
        """综合留存率分析"""
        retention_results = self.calculate_retention_rates(data, group_col, retention_col)

        return {
            'retention_analysis': retention_results,
            'summary': {
                'total_users': len(data),
                'groups': list(retention_results.keys()),
                'overall_retention_rate': data[retention_col].mean() if data[retention_col].dtype != 'object'
                else data[retention_col].apply(lambda x: 1 if str(x).strip() in ['TRUE', 'true', '1', '是'] else 0).mean()
            }
        }

    def bayesian_ab_test(self,
                        data: pd.DataFrame,
                        group_col: str,
                        conversion_col: str,
                        prior_type: str = 'jeffreys',
                        n_simulations: int = 10000) -> Dict:
        """贝叶斯AB测试分析"""
        df = data.copy()

        # 确保转化列是数值类型
        if df[conversion_col].dtype == 'object':
            df[conversion_col] = df[conversion_col].apply(lambda x: 1 if str(x).strip() in ['是', 'True', 'true', '1', 'yes'] else 0)

        results = {}

        for group in df[group_col].unique():
            group_data = df[df[group_col] == group]
            successes = int(group_data[conversion_col].sum())
            trials = len(group_data)

            # 设置先验分布
            if prior_type == 'jeffreys':
                alpha_prior = 0.5
                beta_prior = 0.5
            elif prior_type == 'uniform':
                alpha_prior = 1
                beta_prior = 1
            else:  # 使用经验贝叶斯
                alpha_prior = successes + 1
                beta_prior = trials - successes + 1

            # 后验分布参数
            alpha_post = alpha_prior + successes
            beta_post = beta_prior + trials - successes

            # 从后验分布采样
            samples = np.random.beta(alpha_post, beta_post, n_simulations)

            results[group] = {
                'successes': successes,
                'trials': trials,
                'posterior_alpha': alpha_post,
                'posterior_beta': beta_post,
                'posterior_mean': alpha_post / (alpha_post + beta_post),
                'posterior_samples': samples,
                'credible_interval': (
                    np.percentile(samples, 2.5),
                    np.percentile(samples, 97.5)
                )
            }

        # 计算组间比较
        if len(results) == 2:
            groups = list(results.keys())
            group1_samples = results[groups[0]]['posterior_samples']
            group2_samples = results[groups[1]]['posterior_samples']

            # 计算概率
            prob_group1_better = np.mean(group1_samples > group2_samples)
            prob_group2_better = 1 - prob_group1_better

            # 计算差异分布
            diff_samples = group1_samples - group2_samples

            results['comparison'] = {
                'prob_' + groups[0] + '_better': prob_group1_better,
                'prob_' + groups[1] + '_better': prob_group2_better,
                'difference_distribution': diff_samples,
                'difference_mean': np.mean(diff_samples),
                'difference_ci': (
                    np.percentile(diff_samples, 2.5),
                    np.percentile(diff_samples, 97.5)
                )
            }

        return results

    def calculate_win_probability(self, bayesian_results: Dict) -> Dict:
        """计算贝叶斯AB测试的获胜概率"""
        if 'comparison' not in bayesian_results:
            return {"error": "需要先运行贝叶斯AB测试"}

        comparison = bayesian_results['comparison']

        # 找出获胜概率最大的组
        win_probabilities = {}
        for key, value in comparison.items():
            if key.startswith('prob_') and key.endswith('_better'):
                group_name = key.replace('prob_', '').replace('_better', '')
                win_probabilities[group_name] = value

        # 找出获胜组
        winner = max(win_probabilities, key=win_probabilities.get)

        return {
            'win_probabilities': win_probabilities,
            'winner': winner,
            'winner_probability': win_probabilities[winner]
        }

"""
用户分群分析模块

提供用户分群和交互效应分析功能，包括：
- 基于价值的用户分群
- RFM分群分析
- 行为模式分群
- 交互效应分析
- 自定义分群策略
"""

import pandas as pd
import numpy as np
from scipy import stats
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import silhouette_score
from typing import Dict, List, Tuple, Optional, Union
import warnings
warnings.filterwarnings('ignore')


class SegmentAnalyzer:
    """用户分群分析器"""

    def __init__(self):
        self.scaler = StandardScaler()
        self.segment_labels = {}
        self.clustering_model = None

    def value_based_segmentation(self,
                                data: pd.DataFrame,
                                value_col: str,
                                n_tiers: int = 3,
                                method: str = 'quantile',
                                custom_boundaries: Optional[List[float]] = None) -> Dict:
        """
        基于价值的用户分群

        Parameters:
        - n_tiers: 分层数量
        - method: 'quantile', 'equal_width', 'custom'
        - custom_boundaries: 自定义分界值列表
        """
        df = data.copy()

        # 确保价值列是数值类型
        if df[value_col].dtype == 'object':
            df[value_col] = pd.to_numeric(df[value_col], errors='coerce')

        if method == 'custom' and custom_boundaries:
            # 使用自定义边界
            boundaries = [float('-inf')] + custom_boundaries + [float('inf')]
            labels = [f'Tier_{i+1}' for i in range(len(boundaries)-1)]
            df['value_segment'] = pd.cut(df[value_col], bins=boundaries, labels=labels)

        elif method == 'quantile':
            # 使用分位数
            quantiles = np.linspace(0, 1, n_tiers + 1)
            boundaries = df[value_col].quantile(quantiles).tolist()
            labels = [f'Value_Tier_{i+1}' for i in range(n_tiers)]
            df['value_segment'] = pd.qcut(df[value_col], q=n_tiers, labels=labels, duplicates='drop')

        elif method == 'equal_width':
            # 等宽分箱
            labels = [f'Value_Tier_{i+1}' for i in range(n_tiers)]
            df['value_segment'] = pd.cut(df[value_col], bins=n_tiers, labels=labels)

        else:
            raise ValueError("method必须是'quantile', 'equal_width'或'custom'")

        # 计算分层统计
        segment_stats = {}
        for segment in df['value_segment'].cat.categories:
            segment_data = df[df['value_segment'] == segment]
            segment_stats[segment] = {
                'count': len(segment_data),
                'percentage': len(segment_data) / len(df) * 100,
                'mean_value': segment_data[value_col].mean(),
                'median_value': segment_data[value_col].median(),
                'std_value': segment_data[value_col].std(),
                'min_value': segment_data[value_col].min(),
                'max_value': segment_data[value_col].max()
            }

        return {
            'segmented_data': df,
            'segment_statistics': segment_stats,
            'segmentation_method': method,
            'n_segments': len(df['value_segment'].cat.categories),
            'value_column': value_col
        }

    def rfm_segmentation(self,
                        data: pd.DataFrame,
                        customer_id_col: str,
                        recency_col: str,
                        frequency_col: str,
                        monetary_col: str,
                        n_clusters: int = 5) -> Dict:
        """
        RFM分群分析

        Parameters:
        - recency_col: 最近消费时间距今天数 (越小越好)
        - frequency_col: 消费频率
        - monetary_col: 消费金额
        """
        df = data.copy()

        # 确保数值列是数值类型
        for col in [recency_col, frequency_col, monetary_col]:
            if df[col].dtype == 'object':
                df[col] = pd.to_numeric(df[col], errors='coerce')

        # 计算RFM分数
        rf_data = df.groupby(customer_id_col)[[recency_col, frequency_col, monetary_col]].agg({
            recency_col: 'min',  # 最近消费时间
            frequency_col: 'sum',  # 总消费次数
            monetary_col: 'sum'   # 总消费金额
        }).reset_index()

        # 重命名列
        rf_data.columns = [customer_id_col, 'Recency', 'Frequency', 'Monetary']

        # RFM评分 (1-5分制，分数越高越好)
        rf_data['R_Score'] = pd.qcut(rf_data['Recency'].rank(method='first'), 5, labels=[5,4,3,2,1])
        rf_data['F_Score'] = pd.qcut(rf_data['Frequency'].rank(method='first'), 5, labels=[1,2,3,4,5])
        rf_data['M_Score'] = pd.qcut(rf_data['Monetary'].rank(method='first'), 5, labels=[1,2,3,4,5])

        # 计算RFM总分
        rf_data['RFM_Score'] = rf_data['R_Score'].astype(str) + rf_data['F_Score'].astype(str) + rf_data['M_Score'].astype(str)
        rf_data['RFM_Total'] = rf_data['R_Score'].astype(int) + rf_data['F_Score'].astype(int) + rf_data['M_Score'].astype(int)

        # 聚类分析
        features = rf_data[['Recency', 'Frequency', 'Monetary']]
        features_scaled = self.scaler.fit_transform(features)

        # K-means聚类
        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
        rf_data['Cluster'] = kmeans.fit_predict(features_scaled)

        # 计算轮廓系数
        silhouette_avg = silhouette_score(features_scaled, kmeans.labels_)

        # 聚类统计
        cluster_stats = {}
        for cluster in range(n_clusters):
            cluster_data = rf_data[rf_data['Cluster'] == cluster]
            cluster_stats[f'Cluster_{cluster}'] = {
                'count': len(cluster_data),
                'percentage': len(cluster_data) / len(rf_data) * 100,
                'avg_recency': cluster_data['Recency'].mean(),
                'avg_frequency': cluster_data['Frequency'].mean(),
                'avg_monetary': cluster_data['Monetary'].mean(),
                'avg_rfm_total': cluster_data['RFM_Total'].mean()
            }

        # RFM客户类型定义
        def classify_rfm(row):
            if row['R_Score'] >= 4 and row['F_Score'] >= 4 and row['M_Score'] >= 4:
                return 'Champions'
            elif row['R_Score'] >= 3 and row['F_Score'] >= 3 and row['M_Score'] >= 3:
                return 'Loyal Customers'
            elif row['R_Score'] >= 3 and row['F_Score'] >= 2:
                return 'Potential Loyalists'
            elif row['R_Score'] >= 4 and row['F_Score'] <= 2:
                return 'New Customers'
            elif row['R_Score'] <= 2 and row['F_Score'] >= 3:
                return 'At Risk'
            elif row['R_Score'] <= 2 and row['F_Score'] <= 2 and row['M_Score'] >= 3:
                return 'Cannot Lose Them'
            else:
                return 'Lost'

        rf_data['RFM_Segment'] = rf_data.apply(classify_rfm, axis=1)

        # 保存模型
        self.clustering_model = kmeans

        return {
            'rfm_data': rf_data,
            'cluster_statistics': cluster_stats,
            'silhouette_score': silhouette_avg,
            'n_clusters': n_clusters,
            'rfm_segments': rf_data['RFM_Segment'].value_counts().to_dict()
        }

    def behavioral_segmentation(self,
                               data: pd.DataFrame,
                               customer_id_col: str,
                               behavior_cols: List[str],
                               n_clusters: int = 4) -> Dict:
        """基于行为的用户分群"""
        df = data.copy()

        # 聚合用户行为数据
        behavior_data = df.groupby(customer_id_col)[behavior_cols].agg(['mean', 'sum', 'count']).reset_index()

        # 扁平化列名
        behavior_data.columns = [f'{col[0]}_{col[1]}' if col[1] else col[0] for col in behavior_data.columns]

        # 确保数值列是数值类型
        numeric_cols = behavior_data.select_dtypes(include=[np.number]).columns.tolist()
        for col in numeric_cols:
            if behavior_data[col].dtype == 'object':
                behavior_data[col] = pd.to_numeric(behavior_data[col], errors='coerce')

        # 填充缺失值
        behavior_data[numeric_cols] = behavior_data[numeric_cols].fillna(0)

        # 标准化
        features = behavior_data[numeric_cols]
        features_scaled = self.scaler.fit_transform(features)

        # K-means聚类
        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
        behavior_data['Behavior_Cluster'] = kmeans.fit_predict(features_scaled)

        # 计算轮廓系数
        silhouette_avg = silhouette_score(features_scaled, kmeans.labels_)

        # 聚类统计
        cluster_stats = {}
        for cluster in range(n_clusters):
            cluster_data = behavior_data[behavior_data['Behavior_Cluster'] == cluster]
            cluster_stats[f'Behavior_Cluster_{cluster}'] = {
                'count': len(cluster_data),
                'percentage': len(cluster_data) / len(behavior_data) * 100
            }

            # 为每个行为特征添加统计
            for col in numeric_cols:
                if col in cluster_data.columns:
                    cluster_stats[f'Behavior_Cluster_{cluster}'][f'avg_{col}'] = cluster_data[col].mean()

        # 保存模型
        self.clustering_model = kmeans

        return {
            'behavior_segmented_data': behavior_data,
            'cluster_statistics': cluster_stats,
            'silhouette_score': silhouette_avg,
            'n_clusters': n_clusters,
            'behavior_features': behavior_cols
        }

    def segment_conversion_analysis(self,
                                  data: pd.DataFrame,
                                  segment_col: str,
                                  group_col: str,
                                  conversion_col: str) -> Dict:
        """分群转化率分析"""
        df = data.copy()

        # 确保转化列是数值类型
        if df[conversion_col].dtype == 'object':
            df[conversion_col] = df[conversion_col].apply(lambda x: 1 if str(x).strip() in ['是', 'True', 'true', '1', 'yes'] else 0)

        # 计算每个分群的转化率
        results = {}
        for segment in df[segment_col].unique():
            segment_data = df[df[segment_col] == segment]
            segment_results = {}

            for group in segment_data[group_col].unique():
                group_data = segment_data[segment_data[group_col] == group]
                conversion_rate = group_data[conversion_col].mean()
                sample_size = len(group_data)
                conversions = group_data[conversion_col].sum()

                # 计算置信区间
                se = np.sqrt(conversion_rate * (1 - conversion_rate) / sample_size)
                ci_lower = max(0, conversion_rate - 1.96 * se)
                ci_upper = min(1, conversion_rate + 1.96 * se)

                segment_results[group] = {
                    'conversion_rate': conversion_rate,
                    'sample_size': sample_size,
                    'conversions': int(conversions),
                    'standard_error': se,
                    'confidence_interval': (ci_lower, ci_upper)
                }

            # 计算分群内提升率
            lift_results = {}
            groups = list(segment_results.keys())
            if len(groups) == 2:
                control_rate = segment_results[groups[0]]['conversion_rate']
                test_rate = segment_results[groups[1]]['conversion_rate']
                absolute_lift = test_rate - control_rate
                relative_lift = absolute_lift / control_rate if control_rate > 0 else 0

                lift_results = {
                    'absolute_lift': absolute_lift,
                    'relative_lift': relative_lift,
                    'lift_percentage': relative_lift * 100
                }

            results[segment] = {
                'group_results': segment_results,
                'lift_analysis': lift_results
            }

        # 总体统计
        overall_stats = {}
        for group in df[group_col].unique():
            group_data = df[df[group_col] == group]
            overall_stats[group] = {
                'conversion_rate': group_data[conversion_col].mean(),
                'sample_size': len(group_data),
                'conversions': int(group_data[conversion_col].sum())
            }

        return {
            'segment_conversion_analysis': results,
            'overall_statistics': overall_stats,
            'segment_column': segment_col,
            'group_column': group_col,
            'conversion_column': conversion_col
        }

    def interaction_analysis(self,
                            data: pd.DataFrame,
                            group_col: str,
                            segment_col: str,
                            outcome_col: str,
                            method: str = 'logistic') -> Dict:
        """交互效应分析"""
        df = data.copy()

        # 确保结果列是数值类型
        if df[outcome_col].dtype == 'object':
            df[outcome_col] = df[outcome_col].apply(lambda x: 1 if str(x).strip() in ['是', 'True', 'true', '1', 'yes'] else 0)

        results = {}

        # 分析每个分群内的组间差异
        for segment in df[segment_col].unique():
            segment_data = df[df[segment_col] == segment]

            # 检查组别数量
            groups = segment_data[group_col].unique()
            if len(groups) == 2:
                # 两组比较
                group1_data = segment_data[segment_data[group_col] == groups[0]][outcome_col]
                group2_data = segment_data[segment_data[group_col] == groups[1]][outcome_col]

                # t检验
                t_stat, t_p = stats.ttest_ind(group1_data, group2_data)

                # 卡方检验
                contingency = pd.crosstab(segment_data[group_col], segment_data[outcome_col])
                chi2_stat, chi2_p, _, _ = stats.chi2_contingency(contingency)

                results[segment] = {
                    'group1': groups[0],
                    'group2': groups[1],
                    'group1_conversion': group1_data.mean(),
                    'group2_conversion': group2_data.mean(),
                    'absolute_difference': group2_data.mean() - group1_data.mean(),
                    'relative_difference': (group2_data.mean() - group1_data.mean()) / group1_data.mean() if group1_data.mean() > 0 else 0,
                    't_test_statistic': t_stat,
                    't_test_p_value': t_p,
                    'chi2_statistic': chi2_stat,
                    'chi2_p_value': chi2_p,
                    'is_significant_t': t_p < 0.05,
                    'is_significant_chi2': chi2_p < 0.05
                }
            else:
                # 多组比较 (ANOVA)
                group_outcomes = [segment_data[segment_data[group_col] == group][outcome_col] for group in groups]
                f_stat, f_p = stats.f_oneway(*group_outcomes)

                # 各组转化率
                conversion_rates = {}
                for group in groups:
                    group_data = segment_data[segment_data[group_col] == group]
                    conversion_rates[group] = group_data[outcome_col].mean()

                results[segment] = {
                    'groups': list(groups),
                    'conversion_rates': conversion_rates,
                    'f_statistic': f_stat,
                    'f_p_value': f_p,
                    'is_significant': f_p < 0.05
                }

        # 整体交互效应检验
        # 创建交互项并进行方差分析
        if method == 'logistic':
            from sklearn.linear_model import LogisticRegression
            from sklearn.preprocessing import LabelEncoder

            # 编码分类变量
            le_group = LabelEncoder()
            le_segment = LabelEncoder()

            df_encoded = df.copy()
            df_encoded[f'{group_col}_encoded'] = le_group.fit_transform(df[group_col])
            df_encoded[f'{segment_col}_encoded'] = le_segment.fit_transform(df[segment_col])

            # 创建交互项
            df_encoded['interaction'] = df_encoded[f'{group_col}_encoded'] * df_encoded[f'{segment_col}_encoded']

            # 逻辑回归
            X = df_encoded[[f'{group_col}_encoded', f'{segment_col}_encoded', 'interaction']]
            y = df_encoded[outcome_col]

            logreg = LogisticRegression()
            logreg.fit(X, y)

            # 计算交互效应的统计显著性
            interaction_coef = logreg.coef_[0][2]
            interaction_se = np.sqrt(np.diag(np.linalg.inv(X.T @ X)))[2] * np.std(y) / np.sqrt(len(y))
            interaction_z = interaction_coef / interaction_se
            interaction_p = 2 * (1 - stats.norm.cdf(abs(interaction_z)))

            results['overall_interaction'] = {
                'interaction_coefficient': interaction_coef,
                'interaction_z_score': interaction_z,
                'interaction_p_value': interaction_p,
                'is_significant_interaction': interaction_p < 0.05
            }

        return {
            'segment_interactions': results,
            'group_column': group_col,
            'segment_column': segment_col,
            'outcome_column': outcome_col,
            'analysis_method': method
        }

    def apply_custom_segments(self,
                            data: pd.DataFrame,
                            segment_rules: Dict[str, Dict[str, Tuple[float, float]]]) -> pd.DataFrame:
        """
        应用自定义分群规则

        Parameters:
        - segment_rules: 分群规则字典
          {
              'segment_name': {'column_name': (min_value, max_value)}
          }
        """
        df = data.copy()
        df['custom_segment'] = 'Uncategorized'

        for segment_name, rules in segment_rules.items():
            mask = pd.Series(True, index=df.index)

            for column, (min_val, max_val) in rules.items():
                if column in df.columns:
                    # 确保数值类型
                    if df[column].dtype == 'object':
                        df[column] = pd.to_numeric(df[column], errors='coerce')

                    if min_val is not None and max_val is not None:
                        mask &= (df[column] >= min_val) & (df[column] < max_val)
                    elif min_val is not None:
                        mask &= df[column] >= min_val
                    elif max_val is not None:
                        mask &= df[column] < max_val

            df.loc[mask, 'custom_segment'] = segment_name

        return df

    def calculate_segment_insights(self,
                                  segmented_data: pd.DataFrame,
                                  segment_col: str,
                                  metrics_cols: List[str]) -> Dict:
        """计算分群洞察"""
        df = segmented_data.copy()

        # 确保数值列是数值类型
        for col in metrics_cols:
            if col in df.columns and df[col].dtype == 'object':
                df[col] = pd.to_numeric(df[col], errors='coerce')

        insights = {}

        for segment in df[segment_col].unique():
            segment_data = df[df[segment_col] == segment]
            insights[segment] = {
                'count': len(segment_data),
                'percentage': len(segment_data) / len(df) * 100
            }

            # 为每个指标计算统计
            for metric in metrics_cols:
                if metric in segment_data.columns:
                    insights[segment][f'{metric}_mean'] = segment_data[metric].mean()
                    insights[segment][f'{metric}_median'] = segment_data[metric].median()
                    insights[segment][f'{metric}_std'] = segment_data[metric].std()
                    insights[segment][f'{metric}_min'] = segment_data[metric].min()
                    insights[segment][f'{metric}_max'] = segment_data[metric].max()

        return {
            'segment_insights': insights,
            'segment_column': segment_col,
            'metrics_analyzed': metrics_cols,
            'total_segments': len(df[segment_col].unique())
        }

"""
统计检验模块

提供全面的统计显著性检验功能，包括：
- t检验 (独立样本、配对样本)
- 卡方检验 (拟合优度、独立性)
- 方差分析 (ANOVA)
- 非参数检验
- 效应量计算
"""

import pandas as pd
import numpy as np
from scipy import stats
from typing import Dict, List, Tuple, Optional, Union
import warnings
warnings.filterwarnings('ignore')


class StatisticalTests:
    """统计检验类"""

    def __init__(self, alpha: float = 0.05):
        self.alpha = alpha
        self.correction_method = None

    def set_alpha(self, alpha: float):
        """设置显著性水平"""
        if not 0 < alpha < 1:
            raise ValueError("显著性水平必须在0和1之间")
        self.alpha = alpha

    def set_correction_method(self, method: str):
        """设置多重比较校正方法"""
        valid_methods = ['bonferroni', 'fdr_bh', 'holm', 'sidak']
        if method not in valid_methods:
            raise ValueError(f"校正方法必须是: {valid_methods}")
        self.correction_method = method

    def t_test(self,
               data: pd.DataFrame,
               group_col: str,
               metric_col: str,
               test_type: str = 'independent',
               equal_var: bool = True,
               alternative: str = 'two-sided') -> Dict:
        """
        t检验

        Parameters:
        - test_type: 'independent' 独立样本t检验, 'paired' 配对样本t检验
        - equal_var: 是否假设方差相等 (仅用于独立样本t检验)
        - alternative: 'two-sided', 'less', 'greater'
        """
        df = data.copy()

        # 确保数值列是数值类型
        if df[metric_col].dtype == 'object':
            df[metric_col] = pd.to_numeric(df[metric_col], errors='coerce')

        if test_type == 'independent':
            groups = df[group_col].unique()
            if len(groups) != 2:
                raise ValueError("独立样本t检验需要恰好2个组")

            group1_data = df[df[group_col] == groups[0]][metric_col].dropna()
            group2_data = df[df[group_col] == groups[1]][metric_col].dropna()

            # 执行t检验
            statistic, p_value = stats.ttest_ind(
                group1_data, group2_data,
                equal_var=equal_var,
                alternative=alternative
            )

            # 计算效应量 (Cohen's d)
            pooled_std = np.sqrt(((len(group1_data) - 1) * group1_data.var() +
                                 (len(group2_data) - 1) * group2_data.var()) /
                                (len(group1_data) + len(group2_data) - 2))
            cohens_d = (group1_data.mean() - group2_data.mean()) / pooled_std

            # 计算置信区间
            se = pooled_std * np.sqrt(1/len(group1_data) + 1/len(group2_data))
            mean_diff = group1_data.mean() - group2_data.mean()
            ci_lower = mean_diff - 1.96 * se
            ci_upper = mean_diff + 1.96 * se

            result = {
                'test_type': 'Independent t-test',
                'statistic': statistic,
                'p_value': p_value,
                'degrees_of_freedom': len(group1_data) + len(group2_data) - 2,
                'effect_size': cohens_d,
                'effect_size_interpretation': self._interpret_cohens_d(cohens_d),
                'mean_difference': mean_diff,
                'confidence_interval': (ci_lower, ci_upper),
                'group1_stats': {
                    'name': groups[0],
                    'n': len(group1_data),
                    'mean': group1_data.mean(),
                    'std': group1_data.std()
                },
                'group2_stats': {
                    'name': groups[1],
                    'n': len(group2_data),
                    'mean': group2_data.mean(),
                    'std': group2_data.std()
                },
                'is_significant': p_value < self.alpha
            }

        elif test_type == 'paired':
            if len(df[group_col].unique()) != 2:
                raise ValueError("配对样本t检验需要恰好2个组")

            # 确保配对数据
            group1_data = df[df[group_col] == df[group_col].unique()[0]][metric_col].dropna()
            group2_data = df[df[group_col] == df[group_col].unique()[1]][metric_col].dropna()

            if len(group1_data) != len(group2_data):
                raise ValueError("配对样本t检验需要两组数据长度相同")

            # 执行配对t检验
            statistic, p_value = stats.ttest_rel(group1_data, group2_data, alternative=alternative)

            # 计算效应量
            differences = group1_data - group2_data
            cohens_d = differences.mean() / differences.std()

            # 计算置信区间
            se = differences.std() / np.sqrt(len(differences))
            mean_diff = differences.mean()
            ci_lower = mean_diff - 1.96 * se
            ci_upper = mean_diff + 1.96 * se

            result = {
                'test_type': 'Paired t-test',
                'statistic': statistic,
                'p_value': p_value,
                'degrees_of_freedom': len(differences) - 1,
                'effect_size': cohens_d,
                'effect_size_interpretation': self._interpret_cohens_d(cohens_d),
                'mean_difference': mean_diff,
                'confidence_interval': (ci_lower, ci_upper),
                'paired_differences_stats': {
                    'n': len(differences),
                    'mean': differences.mean(),
                    'std': differences.std()
                },
                'is_significant': p_value < self.alpha
            }

        else:
            raise ValueError("test_type必须是'independent'或'paired'")

        return result

    def chi_square_test(self,
                        data: pd.DataFrame,
                        group_col: str = None,
                        outcome_col: str = None,
                        contingency_table: pd.DataFrame = None,
                        test_type: str = 'independence') -> Dict:
        """
        卡方检验

        Parameters:
        - test_type: 'independence' 独立性检验, 'goodness_of_fit' 拟合优度检验
        """
        if test_type == 'independence':
            if not group_col or not outcome_col:
                raise ValueError("独立性检验需要group_col和outcome_col参数")

            # 创建列联表
            observed = pd.crosstab(data[group_col], data[outcome_col])

        elif test_type == 'goodness_of_fit':
            if contingency_table is None:
                raise ValueError("拟合优度检验需要contingency_table参数")
            observed = contingency_table

        else:
            raise ValueError("test_type必须是'independence'或'goodness_of_fit'")

        # 执行卡方检验
        chi2, p_value, dof, expected = stats.chi2_contingency(observed)

        # 计算效应量 (Cramer's V)
        n = observed.sum().sum()
        cramers_v = np.sqrt(chi2 / (n * (min(observed.shape) - 1)))

        # 计算标准化残差
        std_residuals = (observed - expected) / np.sqrt(expected)

        result = {
            'test_type': f'Chi-square {test_type.replace("_", " ").title()} Test',
            'statistic': chi2,
            'p_value': p_value,
            'degrees_of_freedom': dof,
            'effect_size': cramers_v,
            'effect_size_interpretation': self._interpret_cramers_v(cramers_v),
            'observed_frequencies': observed.to_dict(),
            'expected_frequencies': pd.DataFrame(expected,
                                               index=observed.index,
                                               columns=observed.columns).to_dict(),
            'standardized_residuals': std_residuals.to_dict(),
            'is_significant': p_value < self.alpha
        }

        return result

    def one_way_anova(self,
                      data: pd.DataFrame,
                      group_col: str,
                      metric_col: str) -> Dict:
        """单因素方差分析"""
        df = data.copy()

        # 确保数值列是数值类型
        if df[metric_col].dtype == 'object':
            df[metric_col] = pd.to_numeric(df[metric_col], errors='coerce')

        # 获取各组数据
        groups = [df[df[group_col] == group][metric_col].dropna()
                 for group in df[group_col].unique()]

        # 执行ANOVA
        statistic, p_value = stats.f_oneway(*groups)

        # 计算效应量 (Eta-squared)
        total_var = np.var(df[metric_col].dropna())
        group_means = [group.mean() for group in groups]
        group_sizes = [len(group) for group in groups]
        grand_mean = np.mean(df[metric_col].dropna())

        ss_between = sum(group_sizes[i] * (group_means[i] - grand_mean)**2 for i in range(len(groups)))
        ss_total = (len(df[metric_col].dropna()) - 1) * total_var
        eta_squared = ss_between / ss_total

        # 计算组间统计
        group_stats = {}
        for i, group_name in enumerate(df[group_col].unique()):
            group_stats[group_name] = {
                'n': group_sizes[i],
                'mean': group_means[i],
                'std': groups[i].std(),
                'var': groups[i].var()
            }

        result = {
            'test_type': 'One-way ANOVA',
            'statistic': statistic,
            'p_value': p_value,
            'degrees_of_freedom_between': len(groups) - 1,
            'degrees_of_freedom_within': len(df[metric_col].dropna()) - len(groups),
            'effect_size': eta_squared,
            'effect_size_interpretation': self._interpret_eta_squared(eta_squared),
            'group_statistics': group_stats,
            'is_significant': p_value < self.alpha
        }

        return result

    def mann_whitney_u_test(self,
                           data: pd.DataFrame,
                           group_col: str,
                           metric_col: str,
                           alternative: str = 'two-sided') -> Dict:
        """Mann-Whitney U检验 (非参数检验)"""
        df = data.copy()

        # 确保数值列是数值类型
        if df[metric_col].dtype == 'object':
            df[metric_col] = pd.to_numeric(df[metric_col], errors='coerce')

        groups = df[group_col].unique()
        if len(groups) != 2:
            raise ValueError("Mann-Whitney U检验需要恰好2个组")

        group1_data = df[df[group_col] == groups[0]][metric_col].dropna()
        group2_data = df[df[group_col] == groups[1]][metric_col].dropna()

        # 执行Mann-Whitney U检验
        statistic, p_value = stats.mannwhitneyu(
            group1_data, group2_data,
            alternative=alternative
        )

        # 计算效应量 (Rank-biserial correlation)
        n1, n2 = len(group1_data), len(group2_data)
        z_score = (statistic - n1 * n2 / 2) / np.sqrt(n1 * n2 * (n1 + n2 + 1) / 12)
        rank_biserial = z_score / np.sqrt((n1 + n2))

        result = {
            'test_type': 'Mann-Whitney U Test',
            'statistic': statistic,
            'p_value': p_value,
            'z_score': z_score,
            'effect_size': abs(rank_biserial),
            'effect_size_interpretation': self._interpret_rank_biserial(abs(rank_biserial)),
            'group1_stats': {
                'name': groups[0],
                'n': len(group1_data),
                'median': group1_data.median(),
                'mean_rank': group1_data.rank().mean()
            },
            'group2_stats': {
                'name': groups[1],
                'n': len(group2_data),
                'median': group2_data.median(),
                'mean_rank': group2_data.rank().mean()
            },
            'is_significant': p_value < self.alpha
        }

        return result

    def wilcoxon_signed_rank_test(self,
                                 data: pd.DataFrame,
                                 group_col: str,
                                 metric_col: str) -> Dict:
        """Wilcoxon符号秩检验 (配对样本非参数检验)"""
        df = data.copy()

        # 确保数值列是数值类型
        if df[metric_col].dtype == 'object':
            df[metric_col] = pd.to_numeric(df[metric_col], errors='coerce')

        if len(df[group_col].unique()) != 2:
            raise ValueError("Wilcoxon符号秩检验需要恰好2个组")

        groups = df[group_col].unique()
        group1_data = df[df[group_col] == groups[0]][metric_col].dropna()
        group2_data = df[df[group_col] == groups[1]][metric_col].dropna()

        if len(group1_data) != len(group2_data):
            raise ValueError("配对检验需要两组数据长度相同")

        # 执行Wilcoxon符号秩检验
        statistic, p_value = stats.wilcoxon(group1_data, group2_data)

        # 计算效应量
        differences = group1_data - group2_data
        non_zero_diffs = differences[differences != 0]
        z_score = (statistic - len(non_zero_diffs) * (len(non_zero_diffs) + 1) / 4) / np.sqrt(len(non_zero_diffs) * (len(non_zero_diffs) + 1) * (2 * len(non_zero_diffs) + 1) / 24)
        rank_biserial = z_score / np.sqrt(len(differences))

        result = {
            'test_type': 'Wilcoxon Signed-Rank Test',
            'statistic': statistic,
            'p_value': p_value,
            'z_score': z_score,
            'effect_size': abs(rank_biserial),
            'effect_size_interpretation': self._interpret_rank_biserial(abs(rank_biserial)),
            'sample_size': len(differences),
            'zero_differences_removed': len(differences) - len(non_zero_diffs),
            'is_significant': p_value < self.alpha
        }

        return result

    def cohens_d(self, data: pd.DataFrame, group_col: str, metric_col: str) -> float:
        """计算Cohen's d效应量"""
        groups = data[group_col].unique()
        if len(groups) != 2:
            raise ValueError("Cohen's d计算需要恰好2个组")

        group1_data = data[data[group_col] == groups[0]][metric_col].dropna()
        group2_data = data[data[group_col] == groups[1]][metric_col].dropna()

        # 计算合并标准差
        pooled_std = np.sqrt(((len(group1_data) - 1) * group1_data.var() +
                             (len(group2_data) - 1) * group2_data.var()) /
                            (len(group1_data) + len(group2_data) - 2))

        # 计算Cohen's d
        cohens_d = (group1_data.mean() - group2_data.mean()) / pooled_std

        return cohens_d

    def cramers_v(self, data: pd.DataFrame, col1: str, col2: str) -> float:
        """计算Cramer's V效应量"""
        # 创建列联表
        contingency_table = pd.crosstab(data[col1], data[col2])

        # 执行卡方检验
        chi2, _, _, _ = stats.chi2_contingency(contingency_table)

        # 计算Cramer's V
        n = contingency_table.sum().sum()
        cramers_v = np.sqrt(chi2 / (n * (min(contingency_table.shape) - 1)))

        return cramers_v

    def multiple_comparison_correction(self, p_values: List[float], method: str = None) -> List[float]:
        """多重比较校正"""
        method = method or self.correction_method

        if method is None:
            return p_values

        p_values = np.array(p_values)

        if method == 'bonferroni':
            corrected_p = p_values * len(p_values)
            corrected_p = np.minimum(corrected_p, 1.0)

        elif method == 'fdr_bh':
            # Benjamini-Hochberg FDR校正
            ranked_p_values = stats.rankdata(p_values)
            corrected_p = p_values * len(p_values) / ranked_p_values
            corrected_p = np.minimum.accumulate(corrected_p[::-1])[::-1]
            corrected_p = np.minimum(corrected_p, 1.0)

        elif method == 'holm':
            # Holm-Bonferroni校正
            sorted_indices = np.argsort(p_values)
            corrected_p = np.zeros_like(p_values)
            for i, idx in enumerate(sorted_indices):
                corrected_p[idx] = p_values[idx] * (len(p_values) - i)
            corrected_p = np.minimum(corrected_p, 1.0)

        elif method == 'sidak':
            # Šidák校正
            corrected_p = 1 - (1 - p_values) ** len(p_values)

        else:
            raise ValueError(f"未知的校正方法: {method}")

        return corrected_p.tolist()

    def _interpret_cohens_d(self, d: float) -> str:
        """解释Cohen's d效应量"""
        abs_d = abs(d)
        if abs_d < 0.2:
            return "negligible"
        elif abs_d < 0.5:
            return "small"
        elif abs_d < 0.8:
            return "medium"
        else:
            return "large"

    def _interpret_cramers_v(self, v: float) -> str:
        """解释Cramer's V效应量"""
        if v < 0.1:
            return "negligible"
        elif v < 0.3:
            return "small"
        elif v < 0.5:
            return "medium"
        else:
            return "large"

    def _interpret_eta_squared(self, eta2: float) -> str:
        """解释Eta-squared效应量"""
        if eta2 < 0.01:
            return "negligible"
        elif eta2 < 0.06:
            return "small"
        elif eta2 < 0.14:
            return "medium"
        else:
            return "large"

    def _interpret_rank_biserial(self, r: float) -> str:
        """解释Rank-biserial相关效应量"""
        if r < 0.1:
            return "negligible"
        elif r < 0.3:
            return "small"
        elif r < 0.5:
            return "medium"
        else:
            return "large"