当前位置：首页 > news >正文

Python实现经验分布函数(EDF)的完整指南

news 2026/7/7 10:14:49

1. 经验分布函数基础概念

经验分布函数（Empirical Distribution Function, EDF）是统计学中用于描述样本数据分布特征的非参数方法。当我们在Python中处理实际数据时，EDF能够在不假设数据服从任何特定理论分布的情况下，直接基于观测样本构建分布模型。

EDF的定义非常简单直观：对于一个包含n个观测值的样本数据集，EDF在任意点x处的值等于样本中小于或等于x的观测值所占的比例。数学表达式为： F_n(x) = (number of elements ≤ x) / n

这个阶梯函数在每一个数据点处产生1/n的跳跃，形成典型的"阶梯状"图形。与理论分布函数相比，EDF完全由数据驱动，不需要对总体分布做任何假设，这使得它在探索性数据分析和非参数统计推断中具有独特优势。

注意：虽然EDF不依赖分布假设，但当样本量较小时，EDF的估计可能不够精确。一般建议在样本量n≥30时使用EDF进行统计推断。

2. Python中的EDF实现方法

2.1 使用NumPy和Matplotlib手动构建

最基础的EDF实现可以通过NumPy的排序和向量化操作完成：

import numpy as np import matplotlib.pyplot as plt def empirical_dist_func(sample): # 对样本进行排序 sorted_sample = np.sort(sample) # 计算累积概率 n = len(sorted_sample) y = np.arange(1, n+1) / n return sorted_sample, y # 生成随机样本 np.random.seed(42) sample = np.random.normal(0, 1, 100) # 计算EDF x_edf, y_edf = empirical_dist_func(sample) # 绘制EDF plt.step(x_edf, y_edf, where='post') plt.xlabel('Value') plt.ylabel('Cumulative Probability') plt.title('Empirical Distribution Function') plt.grid(True) plt.show()

这段代码的关键点在于：

np.sort()对样本进行排序，这是构建EDF的基础
np.arange()生成均匀间隔的累积概率值
plt.step()用阶梯图准确呈现EDF的离散特性

2.2 使用statsmodels库的高级功能

statsmodels库提供了更专业的EDF实现：

from statsmodels.distributions.empirical_distribution import ECDF import numpy as np # 生成样本数据 sample = np.random.exponential(scale=1, size=200) # 创建ECDF对象 ecdf = ECDF(sample) # 计算特定点的EDF值 print(ecdf(0.5)) # 输出x=0.5处的累积概率 print(ecdf(1.0)) # 输出x=1.0处的累积概率 # 获取EDF的全部x和y值 x = ecdf.x y = ecdf.y

statsmodels的ECDF类优势在于：

自动处理数据排序和概率计算
提供可调用对象，方便计算任意点的EDF值
内置支持绘制图形的方法

2.3 使用Pandas的快速实现

对于DataFrame数据，Pandas提供了便捷的EDF计算方法：

import pandas as pd import numpy as np # 创建示例DataFrame df = pd.DataFrame({ 'normal': np.random.normal(0, 1, 1000), 'exponential': np.random.exponential(1, 1000) }) # 计算每列的EDF for col in df.columns: sorted_data = df[col].sort_values() df[f'{col}_edf'] = np.arange(1, len(df)+1) / len(df)

这种方法特别适合处理多变量数据，可以一次性计算多个特征的EDF。

3. EDF的进阶应用场景

3.1 分布比较与拟合优度检验

EDF常用于比较样本分布与理论分布的差异。Kolmogorov-Smirnov检验就是基于EDF的典型方法：

from scipy import stats # 生成样本数据 sample = np.random.normal(0, 1, 100) # KS检验样本是否来自标准正态分布 ks_statistic, p_value = stats.kstest(sample, 'norm') print(f"KS统计量: {ks_statistic:.4f}") print(f"P值: {p_value:.4f}")

当p值大于显著性水平(如0.05)时，我们不能拒绝样本来自指定分布的假设。

3.2 非参数密度估计

结合EDF可以进行核密度估计(KDE)：

from sklearn.neighbors import KernelDensity # 生成样本数据 sample = np.random.normal(0, 1, 100).reshape(-1, 1) # 创建并拟合KDE模型 kde = KernelDensity(kernel='gaussian', bandwidth=0.5).fit(sample) # 生成测试点 x_test = np.linspace(-3, 3, 1000).reshape(-1, 1) # 计算对数密度 log_dens = kde.score_samples(x_test) # 绘制结果 plt.fill_between(x_test.flatten(), np.exp(log_dens)) plt.title('Kernel Density Estimation') plt.show()

3.3 生存分析中的应用

在生存分析中，EDF的变体Kaplan-Meier估计器广泛用于估计生存函数：

from lifelines import KaplanMeierFitter # 示例数据：时间(月)和事件指示(1=发生事件) durations = [3, 5, 7, 9, 10, 12, 15, 18, 20, 24] events = [1, 1, 1, 1, 0, 1, 0, 1, 1, 1] # 创建并拟合KM模型 kmf = KaplanMeierFitter() kmf.fit(durations, events, label='KM Estimate') # 绘制生存曲线 kmf.plot_survival_function() plt.title('Kaplan-Meier Survival Estimate') plt.ylabel('Survival Probability') plt.xlabel('Time (months)') plt.show()

4. 性能优化与大规模数据处理

