dtm_python/app/services/spu_lifecycle_service.py

"""
SPU生命周期核心分析服务

负责执行生命周期分析的核心逻辑
"""
import pandas as pd
from typing import Dict, Any

from .lifecycle_analyzer_service import (
    get_dataframe_hash,
    enhanced_data_preprocessing,
    calculate_lifecycle_indicators,
    assess_lifecycle_completeness,
    calculate_hot_product_coefficient,
    MIN_DATA_DAYS
)
from .stage_divider_service import (
    intelligent_stage_division,
    analyze_incomplete_lifecycle_stages,
    determine_current_stage_intelligent,
    calculate_stage_statistics,
    LIFECYCLE_STAGES,
    MIN_STAGE_DURATION
)
from .cache_service import (
    get_redis_connection,
    generate_cache_key_with_prefix,
    save_to_cache,
    load_from_cache,
    validate_analysis_results
)
from .utils import convert_to_native_types, ensure_native_type

# 分析参数
HOT_PRODUCT_THRESHOLD = 0.6  # 爆款阈值


def analyze_spu_lifecycle(df: pd.DataFrame, filename: str = None) -> Dict[str, Any]:
    """
    执行SPU生命周期分析
    
    Args:
        df: 清洗后的数据DataFrame
        filename: 文件名（可选，用于缓存）
    
    Returns:
        分析结果字典
    """
    # 初始化Redis并尝试从缓存加载
    redis_client = get_redis_connection()
    df_hash = get_dataframe_hash(df)
    cache_key = generate_cache_key_with_prefix(df_hash, filename, 'spu_analysis')
    
    if redis_client:
        cached_results = load_from_cache(redis_client, cache_key)
        if cached_results and validate_analysis_results(cached_results):
            print("🎯 从缓存中加载分析结果")
            return cached_results
    
    results = {}
    print(f"📊 开始分析，共 {len(df):,} 条记录")
    
    # 按SPU和日期重采样
    daily_revenue = df.set_index('订单付款时间').groupby('SPU')['单笔销售额'].resample('D').sum()
    daily_quantity = df.set_index('订单付款时间').groupby('SPU')['购买数量'].resample('D').sum()
    
    spus = df['SPU'].unique()
    total_spus = len(spus)
    print(f"📊 识别到 {total_spus} 个SPU")
    
    hot_product_count = 0
    
    for idx, spu in enumerate(spus):
        if (idx + 1) % 10 == 0 or (idx + 1) == total_spus:
            print(f"📊 分析进度: {idx + 1}/{total_spus}")
        
        revenue_series = daily_revenue.get(spu, None)
        quantity_series = daily_quantity.get(spu, None)
        
        # 数据验证
        if revenue_series is None or quantity_series is None or len(revenue_series) < MIN_DATA_DAYS:
            results[spu] = _create_insufficient_data_result()
            continue
        
        # 去除无效数据
        valid_days = (revenue_series > 0) | (quantity_series > 0)
        revenue_series = revenue_series[valid_days]
        quantity_series = quantity_series[valid_days]
        
        if len(revenue_series) < MIN_DATA_DAYS:
            results[spu] = _create_insufficient_data_result()
            continue
        
        # 执行分析
        result_data = _analyze_single_spu(
            spu, revenue_series, quantity_series
        )
        
        if result_data['is_hot_product']:
            hot_product_count += 1
        
        results[spu] = result_data
    
    # 添加分析概要
    results['_analysis_summary_'] = _create_analysis_summary(
        results, total_spus, hot_product_count
    )
    
    # 转换数据类型
    results = convert_to_native_types(results)
    
    # 保存到缓存
    if redis_client:
        save_to_cache(redis_client, cache_key, results)
    
    return results


def _create_insufficient_data_result() -> Dict[str, Any]:
    """创建数据不足的结果"""
    return {
        'stage': '数据不足',
        'current_stage': '数据不足',
        'details': f'有效销售天数少于{MIN_DATA_DAYS}天',
        'is_complete': False,
        'hot_product_coefficient': 0,
        'is_hot_product': False
    }


