Growth Model Analyzer

#!/usr/bin/env python3
"""
增长分析完整示例

演示增长模型分析技能的核心功能：
- 营销策略效果评估
- RFM用户分群
- 用户行为分析
- 转化漏斗分析
- 增长策略洞察
"""

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.growth_analyzer import GrowthModelAnalyzer
from scripts.growth_visualizer import GrowthVisualizer


def create_comprehensive_sample_data():
    """创建全面的样本数据"""
    np.random.seed(42)
    n_users = 2000

    # 创建更真实的用户数据
    data = {
        '用户码': [f'USER_{i:06d}' for i in range(n_users)],
        '裂变类型': np.random.choice(
            ['无裂变页面', '助力砍价', '拼团狂买', '裂变海报', '好友助力'],
            n_users, p=[0.3, 0.25, 0.2, 0.15, 0.1]
        ),
        '城市类型': np.random.choice(
            ['一线城市', '二线城市', '三线城市', '四线城市'],
            n_users, p=[0.2, 0.3, 0.3, 0.2]
        ),
        '设备类型': np.random.choice(['iOS', 'Android'], n_users, p=[0.55, 0.45]),
        '年龄': np.random.randint(18, 65, n_users),
        '性别': np.random.choice(['男', '女'], n_users, p=[0.48, 0.52])
    }

    # 创建行为数据
    df = pd.DataFrame(data)

    # 根据用户特征调整行为
    def calculate_conversion_probability(row):
        base_prob = 0.15

        # 裂变类型影响
        campaign_boost = {
            '无裂变页面': 0.0,
            '助力砍价': 0.12,
            '拼团狂买': 0.15,
            '裂变海报': 0.08,
            '好友助力': 0.20
        }

        # 城市影响
        city_boost = {
            '一线城市': 0.05,
            '二线城市': 0.03,
            '三线城市': 0.01,
            '四线城市': -0.02
        }

        # 年龄影响
        age_boost = -0.001 * (row['年龄'] - 35) ** 2 / 100

        # 设备影响
        device_boost = 0.02 if row['设备类型'] == 'iOS' else 0.0

        prob = base_prob + campaign_boost[row['裂变类型']] + city_boost[row['城市类型']] + age_boost + device_boost
        return max(0, min(1, prob))

    df['转化概率'] = df.apply(calculate_conversion_probability, axis=1)
    df['是否转化'] = np.random.random(len(df)) < df['转化概率']

    # 生成RFM数据
    df['R值'] = np.where(
        df['是否转化'],
        np.random.exponential(7, len(df)).astype(int) + 1,
        np.random.exponential(30, len(df)).astype(int) + 1
    )

    df['F值'] = np.where(
        df['是否转化'],
        np.random.poisson(3, len(df)) + 1,
        np.random.poisson(1, len(df)) + 1
    )

    df['M值'] = np.where(
        df['是否转化'],
        np.random.gamma(2, 50, len(df)),
        np.random.gamma(1, 20, len(df))
    )

    # 生成其他行为数据
    df['曾助力'] = np.where(
        df['裂变类型'].isin(['助力砍价', '好友助力']),
        np.random.random(len(df)) < 0.6,
        np.random.random(len(df)) < 0.15
    )

    df['曾拼团'] = np.where(
        df['裂变类型'] == '拼团狂买',
        np.random.random(len(df)) < 0.7,
        np.random.random(len(df)) < 0.1
    )

    df['曾推荐'] = np.random.random(len(df)) < (df['是否转化'] * 0.5 + 0.05)

    # 生成财务数据
    df['收入'] = np.where(
        df['是否转化'],
        np.random.gamma(2, 80, len(df)),
        np.random.gamma(1, 15, len(df))
    )

    df['成本'] = np.random.uniform(20, 120, len(df))

    return df.drop('转化概率', axis=1)


