1. PyTorch中的统计学函数概述在深度学习项目中数据统计分析是不可或缺的一环。PyTorch作为主流的深度学习框架提供了丰富的统计学函数这些函数不仅支持常规的统计分析还能充分利用GPU加速计算。与NumPy等传统科学计算库相比PyTorch的统计函数具有以下优势自动微分支持所有统计函数都可无缝集成到神经网络计算图中GPU加速大规模数据统计计算效率提升显著批量处理天然支持对多维张量的批量统计运算动态计算适用于可变长度序列数据的统计分析提示PyTorch的统计函数主要分布在torch和torch.nn.functional模块中使用时需注意函数输入的数据类型要求。2. 基础统计量计算函数2.1 中心趋势度量PyTorch提供了计算数据集中趋势的核心函数import torch data torch.tensor([1.0, 2.0, 3.0, 4.0, 5.0]) # 平均值计算 mean torch.mean(data) # 输出: tensor(3.) # 中位数计算 median torch.median(data) # 输出: tensor(3.) # 众数计算需要额外处理 values, counts torch.unique(data, return_countsTrue) mode values[torch.argmax(counts)] # 输出: tensor(1.)对于高维张量可以指定计算维度matrix torch.randn(3, 4) # 3x4矩阵 row_means torch.mean(matrix, dim1) # 计算每行的均值 col_means torch.mean(matrix, dim0) # 计算每列的均值2.2 离散程度度量衡量数据离散程度的关键函数# 方差计算 variance torch.var(data) # 默认计算总体方差 sample_variance torch.var(data, unbiasedTrue) # 计算样本方差 # 标准差计算 std torch.std(data) # 总体标准差 sample_std torch.std(data, unbiasedTrue) # 样本标准差 # 极差计算 range_val torch.max(data) - torch.min(data)注意方差和标准差函数的unbiased参数控制是否使用无偏估计n-1修正在样本统计时通常设为True。2.3 分位数计算PyTorch提供了灵活的分位数计算功能# 计算特定分位数 quantiles torch.quantile(data, torch.tensor([0.25, 0.5, 0.75])) # 四分位数 # 多维数据分位数计算 matrix torch.randn(100, 5) # 100个样本5个特征 feature_quartiles torch.quantile(matrix, torch.tensor([0.25, 0.5, 0.75]), dim0)3. 概率分布相关函数3.1 常见分布采样PyTorch支持从多种概率分布中采样# 正态分布采样 normal_samples torch.normal(mean0.0, std1.0, size(1000,)) # 均匀分布采样 uniform_samples torch.rand(1000) # [0,1)区间均匀分布 # 泊松分布采样 poisson_samples torch.poisson(torch.tensor(5.0), (1000,)) # 多项分布采样 probs torch.tensor([0.2, 0.3, 0.5]) multinomial_samples torch.multinomial(probs, 1000, replacementTrue)3.2 分布统计量计算对于给定的分布参数可以直接计算其统计量# 计算正态分布的熵 normal_entropy torch.distributions.Normal(0, 1).entropy() # 计算两个分布间的KL散度 p torch.distributions.Normal(0, 1) q torch.distributions.Normal(1, 1) kl_divergence torch.distributions.kl_divergence(p, q)3.3 累积分布与分位函数# 正态分布的CDF计算 normal torch.distributions.Normal(0, 1) cdf_values normal.cdf(torch.linspace(-3, 3, 100)) # 分位函数逆CDF quantile_values normal.icdf(torch.linspace(0.01, 0.99, 100))4. 相关性与协方差分析4.1 协方差矩阵计算PyTorch提供了高效的协方差矩阵计算函数# 生成随机数据矩阵 (100个样本5个特征) data torch.randn(100, 5) # 计算协方差矩阵 cov_matrix torch.cov(data.T) # 注意需要转置输入 # 带修正的协方差计算 cov_matrix_corrected torch.cov(data.T, correction1) # 使用n-1修正4.2 相关系数计算Pearson相关系数计算# 计算两变量的相关系数 x torch.randn(100) y torch.randn(100) corr_coef torch.corrcoef(torch.stack([x, y]))[0, 1] # 计算特征间的相关系数矩阵 feature_corr torch.corrcoef(data.T)4.3 互相关计算对于信号处理和时间序列分析# 一维互相关计算 signal torch.randn(100) kernel torch.randn(10) cross_corr torch.nn.functional.conv1d( signal.view(1,1,-1), kernel.view(1,1,-1) ).squeeze()5. 假设检验与统计显著性5.1 t检验实现虽然PyTorch没有直接提供假设检验函数但可以轻松实现def t_test(sample1, sample2): n1, n2 len(sample1), len(sample2) mean1, mean2 torch.mean(sample1), torch.mean(sample2) var1, var2 torch.var(sample1, unbiasedTrue), torch.var(sample2, unbiasedTrue) pooled_var ((n1-1)*var1 (n2-1)*var2) / (n1 n2 - 2) t_value (mean1 - mean2) / torch.sqrt(pooled_var * (1/n1 1/n2)) return t_value # 使用示例 group1 torch.normal(5.0, 1.5, size(30,)) group2 torch.normal(6.0, 1.5, size(35,)) t_stat t_test(group1, group2)5.2 卡方统计量计算对于分类数据def chi_square(observed, expected): return torch.sum((observed - expected)**2 / expected) # 使用示例 obs torch.tensor([25, 30, 45]) exp torch.tensor([30, 30, 40]) chi2 chi_square(obs, exp)5.3 自助法(Bootstrap)实现def bootstrap_ci(data, stat_func, n_bootstraps1000, ci95): stats [] n len(data) for _ in range(n_bootstraps): sample data[torch.randint(0, n, (n,))] stats.append(stat_func(sample)) stats torch.tensor(stats) lower (100 - ci) / 2 upper 100 - lower return torch.quantile(stats, torch.tensor([lower/100, upper/100])) # 使用示例 data torch.randn(500) # 模拟数据 ci bootstrap_ci(data, torch.mean) # 计算均值的95%置信区间6. 高级统计函数与应用6.1 移动窗口统计对于时间序列分析特别有用def rolling_stat(data, window_size, stat_func): return torch.cat([ stat_func(data[i:iwindow_size]).unsqueeze(0) for i in range(len(data) - window_size 1) ]) # 使用示例 ts_data torch.randn(1000) # 时间序列数据 rolling_mean rolling_stat(ts_data, 30, torch.mean) # 30点移动平均 rolling_std rolling_stat(ts_data, 30, torch.std) # 30点移动标准差6.2 直方图与分箱统计# 一维直方图计算 values torch.randn(10000) hist torch.histc(values, bins50, min-3, max3) # 多维直方图通过联合分箱 def multi_dim_hist(data, bins10, rangesNone): if ranges is None: ranges [(torch.min(data[:,i]), torch.max(data[:,i])) for i in range(data.shape[1])] hist torch.zeros(*[bins]*data.shape[1]) bin_indices [torch.bucketize(data[:,i], torch.linspace(ranges[i][0], ranges[i][1], bins1))-1 for i in range(data.shape[1])] for idx in zip(*bin_indices): hist[idx] 1 return hist # 使用示例 data_2d torch.randn(1000, 2) # 二维数据 hist_2d multi_dim_hist(data_2d, bins20)6.3 核密度估计def kde(x, samples, bandwidth0.1): 高斯核密度估计 n len(samples) x x.unsqueeze(1) # (m,1) samples samples.unsqueeze(0) # (1,n) return torch.exp(-0.5 * ((x - samples)/bandwidth)**2).sum(1) / (n * bandwidth * math.sqrt(2*math.pi)) # 使用示例 data torch.randn(500) # 样本数据 x_grid torch.linspace(-3, 3, 100) # 估计点 density kde(x_grid, data) # 密度估计7. 统计函数在深度学习中的应用7.1 数据标准化与归一化# 批量归一化统计 def batch_norm_stats(data, eps1e-5): mean torch.mean(data, dim0) var torch.var(data, dim0, unbiasedFalse) return (data - mean) / torch.sqrt(var eps) # 层归一化统计 def layer_norm_stats(data, eps1e-5): mean torch.mean(data, dim-1, keepdimTrue) var torch.var(data, dim-1, keepdimTrue, unbiasedFalse) return (data - mean) / torch.sqrt(var eps)7.2 损失函数中的统计量# 自定义基于统计的损失函数 class StatisticalLoss(nn.Module): def __init__(self): super().__init__() def forward(self, preds, targets): # 均值匹配损失 mean_loss F.mse_loss(preds.mean(), targets.mean()) # 方差匹配损失 var_loss F.mse_loss(preds.var(unbiasedTrue), targets.var(unbiasedTrue)) # 分布形状损失通过分位数 quantiles torch.linspace(0.1, 0.9, 5) pred_q torch.quantile(preds, quantiles) target_q torch.quantile(targets, quantiles) shape_loss F.mse_loss(pred_q, target_q) return mean_loss var_loss shape_loss7.3 模型输出的统计评估def model_stat_eval(model, dataloader): model.eval() all_outputs [] all_targets [] with torch.no_grad(): for inputs, targets in dataloader: outputs model(inputs) all_outputs.append(outputs) all_targets.append(targets) outputs torch.cat(all_outputs) targets torch.cat(all_targets) stats { mean_absolute_error: torch.mean(torch.abs(outputs - targets)), correlation: torch.corrcoef(torch.stack([outputs.squeeze(), targets.squeeze()]))[0,1], std_ratio: outputs.std() / targets.std(), quantile_alignment: torch.mean( torch.abs( torch.quantile(outputs, torch.linspace(0.1,0.9,5)) - torch.quantile(targets, torch.linspace(0.1,0.9,5)) ) ) } return stats8. 性能优化与注意事项8.1 GPU加速技巧# 确保使用CUDA设备 device torch.device(cuda if torch.cuda.is_available() else cpu) # 大批量统计计算优化 def batch_statistics(data, chunk_size10000): results [] for i in range(0, len(data), chunk_size): chunk data[i:ichunk_size].to(device) results.append(torch.mean(chunk, dim0)) return torch.mean(torch.stack(results), dim0) # 使用示例 large_data torch.randn(1000000, 100) # 大矩阵 mean_vector batch_statistics(large_data)8.2 常见问题排查NaN值处理def safe_stat(data, stat_func, nan_policyomit): if nan_policy omit: valid_data data[~torch.isnan(data)] return stat_func(valid_data) elif nan_policy propagate: return stat_func(data) else: raise ValueError(nan_policy must be omit or propagate)数据类型不一致# 确保输入数据类型一致 def consistent_type_stat(data, stat_func): if data.dtype ! torch.float32: data data.float() return stat_func(data)空张量处理def robust_stat(data, stat_func, default0.0): if len(data) 0: return torch.tensor(default) return stat_func(data)8.3 内存优化策略# 增量统计计算 class OnlineStatistics: def __init__(self): self.n 0 self.mean 0.0 self.M2 0.0 def update(self, x): self.n 1 delta x - self.mean self.mean delta / self.n delta2 x - self.mean self.M2 delta * delta2 def variance(self): return self.M2 / self.n if self.n 1 else float(nan) def std(self): return torch.sqrt(self.variance()) # 使用示例 stats OnlineStatistics() for batch in dataloader: stats.update(torch.mean(batch)) final_mean stats.mean final_std stats.std()9. 实际案例图像数据统计分析9.1 图像通道统计def image_channel_stats(image_batch): image_batch: (B,C,H,W) mean torch.mean(image_batch, dim(0,2,3)) # 各通道均值 std torch.std(image_batch, dim(0,2,3)) # 各通道标准差 return {mean: mean, std: std} # 使用示例 batch torch.rand(32, 3, 256, 256) # 32张RGB图像 stats image_channel_stats(batch)9.2 数据集统计分析def dataset_statistics(dataloader): mean 0. std 0. nb_samples 0. for data, _ in dataloader: batch_samples data.size(0) data data.view(batch_samples, -1) mean data.mean(dim1).sum() std data.std(dim1).sum() nb_samples batch_samples mean / nb_samples std / nb_samples return {mean: mean, std: std}9.3 数据增强效果评估def augmentation_effect(augmentation, original_images, n_trials10): original_stats image_channel_stats(original_images) augmented_stats { mean: [], std: [], corr_with_original: [] } for _ in range(n_trials): augmented augmentation(original_images) stats image_channel_stats(augmented) augmented_stats[mean].append(stats[mean]) augmented_stats[std].append(stats[std]) # 计算与原始图像的相关性 flat_original original_images.flatten(1) flat_augmented augmented.flatten(1) corr torch.corrcoef(torch.cat([ flat_original.mean(dim1, keepdimTrue), flat_augmented.mean(dim1, keepdimTrue) ], dim1).T)[0,1] augmented_stats[corr_with_original].append(corr) # 汇总统计 return { mean_change: torch.std(torch.stack(augmented_stats[mean]), dim0), std_change: torch.std(torch.stack(augmented_stats[std]), dim0), avg_correlation: torch.mean(torch.tensor(augmented_stats[corr_with_original])) }10. 与NumPy统计函数的对比10.1 函数对应关系PyTorch函数NumPy对应函数主要差异torch.mean()np.mean()PyTorch支持自动微分torch.var()np.var()PyTorch的unbiased参数控制n-1修正torch.std()np.std()同上torch.median()np.median()PyTorch返回张量而非标量torch.histc()np.histogram()接口和返回值格式不同torch.unique()np.unique()功能类似返回类型不同10.2 性能对比import numpy as np import time # 创建数据 size (10000, 10000) torch_data torch.randn(size) numpy_data np.random.randn(*size) # 均值计算对比 start time.time() torch_mean torch.mean(torch_data) torch_time time.time() - start start time.time() numpy_mean np.mean(numpy_data) numpy_time time.time() - start print(fPyTorch耗时: {torch_time:.4f}s, NumPy耗时: {numpy_time:.4f}s)提示对于CPU计算NumPy通常更快但对于GPU上的大规模数据PyTorch优势明显。10.3 互操作性# NumPy数组转PyTorch张量 numpy_array np.random.randn(100) torch_tensor torch.from_numpy(numpy_array) # PyTorch张量转NumPy数组 torch_tensor torch.randn(100) numpy_array torch_tensor.numpy() # 混合使用示例 def hybrid_statistics(data): if isinstance(data, np.ndarray): data torch.from_numpy(data) mean torch.mean(data) std torch.std(data) return { mean: mean.numpy() if data.device.type cpu else mean.cpu().numpy(), std: std.numpy() if data.device.type cpu else std.cpu().numpy() }11. 自定义统计函数的实现11.1 扩展torch.autograd.Functionclass WeightedMean(torch.autograd.Function): staticmethod def forward(ctx, data, weights): ctx.save_for_backward(data, weights) return torch.sum(data * weights) / torch.sum(weights) staticmethod def backward(ctx, grad_output): data, weights ctx.saved_tensors grad_data grad_weights None if ctx.needs_input_grad[0]: grad_data grad_output * weights / torch.sum(weights) if ctx.needs_input_grad[1]: grad_weights grad_output * (data * torch.sum(weights) - torch.sum(data * weights)) / (torch.sum(weights)**2) return grad_data, grad_weights # 使用示例 data torch.randn(10, requires_gradTrue) weights torch.rand(10, requires_gradTrue) weighted_mean WeightedMean.apply(data, weights) weighted_mean.backward()11.2 实现滑动窗口统计def sliding_window_statistic(data, window_size, statistic, stride1): data: 输入张量 window_size: 窗口大小 statistic: 统计函数 (如torch.mean) stride: 滑动步长 n data.shape[0] num_windows (n - window_size) // stride 1 result torch.zeros(num_windows) for i in range(num_windows): start i * stride end start window_size window data[start:end] result[i] statistic(window) return result # 使用示例 ts_data torch.randn(1000) # 时间序列数据 rolling_median sliding_window_statistic(ts_data, 30, torch.median)11.3 实现核回归估计def kernel_regression(x, y, x_grid, bandwidth0.1): x: 自变量观测值 y: 因变量观测值 x_grid: 估计点 bandwidth: 核带宽 x x.unsqueeze(1) # (n,1) x_grid x_grid.unsqueeze(0) # (1,m) weights torch.exp(-0.5 * ((x_grid - x)/bandwidth)**2) # (m,n) weights weights / weights.sum(dim1, keepdimTrue) # 归一化 return weights y # (m,) # 使用示例 x torch.rand(100) * 10 y torch.sin(x) torch.randn(100) * 0.2 x_grid torch.linspace(0, 10, 100) y_smooth kernel_regression(x, y, x_grid, bandwidth0.5)12. 统计可视化与结果解释12.1 统计结果可视化虽然PyTorch本身不提供可视化功能但可以配合Matplotlib使用import matplotlib.pyplot as plt def plot_distribution(data, bins30, title): # 转换为CPU numpy数组 if isinstance(data, torch.Tensor): data data.cpu().numpy() # 计算统计量 mean np.mean(data) median np.median(data) std np.std(data) # 绘制直方图 plt.figure(figsize(10,6)) plt.hist(data, binsbins, densityTrue, alpha0.7, colorblue) # 添加统计信息 plt.axvline(mean, colorred, linestyledashed, linewidth2, labelfMean: {mean:.2f}) plt.axvline(median, colorgreen, linestyledashed, linewidth2, labelfMedian: {median:.2f}) plt.axvline(mean-std, colorgray, linestyledotted, linewidth2) plt.axvline(meanstd, colorgray, linestyledotted, linewidth2, labelf±1 std: {std:.2f}) plt.title(title) plt.legend() plt.show() # 使用示例 data torch.randn(1000) plot_distribution(data, titleNormal Distribution Sample)12.2 统计检验结果解释def interpret_t_test(t_value, df, alpha0.05): from scipy.stats import t p_value 2 * (1 - t.cdf(abs(t_value), df)) critical_value t.ppf(1 - alpha/2, df) print(ft统计量: {t_value:.3f}) print(f自由度(df): {df}) print(fp值: {p_value:.4f}) print(f临界值(α{alpha}): ±{critical_value:.3f}) if p_value alpha: print(结论: 拒绝原假设差异具有统计学意义) else: print(结论: 无法拒绝原假设差异无统计学意义) # 使用示例 t_value 2.56 # 假设计算得到的t值 df 45 # 自由度 interpret_t_test(t_value, df)12.3 相关性矩阵可视化def plot_correlation_matrix(corr_matrix, feature_namesNone): if isinstance(corr_matrix, torch.Tensor): corr_matrix corr_matrix.cpu().numpy() fig, ax plt.subplots(figsize(10,8)) cax ax.matshow(corr_matrix, cmapcoolwarm, vmin-1, vmax1) fig.colorbar(cax) if feature_names: ax.set_xticks(range(len(feature_names))) ax.set_yticks(range(len(feature_names))) ax.set_xticklabels(feature_names, rotation90) ax.set_yticklabels(feature_names) for i in range(corr_matrix.shape[0]): for j in range(corr_matrix.shape[1]): ax.text(j, i, f{corr_matrix[i,j]:.2f}, hacenter, vacenter, colorblack) plt.title(Feature Correlation Matrix) plt.show() # 使用示例 data torch.randn(100, 5) # 100个样本5个特征 corr_matrix torch.corrcoef(data.T) feature_names [Feature1, Feature2, Feature3, Feature4, Feature5] plot_correlation_matrix(corr_matrix, feature_names)13. 统计函数在模型训练中的应用13.1 学习率统计分析class LearningRateMonitor: def __init__(self): self.lr_history [] def record(self, optimizer): current_lr optimizer.param_groups[0][lr] self.lr_history.append(current_lr) def stats(self): lr_tensor torch.tensor(self.lr_history) return { mean: torch.mean(lr_tensor), std: torch.std(lr_tensor), min: torch.min(lr_tensor), max: torch.max(lr_tensor), final: lr_tensor[-1] } def plot(self): plt.plot(self.lr_history) plt.title(Learning Rate Schedule) plt.xlabel(Step) plt.ylabel(Learning Rate) plt.show() # 使用示例 optimizer torch.optim.SGD(model.parameters(), lr0.1) scheduler torch.optim.lr_scheduler.StepLR(optimizer, step_size30, gamma0.1) monitor LearningRateMonitor() for epoch in range(100): # 训练步骤... scheduler.step() monitor.record(optimizer) print(monitor.stats()) monitor.plot()13.2 梯度统计分析def gradient_statistics(model): total_norm 0 param_stats [] for name, param in model.named_parameters(): if param.grad is not None: grad_norm torch.norm(param.grad) total_norm grad_norm.item() ** 2 param_stats.append({ name: name, mean: torch.mean(param.grad).item(), std: torch.std(param.grad).item(), norm: grad_norm.item() }) total_norm total_norm ** 0.5 return { total_gradient_norm: total_norm, parameter_stats: param_stats } # 使用示例 # 在训练循环中调用 for inputs, targets in dataloader: optimizer.zero_grad() outputs model(inputs) loss criterion(outputs, targets) loss.backward() grad_stats gradient_statistics(model) print(fTotal gradient norm: {grad_stats[total_gradient_norm]:.4f}) optimizer.step()13.3 激活值统计分析class ActivationStats: def __init__(self, model): self.model model self.stats {} self.hooks [] def hook_factory(name): def hook(module, input, output): if isinstance(output, torch.Tensor): self.stats[name] { mean: output.mean().item(), std: output.std().item(), min: output.min().item(), max: output.max().item() } return hook for name, module in self.model.named_modules(): if isinstance(module, (nn.ReLU, nn.Sigmoid, nn.Tanh, nn.Linear, nn.Conv2d)): self.hooks.append(module.register_forward_hook(hook_factory(name))) def remove_hooks(self): for hook in self.hooks: hook.remove() def get_stats(self): return self.stats # 使用示例 stats ActivationStats(model) with torch.no_grad(): model(test_input) activation_stats stats.get_stats() stats.remove_hooks() for layer, stat in activation_stats.items(): print(f{layer}: mean{stat[mean]:.4f}, std{stat[std]:.4f})14. 统计质量控制与异常检测14.1 基于统计的过程控制class SPCControl: def __init__(self, window_size100): self.window_size window_size self.data_window [] self.mean_history [] self.std_history [] def update(self, new_value): self.data_window.append(new_value) if len(self.data_window) self.window_size: self.data_window.pop(0) current_data torch.tensor(self.data_window) self.mean_history.append(torch.mean(current_data).item()) self.std_history.append(torch.std(current_data).item()) def check_control_limits(self, new_value, sigma_level3): if len(self.mean_history) 0: return False current_mean self.mean_history[-1] current_std self.std_history[-1] upper_limit current_mean sigma_level * current_std lower_limit current_mean - sigma_level * current_std return new_value upper_limit or new_value lower_limit def plot_control_chart(self): plt.figure(figsize(12,6)) plt.plot(self.mean_history, labelMoving Mean) plt.plot([m 3*s for m, s in zip(self.mean_history, self.std_history)], r--, labelUpper Control Limit) plt.plot([m - 3*s for m, s in zip(self.mean_history, self.std_history)], r--, labelLower Control Limit) plt.title(Statistical Process Control Chart) plt.xlabel(Sample) plt.ylabel(Value) plt.legend() plt.show() # 使用示例 spc SPCControl(window_size50) for i in range(200): value torch.randn(1).item() * (1 0.1*(i100)) # 模拟过程偏移 spc.update(value) if spc.check_control_limits(value): print(fAlert at step {i}: Value {value:.2f} out of control limits) spc.plot_control_chart()14.2 异常值检测def detect_outliers(data, methodiqr, threshold1.5): 异常值检测方法: iqr - 四分位距法 zscore - Z分数法 modified_zscore - 修正Z分数法 if method iqr: q1 torch.quantile(data, 0.25) q3 torch.quantile(data, 0.75) iqr q3 - q1 lower_bound q1 - threshold * iqr upper_bound q3 threshold * iqr elif method zscore: mean torch.mean(data) std torch.std(data) lower_bound mean - threshold * std upper_bound mean threshold * std elif method modified_zscore: median torch.median(data) mad torch.median(torch.abs(data - median)) modified_z 0.6745 * (data - median) / mad outliers torch.abs(modified_z) threshold return outliers return (data lower_bound) | (data upper_bound) # 使用示例 data torch.cat([torch.randn(100), torch.tensor([10.0, -5.0])]) # 添加两个异常值 outliers detect_outliers(data, methodiqr) print(f检测到的异常值索引: {torch.where(outliers)[0].tolist()})14.3 数据漂移检测class DataDriftDetector: def __init__(self, reference_data, n_bins10): self.reference_data reference_data self.n_bins n_bins self.bin_edges torch.linspace( torch.min(reference_data), torch.max(reference_data), n_bins 1 ) def compute_distribution(self, data): hist torch.histc(data, binsself.n_bins, minself.bin_edges[0], maxself.bin_edges[-1]) return hist / hist.sum() def detect_drift(self, new_data, threshold0.1): ref_dist self