data-report/generate-data-report-ppt/scripts/metrics_calculator.py

"""
Metrics calculation engine for the universal data report generator.
Provides generic metric computation, trend analysis, distribution, ranking, and insights.
"""
import pandas as pd


def _pct_change(curr, prev):
    """Calculate percentage change. Returns None if previous base is 0 or None."""
    if prev is None or prev == 0:
        return None
    return round((curr - prev) / prev * 100, 1)


def _safe_div(a, b):
    return round(a / b, 1) if b else 0


# ==============================================================================
# HELPER FUNCTIONS
# ==============================================================================

def _fmt_pct(val):
    if val is None:
        return '—'
    sign = '+' if val >= 0 else ''
    return f'{sign}{val:.1f}%'


def _fmt_chg_dir(val):
    if val is None:
        return ''
    return '增加' if val >= 0 else '减少'


def avg(lst):
    return sum(lst) / len(lst) if lst else 0


# ==============================================================================
# GENERIC METRICS
# ==============================================================================

def calc_generic_metrics(df: pd.DataFrame, config) -> dict:
    metrics = {}

    for metric_def in config.metrics:
        col = metric_def.column
        if col not in df.columns:
            metrics[metric_def.name] = 0
            continue

        series = df[col].dropna()
        agg = metric_def.aggregation

        if not pd.api.types.is_numeric_dtype(series):
            coerced = pd.to_numeric(series, errors='coerce').dropna()
            if len(coerced) > 0:
                series = coerced

        if agg == 'sum':
            val = int(series.sum()) if pd.api.types.is_numeric_dtype(series) else len(series)
        elif agg == 'count':
            val = int(series.count())
        elif agg == 'avg':
            val = round(float(series.mean()), 1) if pd.api.types.is_numeric_dtype(series) else 0
        elif agg == 'max':
            val = round(float(series.max()), 1) if pd.api.types.is_numeric_dtype(series) else 0
        elif agg == 'min':
            val = round(float(series.min()), 1) if pd.api.types.is_numeric_dtype(series) else 0
        elif agg == 'distinct_count':
            val = int(series.nunique())
        else:
            val = len(series)

        metrics[metric_def.name] = val
        metrics[f'{metric_def.name}_label'] = metric_def.label
        metrics[f'{metric_def.name}_unit'] = metric_def.unit

    if hasattr(config, 'comparison') and config.comparison:
        pass

    return metrics


def calc_generic_trend(df: pd.DataFrame, time_col: str, metric_col: str,
                       aggregation: str = 'sum') -> dict:
    if time_col not in df.columns or metric_col not in df.columns:
        return {}

    if aggregation == 'sum':
        trend = df.groupby(time_col)[metric_col].sum().sort_index()
    elif aggregation == 'count':
        trend = df.groupby(time_col)[metric_col].count().sort_index()
    else:
        trend = df.groupby(time_col)[metric_col].mean().sort_index()

    dates = []
    for d in trend.index:
        try:
            dates.append(pd.Timestamp(d).strftime('%m/%d'))
        except Exception:
            dates.append(str(d))

    return {
        'dates': dates,
        'values': [int(v) if aggregation != 'avg' else round(float(v), 1) for v in trend.values],
    }


def calc_generic_distribution(df: pd.DataFrame, cat_col: str, metric_col: str = None,
                              aggregation: str = 'sum', top_n: int = 10) -> dict:
    if cat_col not in df.columns:
        return {}

    if metric_col and metric_col in df.columns:
        if aggregation == 'sum':
            dist = df.groupby(cat_col)[metric_col].sum().sort_values(ascending=False).head(top_n)
        elif aggregation == 'count':
            dist = df.groupby(cat_col)[metric_col].count().sort_values(ascending=False).head(top_n)
        else:
            dist = df.groupby(cat_col)[metric_col].mean().sort_values(ascending=False).head(top_n)
    else:
        dist = df[cat_col].value_counts().head(top_n)

    total = sum(dist.values)
    return {
        'categories': [str(k) for k in dist.index],
        'values': [int(v) for v in dist.values],
        'percentages': [round(v / total * 100, 1) if total else 0 for v in dist.values],
    }


def calc_generic_ranking(df: pd.DataFrame, rank_col: str, metric_col: str,
                         aggregation: str = 'sum', top_n: int = 15) -> list[dict]:
    if rank_col not in df.columns or metric_col not in df.columns:
        return []

    if aggregation == 'sum':
        ranked = df.groupby(rank_col)[metric_col].sum().sort_values(ascending=False).head(top_n)
    elif aggregation == 'count':
        ranked = df.groupby(rank_col)[metric_col].count().sort_values(ascending=False).head(top_n)
    else:
        ranked = df.groupby(rank_col)[metric_col].mean().sort_values(ascending=False).head(top_n)

    return [{'name': str(k), 'value': int(v), 'rank': i + 1}
            for i, (k, v) in enumerate(ranked.items())]