def main():
    """主函数：演示完整增长分析流程"""
    print("=" * 80)
    print("增长模型分析技能 - 完整增长分析示例")
    print("=" * 80)

    # 1. 初始化分析器
    print("\n🚀 1. 初始化增长分析组件...")
    analyzer = GrowthModelAnalyzer()
    visualizer = GrowthVisualizer()

    # 2. 创建和加载样本数据
    print("\n📊 2. 创建和加载样本数据...")
    data = create_comprehensive_sample_data()
    print(f"✅ 样本数据创建成功：{len(data):,} 条用户记录")
    print(f"   - 裂变策略类型: {data['裂变类型'].nunique()} 种")
    print(f"   - 整体转化率: {data['是否转化'].mean():.2%}")
    print(f"   - 平均用户价值: {data['M值'].mean():.2f}")

    # 3. 数据质量检查
    print("\n🔍 3. 数据质量检查...")
    quality_report = analyzer.data_quality_check(
        data,
        required_cols=['用户码', '裂变类型', '是否转化', 'R值', 'F值', 'M值']
    )

    print(f"   - 总记录数: {quality_report['total_rows']:,}")
    print(f"   - 总字段数: {quality_report['total_columns']}")
    print(f"   - 重复记录: {quality_report['duplicate_rows']}")
    print(f"   - 数据完整性: {'✅ 良好' if quality_report['required_columns_check'] else '❌ 有问题'}")

    # 4. 营销活动效果分析
    print("\n📈 4. 营销活动效果分析...")
    campaign_results = analyzer.analyze_campaign_effectiveness(
        data=data,
        campaign_col='裂变类型',
        conversion_col='是否转化',
        control_group='无裂变页面'
    )

    print("\n   各策略转化效果:")
    for campaign, stats in campaign_results['campaign_statistics'].items():
        print(f"   • {campaign}:")
        print(f"     - 用户数: {stats['用户数']:,}")
        print(f"     - 转化率: {stats['转化率']:.2%}")
        print(f"     - 转化数: {stats['转化数']:,}")

    if campaign_results['lift_analysis']:
        print("\n   策略提升效果:")
        for campaign, lift in campaign_results['lift_analysis'].items():
            print(f"   • {campaign} vs 对照组:")
            print(f"     - 绝对提升: {lift['absolute_lift']:.2%}")
            print(f"     - 相对提升: {lift['relative_lift']:.2%}")
            print(f"     - 提升百分比: {lift['lift_percentage']:.1f}%")

    print(f"\n   统计显著性检验:")
    stats_test = campaign_results['statistical_test']
    print(f"   • 卡方统计量: {stats_test['chi2_statistic']:.4f}")
    print(f"   • P值: {stats_test['p_value']:.6f}")
    print(f"   • 显著性: {'✅ 显著' if stats_test['is_significant'] else '❌ 不显著'}")
    print(f"   • 效应量: {campaign_results['effect_size']['cramers_v']:.4f} ({campaign_results['effect_size']['interpretation']})")

    # 5. RFM用户分群分析
    print("\n👥 5. RFM用户分群分析...")
    rfm_results = analyzer.rfm_segmentation(
        data=data,
        user_col='用户码',
        recency_col='R值',
        frequency_col='F值',
        monetary_col='M值',
        n_clusters=5
    )

    print(f"   - 聚类数量: {rfm_results['n_clusters']}")
    print(f"   - 轮廓系数: {rfm_results['silhouette_score']:.3f}")

    print("\n   RFM分群统计:")
    for segment, count in rfm_results['rfm_segments'].items():
        print(f"   • {segment}: {count:,} 用户 ({count/len(data):.1%})")

    print("\n   聚类特征分析:")
    for cluster, stats in rfm_results['cluster_statistics'].items():
        print(f"   • {cluster}:")
        print(f"     - 用户数: {stats['count']:,} ({stats['percentage']:.1f}%)")
        print(f"     - 平均R值: {stats['avg_recency']:.1f}")
        print(f"     - 平均F值: {stats['avg_frequency']:.1f}")
        print(f"     - 平均M值: {stats['avg_monetary']:.2f}")

    # 6. 用户行为分析
    print("\n🎯 6. 用户行为分析...")
    behavior_results = analyzer.user_behavior_analysis(
        data=data,
        user_col='用户码',
        behavior_cols=['曾助力', '曾拼团', '曾推荐']
    )

    print("   单项行为参与率:")
    for behavior, stats in behavior_results['individual_behavior'].items():
        print(f"   • {behavior}: {stats['behavior_rate']:.1%} ({stats['behavior_users']:,} 用户)")

    if 'behavior_insights' in behavior_results:
        print("\n   行为洞察:")
        for insight in behavior_results['behavior_insights']:
            print(f"   • {insight}")

    # 7. 转化漏斗分析
    print("\n🔥 7. 转化漏斗分析...")
    funnel_results = analyzer.conversion_funnel_analysis(
        data=data,
        funnel_stages=['曾助力', '曾拼团', '是否转化']
    )

    print("   转化漏斗数据:")
    for stage_data in funnel_results['funnel_data']:
        print(f"   • {stage_data['stage']}: {stage_data['users']:,} 用户 ({stage_data['conversion_rate']:.1%})")
        if 'stage_conversion_rate' in stage_data:
            print(f"     - 阶段转化率: {stage_data['stage_conversion_rate']:.1%}")
            print(f"     - 阶段流失率: {stage_data['stage_dropoff_rate']:.1%}")

    print(f"\n   整体转化率: {funnel_results['overall_conversion_rate']:.1%}")

    if funnel_results['insights']:
        print("\n   漏斗洞察:")
        for insight in funnel_results['insights']:
            print(f"   • {insight}")

    # 8. 队列分析
    print("\n⏰ 8. 队列分析...")
    cohort_results = analyzer.cohort_analysis(
        data=data,
        user_col='用户码',
        time_col='R值',
        conversion_col='是否转化'
    )

    print("   时间队列转化率:")
    for cohort, stats in cohort_results['cohort_conversion'].items():
        print(f"   • {cohort}: {stats['conversion_rate']:.1%} ({stats['conversions']:,}/{stats['total_users']:,})")

    print(f"\n   整体转化率: {cohort_results['overall_conversion_rate']:.1%}")

    # 9. 生成综合洞察报告
    print("\n💡 9. 生成综合洞察报告...")
    insights_report = analyzer.generate_insights_report()

    print("   关键发现:")
    for finding in insights_report['key_findings']:
        print(f"   • {finding}")

    print("\n   策略建议:")
    for recommendation in insights_report['recommendations']:
        print(f"   • {recommendation}")

    # 10. 创建可视化图表
    print("\n📊 10. 创建可视化图表...")
    output_dir = Path(__file__).parent / "growth_analysis_output"
    output_dir.mkdir(exist_ok=True)

    try:
        # 转化漏斗图
        funnel_fig = visualizer.plot_conversion_funnel(
            funnel_results['funnel_data'],
            title="用户转化漏斗分析",
            interactive=False
        )
        if funnel_fig:
            funnel_fig.savefig(output_dir / "conversion_funnel.png", dpi=300, bbox_inches='tight')
            print("   ✅ 转化漏斗图已保存")
            import matplotlib.pyplot as plt
            plt.close(funnel_fig)

        # RFM分群散点图
        rfm_fig = visualizer.plot_rfm_segments(
            rfm_results['rfm_data'],
            title="RFM用户分群分析"
        )
        rfm_fig.write_html(str(output_dir / "rfm_segments.html"))
        print("   ✅ RFM分群图已保存")

        # 营销活动对比雷达图
        campaign_fig = visualizer.plot_campaign_comparison(
            campaign_results,
            title="营销活动效果对比"
        )
        campaign_fig.write_html(str(output_dir / "campaign_comparison.html"))
        print("   ✅ 营销活动对比图已保存")

    except Exception as e:
        print(f"   ⚠️ 图表生成遇到问题: {str(e)}")

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

    # 保存营销活动分析结果
    campaign_df = pd.DataFrame(campaign_results['campaign_statistics']).T
    campaign_df.to_csv(output_dir / "campaign_analysis.csv", encoding='utf-8-sig')

    # 保存RFM分群结果
    rfm_data = rfm_results['rfm_data']
    rfm_data.to_csv(output_dir / "rfm_segments.csv", index=False, encoding='utf-8-sig')

    # 保存转化漏斗结果
    funnel_df = pd.DataFrame(funnel_results['funnel_data'])
    funnel_df.to_csv(output_dir / "conversion_funnel.csv", index=False, encoding='utf-8-sig')

    # 保存队列分析结果
    cohort_df = pd.DataFrame(cohort_results['cohort_conversion']).T
    cohort_df.to_csv(output_dir / "cohort_analysis.csv", encoding='utf-8-sig')

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

    # 12. 总结报告
    print("\n" + "=" * 80)
    print("🎉 增长分析完成!")
    print("=" * 80)

    print(f"\n📁 生成的文件:")
    print(f"   - 营销活动分析: {output_dir}/campaign_analysis.csv")
    print(f"   - RFM分群结果: {output_dir}/rfm_segments.csv")
    print(f"   - 转化漏斗分析: {output_dir}/conversion_funnel.csv")
    print(f"   - 队列分析结果: {output_dir}/cohort_analysis.csv")
    print(f"   - 转化漏斗图: {output_dir}/conversion_funnel.png")
    print(f"   - RFM分群图: {output_dir}/rfm_segments.html")
    print(f"   - 营销对比图: {output_dir}/campaign_comparison.html")

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

    # 获取最佳和最差策略
    if campaign_results['lift_analysis']:
        best_campaign = max(campaign_results['lift_analysis'].items(),
                          key=lambda x: x[1]['lift_percentage'])
        print(f"   - 最佳策略: {best_campaign[0]} (提升 {best_campaign[1]['lift_percentage']:.1f}%)")
        print(f"   - 策略显著性: {'显著' if campaign_results['statistical_test']['is_significant'] else '不显著'}")

    # RFM分群洞察
    top_segment = max(rfm_results['segment_analysis'].items(),
                     key=lambda x: x[1]['Monetary'])
    print(f"   - 最有价值用户群: {top_segment[0]}")
    print(f"   - 高价值用户平均消费: {top_segment[1]['Monetary']:.2f}")

    # 转化漏斗洞察
    print(f"   - 整体转化率: {funnel_results['overall_conversion_rate']:.1%}")
    if len(funnel_results['funnel_data']) > 1:
        max_dropoff = max(funnel_results['funnel_data'][1:],
                         key=lambda x: x.get('stage_dropoff_rate', 0))
        print(f"   - 最大流失环节: {max_dropoff['stage']} ({max_dropoff.get('stage_dropoff_rate', 0)*100:.1f}%)")

    print(f"\n💡 增长模型分析技能特性:")
    print(f"   ✅ 全面的营销策略效果评估")
    print(f"   ✅ 智能RFM用户分群和价值分析")
    print(f"   ✅ 深度用户行为洞察和画像")
    print(f"   ✅ 专业的转化漏斗和队列分析")
    print(f"   ✅ 统计显著性检验和效应量分析")
    print(f"   ✅ 丰富的可视化图表和报告")
    print(f"   ✅ 数据驱动的增长策略建议")

    return True


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

"""
增长模型分析器 - 核心增长分析模块

提供全面的增长黑客分析功能，包括：
- 裂变策略效果评估
- 用户细分和RFM分析
- 转化率分析和统计检验
- 增长策略效果对比
- 用户画像和行为分析
"""

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