def _analyze_single_spu(spu: str, revenue_series, quantity_series) -> Dict[str, Any]:
    """分析单个SPU"""
    # 数据预处理
    smoothed_revenue, smoothed_quantity = enhanced_data_preprocessing(
        revenue_series, quantity_series
    )
    
    # 计算指标
    indicators = calculate_lifecycle_indicators(smoothed_revenue, smoothed_quantity)
    is_complete, completeness_score, completion_details = assess_lifecycle_completeness(indicators)
    
    # 阶段划分
    stages, stage_boundaries, current_stage, details, next_stage_prediction = _divide_stages(
        is_complete, spu, revenue_series, quantity_series,
        smoothed_revenue, smoothed_quantity, indicators, completeness_score
    )
    
    # 确保边界包含日期
    for boundary in stage_boundaries:
        if 'date' not in boundary and 'index' in boundary:
            idx = boundary['index']
            if 0 <= idx < len(revenue_series):
                boundary['date'] = revenue_series.index[idx].isoformat()
    
    # 构建结果
    result_data = {
        'stage': '完整生命周期' if is_complete else '不完整生命周期',
        'details': details,
        'revenue_series': [ensure_native_type(v, decimal_places=2) for v in revenue_series.tolist()],
        'quantity_series': [ensure_native_type(v, decimal_places=2) for v in quantity_series.tolist()],
        'smoothed_revenue': [ensure_native_type(v, decimal_places=2) for v in smoothed_revenue.tolist()],
        'smoothed_quantity': [ensure_native_type(v, decimal_places=2) for v in smoothed_quantity.tolist()],
        'is_complete': bool(is_complete),
        'peak_revenue': ensure_native_type(indicators['revenue_peak'], decimal_places=2),
        'peak_quantity': ensure_native_type(indicators['quantity_peak'], decimal_places=2),
        'peak_revenue_date': revenue_series.index[indicators['revenue_peak_idx']].isoformat(),
        'peak_quantity_date': quantity_series.index[indicators['quantity_peak_idx']].isoformat(),
        'revenue_peak_idx': indicators['revenue_peak_idx'],
        'quantity_peak_idx': indicators['quantity_peak_idx'],
        'stages_map': stages,
        'stage_boundaries': stage_boundaries,
        'completeness_score': ensure_native_type(completeness_score, decimal_places=2),
        'completion_details': completion_details,
        'next_stage_prediction': next_stage_prediction if not is_complete else "",
        'date_series': [d.isoformat() for d in revenue_series.index],
        'current_stage': stages[-1] if is_complete and stages else current_stage,
        'total_revenue': ensure_native_type(sum(revenue_series.tolist()), decimal_places=2),
        'total_quantity': ensure_native_type(sum(quantity_series.tolist()), decimal_places=2)
    }
    
    # 计算阶段统计
    if stages:
        stage_statistics = calculate_stage_statistics(
            revenue_series.tolist(), quantity_series.tolist(),
            [d.isoformat() for d in revenue_series.index], stage_boundaries, stages
        )
        result_data['stage_statistics'] = stage_statistics
    else:
        result_data['stage_statistics'] = {}
    
    # 计算爆款系数
    hot_product_coefficient = ensure_native_type(
        calculate_hot_product_coefficient(result_data, indicators, len(revenue_series)), decimal_places=2
    )
    result_data['hot_product_coefficient'] = hot_product_coefficient
    result_data['is_hot_product'] = bool(hot_product_coefficient > HOT_PRODUCT_THRESHOLD)
    
    return result_data


