Python数据统计案例如何计算数据指标

本文目录导读：

1. 数据准备
2. 描述性统计指标
3. 变异系数和离散程度
4. 相关性分析
5. 异常值检测指标
6. 趋势和季节性指标
7. 汇总统计报告
8. 可视化展示

我来分享一个完整的Python数据统计案例,涵盖常用数据指标的计算方法。

数据准备

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats
# 创建示例数据
np.random.seed(42)
# 生成销售额数据
data = {
    '日期': pd.date_range('2023-01-01', periods=100, freq='D'),
    '销售额': np.random.normal(50000, 10000, 100),
    '客户数': np.random.poisson(200, 100),
    '客单价': np.random.gamma(10, 50, 100),
    '退货率': np.random.beta(2, 20, 100)
}
df = pd.DataFrame(data)
# 添加一些异常值
df.loc[10, '销售额'] = 150000  # 异常高值
df.loc[15, '客单价'] = 0       # 异常低值
print("数据预览:")
print(df.head())
print(f"\n数据形状: {df.shape}")

描述性统计指标

def calculate_descriptive_stats(dataframe, columns):
    """
    计算描述性统计指标
    """
    stats_dict = {}
    for col in columns:
        series = dataframe[col]
        stats_dict[col] = {
            '计数': len(series),
            '均值': np.mean(series),
            '中位数': np.median(series),
            '众数': series.mode().iloc[0] if not series.mode().empty else None,
            '标准差': np.std(series),
            '方差': np.var(series),
            '最小值': np.min(series),
            '最大值': np.max(series),
            '范围': np.max(series) - np.min(series),
            '四分位距': np.percentile(series, 75) - np.percentile(series, 25),
            '25%分位数': np.percentile(series, 25),
            '50%分位数': np.percentile(series, 50),
            '75%分位数': np.percentile(series, 75),
            '偏度': series.skew(),
            '峰度': series.kurtosis()
        }
    return pd.DataFrame(stats_dict)
# 计算描述性统计
numerical_cols = ['销售额', '客户数', '客单价', '退货率']
descriptive_stats = calculate_descriptive_stats(df, numerical_cols)
print("描述性统计指标:")
print(descriptive_stats)

变异系数和离散程度

def calculate_dispersion_metrics(dataframe, columns):
    """
    计算离散程度指标
    """
    metrics = {}
    for col in columns:
        series = dataframe[col]
        mean = np.mean(series)
        std = np.std(series)
        metrics[col] = {
            '变异系数(%)': (std / mean) * 100 if mean != 0 else 0,
            '平均绝对偏差': np.mean(np.abs(series - mean)),
            '标准差': std,
            '均方根误差': np.sqrt(np.mean(series**2)),
            '离散系数': std / abs(mean) if mean != 0 else 0
        }
    return pd.DataFrame(metrics)
dispersion_metrics = calculate_dispersion_metrics(df, numerical_cols)
print("\n离散程度指标:")
print(dispersion_metrics)

相关性分析

def calculate_correlation_metrics(dataframe, columns):
    """
    计算相关性指标
    """
    # 皮尔逊相关系数
    pearson_corr = dataframe[columns].corr(method='pearson')
    # 斯皮尔曼秩相关系数
    spearman_corr = dataframe[columns].corr(method='spearman')
    # 肯德尔秩相关系数
    kendall_corr = dataframe[columns].corr(method='kendall')
    # 协方差矩阵
    cov_matrix = dataframe[columns].cov()
    return {
        'pearson': pearson_corr,
        'spearman': spearman_corr,
        'kendall': kendall_corr,
        'covariance': cov_matrix
    }
correlation_results = calculate_correlation_metrics(df, numerical_cols)
print("皮尔逊相关系数矩阵:")
print(correlation_results['pearson'])

异常值检测指标

def detect_outliers(dataframe, columns, method='zscore'):
    """
    检测异常值
    """
    outlier_results = {}
    for col in columns:
        series = dataframe[col]
        if method == 'zscore':
            # Z-score方法
            z_scores = np.abs(stats.zscore(series))
            outlier_mask = z_scores > 3
            outliers = series[outlier_mask]
        elif method == 'iqr':
            # IQR方法
            Q1 = series.quantile(0.25)
            Q3 = series.quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            outlier_mask = (series < lower_bound) | (series > upper_bound)
            outliers = series[outlier_mask]
        outlier_results[col] = {
            '异常值数量': len(outliers),
            '异常值比例(%)': len(outliers) / len(series) * 100,
            '异常值': list(outliers.values[:5]) if len(outliers) > 0 else [],
            '阈值': (lower_bound if method == 'iqr' else -3, 
                    upper_bound if method == 'iqr' else 3)
        }
    return outlier_results
