泉州模板建站定制淮北官方网站

泉州模板建站定制,淮北官方网站,南京网站制作公司电话,专注旅游网站网站开发主要包含两个部分一个代理模型#xff08;surrogate model#xff09;#xff0c;用于对目标函数进行建模。代理模型通常有确定的公式或者能计算梯度#xff0c;又或者有已知的凹凸性、线性等特性#xff0c;总之就是更容易用于优化。更泛化地讲#xff0c;其实它就是一个…主要包含两个部分一个代理模型surrogate model用于对目标函数进行建模。代理模型通常有确定的公式或者能计算梯度又或者有已知的凹凸性、线性等特性总之就是更容易用于优化。更泛化地讲其实它就是一个学习模型输入是所有观测到的函数值点训练后可以在给定任意x的情况下给出对f(x)的估计。一个优化策略optimization strategy决定下一个采样点的位置即下一步应在哪个输入x处观测函数值f(x)。通常它是通过采集函数acquisition function 来实现的采集函数通常是一个由代理模型推出的函数它的输入是可行集feasible setA上的任意值输出值衡量了每个输入x有多值得被观测。通常会从以下两方面考虑有多大的可能性在x处取得最优值评估x是否能减少贝叶斯统计模型的不确定性采集函数通常也是容易求最优值的函数例如有公式/能算梯度等下一个采样点就是可行集上的最大值点即使采集函数的取最大值的点。本文主要学习代理模型模型代码参考SCOOT(WWW2025 oral)中hebo/model库的代码实现常用的贝叶斯代理模型GP标准高斯过程class GP(BaseModel):# support_grad True 表示该模型支持梯度计算这对于贝叶斯优化等需要梯度信息的场景非常重要support_grad Truedef __init__(self, num_cont, num_enum, num_out, **conf):初始化高斯过程回归模型参数:num_cont (int): 连续变量的数量num_enum (int): 离散/枚举变量的数量num_out (int): 输出变量的数量通常为1表示单目标优化**conf: 配置参数字典可包含以下键值对- lr (float): 学习率默认为3e-2- num_epochs (int): 训练轮数默认为100- verbose (bool): 是否打印训练过程信息默认为False- print_every (int): 打印训练信息的频率默认为10- pred_likeli (bool): 预测时是否考虑似然噪声默认为True- noise_lb (float): 噪声下界默认为1e-5- optimizer (str): 优化器类型可选lbfgs、psgld或adam默认为psgld- noise_guess (float): 噪声初始猜测值默认为0.01- ard_kernel (bool): 是否使用ARD核自动相关性判定默认为True# 调用父类BaseModel的初始化方法super().__init__(num_cont, num_enum, num_out, **conf)# 从配置中提取参数如果未提供则使用默认值self.lr conf.get(lr, 3e-2) # 学习率self.num_epochs conf.get(num_epochs, 100) # 训练轮数self.verbose conf.get(verbose, False) # 是否打印训练过程信息self.print_every conf.get(print_every, 10) # 打印训练信息的频率self.pred_likeli conf.get(pred_likeli, True) # 预测时是否考虑似然噪声self.noise_lb conf.get(noise_lb, 1e-5) # 噪声下界self.optimizer conf.get(optimizer, psgld) # 优化器类型self.noise_guess conf.get(noise_guess, 0.01) # 噪声初始猜测值self.ard_kernel conf.get(ard_kernel, True) # 是否使用ARD核# 初始化数据标准化器# xscaler: 用于连续变量的最小-最大标准化器将数据缩放到[-1, 1]区间self.xscaler TorchMinMaxScaler((-1, 1))# yscaler: 用于目标变量的标准标准化器使其均值为0标准差为1self.yscaler TorchStandardScaler()def fit_scaler(self, Xc : Tensor, Xe : Tensor, y : Tensor):拟合数据标准化器参数:Xc: 连续特征张量Xe: 离散特征张量y: 目标变量张量# 如果存在连续变量且数量大于0则拟合连续变量标准化器if Xc is not None and Xc.shape[1] 0:self.xscaler.fit(Xc)# 拟合目标变量标准化器self.yscaler.fit(y)def xtrans(self, Xc : Tensor, Xe : Tensor, y : Tensor None):对输入数据进行转换标准化和类型转换参数:Xc: 连续特征张量Xe: 离散特征张量y: 可选的目标变量张量返回:转换后的特征和目标变量如果提供了y# 处理连续变量如果存在则标准化否则创建空张量if Xc is not None and Xc.shape[1] 0:Xc_t self.xscaler.transform(Xc) # 标准化连续变量else:Xc_t torch.zeros(Xe.shape[0], 0) # 创建空的连续变量张量# 处理离散变量如果存在则转换为long类型否则创建空张量if Xe is None:Xe_t torch.zeros(Xc.shape[0], 0).long() # 创建空的离散变量张量else:Xe_t Xe.long() # 转换为long类型通常用于嵌入层# 如果提供了目标变量则也进行标准化if y is not None:y_t self.yscaler.transform(y) # 标准化目标变量return Xc_t, Xe_t, y_telse:return Xc_t, Xe_tdef fit(self, Xc : Tensor, Xe : Tensor, y : Tensor):训练高斯过程模型参数:Xc: 连续特征张量Xe: 离散特征张量y: 目标变量张量# 1. 数据预处理过滤NaN值Xc, Xe, y filter_nan(Xc, Xe, y, all)# 2. 拟合标准化器并转换数据self.fit_scaler(Xc, Xe, y)Xc, Xe, y self.xtrans(Xc, Xe, y)# 3. 验证数据维度assert(Xc.shape[1] self.num_cont)assert(Xe.shape[1] self.num_enum)assert(y.shape[1] self.num_out)# 4. 保存训练数据self.Xc Xcself.Xe Xeself.y y# 5. 设置似然函数噪声模型n_constr GreaterThan(self.noise_lb) # 噪声下界约束n_prior LogNormalPrior(np.log(self.noise_guess), 0.5) # 对数正态先验# 创建高斯似然函数包含噪声约束和先验self.lik GaussianLikelihood(noise_constraint n_constr, noise_prior n_prior)# 6. 创建GPyTorch模型self.gp GPyTorchModel(self.Xc, self.Xe, self.y, self.lik, **self.conf)# 7. 初始化噪声参数self.gp.likelihood.noise max(1e-2, self.noise_lb)# 8. 设置模型为训练模式self.gp.train()self.lik.train()# 9. 选择优化器if self.optimizer.lower() lbfgs:# L-BFGS优化器适用于小批量数据的准牛顿方法opt torch.optim.LBFGS(self.gp.parameters(), lr self.lr, max_iter 5, line_search_fn strong_wolfe)elif self.optimizer psgld:# 预处理随机梯度Langevin动力学用于贝叶斯采样opt pSGLD(self.gp.parameters(), lr self.lr, factor 1. / y.shape[0], pretrain_step self.num_epochs // 10)else:# Adam优化器默认选择opt torch.optim.Adam(self.gp.parameters(), lr self.lr)# 10. 创建边际对数似然目标函数mll gpytorch.mlls.ExactMarginalLogLikelihood(self.lik, self.gp)# 11. 训练循环for epoch in range(self.num_epochs):def closure():优化器的闭包函数计算损失并梯度# 前向传播获取GP的后验分布dist self.gp(self.Xc, self.Xe)# 计算负边际对数似然损失loss -1 * mll(dist, self.y.squeeze())# 清零梯度opt.zero_grad()# 反向传播loss.backward()return loss# 优化器步进opt.step(closure)# 打印训练信息if self.verbose and ((epoch 1) % self.print_every 0 or epoch 0):print(After %d epochs, loss %g % (epoch 1, closure().item()), flush True)# 12. 设置模型为评估模式self.gp.eval()self.lik.eval()def predict(self, Xc, Xe):使用训练好的模型进行预测参数:Xc: 连续特征张量Xe: 离散特征张量返回:mu: 预测均值var: 预测方差# 1. 转换输入数据Xc, Xe self.xtrans(Xc, Xe)# 2. 使用快速预测设置进行预测with gpytorch.settings.fast_pred_var(), gpytorch.settings.debug(False):# 获取GP预测pred self.gp(Xc, Xe)# 如果考虑似然噪声通过似然函数进行预测if self.pred_likeli:pred self.lik(pred)# 提取均值和方差mu_ pred.mean.reshape(-1, self.num_out)var_ pred.variance.reshape(-1, self.num_out)# 3. 将预测结果转换回原始尺度mu self.yscaler.inverse_transform(mu_) # 逆标准化均值var var_ * self.yscaler.std**2 # 调整方差到原始尺度# 4. 确保方差不为零避免数值问题return mu, var.clamp(min torch.finfo(var.dtype).eps)def sample_y(self, Xc, Xe, n_samples 1) - FloatTensor:从后验分布中采样目标变量参数:Xc: 连续特征张量Xe: 离散特征张量n_samples: 采样数量返回:采样结果形状为(n_samples, 样本数, 输出维度)# 1. 转换输入数据Xc, Xe self.xtrans(Xc, Xe)# 2. 进行采样with gpytorch.settings.debug(False):if self.pred_likeli:# 考虑噪声的采样pred self.lik(self.gp(Xc, Xe))else:# 不考虑噪声的采样函数值采样pred self.gp(Xc, Xe)# 从后验分布中采样samp pred.rsample(torch.Size((n_samples,))).view(n_samples, Xc.shape[0], self.num_out)# 将采样结果转换回原始尺度return self.yscaler.inverse_transform(samp)def sample_f(self):采样函数值不支持使用sample_y代替raise NotImplementedError(Thompson sampling is not supported for GP, use sample_y instead)propertydef noise(self):获取估计的噪声水平转换回原始尺度return (self.gp.likelihood.noise * self.yscaler.std**2).view(self.num_out).detach()class GPyTorchModel(gpytorch.models.ExactGP):GPyTorch模型实现类继承自ExactGP精确高斯过程def __init__(self,x : torch.Tensor, # 训练数据的连续特征xe : torch.Tensor, # 训练数据的离散特征y : torch.Tensor, # 训练目标值lik : GaussianLikelihood, # 似然函数**conf): # 配置参数# 调用父类构造函数传入训练数据和似然函数super().__init__((x, xe), y.squeeze(), lik)# 特征提取器处理连续和离散特征的组合self.fe deepcopy(conf.get(fe, DummyFeatureExtractor(x.shape[1], xe.shape[1], conf.get(num_uniqs), conf.get(emb_sizes))))# 均值函数默认使用常数均值self.mean deepcopy(conf.get(mean, ConstantMean()))# 核函数选择根据配置选择默认核或随机分解核if conf.get(rd, False):# 使用随机分解核Random Decompositionself.cov deepcopy(conf.get(kern, default_kern_rd(x, xe, y, self.fe.total_dim, conf.get(ard_kernel, True), conf.get(fe), Econf.get(E, 0.2))))else:# 使用默认核函数self.cov deepcopy(conf.get(kern, default_kern(x, xe, y, self.fe.total_dim, conf.get(ard_kernel, True), conf.get(fe))))def forward(self, x, xe):前向传播定义高斯过程的行为参数:x: 连续特征xe: 离散特征返回:MultivariateNormal: 多元正态分布表示高斯过程的预测# 1. 特征提取将连续和离散特征组合成统一表示x_all self.fe(x, xe)# 2. 计算均值函数m self.mean(x_all)# 3. 计算协方差矩阵核函数K self.cov(x_all)# 4. 返回多元正态分布return MultivariateNormal(m, K)GPyGP:输入扭曲的高斯过程模型class GPyGP(BaseModel):Input warped GP model implemented using GPy instead of GPyTorch使用GPy库实现的输入扭曲高斯过程模型而非GPyTorchWhy doing so: 为什么这样做- Input warped GP 支持输入扭曲的高斯过程def __init__(self, num_cont, num_enum, num_out, **conf):super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化total_dim num_cont # 总维度初始化为连续变量数量if num_enum 0: # 如果存在离散变量self.one_hot OneHotTransform(self.conf[num_uniqs]) # 创建one-hot编码器total_dim self.one_hot.num_out # 增加离散变量编码后的维度self.xscaler TorchMinMaxScaler((-1, 1)) # 连续变量标准化器范围[-1,1]self.yscaler TorchStandardScaler() # 目标变量标准化器self.verbose self.conf.get(verbose, False) # 是否显示训练信息self.num_epochs self.conf.get(num_epochs, 200) # 训练轮数self.warp self.conf.get(warp, True) # 是否使用输入扭曲self.space self.conf.get(space) # 设计空间定义self.num_restarts self.conf.get(num_restarts, 10) # 优化重启次数self.rd self.conf.get(rd, False) # 是否使用随机分解self.E self.conf.get(E, 0.2) # 随机分解的边概率if self.space is None and self.warp: # 如果没有设计空间但启用了扭曲warnings.warn(Space not provided, set warp to False) # 发出警告self.warp False # 禁用扭曲if self.warp: # 如果启用扭曲for i in range(total_dim): # 为每个维度禁用日志记录器logging.getLogger(fa{i}).disabled True # 禁用参数a的日志logging.getLogger(fb{i}).disabled True # 禁用参数b的日志def fit_scaler(self, Xc : FloatTensor, y : FloatTensor):if Xc is not None and Xc.shape[1] 0: # 如果存在连续变量if self.space is not None: # 如果有设计空间信息cont_lb self.space.opt_lb[:self.space.num_numeric].view(1, -1).float() # 获取连续变量下界cont_ub self.space.opt_ub[:self.space.num_numeric].view(1, -1).float() # 获取连续变量上界self.xscaler.fit(torch.cat([Xc, cont_lb, cont_ub], dim 0)) # 结合数据和边界来拟合标准化器else:self.xscaler.fit(Xc) # 仅使用数据拟合标准化器self.yscaler.fit(y) # 拟合目标变量标准化器def trans(self, Xc : Tensor, Xe : Tensor, y : Tensor None):if Xc is not None and Xc.shape[1] 0: # 处理连续变量Xc_t self.xscaler.transform(Xc) # 标准化连续变量else:Xc_t torch.zeros(Xe.shape[0], 0) # 创建空的连续变量张量if Xe is None or Xe.shape[1] 0: # 处理离散变量Xe_t torch.zeros(Xc.shape[0], 0) # 创建空的离散变量张量else:Xe_t self.one_hot(Xe.long()) # one-hot编码离散变量Xall torch.cat([Xc_t, Xe_t], dim 1) # 合并连续和离散特征if y is not None: # 如果提供了目标变量y_t self.yscaler.transform(y) # 标准化目标变量return Xall.numpy(), y_t.numpy() # 转换为numpy数组返回return Xall.numpy() # 只返回特征数据def fit(self, Xc : FloatTensor, Xe : LongTensor, y : LongTensor):Xc, Xe, y filter_nan(Xc, Xe, y, all) # 过滤NaN值self.fit_scaler(Xc, y) # 拟合标准化器X, y self.trans(Xc, Xe, y) # 转换数据格式if self.rd: # 如果使用随机分解cliques get_random_graph(X.shape[1], self.E) # 生成随机图团结构# process first clique 处理第一个团pair cliques[0] # 获取第一个团k1 GPy.kern.Linear(len(pair), active_dimspair, ARD False) # 线性核k2 GPy.kern.Matern32(len(pair), active_dimspair, ARD True) # Matern32核启用ARDk2.lengthscale np.std(X, axis 0)[pair] # 设置长度尺度为特征标准差k2.variance 0.5 # 设置核方差k2.variance.set_prior(GPy.priors.Gamma(0.5, 1)) # 设置方差先验为Gamma分布kern k1 k2 # 线性核 Matern核# process remaining cliques 处理剩余团for pair in cliques[1:]:k1 GPy.kern.Linear(len(pair), active_dimspair, ARD False) # 线性核k2 GPy.kern.Matern32(len(pair), active_dimspair, ARD True) # Matern32核geo_mean 1 # 初始化几何均值for d in pair: # 计算团内特征的几何标准差geo_mean * np.std(X, axis 0)[d]k2.lengthscale geo_mean**(1/len(pair)) # 设置长度尺度为几何均值k2.variance 0.5 # 设置核方差k2.variance.set_prior(GPy.priors.Gamma(0.5, 1)) # 设置方差先验kern k1 k2 # 累加到总核函数else: # 如果不使用随机分解k1 GPy.kern.Linear(X.shape[1], ARD False) # 全局线性核k2 GPy.kern.Matern32(X.shape[1], ARD True) # 全局Matern32核启用ARDk2.lengthscale np.std(X, axis 0).clip(min 0.02) # 设置长度尺度最小0.02k2.variance 0.5 # 设置核方差k2.variance.set_prior(GPy.priors.Gamma(0.5, 1), warning False) # 设置方差先验kern k1 k2 # 线性核 Matern核if not self.warp: # 如果不使用输入扭曲self.gp GPy.models.GPRegression(X, y, kern) # 创建标准高斯过程回归else: # 如果使用输入扭曲xmin np.zeros(X.shape[1]) # 初始化最小边界xmax np.ones(X.shape[1]) # 初始化最大边界xmin[:Xc.shape[1]] -1 # 设置连续变量范围为[-1,1]warp_f GPy.util.input_warping_functions.KumarWarping(X, Xmin xmin, Xmax xmax) # 创建Kumar扭曲函数self.gp GPy.models.InputWarpedGP(X, y, kern, warping_function warp_f) # 创建输入扭曲高斯过程self.gp.likelihood.variance.set_prior(GPy.priors.LogGaussian(-4.63, 0.5), warning False) # 设置噪声先验# 多重启优化最大迭代次数、是否显示信息、重启次数、鲁棒模式self.gp.optimize_restarts(max_iters self.num_epochs, verbose self.verbose, num_restarts self.num_restarts, robust True)return self # 返回自身用于链式调用def predict(self, Xc : FloatTensor, Xe : LongTensor) - (FloatTensor, FloatTensor):Xall self.trans(Xc, Xe) # 转换输入数据py, ps2 self.gp.predict(Xall) # 使用GPy模型预测均值和方差mu self.yscaler.inverse_transform(FloatTensor(py).view(-1, 1)) # 逆标准化均值var self.yscaler.std**2 * FloatTensor(ps2).view(-1, 1) # 调整方差到原始尺度return mu, var.clamp(torch.finfo(var.dtype).eps) # 返回均值和确保非负的方差def sample_f(self):raise NotImplementedError(Thompson sampling is not supported for GP, use sample_y instead) # 不支持函数采样propertydef noise(self):var_normalized self.gp.likelihood.variance[0] # 获取标准化后的噪声方差return (var_normalized * self.yscaler.std**2).view(self.num_out) # 转换到原始尺度并调整形状RF随机森林回归class RF(BaseModel):随机森林回归模型实现def __init__(self, num_cont, num_enum, num_out, **conf):super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化self.n_estimators self.conf.get(n_estimators, 100) # 树的数量默认100self.rf RandomForestRegressor(n_estimators self.n_estimators) # 创建随机森林回归器self.est_noise torch.zeros(self.num_out) # 初始化估计噪声为零张量if self.num_enum 0: # 如果存在离散变量self.one_hot OneHotTransform(self.conf[num_uniqs]) # 创建one-hot编码器def xtrans(self, Xc : FloatTensor, Xe: LongTensor) - np.ndarray:转换输入数据为numpy数组格式if self.num_enum 0: # 如果没有离散变量return Xc.detach().numpy() # 直接返回连续变量的numpy数组else: # 如果有离散变量Xe_one_hot self.one_hot(Xe) # 对离散变量进行one-hot编码if Xc is None: # 如果没有连续变量Xc torch.zeros(Xe.shape[0], 0) # 创建空的连续变量张量return torch.cat([Xc, Xe_one_hot], dim 1).numpy() # 合并连续和离散特征并转为numpydef fit(self, Xc : torch.Tensor, Xe : torch.Tensor, y : torch.Tensor):训练随机森林模型Xc, Xe, y filter_nan(Xc, Xe, y, all) # 过滤包含NaN的数据点Xtr self.xtrans(Xc, Xe) # 转换输入特征为numpy格式ytr y.numpy().reshape(-1) # 转换目标变量为一维numpy数组self.rf.fit(Xtr, ytr) # 训练随机森林模型# 计算训练集上的MSE作为噪声估计mse np.mean((self.rf.predict(Xtr).reshape(-1) - ytr)**2).reshape(self.num_out)self.est_noise torch.FloatTensor(mse) # 保存估计的噪声水平propertydef noise(self):返回估计的噪声水平return self.est_noisedef predict(self, Xc : torch.Tensor, Xe : torch.Tensor):使用训练好的模型进行预测返回: (预测均值, 预测方差 估计噪声)X self.xtrans(Xc, Xe) # 转换输入数据mean self.rf.predict(X).reshape(-1, 1) # 预测均值所有树的平均preds [] # 存储每棵树的独立预测for estimator in self.rf.estimators_: # 遍历所有决策树preds.append(estimator.predict(X).reshape([-1,1])) # 收集每棵树的预测var np.var(np.concatenate(preds, axis1), axis1) # 计算树间预测方差# 返回均值和总方差模型方差 估计噪声return torch.FloatTensor(mean.reshape([-1,1])), torch.FloatTensor(var.reshape([-1,1])) self.noiseSVGP稀疏变分高斯过程(Stochastic Variational Gaussian Process)class SVGP(BaseModel):稀疏变分高斯过程 (Stochastic Variational Gaussian Process)适用于大规模数据的近似高斯过程模型support_grad True # 支持梯度计算support_multi_output True # 支持多输出任务def __init__(self, num_cont, num_enum, num_out, **conf):super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化# 配置参数self.use_ngd conf.get(use_ngd, False) # 是否使用自然梯度下降self.lr conf.get(lr, 1e-2) # 基础学习率self.lr_vp conf.get(lr_vp, 1e-1) # 变分参数学习率self.lr_fe conf.get(lr_fe, 1e-3) # 特征提取器学习率self.num_inducing conf.get(num_inducing, 128) # 诱导点数量self.ard_kernel conf.get(ard_kernel, True) # 是否使用ARD核self.pred_likeli conf.get(pred_likeli, True) # 预测时是否考虑噪声self.beta conf.get(beta, 1.0) # ELBO的beta参数# 训练参数self.batch_size conf.get(batch_size, 64) # 批大小self.num_epochs conf.get(num_epochs, 300) # 训练轮数self.verbose conf.get(verbose, False) # 是否显示训练信息self.print_every conf.get(print_every, 10) # 打印频率self.noise_lb conf.get(noise_lb, 1e-5) # 噪声下界# 数据标准化器self.xscaler TorchMinMaxScaler((-1, 1)) # 输入标准化器self.yscaler TorchStandardScaler() # 输出标准化器def fit_scaler(self, Xc : FloatTensor, Xe : LongTensor, y : FloatTensor):拟合数据标准化器if Xc is not None and Xc.shape[1] 0: # 如果有连续变量self.xscaler.fit(Xc) # 拟合输入标准化器self.yscaler.fit(y) # 拟合输出标准化器def xtrans(self, Xc : FloatTensor, Xe : LongTensor, y : FloatTensor None):转换输入数据格式if Xc is not None and Xc.shape[1] 0: # 处理连续变量Xc_t self.xscaler.transform(Xc) # 标准化连续变量else:Xc_t torch.zeros(Xe.shape[0], 0) # 创建空的连续变量张量if Xe is None: # 处理离散变量Xe_t torch.zeros(Xc.shape[0], 0).long()else:Xe_t Xe.long() # 确保离散变量为long类型if y is not None: # 如果提供了目标变量y_t self.yscaler.transform(y) # 标准化目标变量return Xc_t, Xe_t, y_telse:return Xc_t, Xe_tdef fit(self, Xc : FloatTensor, Xe : LongTensor, y : FloatTensor):训练SVGP模型Xc, Xe, y filter_nan(Xc, Xe, y, any) # 过滤包含NaN的数据点self.fit_scaler(Xc, Xe, y) # 拟合标准化器Xc, Xe, y self.xtrans(Xc, Xe, y) # 转换数据格式# 验证数据维度assert(Xc.shape[1] self.num_cont)assert(Xe.shape[1] self.num_enum)assert(y.shape[1] self.num_out)# 设置噪声约束和创建模型n_constr GreaterThan(self.noise_lb) # 噪声下界约束self.gp SVGPModel(Xc, Xe, y, **self.conf) # 创建SVGP模型# 为每个输出创建独立的高斯似然函数self.lik nn.ModuleList([GaussianLikelihood(noise_constraint n_constr) for _ in range(self.num_out)])# 设置模型为训练模式self.gp.train()self.lik.train()# 创建数据加载器ds TensorDataset(Xc, Xe, y) # 创建Tensor数据集dl DataLoader(ds, batch_size self.batch_size, shuffle True, drop_last y.shape[0] self.batch_size) # 数据加载器# 配置优化器if self.use_ngd: # 如果使用自然梯度下降opt torch.optim.Adam([{params : self.gp.fe.parameters(), lr : self.lr_fe}, # 特征提取器参数{params : self.gp.gp.hyperparameters()}, # GP超参数{params : self.lik.parameters()}, # 似然函数参数], lr self.lr)opt_ng gpytorch.optim.NGD(self.gp.variational_parameters(), lr self.lr_vp, num_data y.shape[0]) # 自然梯度优化器else: # 使用标准Adam优化器opt torch.optim.Adam([{params : self.gp.fe.parameters(), lr : self.lr_fe}, # 特征提取器参数{params : self.gp.gp.hyperparameters()}, # GP超参数{params : self.gp.gp.variational_parameters(), lr : self.lr_vp}, # 变分参数{params : self.lik.parameters()}, # 似然函数参数], lr self.lr)# 为每个输出创建变分ELBO目标函数mll [gpytorch.mlls.VariationalELBO(self.lik[i], self.gp.gp[i], num_data y.shape[0], beta self.beta) for i in range(self.num_out)]# 训练循环for epoch in range(self.num_epochs):epoch_loss 0. # 累计损失epoch_cnt 1e-6 # 批次计数避免除零for bxc, bxe, by in dl: # 遍历数据批次dist_list self.gp(bxc, bxe, by) # 前向传播获取分布列表loss 0 # 初始化损失valid torch.isfinite(by) # 检查有效值# 计算每个输出的损失for i, dist in enumerate(dist_list):loss -1 * mll[i](dist, by[valid[:, i], i]) * valid[:, i].sum() # 加权ELBO损失loss / by.shape[0] # 平均损失# 反向传播和优化if self.use_ngd:opt.zero_grad()opt_ng.zero_grad()loss.backward()opt.step()opt_ng.step()else:opt.zero_grad()loss.backward()opt.step()# 累计损失和计数epoch_loss loss.item()epoch_cnt 1epoch_loss / epoch_cnt # 计算平均epoch损失if self.verbose and ((epoch 1) % self.print_every 0 or epoch 0):print(After %d epochs, loss %g % (epoch 1, epoch_loss), flush True)# 设置模型为评估模式self.gp.eval()self.lik.eval()def predict(self, Xc, Xe):模型预测Xc, Xe self.xtrans(Xc, Xe) # 转换输入数据with gpytorch.settings.fast_pred_var(), gpytorch.settings.debug(False): # 快速预测设置pred self.gp(Xc, Xe) # 获取预测分布if self.pred_likeli: # 如果考虑似然噪声for i in range(self.num_out):pred[i] self.lik[i](pred[i]) # 通过似然函数转换# 合并所有输出的均值和方差mu_ torch.cat([pred[i].mean.reshape(-1, 1) for i in range(self.num_out)], dim 1)var_ torch.cat([pred[i].variance.reshape(-1, 1) for i in range(self.num_out)], dim 1)# 逆标准化到原始尺度mu self.yscaler.inverse_transform(mu_)var var_ * self.yscaler.std**2return mu, var.clamp(min torch.finfo(var.dtype).eps) # 确保方差非负def sample_y(self, Xc, Xe, n_samples 1) - FloatTensor:从后验分布采样目标变量返回: (n_samples, 样本数, 输出维度)Xc, Xe self.xtrans(Xc, Xe) # 转换输入数据with gpytorch.settings.debug(False):pred self.gp(Xc, Xe) # 获取预测分布if self.pred_likeli: # 如果考虑似然噪声for i in range(self.num_out):pred[i] self.lik[i](pred[i]) # 通过似然函数转换# 从每个输出分布采样并合并samp [pred[i].rsample(torch.Size((n_samples, ))).reshape(n_samples, -1, 1) for i in range(self.num_out)]samp torch.cat(samp, dim -1)return self.yscaler.inverse_transform(samp) # 逆标准化采样结果def sample_f(self):不支持函数采样raise NotImplementedError(Thompson sampling is not supported for GP, use sample_y instead)propertydef noise(self):获取估计的噪声水平noise torch.FloatTensor([lik.noise for lik in self.lik]).view(self.num_out).detach() # 各输出噪声return noise * self.yscaler.std**2 # 转换到原始尺度SVIDKL稀疏变分深度核学习(Stochastic Variational Deep Kernel Learning)class DKLFe(nn.Module):深度核学习特征提取器 (Deep Kernel Learning Feature Extractor)使用神经网络自动学习特征表示替代手动设计的核函数def __init__(self, num_cont, num_enum, num_out, **conf):super().__init__()# 神经网络配置参数self.num_hiddens conf.get(num_hiddens, 64) # 隐藏层维度默认64self.num_layers conf.get(num_layers, 2) # 隐藏层数量默认2层self.act conf.get(act, nn.LeakyReLU()) # 激活函数默认LeakyReLUself.sn_norm conf.get(sn_norm) # 谱归一化默认不使用# 特征预处理处理连续和离散变量的嵌入转换self.emb_trans DummyFeatureExtractor(num_cont, num_enum, conf.get(num_uniqs), conf.get(emb_sizes))# 构建深度神经网络特征提取器self.fe construct_hidden(self.emb_trans.total_dim, self.num_layers, self.num_hiddens, self.act, self.sn_norm)self.total_dim self.num_hiddens # 输出特征维度def forward(self, x, xe):前向传播将原始输入转换为深度特征表示x_all self.emb_trans(x, xe) # 首先进行嵌入转换处理混合类型输入return self.fe(x_all) # 通过深度网络提取高级特征class SVIDKL(SVGP):稀疏变分深度核学习 (Stochastic Variational Deep Kernel Learning)在SVGP基础上集成深度特征学习的扩展def __init__(self, num_cont, num_enum, num_out, **conf):# 创建配置的深拷贝避免修改原始配置new_conf deepcopy(conf)# 关键变化1禁用ARD核new_conf.setdefault(ard_kernel, False)# 解释当有神经网络特征提取器时不需要使用ARD核# 因为神经网络已经自动学习了特征的重要性权重# 关键变化2使用深度特征提取器new_conf.setdefault(fe, DKLFe(num_cont, num_enum, num_out, **new_conf))# 解释用深度神经网络替代简单的特征转换自动学习数据表示# 关键变化3使用简化的核函数new_conf.setdefault(kern, ScaleKernel(MaternKernel(nu 2.5)))# 解释由于特征提取器已经处理了复杂性可以使用更简单的核函数# 调用父类SVGP的初始化传入修改后的配置super().__init__(num_cont, num_enum, num_out, **new_conf)CatBoost梯度提升树模型class CatBoost(BaseModel):CatBoost梯度提升树模型实现专门优化类别特征处理的梯度提升算法def __init__(self, num_cont, num_enum, num_out, **conf):super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化# CatBoost配置参数self.num_epochs self.conf.get(num_epochs, 100) # 最大树的数量迭代次数self.lr self.conf.get(lr, 0.2) # 学习率self.depth self.conf.get(depth, 10) # 树深度推荐范围[1, 10]self.loss_function self.conf.get(loss_function, RMSEWithUncertainty) # 损失函数支持不确定性估计self.posterior_sampling self.conf.get(posterior_sampling, True) # 是否使用后验采样self.verbose self.conf.get(verbose, False) # 是否显示训练过程self.random_seed self.conf.get(random_seed, 42) # 随机种子self.num_ensembles self.conf.get(num_ensembles, 10) # 虚拟集成数量# 确保迭代次数足够用于集成if self.num_epochs 2 * self.num_ensembles:self.num_epochs self.num_ensembles * 2 # 至少是集成数量的2倍# 创建CatBoost回归器实例self.model CatBoostRegressor(iterationsself.num_epochs, # 迭代次数learning_rateself.lr, # 学习率depthself.depth, # 树深度loss_functionself.loss_function, # 损失函数支持不确定性posterior_samplingself.posterior_sampling, # 后验采样verboseself.verbose, # 是否显示训练信息random_seedself.random_seed, # 随机种子allow_writing_filesFalse) # 禁止写入文件def xtrans(self, Xc: FloatTensor, Xe: LongTensor) - FeaturesData:转换输入数据为CatBoost所需的格式# 处理连续变量转换为numpy float32格式num_feature_data Xc.numpy().astype(np.float32) if self.num_cont ! 0 else None# 处理离散变量转换为字符串格式CatBoost要求cat_feature_data Xe.numpy().astype(str).astype(object) if self.num_enum ! 0 else None# 返回CatBoost特征数据对象return FeaturesData(num_feature_datanum_feature_data, # 数值特征cat_feature_datacat_feature_data) # 类别特征def fit(self, Xc: FloatTensor, Xe: LongTensor, y: FloatTensor):训练CatBoost模型Xc, Xe, y filter_nan(Xc, Xe, y, all) # 过滤NaN值# 创建训练数据池train_data Pool(dataself.xtrans(XcXc, XeXe), # 转换特征数据labely.numpy().reshape(-1)) # 转换标签为一维数组self.model.fit(train_data) # 训练模型def predict(self, Xc: FloatTensor, Xe: LongTensor) - (FloatTensor, FloatTensor):使用CatBoost进行预测返回均值和方差test_data Pool(dataself.xtrans(XcXc, XeXe)) # 创建测试数据池# 使用虚拟集成进行预测获取不确定性估计preds self.model.virtual_ensembles_predict(datatest_data, # 测试数据prediction_typeTotalUncertainty, # 预测类型总不确定性virtual_ensembles_countself.num_ensembles) # 虚拟集成数量# 解析预测结果# preds[:, 0] 预测均值# preds[:, 1] 知识不确定性模型不确定性# preds[:, 2] 偶然不确定性数据噪声mean preds[:, 0] # 预测均值var preds[:, 1] preds[:, 2] # 总方差 模型方差 数据噪声# 返回PyTorch张量格式的结果return torch.FloatTensor(mean.reshape([-1,1])), \torch.FloatTensor(var.reshape([-1,1]))DeepEnsemble:深度集成模型class DeepEnsemble(BaseModel):深度集成模型 - 通过多个独立训练的神经网络集成来估计不确定性基于Lakshminarayanan et al. (2017) Simple and Scalable Predictive Uncertainty Estimation using Deep Ensembles# 模型能力标识support_ts True # 支持Thompson采样support_grad True # 支持梯度计算support_multi_output True # 支持多输出任务support_warm_start True # 支持热启动从已有模型继续训练def __init__(self, num_cont, num_enum, num_out, **conf):初始化深度集成模型参数:num_cont: 连续变量数量num_enum: 离散变量数量num_out: 输出维度**conf: 配置参数super().__init__(num_cont, num_enum, num_out, **conf) # 调用父类初始化# 集成策略配置self.bootstrap self.conf.setdefault(bootstrap, False) # 是否使用自助采样创建数据多样性self.rand_prior self.conf.setdefault(rand_prior, False) # 是否使用随机先验网络self.output_noise self.conf.setdefault(output_noise, True) # 是否输出噪声估计self.num_ensembles self.conf.setdefault(num_ensembles, 5) # 集成模型数量self.num_process self.conf.setdefault(num_processes, 1) # 并行训练进程数self.num_epochs self.conf.setdefault(num_epochs, 500) # 每个模型的训练轮数self.print_every self.conf.setdefault(print_every, 50) # 训练信息打印频率# 网络架构配置self.num_layers self.conf.setdefault(num_layers, 1) # 隐藏层数量self.num_hiddens self.conf.setdefault(num_hiddens, 128) # 隐藏层维度self.l1 self.conf.setdefault(l1, 1e-3) # L1正则化系数self.batch_size self.conf.setdefault(batch_size, 32) # 批大小self.lr self.conf.setdefault(lr, 5e-3) # 学习率self.adv_eps self.conf.setdefault(adv_eps, 0.) # 对抗训练扰动大小未使用self.verbose self.conf.setdefault(verbose, False) # 是否显示训练详情self.basenet_cls self.conf.setdefault(basenet_cls, BaseNet) # 基础网络类