def _divide_stages(is_complete, spu, revenue_series, quantity_series,
                   smoothed_revenue, smoothed_quantity, indicators, completeness_score):
    """执行阶段划分"""
    current_stage = '引入期'
    stage_boundaries = []
    stages = []
    details = ""
    next_stage_prediction = ""
    
    if is_complete:
        peak_time_revenue = revenue_series.index[indicators['revenue_peak_idx']]
        peak_time_quantity = quantity_series.index[indicators['quantity_peak_idx']]
        details = f"该SPU在 {peak_time_revenue.strftime('%Y-%m-%d')} 达到销售额峰值。完整性得分：{completeness_score}/100"
        
        stage_boundaries = intelligent_stage_division(
            smoothed_revenue, smoothed_quantity, indicators, revenue_series, MIN_STAGE_DURATION
        )
        
        if len(stage_boundaries) < 3:
            is_complete = False
            details += f" 但阶段划分不完整"
            stages = [LIFECYCLE_STAGES[0]] * len(revenue_series)
        else:
            stages = _build_stages_array(revenue_series, stage_boundaries)
            
            # 验证阶段持续时间
            stage_durations = {s: sum(1 for st in stages if st == s) for s in LIFECYCLE_STAGES}
            insufficient_stages = [s for s, d in stage_durations.items() 
                                  if d > 0 and d < MIN_STAGE_DURATION]
            if insufficient_stages:
                is_complete = False
                details += f" 某些阶段持续时间不足{MIN_STAGE_DURATION}天"
    
    if not is_complete:
        stage_boundaries = analyze_incomplete_lifecycle_stages(
            smoothed_revenue, smoothed_quantity, indicators, revenue_series
        )
        current_stage, stage_description, next_stage_prediction = determine_current_stage_intelligent(
            smoothed_revenue, smoothed_quantity, indicators, revenue_series
        )
        details = stage_description
        
        stages = _build_stages_array(revenue_series, stage_boundaries)
        
        # 只保留当前阶段及之前的阶段
        if current_stage in LIFECYCLE_STAGES:
            current_stage_index = LIFECYCLE_STAGES.index(current_stage)
            valid_stages = LIFECYCLE_STAGES[:current_stage_index+1]
            for i in range(len(stages)):
                if stages[i] not in valid_stages:
                    stages[i] = current_stage
            stage_boundaries = [b for b in stage_boundaries 
                               if b.get('type', '').split('→')[-1] in valid_stages]
    
    return stages, stage_boundaries, current_stage, details, next_stage_prediction


def _build_stages_array(revenue_series, stage_boundaries):
    """根据边界构建阶段数组"""
    stages = [LIFECYCLE_STAGES[0]] * len(revenue_series)
    for boundary in stage_boundaries:
        boundary_type = boundary.get('type', '')
        boundary_index = boundary.get('index', 0)
        if boundary_index < len(stages):
            if '引入期→成长期' in boundary_type:
                for i in range(boundary_index, len(stages)):
                    if stages[i] == '引入期':
                        stages[i] = '成长期'
            elif '成长期→成熟期' in boundary_type:
                for i in range(boundary_index, len(stages)):
                    if stages[i] == '成长期':
                        stages[i] = '成熟期'
            elif '成熟期→衰退期' in boundary_type:
                for i in range(boundary_index, len(stages)):
                    if stages[i] == '成熟期':
                        stages[i] = '衰退期'
    return stages


def _create_analysis_summary(results, total_spus, hot_product_count):
    """创建分析概要"""
    complete_spus = sum(1 for r in results.values() if r.get('is_complete', False))
    incomplete_spus = sum(1 for r in results.values() 
                         if not r.get('is_complete', True) and r.get('stage') != '数据不足')
    insufficient_spus = sum(1 for r in results.values() if r.get('stage') == '数据不足')
    
    return {
        'total_spus': total_spus,
        'complete_spus': complete_spus,
        'incomplete_spus': incomplete_spus,
        'insufficient_spus': insufficient_spus,
        'hot_product_count': hot_product_count,
        'hot_product_rate': hot_product_count / total_spus if total_spus > 0 else 0
    }