4.1 内存高效的EDF计算

对于大型数据集，可以使用生成器或分块处理：

def chunked_edf(data, chunk_size=10000): """分块计算EDF以节省内存""" sorted_data = np.sort(data) n = len(sorted_data) for i in range(0, n, chunk_size): chunk = sorted_data[i:i+chunk_size] yield chunk, np.arange(i+1, i+len(chunk)+1) / n # 使用示例 large_data = np.random.normal(0, 1, 10**6) for chunk, edf_values in chunked_edf(large_data): # 处理每个数据块 pass

4.2 并行计算加速

利用多核处理器加速EDF相关计算：

from joblib import Parallel, delayed import multiprocessing def parallel_edf(data, n_jobs=None): if n_jobs is None: n_jobs = multiprocessing.cpu_count() # 将数据分割为n_jobs份 chunks = np.array_split(data, n_jobs) # 并行排序各子集 sorted_chunks = Parallel(n_jobs=n_jobs)( delayed(np.sort)(chunk) for chunk in chunks ) # 合并排序结果 return np.concatenate(sorted_chunks) # 使用示例 large_data = np.random.normal(0, 1, 10**6) sorted_data = parallel_edf(large_data)

4.3 稀疏数据的EDF处理

对于包含大量重复值的数据，可以使用优化算法：

def sparse_edf(data): unique_values, counts = np.unique(data, return_counts=True) cum_counts = np.cumsum(counts) n = len(data) return unique_values, cum_counts / n # 使用示例 repeated_data = np.random.randint(0, 100, 10000) x, y = sparse_edf(repeated_data)

5. 常见问题与解决方案

5.1 处理含缺失值的数据

当数据包含NaN时，需要先处理缺失值：

def edf_with_nan(data): # 移除NaN值 clean_data = data[~np.isnan(data)] # 计算EDF sorted_data = np.sort(clean_data) n = len(sorted_data) return sorted_data, np.arange(1, n+1) / n # 使用示例 data_with_nan = np.array([1, 2, np.nan, 3, 4, np.nan, 5]) x, y = edf_with_nan(data_with_nan)

5.2 处理多维数据的EDF

对于多维数据，可以计算边缘EDF或使用Copula方法：

def multivariate_edf(data): """计算每个维度的边缘EDF""" n_dims = data.shape[1] edfs = [] for i in range(n_dims): sorted_dim = np.sort(data[:, i]) y = np.arange(1, len(sorted_dim)+1) / len(sorted_dim) edfs.append((sorted_dim, y)) return edfs # 使用示例 multi_data = np.random.multivariate_normal( mean=[0, 0], cov=[[1, 0.5], [0.5, 1]], size=100 ) dim_edfs = multivariate_edf(multi_data)

5.3 EDF反演与分位数计算

从EDF获取分位数：

def edf_quantile(sorted_data, quantile): """基于排序数据计算分位数""" n = len(sorted_data) index = int(np.floor(quantile * n)) return sorted_data[min(index, n-1)] # 使用示例 sample = np.random.normal(0, 1, 1000) sorted_sample = np.sort(sample) median = edf_quantile(sorted_sample, 0.5) percentile_95 = edf_quantile(sorted_sample, 0.95)

6. 可视化技巧与最佳实践

6.1 多EDF比较可视化

比较多个分布的EDF：

# 生成三种不同分布 sample1 = np.random.normal(0, 1, 100) sample2 = np.random.exponential(1, 100) sample3 = np.random.uniform(-2, 2, 100) # 计算各自的EDF x1, y1 = empirical_dist_func(sample1) x2, y2 = empirical_dist_func(sample2) x3, y3 = empirical_dist_func(sample3) # 绘制比较图 plt.figure(figsize=(10, 6)) plt.step(x1, y1, where='post', label='Normal') plt.step(x2, y2, where='post', label='Exponential') plt.step(x3, y3, where='post', label='Uniform') plt.legend() plt.title('Comparison of EDFs for Different Distributions') plt.xlabel('Value') plt.ylabel('Cumulative Probability') plt.grid(True) plt.show()

6.2 EDF与理论CDF对比

将EDF与理论累积分布函数(CDF)叠加：

from scipy.stats import norm # 生成正态样本 sample = np.random.normal(0, 1, 100) x_edf, y_edf = empirical_dist_func(sample) # 理论CDF点 x_cdf = np.linspace(-3, 3, 100) y_cdf = norm.cdf(x_cdf) # 绘制对比图 plt.step(x_edf, y_edf, where='post', label='EDF') plt.plot(x_cdf, y_cdf, label='Theoretical CDF') plt.legend() plt.title('EDF vs Theoretical CDF') plt.xlabel('Value') plt.ylabel('Cumulative Probability') plt.grid(True) plt.show()

6.3 交互式EDF可视化

使用Plotly创建交互式EDF图：

import plotly.express as px import numpy as np # 生成样本数据 sample = np.random.normal(0, 1, 500) # 创建DataFrame df = pd.DataFrame({ 'value': np.sort(sample), 'cum_prob': np.arange(1, 501) / 500 }) # 创建交互式EDF图 fig = px.line(df, x='value', y='cum_prob', title='Interactive EDF Plot', labels={'value': 'Value', 'cum_prob': 'Cumulative Probability'}) fig.update_traces(line_shape='hv') # 阶梯线 fig.show()