outlier_results = detect_outliers(df, numerical_cols, method='iqr')
print("\n异常值检测结果 (IQR方法):")
for col, result in outlier_results.items():
    print(f"{col}: {result}")

趋势和季节性指标

def calculate_trend_metrics(dataframe, date_col, value_col):
    """
    计算时间序列趋势指标
    """
    df_temp = dataframe.copy()
    df_temp = df_temp.set_index(date_col)
    # 移动平均
    rolling_mean_7 = df_temp[value_col].rolling(window=7).mean()
    rolling_mean_30 = df_temp[value_col].rolling(window=30).mean()
    # 增长率
    df_temp['增长率'] = df_temp[value_col].pct_change() * 100
    # 累计值
    df_temp['累计值'] = df_temp[value_col].cumsum()
    # 最大回撤
    df_temp['累计最大值'] = df_temp[value_col].cummax()
    df_temp['回撤'] = (df_temp[value_col] - df_temp['累计最大值']) / df_temp['累计最大值'] * 100
    max_drawdown = df_temp['回撤'].min()
    # 变异系数（衡量波动性）
    cv = (df_temp[value_col].std() / df_temp[value_col].mean()) * 100
    metrics = {
        '总趋势': '上升' if df_temp[value_col].iloc[-1] > df_temp[value_col].iloc[0] else '下降',
        '总增长率(%)': df_temp['增长率'].mean(),
        '最大增长率(%)': df_temp['增长率'].max(),
        '最小增长率(%)': df_temp['增长率'].min(),
        '波动率(CV)(%)': cv,
        '最大回撤(%)': max_drawdown,
        '7日移动平均(': rolling_mean_7.iloc[-1],
        '30日移动平均(': rolling_mean_30.iloc[-1]
    }
    return pd.DataFrame([metrics])
trend_results = calculate_trend_metrics(df, '日期', '销售额')
print("\n趋势分析指标:")
print(trend_results.to_string(index=False))

汇总统计报告

def generate_summary_report(dataframe):
    """
    生成综合统计报告
    """
    report = {}
    # 基本统计
    report['基本统计'] = dataframe.describe()
    # 缺失值情况
    report['缺失值统计'] = dataframe.isnull().sum()
    # 数据类型统计
    report['数据类型'] = dataframe.dtypes
    # 独特值统计
    categorical_cols = dataframe.select_dtypes(include=['object']).columns
    report['独特值统计'] = {col: dataframe[col].nunique() for col in categorical_cols}
    # 内存使用
    report['内存使用(MB)'] = dataframe.memory_usage(deep=True).sum() / 1024 / 1024
    return report
summary_report = generate_summary_report(df)
print("\n=== 综合统计报告 ===")
for key, value in summary_report.items():
    print(f"\n{key}:")
    if isinstance(value, pd.DataFrame):
        print(value)
    elif isinstance(value, dict):
        for k, v in value.items():
            print(f"  {k}: {v}")
    else:
        print(f"  {value:.2f}")

可视化展示

def plot_statistical_summary(dataframe, columns):
    """
    可视化统计指标
    """
    fig, axes = plt.subplots(2, 2, figsize=(12, 10))
    for idx, col in enumerate(columns[:4]):
        row = idx // 2
        col_idx = idx % 2
        # 直方图和密度图
        dataframe[col].hist(bins=30, ax=axes[row, col_idx], alpha=0.7, density=True)
        dataframe[col].plot.kde(ax=axes[row, col_idx], color='red')
        # 添加统计信息
        mean = dataframe[col].mean()
        median = dataframe[col].median()
        std = dataframe[col].std()
        axes[row, col_idx].axvline(mean, color='green', linestyle='--', label=f'均值={mean:.2f}')
        axes[row, col_idx].axvline(median, color='blue', linestyle='--', label=f'中位数={median:.2f}')
        axes[row, col_idx].axvline(mean + std, color='orange', linestyle=':', label=f'±1σ')
        axes[row, col_idx].axvline(mean - std, color='orange', linestyle=':')
        axes[row, col_idx].set_title(f'{col} 分布')
        axes[row, col_idx].legend()
    plt.tight_layout()
    plt.show()
# 数据可视化
plot_statistical_summary(df, numerical_cols)

这个案例涵盖了：

• 中心趋势指标：均值、中位数、众数
• 离散程度指标：标准差、方差、变异系数
• 分布形态指标：偏度、峰度
• 相关性指标：皮尔逊、斯皮尔曼相关系数
• 异常值检测：Z-score、IQR方法
• 时间序列指标：趋势、增长率、波动性

你可以根据具体需求选择适用的指标和方法。