Zero-Flow双样本检验：基于归一化流的高维分布差异检测方法

在日常的数据分析和机器学习项目中我们经常需要判断两个样本是否来自同一个分布。无论是A/B测试中的实验组与对照组比较还是模型训练前后的数据分布验证双样本检验都是不可或缺的统计工具。然而传统的检验方法如t检验、KS检验等在面对高维数据或复杂分布时往往显得力不从心。本文将深入探讨一种新兴的双样本检验方法——Zero-Flow Two-Sample Tests。我们将从基础概念入手逐步解析其核心原理并通过完整的Python代码示例展示如何在实际项目中应用这一技术。无论你是统计学初学者还是有一定经验的数据科学家都能从本文中获得实用的知识和可复现的代码方案。1. 双样本检验的背景与挑战1.1 什么是双样本检验双样本检验Two-Sample Tests是统计学中用于判断两个独立样本是否来自同一概率分布的假设检验方法。在实际应用中我们通常设定零假设H0两个样本来自同一分布备择假设H1两个样本来自不同分布传统的双样本检验方法包括参数检验如t检验和非参数检验如Kolmogorov-Smirnov检验、Mann-Whitney U检验等。这些方法各有优缺点但普遍存在对数据分布假设敏感、高维数据效果不佳等问题。1.2 传统方法的局限性随着数据科学的发展我们面临的数据越来越复杂维度灾难当数据维度增加时传统检验方法的功效power急剧下降。例如KS检验在一维数据上表现良好但在高维空间中几乎无法有效区分分布差异。非线性关系许多传统检验方法基于线性假设无法捕捉复杂的非线性分布差异。计算效率对于大规模数据集某些检验方法如置换检验计算成本过高难以在实际项目中应用。1.3 Zero-Flow方法的出现Zero-Flow Two-Sample Tests是基于归一化流Normalizing Flows和深度学习的最新进展。它通过训练一个可逆的神经网络将复杂的数据分布转换为简单的基准分布如标准正态分布从而将复杂的分布比较问题简化为简单的分布差异检测。2. Zero-Flow的核心原理2.1 归一化流基础归一化流是一类可逆的神经网络架构能够学习从复杂分布到简单分布的双射变换。其核心思想是通过一系列可逆变换将原始数据空间中的复杂分布p(x)映射到潜空间中的简单分布p(z)其中变换满足z f(x), x f⁻¹(z)变换的雅可比行列式必须可计算以便进行概率密度函数的转换p(x) p(z) |det ∂f/∂x|2.2 Zero-Flow的检验框架Zero-Flow检验的基本框架包含三个主要步骤步骤1训练流模型使用两个样本的合并数据训练一个归一化流模型目标是让模型能够将数据转换为标准正态分布。步骤2转换样本将两个原始样本分别通过训练好的流模型转换到潜空间。步骤3检验转换结果在潜空间中检验两个转换后的样本是否都来自标准正态分布。如果原假设成立两个样本来自同一分布那么转换后的两个样本都应该近似服从标准正态分布。2.3 数学理论基础设X和Y是两个待检验的样本集。我们训练一个流模型f使得f: ℝ^d → ℝ^d, 且 f(X ∪ Y) ∼ N(0, I)如果X和Y来自同一分布那么f(X) ∼ N(0, I) 且 f(Y) ∼ N(0, I)检验统计量可以基于转换后样本与标准正态分布的差异来构建例如使用能量距离或最大均值差异MMD。3. 环境准备与依赖配置3.1 Python环境要求本文示例基于Python 3.8环境主要依赖包包括# requirements.txt torch1.9.0 numpy1.21.0 scipy1.7.0 scikit-learn1.0.0 matplotlib3.5.0 seaborn0.11.03.2 核心库安装pip install torch numpy scipy scikit-learn matplotlib seaborn3.3 检查环境配置import torch import numpy as np import scipy import sklearn import matplotlib.pyplot as plt import seaborn as sns print(fPyTorch版本: {torch.__version__}) print(fNumPy版本: {np.__version__}) print(fSciPy版本: {scipy.__version__})4. 实现基础的归一化流模型4.1 构建可逆神经网络层import torch import torch.nn as nn import torch.nn.functional as F class AffineCouplingLayer(nn.Module): 仿射耦合层 - 归一化流的基本构建块 def __init__(self, dim, hidden_dim64): super(AffineCouplingLayer, self).__init__() self.dim dim self.split_idx dim // 2 # 缩放和平移网络 self.scale_translate_net nn.Sequential( nn.Linear(self.split_idx, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, (dim - self.split_idx) * 2) ) # 初始化权重 self.scale_translate_net[-1].weight.data.zero_() self.scale_translate_net[-1].bias.data.zero_() def forward(self, x, reverseFalse): 前向传播变换或反向传播逆变换 if not reverse: return self._transform(x) else: return self._inverse_transform(x) def _transform(self, x): x1, x2 x[:, :self.split_idx], x[:, self.split_idx:] st self.scale_translate_net(x1) scale, translate st.chunk(2, dim1) scale torch.tanh(scale) z2 x2 * torch.exp(scale) translate z1 x1 z torch.cat([z1, z2], dim1) log_det scale.sum(dim1) return z, log_det def _inverse_transform(self, z): z1, z2 z[:, :self.split_idx], z[:, self.split_idx:] st self.scale_translate_net(z1) scale, translate st.chunk(2, dim1) scale torch.tanh(scale) x2 (z2 - translate) * torch.exp(-scale) x1 z1 x torch.cat([x1, x2], dim1) log_det -scale.sum(dim1) return x, log_det4.2 构建完整的归一化流模型class NormalizingFlow(nn.Module): 完整的归一化流模型 def __init__(self, dim, num_layers10, hidden_dim64): super(NormalizingFlow, self).__init__() self.dim dim self.layers nn.ModuleList([ AffineCouplingLayer(dim, hidden_dim) for _ in range(num_layers) ]) # 随机排列层 self.permutations [ torch.randperm(dim) for _ in range(num_layers) ] def forward(self, x): 将数据变换到潜空间 log_det_total torch.zeros(x.shape[0]) for i, layer in enumerate(self.layers): # 应用随机排列 perm self.permutations[i] x x[:, perm] x, log_det layer(x) log_det_total log_det return x, log_det_total def inverse(self, z): 从潜空间变换回数据空间 log_det_total torch.zeros(z.shape[0]) for i, layer in reversed(list(enumerate(self.layers))): z, log_det layer(z, reverseTrue) log_det_total log_det # 应用逆排列 perm self.permutations[i] inv_perm torch.argsort(perm) z z[:, inv_perm] return z, log_det_total def log_prob(self, x): 计算数据的对数似然 z, log_det self.forward(x) # 标准正态分布的对数概率 log_prob_z -0.5 * (z ** 2).sum(dim1) - 0.5 * self.dim * np.log(2 * np.pi) log_prob_x log_prob_z log_det return log_prob_x5. 实现Zero-Flow双样本检验5.1 检验统计量实现def energy_distance(z1, z2): 计算两个样本集之间的能量距离 n1, n2 z1.shape[0], z2.shape[0] # 样本内距离 dist_z1z1 torch.cdist(z1, z1).mean() dist_z2z2 torch.cdist(z2, z2).mean() # 样本间距离 dist_z1z2 torch.cdist(z1, z2).mean() energy_dist 2 * dist_z1z2 - dist_z1z1 - dist_z2z2 return energy_dist def mmd_rbf(z1, z2, sigma1.0): 使用RBF核计算最大均值差异MMD n1, n2 z1.shape[0], z2.shape[0] # 计算核矩阵 z1z1 torch.cdist(z1, z1) z2z2 torch.cdist(z2, z2) z1z2 torch.cdist(z1, z2) k_z1z1 torch.exp(-z1z1 ** 2 / (2 * sigma ** 2)) k_z2z2 torch.exp(-z2z2 ** 2 / (2 * sigma ** 2)) k_z1z2 torch.exp(-z1z2 ** 2 / (2 * sigma ** 2)) mmd (k_z1z1.sum() / (n1 * n1) k_z2z2.sum() / (n2 * n2) - 2 * k_z1z2.sum() / (n1 * n2)) return mmd5.2 Zero-Flow检验主函数class ZeroFlowTwoSampleTest: Zero-Flow双样本检验实现 def __init__(self, dim, flow_layers10, hidden_dim64, lr1e-3): self.dim dim self.flow NormalizingFlow(dim, flow_layers, hidden_dim) self.optimizer torch.optim.Adam(self.flow.parameters(), lrlr) def train_flow(self, x, y, epochs1000, batch_size128): 训练归一化流模型 # 合并两个样本 data torch.cat([x, y], dim0) n_data data.shape[0] losses [] for epoch in range(epochs): # 随机打乱数据 indices torch.randperm(n_data) data_shuffled data[indices] epoch_loss 0 for i in range(0, n_data, batch_size): batch data_shuffled[i:ibatch_size] # 计算负对数似然 log_prob self.flow.log_prob(batch) loss -log_prob.mean() self.optimizer.zero_grad() loss.backward() self.optimizer.step() epoch_loss loss.item() losses.append(epoch_loss / (n_data // batch_size)) if epoch % 100 0: print(fEpoch {epoch}, Loss: {losses[-1]:.4f}) return losses def permutation_test(self, z1, z2, test_statistic, n_permutations1000): 执行置换检验计算p值 # 计算观察到的检验统计量 observed_stat test_statistic(z1, z2) # 合并样本 z_combined torch.cat([z1, z2], dim0) n1, n2 z1.shape[0], z2.shape[0] n_total n1 n2 # 生成置换样本 perm_stats [] for _ in range(n_permutations): # 随机置换标签 perm_indices torch.randperm(n_total) perm_z1 z_combined[perm_indices[:n1]] perm_z2 z_combined[perm_indices[n1:]] perm_stat test_statistic(perm_z1, perm_z2) perm_stats.append(perm_stat.item()) # 计算p值 p_value (np.sum(perm_stats observed_stat.item()) 1) / (n_permutations 1) return observed_stat.item(), p_value def test(self, x, y, test_statisticenergy, n_permutations1000): 执行双样本检验 # 训练流模型 print(训练归一化流模型...) self.train_flow(x, y) # 转换样本到潜空间 with torch.no_grad(): z1, _ self.flow.forward(x) z2, _ self.flow.forward(y) # 选择检验统计量 if test_statistic energy: stat_func energy_distance elif test_statistic mmd: stat_func mmd_rbf else: raise ValueError(支持的检验统计量: energy 或 mmd) # 执行置换检验 print(执行置换检验...) stat_value, p_value self.permutation_test(z1, z2, stat_func, n_permutations) return { test_statistic: stat_value, p_value: p_value, z1: z1, z2: z2 }6. 完整实战案例6.1 生成模拟数据def generate_simulation_data(n_samples1000, dim10, scenariosame): 生成模拟数据用于测试 torch.manual_seed(42) if scenario same: # 相同分布均来自多元正态分布 mean torch.zeros(dim) cov torch.eye(dim) x torch.distributions.MultivariateNormal(mean, cov).sample((n_samples,)) y torch.distributions.MultivariateNormal(mean, cov).sample((n_samples,)) elif scenario mean_shift: # 均值偏移相同的协方差不同的均值 mean1 torch.zeros(dim) mean2 0.5 * torch.ones(dim) cov torch.eye(dim) x torch.distributions.MultivariateNormal(mean1, cov).sample((n_samples,)) y torch.distributions.MultivariateNormal(mean2, cov).sample((n_samples,)) elif scenario cov_shift: # 协方差偏移相同的均值不同的协方差 mean torch.zeros(dim) cov1 torch.eye(dim) cov2 torch.eye(dim) * 2.0 x torch.distributions.MultivariateNormal(mean, cov1).sample((n_samples,)) y torch.distributions.MultivariateNormal(mean, cov2).sample((n_samples,)) elif scenario nonlinear: # 非线性变换相同的潜变量不同的非线性变换 z torch.distributions.MultivariateNormal(torch.zeros(dim), torch.eye(dim)).sample((n_samples,)) x z 0.1 * torch.randn_like(z) y torch.sin(z) 0.1 * torch.randn_like(z) return x.float(), y.float() # 生成测试数据 print(生成模拟数据...) x_same, y_same generate_simulation_data(scenariosame) x_mean, y_mean generate_simulation_data(scenariomean_shift) x_cov, y_cov generate_simulation_data(scenariocov_shift) x_nonlinear, y_nonlinear generate_simulation_data(scenariononlinear)6.2 执行Zero-Flow检验# 初始化检验器 dim x_same.shape[1] tester ZeroFlowTwoSampleTest(dimdim, flow_layers8, hidden_dim64) # 测试相同分布的情况 print( 测试相同分布 ) result_same tester.test(x_same, y_same, test_statisticenergy) print(f检验统计量: {result_same[test_statistic]:.4f}) print(fP值: {result_same[p_value]:.4f}) # 测试均值偏移的情况 print(\n 测试均值偏移 ) result_mean tester.test(x_mean, y_mean, test_statisticenergy) print(f检验统计量: {result_mean[test_statistic]:.4f}) print(fP值: {result_mean[p_value]:.4f}) # 测试协方差偏移的情况 print(\n 测试协方差偏移 ) result_cov tester.test(x_cov, y_cov, test_statisticenergy) print(f检验统计量: {result_cov[test_statistic]:.4f}) print(fP值: {result_cov[p_value]:.4f}) # 测试非线性变换的情况 print(\n 测试非线性变换 ) result_nonlinear tester.test(x_nonlinear, y_nonlinear, test_statisticenergy) print(f检验统计量: {result_nonlinear[test_statistic]:.4f}) print(fP值: {result_nonlinear[p_value]:.4f})6.3 可视化结果def plot_results(results_dict, scenarios): 可视化检验结果 fig, axes plt.subplots(2, 2, figsize(12, 10)) axes axes.ravel() for i, scenario in enumerate(scenarios): result results_dict[scenario] z1, z2 result[z1], result[z2] # 只可视化前两个维度 z1_2d z1[:, :2].numpy() z2_2d z2[:, :2].numpy() axes[i].scatter(z1_2d[:, 0], z1_2d[:, 1], alpha0.6, label样本1) axes[i].scatter(z2_2d[:, 0], z2_2d[:, 1], alpha0.6, label样本2) axes[i].set_title(f{scenario}\nP值: {result[p_value]:.4f}) axes[i].legend() axes[i].set_xlabel(潜变量维度1) axes[i].set_ylabel(潜变量维度2) plt.tight_layout() plt.show() # 收集所有结果 results { 相同分布: result_same, 均值偏移: result_mean, 协方差偏移: result_cov, 非线性变换: result_nonlinear } scenarios [相同分布, 均值偏移, 协方差偏移, 非线性变换] plot_results(results, scenarios)6.4 与传统方法对比from scipy import stats from scipy.spatial.distance import cdist def traditional_t_test(x, y): 传统t检验 t_stat, p_value stats.ttest_ind(x.numpy(), y.numpy(), axis0) return np.mean(p_value) def traditional_ks_test(x, y): 传统KS检验逐维度 p_values [] for i in range(x.shape[1]): stat, p_val stats.ks_2samp(x[:, i].numpy(), y[:, i].numpy()) p_values.append(p_val) return np.mean(p_values) # 对比不同方法的性能 print( 方法对比 ) methods { Zero-Flow: None, T检验: traditional_t_test, KS检验: traditional_ks_test } scenarios_data { 相同分布: (x_same, y_same), 均值偏移: (x_mean, y_mean), 协方差偏移: (x_cov, y_cov), 非线性变换: (x_nonlinear, y_nonlinear) } results_comparison {} for scenario, (x, y) in scenarios_data.items(): print(f\n--- {scenario} ---) results_comparison[scenario] {} # Zero-Flow检验 result_zf tester.test(x, y, test_statisticenergy, n_permutations500) results_comparison[scenario][Zero-Flow] result_zf[p_value] print(fZero-Flow P值: {result_zf[p_value]:.4f}) # 传统方法 for method_name, method_func in methods.items(): if method_name ! Zero-Flow: p_val method_func(x, y) results_comparison[scenario][method_name] p_val print(f{method_name} P值: {p_val:.4f})7. 常见问题与解决方案7.1 训练不收敛问题问题现象流模型的损失函数震荡或不下降检验结果不稳定。解决方案调整学习率尝试更小的学习率如1e-4或使用学习率调度器增加网络容量增大隐藏层维度或增加流层数数据标准化确保输入数据经过适当的标准化处理检查梯度监控梯度范数避免梯度爆炸或消失# 改进的训练配置 class ImprovedZeroFlowTest(ZeroFlowTwoSampleTest): def __init__(self, dim, flow_layers10, hidden_dim128, lr1e-4): super().__init__(dim, flow_layers, hidden_dim, lr) # 添加学习率调度 self.scheduler torch.optim.lr_scheduler.ReduceLROnPlateau( self.optimizer, modemin, patience50, factor0.5 ) def train_flow(self, x, y, epochs1000, batch_size128): data torch.cat([x, y], dim0) n_data data.shape[0] # 数据标准化 self.data_mean data.mean(dim0) self.data_std data.std(dim0) data_normalized (data - self.data_mean) / (self.data_std 1e-8) losses [] for epoch in range(epochs): indices torch.randperm(n_data) data_shuffled data_normalized[indices] epoch_loss 0 for i in range(0, n_data, batch_size): batch data_shuffled[i:ibatch_size] log_prob self.flow.log_prob(batch) loss -log_prob.mean() self.optimizer.zero_grad() loss.backward() # 梯度裁剪 torch.nn.utils.clip_grad_norm_(self.flow.parameters(), max_norm1.0) self.optimizer.step() epoch_loss loss.item() avg_loss epoch_loss / (n_data // batch_size) losses.append(avg_loss) self.scheduler.step(avg_loss) if epoch % 100 0: current_lr self.optimizer.param_groups[0][lr] print(fEpoch {epoch}, Loss: {avg_loss:.4f}, LR: {current_lr:.6f}) return losses7.2 高维数据处理问题现象当数据维度很高时模型训练缓慢检验功效下降。解决方案维度缩减使用PCA或自动编码器进行降维流模型优化使用更高效的流架构如Glow或RealNVP小批量处理确保批量大小足够大以估计高维分布正则化添加适当的正则化防止过拟合def pca_preprocessing(x, y, explained_variance0.95): 使用PCA进行降维预处理 from sklearn.decomposition import PCA data torch.cat([x, y], dim0).numpy() pca PCA(n_componentsexplained_variance) data_reduced pca.fit_transform(data) n_components data_reduced.shape[1] print(f原始维度: {data.shape[1]}, 降维后: {n_components}) x_reduced torch.tensor(data_reduced[:x.shape[0]], dtypetorch.float32) y_reduced torch.tensor(data_reduced[x.shape[0]:], dtypetorch.float32) return x_reduced, y_reduced, pca # 使用PCA预处理高维数据 x_highdim torch.randn(1000, 100) # 100维数据 y_highdim torch.randn(1000, 100) 0.1 # 轻微差异 x_reduced, y_reduced, pca pca_preprocessing(x_highdim, y_highdim) tester_reduced ZeroFlowTwoSampleTest(dimx_reduced.shape[1]) result_reduced tester_reduced.test(x_reduced, y_reduced)7.3 小样本问题问题现象当样本量较小时置换检验的p值估计不准确。解决方案增加置换次数使用更多的置换样本提高估计精度精确检验对于小样本使用精确置换检验参数化检验在潜空间中使用参数检验方法自助法使用自助法估计检验统计量的分布def exact_permutation_test(z1, z2, test_statistic, max_permutations10000): 精确置换检验适用于小样本 n1, n2 z1.shape[0], z2.shape[0] n_total n1 n2 # 如果可能排列数太大使用随机抽样 total_permutations np.math.comb(n_total, n1) if total_permutations max_permutations: return tester.permutation_test(z1, z2, test_statistic, max_permutations) # 精确检验枚举所有可能排列 from itertools import combinations z_combined torch.cat([z1, z2], dim0) observed_stat test_statistic(z1, z2) extreme_count 0 total_count 0 # 枚举所有可能的组合计算量很大仅适用于小样本 for indices in combinations(range(n_total), n1): perm_z1 z_combined[list(indices)] perm_z2 z_combined[[i for i in range(n_total) if i not in indices]] perm_stat test_statistic(perm_z1, perm_z2) if perm_stat observed_stat: extreme_count 1 total_count 1 p_value extreme_count / total_count return observed_stat.item(), p_value8. 最佳实践与工程建议8.1 超参数调优策略流模型架构选择数据维度较低时50使用8-12个耦合层隐藏层维度64-128数据维度中等时50-200增加网络容量隐藏层维度128-256数据维度较高时200考虑先降维再应用Zero-Flow训练参数优化def optimize_hyperparameters(x, y, param_grid): 超参数网格搜索优化 best_params None best_score float(inf) for flow_layers in param_grid[flow_layers]: for hidden_dim in param_grid[hidden_dim]: for lr in param_grid[lr]: try: tester ZeroFlowTwoSampleTest( dimx.shape[1], flow_layersflow_layers, hidden_dimhidden_dim, lrlr ) # 使用部分数据快速验证 x_val x[:200] y_val y[:200] losses tester.train_flow(x_val, y_val, epochs100) # 使用最终损失作为评分标准 final_loss losses[-1] if final_loss best_score: best_score final_loss best_params { flow_layers: flow_layers, hidden_dim: hidden_dim, lr: lr } except Exception as e: print(f参数组合失败: {flow_layers}, {hidden_dim}, {lr}, 错误: {e}) return best_params, best_score # 超参数搜索空间 param_grid { flow_layers: [6, 8, 10, 12], hidden_dim: [64, 128, 256], lr: [1e-3, 1e-4, 5e-5] }8.2 生产环境部署考虑计算效率优化模型预热预先训练通用流模型在新数据上微调批量处理对多个检验任务进行批量处理GPU加速利用CUDA进行并行计算缓存机制缓存训练好的流模型避免重复训练代码示例生产级实现class ProductionZeroFlowTester: 生产环境适用的Zero-Flow检验器 def __init__(self, base_flow_pathNone): self.base_flow None if base_flow_path and os.path.exists(base_flow_path): self.load_base_flow(base_flow_path) def load_base_flow(self, path): 加载预训练的基础流模型 self.base_flow torch.load(path) self.base_flow.eval() def fine_tune_flow(self, x, y, fine_tune_epochs100): 在基础模型上微调 if self.base_flow is None: # 如果没有基础模型正常训练 return self.train_flow(x, y, epochs1000) # 微调只训练最后几层 for param in self.base_flow.parameters(): param.requires_grad False # 解冻最后几层 for layer in list(self.base_flow.layers)[-3:]: for param in layer.parameters(): param.requires_grad True # 微调训练 optimizer torch.optim.Adam( filter(lambda p: p.requires_grad, self.base_flow.parameters()), lr1e-5 ) data torch.cat([x, y], dim0) for epoch in range(fine_tune_epochs): # 简化的微调过程 indices torch.randperm(data.shape[0]) batch data[indices[:128]] log_prob self.base_flow.log_prob(batch) loss -log_prob.mean() optimizer.zero_grad() loss.backward() optimizer.step() def batch_test(self, test_pairs): 批量处理多个检验任务 results [] for i, (x, y) in enumerate(test_pairs): print(f处理第 {i1}/{len(test_pairs)} 个检验任务...) result self.test(x, y) results.append(result) return results8.3 结果解释与报告统计显著性判断p值 0.01强证据拒绝原假设分布不同0.01 ≤ p值 0.05中等证据拒绝原假设0.05 ≤ p值 0.1弱证据需要更多数据p值 ≥ 0.1没有足够证据拒绝原假设效应大小评估def effect_size_analysis(z1, z2): 分析分布差异的效应大小 # 计算均值差异 mean_diff (z1.mean(dim0) - z2.mean(dim0)).abs().mean() # 计算方差比率 var_ratio (z1.var(dim0) / (z2.var(dim0) 1e-8)).mean() # 计算MMD距离 mmd_val mmd_rbf(z1, z2).item() return { mean_difference: mean_diff.item(), variance_ratio: var_ratio.item(), mmd_distance: mmd_val } # 在检验后添加效应大小分析 def comprehensive_test(x, y): 综合检验包括显著性检验和效应大小分析 # 执行Zero-Flow检验 basic_result tester.test(x, y) # 效应大小分析 effect_sizes effect_size_analysis(basic_result[z1], basic_result[z2]) # 综合报告 comprehensive_result { **basic_result, effect_sizes: effect_sizes, interpretation: interpret_results(basic_result[p_value], effect_sizes) } return comprehensive_result def interpret_results(p_value, effect_sizes): 结果解释 if p_value 0.05: if effect_sizes[mmd_distance] 0.1: return 强证据表明分布存在显著差异 else: return 分布存在显著差异但效应较小 else: if effect_sizes[mmd_distance] 0.05: return 虽然不显著但观察到中等效应建议增加样本量 else: return 没有足够证据表明分布存在差异Zero-Flow双样本检验为复杂分布比较提供了强大的工具特别适合高维数据和非线性差异的检测。在实际应用中建议根据具体数据特性调整模型架构和超参数并结合效应大小分析来全面理解检验结果。这种方法的真正价值在于其灵活性——能够适应各种复杂的数据分布为数据科学家提供更可靠的分布差异检测能力。