class GrowthModelAnalyzer:
    """增长模型分析器 - 核心分析引擎"""

    def __init__(self, config: Optional[Dict] = None):
        """
        初始化增长分析器

        Parameters:
        - config: 配置参数字典
        """
        self.config = config or {}
        self.scaler = StandardScaler()
        self.data = None
        self.results = {}

    def load_data(self, data_path: str, **kwargs) -> pd.DataFrame:
        """
        加载增长分析数据

        Parameters:
        - data_path: 数据文件路径
        - **kwargs: pandas.read_csv的额外参数

        Returns:
        - 加载的数据DataFrame
        """
        try:
            if data_path.endswith('.csv'):
                self.data = pd.read_csv(data_path, **kwargs)
            elif data_path.endswith('.xlsx'):
                self.data = pd.read_excel(data_path, **kwargs)
            else:
                raise ValueError("支持的文件格式: .csv, .xlsx")

            print(f"✅ 数据加载成功: {len(self.data)} 条记录")
            return self.data

        except Exception as e:
            print(f"❌ 数据加载失败: {str(e)}")
            return None

    def data_quality_check(self, data: pd.DataFrame = None,
                          required_cols: List[str] = None) -> Dict:
        """
        数据质量检查

        Parameters:
        - data: 待检查的数据
        - required_cols: 必需的列名列表

        Returns:
        - 质量检查报告
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        quality_report = {
            'total_rows': len(data),
            'total_columns': len(data.columns),
            'duplicate_rows': data.duplicated().sum(),
            'missing_values': {},
            'data_types': data.dtypes.to_dict(),
            'required_columns_check': True
        }

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

        # 检查必需列
        if required_cols:
            missing_required = [col for col in required_cols if col not in data.columns]
            if missing_required:
                quality_report['required_columns_check'] = False
                quality_report['missing_required_columns'] = missing_required

        return quality_report

    def analyze_campaign_effectiveness(self,
                                     data: pd.DataFrame = None,
                                     campaign_col: str = '裂变类型',
                                     conversion_col: str = '是否转化',
                                     control_group: Optional[str] = None) -> Dict:
        """
        分析营销活动效果

        Parameters:
        - data: 分析数据
        - campaign_col: 活动类型列名
        - conversion_col: 转化结果列名
        - control_group: 对照组名称

        Returns:
        - 效果分析结果
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        results = {}

        # 1. 基础统计
        campaign_stats = data.groupby(campaign_col)[conversion_col].agg([
            'count', 'sum', 'mean', 'std'
        ]).round(4)

        campaign_stats.columns = ['用户数', '转化数', '转化率', '标准差']
        campaign_stats['转化数'] = campaign_stats['转化数'].astype(int)

        results['campaign_statistics'] = campaign_stats.to_dict()

        # 2. 转化率提升分析
        if control_group and control_group in campaign_stats.index:
            control_rate = campaign_stats.loc[control_group, '转化率']

            lift_analysis = {}
            for campaign in campaign_stats.index:
                if campaign != control_group:
                    campaign_rate = campaign_stats.loc[campaign, '转化率']
                    absolute_lift = campaign_rate - control_rate
                    relative_lift = absolute_lift / control_rate if control_rate > 0 else 0

                    lift_analysis[campaign] = {
                        'absolute_lift': absolute_lift,
                        'relative_lift': relative_lift,
                        'lift_percentage': relative_lift * 100,
                        'control_rate': control_rate,
                        'campaign_rate': campaign_rate
                    }

            results['lift_analysis'] = lift_analysis

        # 3. 统计显著性检验
        contingency_table = pd.crosstab(data[campaign_col], data[conversion_col])
        chi2_stat, p_value, dof, expected = stats.chi2_contingency(contingency_table)

        results['statistical_test'] = {
            'chi2_statistic': chi2_stat,
            'p_value': p_value,
            'degrees_of_freedom': dof,
            'is_significant': p_value < 0.05
        }

        # 4. 效应量计算 (Cramer's V)
        n = contingency_table.sum().sum()
        cramer_v = np.sqrt(chi2_stat / (n * (min(contingency_table.shape) - 1)))

        results['effect_size'] = {
            'cramers_v': cramer_v,
            'interpretation': self._interpret_cramers_v(cramer_v)
        }

        self.results['campaign_effectiveness'] = results
        return results

    def rfm_segmentation(self,
                        data: pd.DataFrame = None,
                        user_col: str = '用户码',
                        recency_col: str = 'R值',
                        frequency_col: str = 'F值',
                        monetary_col: str = 'M值',
                        n_clusters: int = 5) -> Dict:
        """
        RFM用户分群分析

        Parameters:
        - data: 分析数据
        - user_col: 用户标识列名
        - recency_col: 近度指标列名
        - frequency_col: 频度指标列名
        - monetary_col: 金额指标列名
        - n_clusters: 聚类数量

        Returns:
        - RFM分群结果
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        # 准备RFM数据
        rfm_data = data.groupby(user_col)[[recency_col, frequency_col, monetary_col]].agg({
            recency_col: 'min',      # R值越小越好 (最近消费时间)
            frequency_col: 'sum',     # F值越大越好 (消费频次)
            monetary_col: 'sum'       # M值越大越好 (消费金额)
        }).reset_index()

        rfm_data.columns = [user_col, 'Recency', 'Frequency', 'Monetary']

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

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

        # K-means聚类
        features = rfm_data[['Recency', 'Frequency', 'Monetary']]
        features_scaled = self.scaler.fit_transform(features)

        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
        rfm_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 = rfm_data[rfm_data['Cluster'] == cluster]
            cluster_stats[f'Cluster_{cluster}'] = {
                'count': len(cluster_data),
                'percentage': len(cluster_data) / len(rfm_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'

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

        results = {
            'rfm_data': rfm_data,
            'cluster_statistics': cluster_stats,
            'silhouette_score': silhouette_avg,
            'n_clusters': n_clusters,
            'rfm_segments': rfm_data['RFM_Segment'].value_counts().to_dict(),
            'segment_analysis': rfm_data.groupby('RFM_Segment')[['Recency', 'Frequency', 'Monetary']].mean().to_dict()
        }

        self.results['rfm_segmentation'] = results
        return results

    def user_behavior_analysis(self,
                             data: pd.DataFrame = None,
                             user_col: str = '用户码',
                             behavior_cols: List[str] = None) -> Dict:
        """
        用户行为分析

        Parameters:
        - data: 分析数据
        - user_col: 用户标识列名
        - behavior_cols: 行为指标列名列表

        Returns:
        - 用户行为分析结果
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        if behavior_cols is None:
            behavior_cols = ['曾助力', '曾拼团', '曾推荐']

        # 过滤存在的列
        existing_cols = [col for col in behavior_cols if col in data.columns]

        # 用户行为统计
        behavior_stats = {}
        for col in existing_cols:
            behavior_stats[col] = {
                'total_users': len(data),
                'behavior_users': data[col].sum(),
                'behavior_rate': data[col].sum() / len(data),
                'non_behavior_users': len(data) - data[col].sum()
            }

        # 行为组合分析
        if len(existing_cols) >= 2:
            behavior_combinations = data.groupby(existing_cols)[user_col].count().reset_index()
            behavior_combinations.columns = existing_cols + ['user_count']
            behavior_combinations['percentage'] = behavior_combinations['user_count'] / len(data)

            results = {
                'individual_behavior': behavior_stats,
                'behavior_combinations': behavior_combinations.to_dict(),
                'behavior_insights': self._analyze_behavior_insights(behavior_stats, behavior_combinations)
            }
        else:
            results = {
                'individual_behavior': behavior_stats
            }

        self.results['user_behavior_analysis'] = results
        return results

    def conversion_funnel_analysis(self,
                                 data: pd.DataFrame = None,
                                 funnel_stages: List[str] = None) -> Dict:
        """
        转化漏斗分析

        Parameters:
        - data: 分析数据
        - funnel_stages: 漏斗阶段列名列表

        Returns:
        - 转化漏斗分析结果
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        if funnel_stages is None:
            funnel_stages = ['曾助力', '曾拼团', '是否转化']

        # 过滤存在的列
        existing_stages = [stage for stage in funnel_stages if stage in data.columns]

        funnel_data = []
        for i, stage in enumerate(existing_stages):
            stage_name = self._get_stage_name(stage)
            users_at_stage = data[stage].sum() if i > 0 else len(data)
            conversion_rate = users_at_stage / len(data)

            funnel_data.append({
                'stage': stage_name,
                'stage_column': stage,
                'users': users_at_stage,
                'conversion_rate': conversion_rate,
                'dropoff_rate': 1 - conversion_rate if i > 0 else 0
            })

        # 计算阶段间转化率
        for i in range(1, len(funnel_data)):
            prev_users = funnel_data[i-1]['users']
            current_users = funnel_data[i]['users']
            stage_conversion = current_users / prev_users if prev_users > 0 else 0
            funnel_data[i]['stage_conversion_rate'] = stage_conversion
            funnel_data[i]['stage_dropoff_rate'] = 1 - stage_conversion

        results = {
            'funnel_data': funnel_data,
            'overall_conversion_rate': funnel_data[-1]['conversion_rate'] if funnel_data else 0,
            'total_users': len(data),
            'insights': self._analyze_funnel_insights(funnel_data)
        }

        self.results['conversion_funnel'] = results
        return results

    def cohort_analysis(self,
                       data: pd.DataFrame = None,
                       user_col: str = '用户码',
                       time_col: str = 'R值',
                       conversion_col: str = '是否转化') -> Dict:
        """
        队列分析

        Parameters:
        - data: 分析数据
        - user_col: 用户标识列名
        - time_col: 时间指标列名
        - conversion_col: 转化结果列名

        Returns:
        - 队列分析结果
        """
        if data is None:
            data = self.data

        if data is None:
            raise ValueError("请先加载数据")

        # 创建时间队列 (使用R值作为时间分组)
        data_copy = data.copy()
        data_copy['time_group'] = pd.qcut(data_copy[time_col], q=4, labels=['Q1', 'Q2', 'Q3', 'Q4'])

        # 计算各队列的转化率
        cohort_conversion = data_copy.groupby('time_group')[conversion_col].agg([
            'count', 'sum', 'mean'
        ]).round(4)

        cohort_conversion.columns = ['total_users', 'conversions', 'conversion_rate']
        cohort_conversion['conversions'] = cohort_conversion['conversions'].astype(int)

        results = {
            'cohort_conversion': cohort_conversion.to_dict(),
            'overall_conversion_rate': data[conversion_col].mean(),
            'cohort_insights': self._analyze_cohort_insights(cohort_conversion)
        }

        self.results['cohort_analysis'] = results
        return results

    def generate_insights_report(self) -> Dict:
        """
        生成洞察报告

        Returns:
        - 综合洞察报告
        """
        if not self.results:
            return {"error": "请先进行分析以生成结果"}

        insights = {
            "summary": {},
            "key_findings": [],
            "recommendations": [],
            "data_quality": "良好"
        }

        # 策略效果洞察
        if 'campaign_effectiveness' in self.results:
            campaign_results = self.results['campaign_effectiveness']
            if campaign_results['statistical_test']['is_significant']:
                insights["key_findings"].append("营销策略对用户转化有显著影响")

                # 找到最佳策略
                if 'lift_analysis' in campaign_results:
                    best_campaign = max(
                        campaign_results['lift_analysis'].items(),
                        key=lambda x: x[1]['lift_percentage']
                    )
                    insights["key_findings"].append(
                        f"{best_campaign[0]}策略效果最佳，提升{best_campaign[1]['lift_percentage']:.1f}%"
                    )
                    insights["recommendations"].append(
                        f"建议优先推广{best_campaign[0]}策略"
                    )

        # RFM分群洞察
        if 'rfm_segmentation' in self.results:
            rfm_results = self.results['rfm_segmentation']
            top_segment = max(rfm_results['segment_analysis'].items(),
                            key=lambda x: x[1]['Monetary'])
            insights["key_findings"].append(
                f"最有价值用户群体是{top_segment[0]}，平均消费金额为{top_segment[1]['Monetary']:.2f}"
            )
            insights["recommendations"].append(
                "针对高价值用户群体提供个性化服务和权益"
            )

        # 转化漏斗洞察
        if 'conversion_funnel' in self.results:
            funnel_results = self.results['conversion_funnel']
            if funnel_results['overall_conversion_rate'] < 0.1:
                insights["recommendations"].append("整体转化率偏低，建议优化用户旅程")

            # 找到最大流失环节
            funnel_data = funnel_results['funnel_data']
            if len(funnel_data) > 1:
                max_dropoff = max(funnel_data[1:], key=lambda x: x.get('stage_dropoff_rate', 0))
                insights["key_findings"].append(
                    f"最大流失环节是{max_dropoff['stage']}，流失率{max_dropoff.get('stage_dropoff_rate', 0)*100:.1f}%"
                )

        return insights

    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 _get_stage_name(self, stage_col: str) -> str:
        """获取漏斗阶段的中文名称"""
        stage_names = {
            '曾助力': '助力行为',
            '曾拼团': '拼团行为',
            '曾推荐': '推荐行为',
            '是否转化': '最终转化'
        }
        return stage_names.get(stage_col, stage_col)

    def _analyze_behavior_insights(self, behavior_stats: Dict, behavior_combinations: pd.DataFrame) -> List[str]:
        """分析用户行为洞察"""
        insights = []

        # 找到最常见的行为
        most_common = max(behavior_stats.items(), key=lambda x: x[1]['behavior_rate'])
        insights.append(f"最常见的行为是{most_common[0]}，参与率{most_common[1]['behavior_rate']:.1%}")

        # 分析行为重叠
        if len(behavior_combinations) > 0:
            multi_behavior = behavior_combinations[
                behavior_combinations[[col for col in behavior_combinations.columns if col in behavior_stats]].sum(axis=1) > 1
            ]
            if len(multi_behavior) > 0:
                multi_behavior_rate = multi_behavior['user_count'].sum() / len(behavior_combinations)
                insights.append(f"多行为用户占比{multi_behavior_rate:.1%}")

        return insights

    def _analyze_funnel_insights(self, funnel_data: List[Dict]) -> List[str]:
        """分析转化漏斗洞察"""
        insights = []

        if len(funnel_data) > 1:
            # 找到最大流失环节
            max_dropoff_stage = max(funnel_data[1:], key=lambda x: x.get('stage_dropoff_rate', 0))
            dropoff_rate = max_dropoff_stage.get('stage_dropoff_rate', 0)
            insights.append(f"最大流失环节是{max_dropoff_stage['stage']}，流失率{dropoff_rate*100:.1f}%")

        # 整体转化率评估
        overall_rate = funnel_data[-1]['conversion_rate'] if funnel_data else 0
        if overall_rate > 0.2:
            insights.append("整体转化率表现良好")
        elif overall_rate > 0.1:
            insights.append("整体转化率有提升空间")
        else:
            insights.append("整体转化率偏低，需要重点优化")

        return insights

    def _analyze_cohort_insights(self, cohort_conversion: pd.DataFrame) -> List[str]:
        """分析队列洞察"""
        insights = []

        # 找到表现最好的队列
        best_cohort = cohort_conversion.loc[cohort_conversion['conversion_rate'].idxmax()]
        insights.append(f"表现最好的队列是{best_cohort.name}，转化率{best_cohort['conversion_rate']:.1%}")

        # 分析队列间差异
        conversion_rates = cohort_conversion['conversion_rate']
        if conversion_rates.max() - conversion_rates.min() > 0.1:
            insights.append("不同队列间转化率差异较大，建议针对性优化")

        return insights

"""
增长可视化模块 - 增长模型图表和仪表板