def generate_generic_insights(data_profile: dict, metrics: dict) -> list[dict]:
    items = []

    num_cols = data_profile.get('numeric_columns', [])
    cat_cols = data_profile.get('category_columns', [])
    time_cols = data_profile.get('time_columns', [])
    q = data_profile.get('data_quality', {})
    score = q.get('score', 100)

    if metrics:
        metric_details = []
        for k, v in metrics.items():
            if isinstance(v, (int, float)):
                metric_details.append(f'{k}: {v:,.0f}')
        if metric_details:
            items.append({
                'title': '核心指标总览',
                'content': f'本期关键指标：{"；".join(metric_details[:6])}。'
                           f'综合来看，业务运行态势可通过这些核心数据进行量化评估，'
                           f'建议结合业务目标与实际值的差距进行针对性分析。',
            })

    if num_cols:
        for nc in num_cols[:2]:
            ns = nc.get('numeric_stats', {}) or {}
            col_name = nc.get('inferred_label', nc['column_name'])
            stats_parts = []
            if 'sum' in ns and ns['sum']:
                stats_parts.append(f'总量 {ns["sum"]:,.0f}')
            if 'mean' in ns and ns['mean']:
                stats_parts.append(f'均值 {ns["mean"]:,.1f}')
            if 'max' in ns and ns['max']:
                stats_parts.append(f'峰值 {ns["max"]:,.0f}')
            if 'min' in ns and ns['min']:
                stats_parts.append(f'最低 {ns["min"]:,.0f}')
            if stats_parts:
                items.append({
                    'title': f'{col_name}数据特征',
                    'content': f'指标"{col_name}"的统计特征：{"，".join(stats_parts)}。'
                               f'标准差 {ns.get("std", "N/A")}，数据波动幅度'
                               f'{"较大" if isinstance(ns.get("std"), (int,float)) and ns["std"] > ns.get("mean", 1) * 0.5 else "适中"}。',
                })

    if cat_cols:
        for cc in cat_cols[:2]:
            uc = cc.get('unique_count', 0)
            items.append({
                'title': f'{cc.get("inferred_label", cc["column_name"])}维度分析',
                'content': f'数据覆盖 {uc} 个不同的{cc.get("inferred_label", cc["column_name"])}类别，'
                           f'丰富的分类维度支持多角度交叉分析。'
                           f'建议重点关注主要类别的集中度与分布均衡性，'
                           f'识别高价值类别与低效类别之间的差异特征。',
            })

    if time_cols:
        tc = time_cols[0]
        items.append({
            'title': '时间维度覆盖',
            'content': f'数据包含时间列"{tc.get("inferred_label", tc["column_name"])}"，'
                       f'支持按时间维度进行趋势分析。通过对时间序列数据的分解，'
                       f'可识别周期性波动、趋势变化及异常时间节点，为预测与规划提供依据。',
        })

    items.append({
        'title': '数据质量评估',
        'content': f'数据质量评分 {score}/100，'
                   f'{"数据完整可靠，" if score >= 90 else "数据质量良好，建议关注缺失值" if score >= 80 else "数据需重点关注质量控制"}'
                   f'缺失率 {q.get("null_rate", 0)*100:.1f}%。'
                   f'本报告中的分析与图表均基于现有数据进行自动化生成，确保数据准确性。',
    })

    high_null = q.get('high_null_columns', [])
    if high_null:
        items.append({
            'title': '数据完整性说明',
            'content': f'以下列缺失值比例较高：{", ".join(high_null[:5])}。'
                       f'在分析涉及这些列时已进行空值排除处理，'
                       f'建议后续数据录入环节关注这些字段的完整填写，以提升分析精度。',
        })

    total_rows = data_profile.get('total_rows', 0)
    if total_rows:
        items.append({
            'title': '数据规模概述',
            'content': f'本期报告基于 {total_rows} 条数据记录进行分析，'
                       f'样本量{"充足，统计结果具有较好的代表性" if total_rows >= 100 else "适中，统计结果可作为参考" if total_rows >= 30 else "有限，分析结果仅供参考"}。',
        })

    return items


if __name__ == '__main__':
    import sys
    if len(sys.argv) > 1:
        from data_loader import load_generic_excel
        from data_profiler import profile_dataframe
        from agent_analyzer import analyze_and_recommend
        fp = sys.argv[1]
        df = load_generic_excel(fp)
        print(f"Loaded: {len(df)} rows x {len(df.columns)} cols")
        profile = profile_dataframe(df)
        recs = analyze_and_recommend(profile)
        print(f"Suggested {len(recs['suggested_metrics'])} metrics, "
              f"{len(recs['suggested_pages'])} pages")