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

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, n1) / 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, wherepost) 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(scale1, size200) # 创建ECDF对象 ecdf ECDF(sample) # 计算特定点的EDF值 print(ecdf(0.5)) # 输出x0.5处的累积概率 print(ecdf(1.0)) # 输出x1.0处的累积概率 # 获取EDF的全部x和y值 x ecdf.x y ecdf.ystatsmodels的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(fKS统计量: {ks_statistic:.4f}) print(fP值: {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(kernelgaussian, bandwidth0.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, labelKM 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_size10000): 分块计算EDF以节省内存 sorted_data np.sort(data) n len(sorted_data) for i in range(0, n, chunk_size): chunk sorted_data[i:ichunk_size] yield chunk, np.arange(i1, ilen(chunk)1) / n # 使用示例 large_data np.random.normal(0, 1, 10**6) for chunk, edf_values in chunked_edf(large_data): # 处理每个数据块 pass4.2 并行计算加速利用多核处理器加速EDF相关计算from joblib import Parallel, delayed import multiprocessing def parallel_edf(data, n_jobsNone): if n_jobs is None: n_jobs multiprocessing.cpu_count() # 将数据分割为n_jobs份 chunks np.array_split(data, n_jobs) # 并行排序各子集 sorted_chunks Parallel(n_jobsn_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_countsTrue) 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, n1) / 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]], size100 ) 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, wherepost, labelNormal) plt.step(x2, y2, wherepost, labelExponential) plt.step(x3, y3, wherepost, labelUniform) 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, wherepost, labelEDF) plt.plot(x_cdf, y_cdf, labelTheoretical 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, xvalue, ycum_prob, titleInteractive EDF Plot, labels{value: Value, cum_prob: Cumulative Probability}) fig.update_traces(line_shapehv) # 阶梯线 fig.show()提示在实际项目中EDF的计算和可视化应根据数据特点和业务需求进行调整。对于非常大的数据集考虑使用近似算法或分布式计算框架如Dask来加速EDF计算。