提供全面的增长分析可视化功能，包括：
- 转化漏斗可视化
- 用户分群图表
- 增长趋势分析
- ROI分析图表
- 交互式仪表板
- Qini曲线和Uplift模型可视化
"""

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
from typing import Dict, List, Optional, Union, Tuple
import warnings
warnings.filterwarnings('ignore')

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False


class GrowthVisualizer:
    """增长可视化器 - 专业图表和仪表板生成工具"""

    def __init__(self, style: str = 'seaborn', palette: str = 'husl'):
        """
        初始化可视化器

        Parameters:
        - style: 图表风格
        - palette: 色彩方案
        """
        self.style = style
        self.palette = palette
        self.setup_style()

    def setup_style(self):
        """设置图表样式"""
        if self.style == 'seaborn':
            sns.set_style("whitegrid")
            sns.set_palette(self.palette)

    def plot_conversion_funnel(self,
                             funnel_data: List[Dict],
                             title: str = "用户转化漏斗",
                             interactive: bool = True) -> Union[plt.Figure, go.Figure]:
        """
        绘制转化漏斗图

        Parameters:
        - funnel_data: 漏斗数据列表
        - title: 图表标题
        - interactive: 是否使用交互式图表

        Returns:
        - 图表对象
        """
        if interactive:
            return self._plot_interactive_funnel(funnel_data, title)
        else:
            return self._plot_static_funnel(funnel_data, title)

    def _plot_interactive_funnel(self, funnel_data: List[Dict], title: str) -> go.Figure:
        """绘制交互式漏斗图"""
        fig = go.Figure(go.Funnel(
            y=[item['stage'] for item in funnel_data],
            x=[item['users'] for item in funnel_data],
            textinfo="value+percent initial",
            textposition="inside",
            textfont={"size": 14},
            marker={"color": ["#1f77b4", "#ff7f0e", "#2ca02c", "#d62728", "#9467bd"][:len(funnel_data)]},
            connector={"line": {"color": "royalblue", "dash": "dot", "width": 2}}
        ))

        fig.update_layout(
            title=f"{title}<br><sup>总用户数: {funnel_data[0]['users']:,}</sup>",
            font_size=12,
            height=600
        )

        return fig

    def _plot_static_funnel(self, funnel_data: List[Dict], title: str) -> plt.Figure:
        """绘制静态漏斗图"""
        fig, ax = plt.subplots(figsize=(10, 8))

        stages = [item['stage'] for item in funnel_data]
        users = [item['users'] for item in funnel_data]
        colors = sns.color_palette(self.palette, len(funnel_data))

        # 绘制漏斗条形图
        bars = ax.barh(stages, users, color=colors, alpha=0.8)

        # 添加转化率标签
        for i, (stage, bar) in enumerate(zip(funnel_data, bars)):
            if i < len(funnel_data) - 1:
                conversion_rate = funnel_data[i+1]['users'] / funnel_data[i]['users']
                ax.text(bar.get_width() + max(users) * 0.01, bar.get_y() + bar.get_height()/2,
                       f"转化率: {conversion_rate:.1%}",
                       va='center', fontsize=10)

        # 美化图表
        ax.set_xlabel('用户数', fontsize=12)
        ax.set_title(f"{title}\n总用户数: {funnel_data[0]['users']:,}", fontsize=14, fontweight='bold')
        ax.grid(True, alpha=0.3)

        # 添加数值标签
        for i, (stage, bar) in enumerate(zip(funnel_data, bars)):
            ax.text(bar.get_width() - max(users) * 0.05, bar.get_y() + bar.get_height()/2,
                   f"{int(bar.get_width()):,}",
                   ha='right', va='center', fontsize=11, fontweight='bold', color='white')

        plt.tight_layout()
        return fig

    def plot_rfm_segments(self,
                         rfm_data: pd.DataFrame,
                         title: str = "RFM用户分群分析") -> go.Figure:
        """
        绘制RFM分群散点图

        Parameters:
        - rfm_data: RFM分析数据
        - title: 图表标题

        Returns:
        - 交互式散点图
        """
        fig = px.scatter(
            rfm_data,
            x='Recency',
            y='Frequency',
            size='Monetary',
            color='RFM_Segment',
            hover_data=['RFM_Total', 'Cluster'],
            title=title,
            labels={
                'Recency': '近度 (R值)',
                'Frequency': '频度 (F值)',
                'Monetary': '金额 (M值)',
                'RFM_Segment': '用户分群'
            },
            color_discrete_map={
                'Champions': '#FF6B6B',
                'Loyal Customers': '#4ECDC4',
                'Potential Loyalists': '#45B7D1',
                'New Customers': '#96CEB4',
                'At Risk': '#FFEAA7',
                'Cannot Lose Them': '#DDA0DD',
                'Lost': '#FFA07A'
            }
        )

        fig.update_layout(
            height=600,
            xaxis_title="近度 (R值) - 越小越好",
            yaxis_title="频度 (F值) - 越大越好"
        )

        return fig

    def plot_campaign_comparison(self,
                               campaign_data: Dict,
                               metrics: List[str] = ['转化率', 'ROI', 'CPA'],
                               title: str = "营销活动对比分析") -> go.Figure:
        """
        绘制营销活动对比雷达图

        Parameters:
        - campaign_data: 活动数据
        - metrics: 对比指标
        - title: 图表标题

        Returns:
        - 雷达图
        """
        if isinstance(campaign_data, dict) and 'campaign_metrics' in campaign_data:
            df = pd.DataFrame(campaign_data['campaign_metrics']).T
        else:
            df = pd.DataFrame(campaign_data).T

        # 标准化数据到0-1范围
        normalized_df = df[metrics].copy()
        for metric in metrics:
            if metric == 'CPA':  # CPA越小越好
                normalized_df[metric] = 1 - (normalized_df[metric] - normalized_df[metric].min()) / (normalized_df[metric].max() - normalized_df[metric].min())
            else:
                normalized_df[metric] = (normalized_df[metric] - normalized_df[metric].min()) / (normalized_df[metric].max() - normalized_df[metric].min())

        fig = go.Figure()

        colors = px.colors.qualitative.Set1
        for i, (campaign, row) in enumerate(normalized_df.iterrows()):
            fig.add_trace(go.Scatterpolar(
                r=row.values.tolist() + row.values.tolist()[:1],  # 闭合雷达图
                theta=metrics + metrics[:1],
                fill='toself',
                name=campaign,
                line_color=colors[i % len(colors)]
            ))

        fig.update_layout(
            polar=dict(
                radialaxis=dict(
                    visible=True,
                    range=[0, 1]
                )
            ),
            title=title,
            height=600
        )

        return fig

    def plot_qini_curve(self,
                       qini_data: Dict,
                       title: str = "Qini曲线分析") -> go.Figure:
        """
        绘制Qini曲线

        Parameters:
        - qini_data: Qini曲线数据
        - title: 图表标题

        Returns:
        - Qini曲线图
        """
        fractions = qini_data['fraction']
        qini_gains = qini_data['qini_gain']
        random_qini = qini_data['random_qini']

        fig = go.Figure()

        # 模型Qini曲线
        fig.add_trace(go.Scatter(
            x=fractions,
            y=qini_gains,
            mode='lines',
            name='Uplift模型',
            line=dict(color='blue', width=3)
        ))

        # 随机模型曲线
        fig.add_trace(go.Scatter(
            x=fractions,
            y=random_qini,
            mode='lines',
            name='随机模型',
            line=dict(color='gray', width=2, dash='dash')
        ))

        # 填充区域
        fig.add_trace(go.Scatter(
            x=fractions + fractions[::-1],
            y=qini_gains + random_qini[::-1],
            fill='toself',
            fillcolor='rgba(0,100,80,0.2)',
            line=dict(color='rgba(255,255,255,0)'),
            name='增益区域'
        ))

        fig.update_layout(
            title=f"{title}<br><sup>AUQC: {qini_data.get('auqc', 0):.4f}</sup>",
            xaxis_title='人口比例',
            yaxis_title='累积增量',
            height=500
        )

        return fig

    def plot_uplift_distribution(self,
                                uplift_data: pd.DataFrame,
                                title: str = "增量分数分布") -> go.Figure:
        """
        绘制增量分数分布图

        Parameters:
        - uplift_data: 增量分析数据
        - title: 图表标题

        Returns:
        - 分布图
        """
        fig = make_subplots(
            rows=2, cols=2,
            subplot_titles=('增量分数分布', '分群用户数', '平均增量分数', '累积分布'),
            specs=[[{"secondary_y": False}, {"secondary_y": False}],
                   [{"secondary_y": False}, {"secondary_y": False}]]
        )

        # 1. 增量分数直方图
        fig.add_trace(
            go.Histogram(x=uplift_data['uplift_score'], name='增量分数', nbinsx=50),
            row=1, col=1
        )

        # 2. 分群用户数
        if 'uplift_segment' in uplift_data.columns:
            segment_counts = uplift_data['uplift_segment'].value_counts()
            fig.add_trace(
                go.Bar(x=segment_counts.index, y=segment_counts.values, name='分群用户数'),
                row=1, col=2
            )

        # 3. 各分群平均增量分数
        if 'uplift_segment' in uplift_data.columns:
            segment_uplift = uplift_data.groupby('uplift_segment')['uplift_score'].mean()
            fig.add_trace(
                go.Bar(x=segment_uplift.index, y=segment_uplift.values, name='平均增量分数'),
                row=2, col=1
            )

        # 4. 累积分布
        sorted_uplift = np.sort(uplift_data['uplift_score'])
        cumulative = np.arange(1, len(sorted_uplift) + 1) / len(sorted_uplift)
        fig.add_trace(
            go.Scatter(x=sorted_uplift, y=cumulative, name='累积分布', mode='lines'),
            row=2, col=2
        )

        fig.update_layout(
            title=title,
            height=800,
            showlegend=False
        )

        return fig

    def plot_roi_dashboard(self,
                          roi_data: Dict,
                          title: str = "ROI分析仪表板") -> go.Figure:
        """
        绘制ROI分析仪表板

        Parameters:
        - roi_data: ROI分析数据
        - title: 仪表板标题

        Returns:
        - 仪表板图表
        """
        if isinstance(roi_data, dict) and 'campaign_metrics' in roi_data:
            df = pd.DataFrame(roi_data['campaign_metrics']).T
        else:
            df = pd.DataFrame(roi_data).T

        fig = make_subplots(
            rows=2, cols=2,
            subplot_titles=('ROI对比', '转化率vs成本', '收入vs投入', '利润率分析'),
            specs=[[{"type": "bar"}, {"type": "scatter"}],
                   [{"type": "bar"}, {"type": "bar"}]]
        )

        # 1. ROI对比
        fig.add_trace(
            go.Bar(x=df.index, y=df['ROI'], name='ROI', marker_color='lightblue'),
            row=1, col=1
        )

        # 2. 转化率vs成本散点图
        fig.add_trace(
            go.Scatter(
                x=df['CPA'], y=df['转化率'],
                mode='markers+text',
                text=df.index,
                textposition="top center",
                name='转化率vsCPA',
                marker=dict(size=df['触达用户数']/100, color=df['ROI'], colorscale='Viridis', showscale=True)
            ),
            row=1, col=2
        )

        # 3. 收入vs投入
        fig.add_trace(
            go.Bar(x=df.index, y=df['总收入'], name='总收入', marker_color='lightgreen'),
            row=2, col=1
        )
        fig.add_trace(
            go.Bar(x=df.index, y=df['总成本'], name='总成本', marker_color='lightcoral'),
            row=2, col=1
        )

        # 4. 利润率分析
        fig.add_trace(
            go.Bar(x=df.index, y=df['利润率'], name='利润率', marker_color='gold'),
            row=2, col=2
        )

        fig.update_layout(
            title=f"{title}<br><sup>整体ROI: {roi_data.get('overall_roi', 0):.2%}</sup>",
            height=800,
            showlegend=True
        )

        return fig

    def plot_growth_trend(self,
                         trend_data: Dict,
                         title: str = "增长趋势分析") -> go.Figure:
        """
        绘制增长趋势图

        Parameters:
        - trend_data: 趋势数据
        - title: 图表标题

        Returns:
        - 趋势图
        """
        if isinstance(trend_data, dict) and 'time_series' in trend_data:
            df = pd.DataFrame(trend_data['time_series']).T
            df.index = pd.to_datetime(df.index)
        else:
            df = trend_data

        fig = make_subplots(
            rows=3, cols=1,
            subplot_titles=('收入成本趋势', 'ROI趋势', '月环比变化'),
            vertical_spacing=0.08
        )

        # 1. 收入成本趋势
        fig.add_trace(
            go.Scatter(x=df.index, y=df['总收入'], name='总收入', line=dict(color='green')),
            row=1, col=1
        )
        fig.add_trace(
            go.Scatter(x=df.index, y=df['总成本'], name='总成本', line=dict(color='red')),
            row=1, col=1
        )

        # 2. ROI趋势
        fig.add_trace(
            go.Scatter(x=df.index, y=df['ROI'], name='ROI', line=dict(color='blue')),
            row=2, col=1
        )
        fig.add_trace(
            go.Scatter(x=df.index, y=df['ROI_3MAvg'], name='3月平均', line=dict(color='orange', dash='dash')),
            row=2, col=1
        )

        # 3. 月环比变化
        fig.add_trace(
            go.Bar(x=df.index, y=df['ROI_MoM'], name='月环比变化'),
            row=3, col=1
        )

        fig.update_layout(
            title=title,
            height=900,
            showlegend=True
        )

        return fig

    def plot_budget_optimization(self,
                                optimization_data: Dict,
                                title: str = "预算优化分配") -> go.Figure:
        """
        绘制预算优化图

        Parameters:
        - optimization_data: 优化数据
        - title: 图表标题

        Returns:
        - 预算优化图
        """
        allocations = optimization_data['allocations']
        campaigns = list(allocations.keys())
        budgets = [allocations[c]['分配预算'] for c in campaigns]
        expected_returns = [allocations[c]['预期回报'] for c in campaigns]

        fig = make_subplots(
            rows=1, cols=2,
            subplot_titles=('预算分配', '预期回报'),
            specs=[[{"type": "pie"}, {"type": "bar"}]]
        )

        # 1. 预算分配饼图
        fig.add_trace(
            go.Pie(labels=campaigns, values=budgets, name="预算分配"),
            row=1, col=1
        )

        # 2. 预期回报柱状图
        fig.add_trace(
            go.Bar(x=campaigns, y=expected_returns, name='预期回报', marker_color='lightblue'),
            row=1, col=2
        )

        fig.update_layout(
            title=f"{title}<br><sup>总预期回报: {optimization_data.get('total_expected_return', 0):,.0f}</sup>",
            height=500
        )

        return fig

    def create_growth_dashboard(self,
                               data_dict: Dict,
                               title: str = "增长分析综合仪表板") -> go.Figure:
        """
        创建综合增长分析仪表板

        Parameters:
        - data_dict: 包含所有分析数据的字典
        - title: 仪表板标题

        Returns:
        - 综合仪表板
        """
        fig = make_subplots(
            rows=3, cols=3,
            subplot_titles=(
                '转化漏斗', 'RFM分群', 'ROI对比',
                'Qini曲线', '增量分布', '增长趋势',
                '预算分配', '活动效果', '关键指标'
            ),
            specs=[
                [{"type": "scatter"}, {"type": "scatter"}, {"type": "bar"}],
                [{"type": "scatter"}, {"type": "histogram"}, {"type": "scatter"}],
                [{"type": "pie"}, {"type": "bar"}, {"type": "indicator"}]
            ]
        )

        # 根据可用数据添加图表
        # 这里简化处理，实际使用时需要根据具体数据结构调整

        fig.update_layout(
            title=title,
            height=1200,
            showlegend=False
        )

        return fig

    def export_charts(self, figs: List[go.Figure], output_dir: str = "charts"):
        """
        导出图表到文件

        Parameters:
        - figs: 图表列表
        - output_dir: 输出目录
        """
        import os
        os.makedirs(output_dir, exist_ok=True)

        for i, fig in enumerate(figs):
            filename = f"chart_{i+1}.html"
            filepath = os.path.join(output_dir, filename)
            fig.write_html(filepath)

        print(f"✅ 已导出 {len(figs)} 个图表到 {output_dir} 目录")

    def set_chinese_font(self):
        """设置中文字体支持"""
        import matplotlib.pyplot as plt
        plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans']
        plt.rcParams['axes.unicode_minus'] = False

"""
ROI分析器模块 - 增长模型成本效益分析

提供全面的ROI分析功能，包括：
- 营销活动ROI计算
- 成本效益分析
- 预算分配优化
- LTV预测
- 增长策略投资回报评估
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Tuple, Optional, Union
from scipy.optimize import minimize
import matplotlib.pyplot as plt
import seaborn as sns
import warnings
warnings.filterwarnings('ignore')


class ROIAnalyzer:
    """ROI分析器 - 成本效益与预算优化核心"""

    def __init__(self, config: Optional[Dict] = None):
        """
        初始化ROI分析器

        Parameters:
        - config: 配置参数字典
        """
        self.config = config or {}
        self.results = {}

    def calculate_campaign_roi(self,
                             data: pd.DataFrame,
                             campaign_col: str = '裂变类型',
                             conversion_col: str = '是否转化',
                             cost_col: str = '成本',
                             revenue_col: str = '收入',
                             user_col: str = '用户码') -> Dict:
        """
        计算营销活动ROI

        Parameters:
        - data: 营销数据
        - campaign_col: 活动类型列名
        - conversion_col: 转化结果列名
        - cost_col: 成本列名
        - revenue_col: 收入列名
        - user_col: 用户标识列名

        Returns:
        - ROI分析结果
        """
        results = {}

        # 按活动分组计算指标
        campaign_metrics = data.groupby(campaign_col).agg({
            user_col: 'nunique',        # 触达用户数
            conversion_col: 'sum',       # 转化数
            cost_col: 'sum',            # 总成本
            revenue_col: 'sum'          # 总收入
        }).round(4)

        campaign_metrics.columns = ['触达用户数', '转化数', '总成本', '总收入']
        campaign_metrics['转化率'] = campaign_metrics['转化数'] / campaign_metrics['触达用户数']
        campaign_metrics['ARPU'] = campaign_metrics['总收入'] / campaign_metrics['触达用户数']  # 每用户平均收入
        campaign_metrics['CPA'] = campaign_metrics['总成本'] / campaign_metrics['转化数']  # 获客成本
        campaign_metrics['ROI'] = (campaign_metrics['总收入'] - campaign_metrics['总成本']) / campaign_metrics['总成本']  # ROI

        # 计算投资回收期
        campaign_metrics['回收期(天)'] = campaign_metrics['总成本'] / (campaign_metrics['总收入'] / 30)  # 假设按30天计算

        # 计算利润率
        campaign_metrics['利润率'] = (campaign_metrics['总收入'] - campaign_metrics['总成本']) / campaign_metrics['总收入']

        # LTV/CPA比率
        campaign_metrics['LTV_CPA_Ratio'] = campaign_metrics['ARPU'] * 12 / campaign_metrics['CPA']  # 假设年度LTV

        results['campaign_metrics'] = campaign_metrics.to_dict()
        results['overall_roi'] = (campaign_metrics['总收入'].sum() - campaign_metrics['总成本'].sum()) / campaign_metrics['总成本'].sum()
        results['best_campaign'] = campaign_metrics['ROI'].idxmax()
        results['worst_campaign'] = campaign_metrics['ROI'].idxmin()

        self.results['campaign_roi'] = results
        return results

    def calculate_ltv(self,
                     data: pd.DataFrame,
                     user_col: str = '用户码',
                     revenue_col: str = '收入',
                     frequency_col: str = 'F值',
                     recency_col: str = 'R值') -> pd.DataFrame:
        """
        计算用户生命周期价值(LTV)

        Parameters:
        - data: 用户数据
        - user_col: 用户标识列名
        - revenue_col: 收入列名
        - frequency_col: 频次指标列名
        - recency_col: 近度指标列名

        Returns:
        - 包含LTV的DataFrame
        """
        # 用户历史价值分析
        user_metrics = data.groupby(user_col).agg({
            revenue_col: ['sum', 'mean'],
            frequency_col: 'sum',
            recency_col: 'min'
        }).round(4)

        user_metrics.columns = ['总收入', '平均收入', '总频次', '最近消费']

        # 计算基础LTV指标
        user_metrics['平均订单价值'] = user_metrics['总收入'] / user_metrics['总频次']
        user_metrics['购买频率'] = user_metrics['总频次']

        # 简单LTV预测 (基于历史数据)
        user_metrics['预测LTV'] = user_metrics['平均订单价值'] * user_metrics['购买频率'] * 12  # 年度预测

        # 考虑客户流失的LTV调整
        churn_rate = 1 / (user_metrics['最近消费'].mean() / 30 + 1)  # 简化的流失率估算
        user_metrics['调整LTV'] = user_metrics['预测LTV'] * (1 - churn_rate)

        # LTV分群
        ltv_quantiles = user_metrics['调整LTV'].quantile([0.2, 0.4, 0.6, 0.8])

        def classify_ltv(ltv):
            if ltv <= ltv_quantiles[0.2]:
                return '低价值'
            elif ltv <= ltv_quantiles[0.4]:
                return '中低价值'
            elif ltv <= ltv_quantiles[0.6]:
                return '中等价值'
            elif ltv <= ltv_quantiles[0.8]:
                return '中高价值'
            else:
                return '高价值'

        user_metrics['LTV分群'] = user_metrics['调整LTV'].apply(classify_ltv)

        self.results['ltv_analysis'] = user_metrics
        return user_metrics

    def optimize_budget_allocation(self,
                                  campaigns_data: Dict[str, Dict],
                                  total_budget: float,
                                  min_allocation: float = 0.05) -> Dict:
        """
        优化预算分配

        Parameters:
        - campaigns_data: 活动数据字典 {活动名: {roi, reach, cost, capacity}}
        - total_budget: 总预算
        - min_allocation: 最小分配比例

        Returns:
        - 预算优化结果
        """
        campaigns = list(campaigns_data.keys())
        n_campaigns = len(campaigns)

        # 目标函数：最大化总回报
        def objective(allocations):
            total_return = 0
            for i, campaign in enumerate(campaigns):
                allocation = allocations[i]
                campaign_data = campaigns_data[campaign]

                # 使用边际效益递减模型
                base_roi = campaign_data.get('roi', 0.1)
                capacity = campaign_data.get('capacity', total_budget)

                # 边际效益递减函数
                marginal_return = base_roi * allocation * (1 - allocation / (2 * capacity))
                total_return += marginal_return

            return -total_return  # 负号因为要最大化

        # 约束条件
        constraints = [
            {'type': 'eq', 'fun': lambda x: sum(x) - total_budget},  # 总预算约束
        ]

        # 边界约束
        bounds = [(total_budget * min_allocation, total_budget * 0.6) for _ in range(n_campaigns)]

        # 初始猜测 (均匀分配)
        initial_guess = [total_budget / n_campaigns] * n_campaigns

        # 优化
        result = minimize(
            objective,
            initial_guess,
            method='SLSQP',
            bounds=bounds,
            constraints=constraints
        )

        if result.success:
            optimized_allocations = result.x
        else:
            # 如果优化失败，使用均匀分配
            optimized_allocations = initial_guess

        # 生成结果
        allocation_results = {}
        for i, campaign in enumerate(campaigns):
            allocation = optimized_allocations[i]
            campaign_data = campaigns_data[campaign]

            allocation_results[campaign] = {
                '分配预算': allocation,
                '分配比例': allocation / total_budget,
                '预期回报': allocation * campaign_data.get('roi', 0.1),
                '预期触达': allocation / campaign_data.get('cost', 1),
                '分配比例提升': self._calculate_allocation_improvement(campaign, allocation, campaigns_data)
            }

        results = {
            'allocations': allocation_results,
            'total_expected_return': sum(r['预期回报'] for r in allocation_results.values()),
            'optimization_success': result.success,
            'optimization_message': result.message if not result.success else 'Optimization successful'
        }

        self.results['budget_optimization'] = results
        return results

    def calculate_marginal_roi(self,
                              data: pd.DataFrame,
                              campaign_col: str = '裂变类型',
                              spend_col: str = '成本',
                              return_col: str = '收入') -> Dict:
        """
        计算边际ROI

        Parameters:
        - data: 营销数据
        - campaign_col: 活动类型列名
        - spend_col: 投入列名
        - return_col: 回报列名

        Returns:
        - 边际ROI分析结果
        """
        marginal_results = {}

        for campaign in data[campaign_col].unique():
            campaign_data = data[data[campaign_col] == campaign].copy()

            if len(campaign_data) < 10:
                continue

            # 按投入排序，计算边际效益
            campaign_data = campaign_data.sort_values(spend_col)
            campaign_data['累计投入'] = campaign_data[spend_col].cumsum()
            campaign_data['累计回报'] = campaign_data[return_col].cumsum()
            campaign_data['边际投入'] = campaign_data[spend_col].diff()
            campaign_data['边际回报'] = campaign_data[return_col].diff()

            # 计算边际ROI
            campaign_data['边际ROI'] = campaign_data['边际回报'] / campaign_data['边际投入']

            # 计算平均边际ROI (排除异常值)
            valid_marginal_roi = campaign_data['边际ROI'].dropna()
            valid_marginal_roi = valid_marginal_roi[valid_marginal_roi.abs() < 10]  # 过滤极端值

            if len(valid_marginal_roi) > 0:
                marginal_results[campaign] = {
                    '平均边际ROI': valid_marginal_roi.mean(),
                    '边际ROI标准差': valid_marginal_roi.std(),
                    '边际ROI中位数': valid_marginal_roi.median(),
                    '最佳投入点': campaign_data.loc[valid_marginal_roi.idxmax(), spend_col] if len(valid_marginal_roi) > 0 else 0,
                    '投入回报弹性': self._calculate_roi_elasticity(campaign_data)
                }

        self.results['marginal_roi'] = marginal_results
        return marginal_results

    def simulate_roi_scenario(self,
                             base_data: Dict,
                             scenarios: List[Dict]) -> Dict:
        """
        ROI情景模拟

        Parameters:
        - base_data: 基础数据
        - scenarios: 情景列表

        Returns:
        - 情景模拟结果
        """
        simulation_results = {}

        for i, scenario in enumerate(scenarios):
            scenario_name = scenario.get('name', f'Scenario_{i+1}')

            # 应用情景参数
            modified_data = self._apply_scenario_parameters(base_data, scenario)

            # 计算情景ROI
            scenario_roi = self._calculate_scenario_roi(modified_data)

            simulation_results[scenario_name] = {
                'scenario_parameters': scenario,
                'roi_result': scenario_roi,
                'roi_vs_base': scenario_roi - base_data.get('base_roi', 0),
                'improvement_percentage': (scenario_roi - base_data.get('base_roi', 0)) / base_data.get('base_roi', 1) * 100
            }

        self.results['roi_simulation'] = simulation_results
        return simulation_results

    def analyze_roi_trend(self,
                         data: pd.DataFrame,
                         time_col: str = '时间',
                         revenue_col: str = '收入',
                         cost_col: str = '成本') -> Dict:
        """
        分析ROI趋势

        Parameters:
        - data: 时间序列数据
        - time_col: 时间列名
        - revenue_col: 收入列名
        - cost_col: 成本列名

        Returns:
        - ROI趋势分析结果
        """
        # 确保时间列格式正确
        if not pd.api.types.is_datetime64_any_dtype(data[time_col]):
            data[time_col] = pd.to_datetime(data[time_col])

        # 按时间分组计算ROI
        time_series = data.groupby(data[time_col].dt.to_period('M')).agg({
            revenue_col: 'sum',
            cost_col: 'sum'
        })

        time_series['ROI'] = (time_series[revenue_col] - time_series[cost_col]) / time_series[cost_col]
        time_series['累计ROI'] = ((time_series[revenue_col].cumsum() - time_series[cost_col].cumsum()) /
                                time_series[cost_col].cumsum())

        # 计算趋势
        time_series['ROI_MoM'] = time_series['ROI'].pct_change() * 100  # 月环比
        time_series['ROI_3MAvg'] = time_series['ROI'].rolling(window=3).mean()  # 3月移动平均

        # 趋势分析
        trend_analysis = {
            'overall_trend': '上升' if time_series['ROI'].iloc[-1] > time_series['ROI'].iloc[0] else '下降',
            'average_roi': time_series['ROI'].mean(),
            'roi_volatility': time_series['ROI'].std(),
            'best_period': time_series['ROI'].idxmax(),
            'worst_period': time_series['ROI'].idxmin(),
            'trend_strength': self._calculate_trend_strength(time_series['ROI'])
        }

        results = {
            'time_series': time_series.to_dict(),
            'trend_analysis': trend_analysis
        }

        self.results['roi_trend'] = results
        return results

    def generate_roi_report(self) -> Dict:
        """
        生成ROI分析报告

        Returns:
        - 综合ROI分析报告
        """
        if not self.results:
            return {"error": "请先进行分析以生成结果"}

        report = {
            "summary": {},
            "key_findings": [],
            "recommendations": [],
            "budget_optimization": {}
        }

        # 营销活动ROI总结
        if 'campaign_roi' in self.results:
            roi_results = self.results['campaign_roi']
            report["summary"]["overall_roi"] = roi_results['overall_roi']
            report["summary"]["best_campaign"] = roi_results['best_campaign']
            report["summary"]["campaign_count"] = len(roi_results['campaign_metrics'])

            # 关键发现
            if roi_results['overall_roi'] > 0.5:
                report["key_findings"].append("整体营销ROI表现优秀")
            elif roi_results['overall_roi'] > 0.2:
                report["key_findings"].append("整体营销ROI表现良好")
            else:
                report["key_findings"].append("营销ROI需要改善")

            report["recommendations"].append(f"重点关注{roi_results['best_campaign']}活动的成功经验")

        # 预算优化建议
        if 'budget_optimization' in self.results:
            optimization = self.results['budget_optimization']
            report["budget_optimization"] = optimization['allocations']
            report["recommendations"].append("根据ROI优化结果调整预算分配")

        # LTV分析洞察
        if 'ltv_analysis' in self.results:
            ltv_data = self.results['ltv_analysis']
            high_value_users = len(ltv_data[ltv_data['LTV分群'] == '高价值'])
            total_users = len(ltv_data)
            high_value_ratio = high_value_users / total_users

            report["key_findings"].append(f"高价值用户占比: {high_value_ratio:.1%}")
            if high_value_ratio < 0.2:
                report["recommendations"].append("加强高价值用户识别和维护策略")

        return report

    def plot_roi_comparison(self, roi_data: Dict, title: str = "ROI对比分析"):
        """
        绘制ROI对比图

        Parameters:
        - roi_data: ROI数据
        - title: 图表标题
        """
        if isinstance(roi_data, dict) and 'campaign_metrics' in roi_data:
            metrics_df = pd.DataFrame(roi_data['campaign_metrics']).T
        else:
            metrics_df = pd.DataFrame(roi_data).T

        fig, axes = plt.subplots(2, 2, figsize=(15, 12))
        fig.suptitle(title, fontsize=16)

        # ROI对比
        metrics_df['ROI'].sort_values().plot(kind='barh', ax=axes[0,0], color='skyblue')
        axes[0,0].set_title('各活动ROI对比')
        axes[0,0].set_xlabel('ROI')

        # 转化率对比
        metrics_df['转化率'].sort_values().plot(kind='barh', ax=axes[0,1], color='lightgreen')
        axes[0,1].set_title('各活动转化率对比')
        axes[0,1].set_xlabel('转化率')

        # CPA对比
        metrics_df['CPA'].sort_values().plot(kind='barh', ax=axes[1,0], color='lightcoral')
        axes[1,0].set_title('各活动获客成本对比')
        axes[1,0].set_xlabel('CPA')

        # 触达用户数对比
        metrics_df['触达用户数'].sort_values().plot(kind='barh', ax=axes[1,1], color='gold')
        axes[1,1].set_title('各活动触达用户数对比')
        axes[1,1].set_xlabel('触达用户数')

        plt.tight_layout()
        return fig

    def _calculate_allocation_improvement(self, campaign: str, allocation: float, campaigns_data: Dict) -> float:
        """计算分配比例改善情况"""
        current_roi = campaigns_data[campaign].get('roi', 0.1)
        total_current_roi = sum(c.get('roi', 0.1) for c in campaigns_data.values())

        if total_current_roi > 0:
            proportional_allocation = allocation / sum(campaigns_data[c].get('cost', 1) for c in campaigns_data)
            roi_based_allocation = current_roi / total_current_roi

            return (proportional_allocation - roi_based_allocation) / roi_based_allocation
        return 0

    def _calculate_roi_elasticity(self, campaign_data: pd.DataFrame) -> float:
        """计算ROI弹性"""
        if len(campaign_data) < 10:
            return 0

        # 简化的弹性计算
        spend_change = campaign_data['累计投入'].pct_change().dropna()
        return_change = campaign_data['累计回报'].pct_change().dropna()

        if len(spend_change) > 1 and len(return_change) > 1:
            correlation = np.corrcoef(spend_change[1:], return_change[1:])[0, 1]
            return correlation if not np.isnan(correlation) else 0
        return 0

    def _apply_scenario_parameters(self, base_data: Dict, scenario: Dict) -> Dict:
        """应用情景参数到基础数据"""
        modified_data = base_data.copy()

        for key, value in scenario.items():
            if key in ['cost_reduction', 'revenue_increase', 'conversion_improvement']:
                if key == 'cost_reduction':
                    modified_data['total_cost'] = modified_data.get('total_cost', 0) * (1 - value)
                elif key == 'revenue_increase':
                    modified_data['total_revenue'] = modified_data.get('total_revenue', 0) * (1 + value)
                elif key == 'conversion_improvement':
                    modified_data['conversion_rate'] = modified_data.get('conversion_rate', 0) * (1 + value)

        return modified_data

    def _calculate_scenario_roi(self, data: Dict) -> float:
        """计算情景ROI"""
        revenue = data.get('total_revenue', 0)
        cost = data.get('total_cost', 1)
        return (revenue - cost) / cost if cost > 0 else 0

    def _calculate_trend_strength(self, roi_series: pd.Series) -> str:
        """计算趋势强度"""
        if len(roi_series) < 3:
            return "数据不足"

        # 简单线性回归计算趋势
        x = np.arange(len(roi_series))
        slope, _ = np.polyfit(x, roi_series, 1)

        if abs(slope) < 0.001:
            return "平稳"
        elif abs(slope) < 0.01:
            return "温和"
        else:
            return "强劲"