with tf.variable_scope(self.scope): self.dense1_mul = dense(self.state, dense1_size, weight_init=tf.random_uniform_initializer((-1/tf.sqrt(tf.to_float(self.state_dims))), 1/tf.sqrt(tf.to_float(self.state_dims))), bias_init=tf.random_uniform_initializer((-1/tf.sqrt(tf.to_float(self.state_dims))), 1/tf.sqrt(tf.to_float(self.state_dims))), scope='dense1') self.dense1 = relu(self.dense1_mul, scope='dense1') self.dense2_mul = dense(self.dense1, dense2_size, weight_init=tf.random_uniform_initializer((-1/tf.sqrt(tf.to_float(dense1_size))), 1/tf.sqrt(tf.to_float(dense1_size))), bias_init=tf.random_uniform_initializer((-1/tf.sqrt(tf.to_float(dense1_size))), 1/tf.sqrt(tf.to_float(dense1_size))), scope='dense2') self.dense2 = relu(self.dense2_mul, scope='dense2') self.output_mul = dense(self.dense2, self.action_dims, weight_init=tf.random_uniform_initializer(-1*final_layer_init, final_layer_init), bias_init=tf.random_uniform_initializer(-1*final_layer_init, final_layer_init), scope='output') self.output_tanh = tanh(self.output_mul, scope='output') # Scale tanh output to lower and upper action bounds self.output = tf.multiply(0.5, tf.multiply(self.output_tanh, (self.action_bound_high-self.action_bound_low)) + (self.action_bound_high+self.action_bound_low)) self.network_params = tf.trainable_variables(scope=self.scope) self.bn_params = []

class SelfAttention(nn.Module): def init(self, input_size=1, num_heads=1): super(SelfAttention, self).init() self.num_heads = 1 self.head_size = 1 self.query = nn.Linear(1, 1) self.key = nn.Linear(1, 1) self.value = nn.Linear(1, 1) self.out = nn.Linear(1, 1) def forward(self, inputs): batch_size, seq_len, input_size = inputs.size() # 128 706 1 # Split inputs into num_heads inputs = inputs.view(batch_size, seq_len, self.num_heads, self.head_size) inputs = inputs.permute(0, 2, 1, 3).contiguous() queries = self.query(inputs).view(batch_size, self.num_heads, seq_len, self.head_size) keys = self.key(inputs).view(batch_size, self.num_heads, seq_len, self.head_size) values = self.value(inputs).view(batch_size, self.num_heads, seq_len, self.head_size) # Compute attention scores scores = torch.matmul(queries, keys.permute(0, 1, 3, 2)) scores = scores / (self.head_size ** 0.5) attention = F.softmax(scores, dim=-1) # Apply attention weights to values attention_output = torch.matmul(attention, values) attention_output = attention_output.view(batch_size, seq_len, input_size) # Apply output linear layer output = self.out(attention_output) return output class DenseAttentionLayer(nn.Module): def init(self, input_size, return_alphas=True, name=None, num_heads=1): super(DenseAttentionLayer, self).init() self.return_alphas = return_alphas self.name = name self.num_heads = num_heads # If input comes with a hidden dimension (e.g. 5 features per gene) # print("len(input_size): ",len(input_size)) # 2 if len(input_size) == 3: self.feature_collapse = nn.Linear(input_size[-1], 1) input_size = (input_size[0], input_size[1]) self.attention = SelfAttention(input_size=1, num_heads=1) def forward(self, inputs): print("inputs.shape: ",inputs.shape) # torch.Size([128, 706]) output = self.attention(inputs) if self.return_alphas: alphas = F.softmax(output, dim=1) return torch.mul(inputs, alphas), alphas else: return output 对于上述代码其中numheads=1 headsize=1

这段代码实现了一个自注意力层（Self-Attention Layer）和一个稠密注意力层（Dense Attention Layer）。在自注意力层中，输入被划分为多个头（num_heads），每个头的大小为head_size。然后，通过三个线性层（query...

import tkinter as tk from tkinter import ttk, filedialog, messagebox import pandas as pd import numpy as np import matplotlib as mpl import matplotlib.pyplot as plt from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg import tensorflow as tf from tensorflow.keras.models import Model from tensorflow.keras.layers import Input, Dense, Lambda from tensorflow.keras.optimizers import Adam from sklearn.preprocessing import MinMaxScaler import os import time import warnings import matplotlib.dates as mdates warnings.filterwarnings('ignore', category=UserWarning, module='tensorflow') mpl.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'Arial Unicode MS'] mpl.rcParams['axes.unicode_minus'] = False # 关键修复：使用 ASCII 减号 # 设置中文字体支持 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False class PINNModel(tf.keras.Model): def init(self, num_layers=4, hidden_units=32, dropout_rate=0.1, kwargs): super(PINNModel, self).init(kwargs) self.dense_layers = [] self.dropout_layers = [] # 创建隐藏层和对应的Dropout层 for _ in range(num_layers): self.dense_layers.append(Dense(hidden_units, activation='tanh')) self.dropout_layers.append(tf.keras.layers.Dropout(dropout_rate)) self.final_layer = Dense(1, activation='linear') # 添加更多带约束的物理参数 # 基本衰减系数 self.k1_raw = tf.Variable(0.1, trainable=True, dtype=tf.float32, name='k1_raw') self.k1 = tf.math.sigmoid(self.k1_raw) * 0.5 # 约束在0-0.5之间 # 水位依赖的衰减系数 self.k2_raw = tf.Variable(0.01, trainable=True, dtype=tf.float32, name='k2_raw') self.k2 = tf.math.sigmoid(self.k2_raw) * 0.1 # 约束在0-0.1之间 # 非线性项系数 self.alpha_raw = tf.Variable(0.1, trainable=True, dtype=tf.float32, name='alpha_raw') self.alpha = tf.math.sigmoid(self.alpha_raw) * 1.0 # 约束在0-1.0之间 # 外部影响系数（如降雨、温度等） self.beta_raw = tf.Variable(0.05, trainable=True, dtype=tf.float32, name='beta_raw') self.beta = tf.math.sigmoid(self.beta_raw) * 0.2 # 约束在0-0.2之间 def call(self, inputs, training=False): t, h, dt = inputs # 添加更多特征交互项 interaction = tf.concat([ t * h, h * dt, t * dt, (t * h * dt) ], axis=1) # 将时间、水位和时间步长作为输入特征 x = tf.concat([t, h, dt, interaction], axis=1) # 依次通过所有隐藏层和Dropout层 for dense_layer, dropout_layer in zip(self.dense_layers, self.dropout_layers): x = dense_layer(x) x = dropout_layer(x, training=training) # 仅在训练时应用Dropout return self.final_layer(x) def physics_loss(self, t, h_current, dt, training=False): """计算物理损失(改进的离散渗流方程)""" # 预测下一时刻的水位 h_next_pred = self([t, h_current, dt], training=training) # 改进的物理方程：非线性衰减模型 + 外部影响项 # 添加数值保护 exponent = - (self.k1 + self.k2 * h_current) * dt exponent = tf.clip_by_value(exponent, -50.0, 50.0) # 防止指数爆炸 decay_term = h_current * tf.exp(exponent) # 同样保护第二个指数项 beta_exp = -self.beta * dt beta_exp = tf.clip_by_value(beta_exp, -50.0, 50.0) external_term = self.alpha * (1 - tf.exp(beta_exp)) residual = h_next_pred - (decay_term + external_term) return tf.reduce_mean(tf.square(residual)) class DamSeepageModel: def init(self, root): self.root = root self.root.title("大坝渗流预测模型(PINNs)") self.root.geometry("1200x800") # 初始化数据 self.train_df = None # 训练集 self.test_df = None # 测试集 self.model = None self.scaler_t = MinMaxScaler(feature_range=(0, 1)) self.scaler_h = MinMaxScaler(feature_range=(0, 1)) self.scaler_dt = MinMaxScaler(feature_range=(0, 1)) self.evaluation_metrics = {} # 创建主界面 self.create_widgets() def create_widgets(self): # 创建主框架 main_frame = ttk.Frame(self.root, padding=10) main_frame.pack(fill=tk.BOTH, expand=True) # 左侧控制面板 control_frame = ttk.LabelFrame(main_frame, text="模型控制", padding=10) control_frame.pack(side=tk.LEFT, fill=tk.Y, padx=5, pady=5) # 文件选择部分 file_frame = ttk.LabelFrame(control_frame, text="数据文件", padding=10) file_frame.pack(fill=tk.X, pady=5) # 训练集选择 ttk.Label(file_frame, text="训练集:").grid(row=0, column=0, sticky=tk.W, pady=5) self.train_file_var = tk.StringVar() ttk.Entry(file_frame, textvariable=self.train_file_var, width=30, state='readonly').grid( row=0, column=1, padx=5) ttk.Button(file_frame, text="选择文件", command=lambda: self.select_file("train")).grid(row=0, column=2) # 测试集选择 ttk.Label(file_frame, text="测试集:").grid(row=1, column=0, sticky=tk.W, pady=5) self.test_file_var = tk.StringVar() ttk.Entry(file_frame, textvariable=self.test_file_var, width=30, state='readonly').grid(row=1, column=1, padx=5) ttk.Button(file_frame, text="选择文件", command=lambda: self.select_file("test")).grid(row=1, column=2) # PINNs参数设置 param_frame = ttk.LabelFrame(control_frame, text="PINNs参数", padding=10) param_frame.pack(fill=tk.X, pady=10) # 验证集切分比例 ttk.Label(param_frame, text="验证集比例:").grid(row=0, column=0, sticky=tk.W, pady=5) self.split_ratio_var = tk.DoubleVar(value=0.2) ttk.Spinbox(param_frame, from_=0, to=1, increment=0.05, textvariable=self.split_ratio_var, width=10).grid(row=0, column=1, padx=5) # 隐藏层数量 ttk.Label(param_frame, text="网络层数:").grid(row=1, column=0, sticky=tk.W, pady=5) self.num_layers_var = tk.IntVar(value=4) ttk.Spinbox(param_frame, from_=2, to=8, increment=1, textvariable=self.num_layers_var, width=10).grid(row=1, column=1, padx=5) # 每层神经元数量 ttk.Label(param_frame, text="神经元数/层:").grid(row=2, column=0, sticky=tk.W, pady=5) self.hidden_units_var = tk.IntVar(value=32) ttk.Spinbox(param_frame, from_=16, to=128, increment=4, textvariable=self.hidden_units_var, width=10).grid(row=2, column=1, padx=5) # 训练轮次 ttk.Label(param_frame, text="训练轮次:").grid(row=3, column=0, sticky=tk.W, pady=5) self.epochs_var = tk.IntVar(value=500) ttk.Spinbox(param_frame, from_=100, to=2000, increment=100, textvariable=self.epochs_var, width=10).grid(row=3, column=1, padx=5) # 物理损失权重 ttk.Label(param_frame, text="物理损失权重:").grid(row=4, column=0, sticky=tk.W, pady=5) self.physics_weight_var = tk.DoubleVar(value=0.5) ttk.Spinbox(param_frame, from_=0.1, to=1.0, increment=0.1, textvariable=self.physics_weight_var, width=10).grid(row=4, column=1, padx=5) # 控制按钮 btn_frame = ttk.Frame(control_frame) btn_frame.pack(fill=tk.X, pady=10) ttk.Button(btn_frame, text="训练模型", command=self.train_model).pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame, text="预测结果", command=self.predict).pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame, text="保存结果", command=self.save_results).pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame, text="重置", command=self.reset).pack(side=tk.RIGHT, padx=5) # 状态栏 self.status_var = tk.StringVar(value="就绪") status_bar = ttk.Label(control_frame, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W) status_bar.pack(fill=tk.X, side=tk.BOTTOM) # 右侧结果显示区域 result_frame = ttk.Frame(main_frame) result_frame.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True, padx=5, pady=5) # 创建标签页 self.notebook = ttk.Notebook(result_frame) self.notebook.pack(fill=tk.BOTH, expand=True) # 损失曲线标签页 self.loss_frame = ttk.Frame(self.notebook) self.notebook.add(self.loss_frame, text="训练损失") # 预测结果标签页 self.prediction_frame = ttk.Frame(self.notebook) self.notebook.add(self.prediction_frame, text="预测结果") # 指标显示 self.metrics_var = tk.StringVar() metrics_label = ttk.Label( self.prediction_frame, textvariable=self.metrics_var, font=('TkDefaultFont', 10, 'bold'), relief='ridge', padding=5 ) metrics_label.pack(fill=tk.X, padx=5, pady=5) # 初始化绘图区域 self.fig, self.ax = plt.subplots(figsize=(10, 6)) self.canvas = FigureCanvasTkAgg(self.fig, master=self.prediction_frame) self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) # 损失曲线画布 self.loss_fig, self.loss_ax = plt.subplots(figsize=(10, 4)) self.loss_canvas = FigureCanvasTkAgg(self.loss_fig, master=self.loss_frame) self.loss_canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True) def select_file(self, file_type): """选择Excel文件并计算时间步长""" try: file_path = filedialog.askopenfilename( title=f"选择{file_type}集Excel文件", filetypes=[("Excel文件", ".xlsx .xls"), ("所有文件", ".")] ) if not file_path: return df = pd.read_excel(file_path) # 验证必需列是否存在 required_cols = ['year', 'month', 'day', '水位'] missing_cols = [col for col in required_cols if col not in df.columns] if missing_cols: messagebox.showerror("列名错误", f"缺少必需列: {', '.join(missing_cols)}") return # 时间特征处理 time_features = ['year', 'month', 'day'] missing_time_features = [feat for feat in time_features if feat not in df.columns] if missing_time_features: messagebox.showerror("列名错误", f"Excel文件缺少预处理后的时间特征列: {', '.join(missing_time_features)}") return # 创建时间戳列 (增强兼容性) time_cols = ['year', 'month', 'day'] if 'hour' in df.columns: time_cols.append('hour') if 'minute' in df.columns: time_cols.append('minute') if 'second' in df.columns: time_cols.append('second') # 填充缺失的时间单位 for col in ['hour', 'minute', 'second']: if col not in df.columns: df[col] = 0 df['datetime'] = pd.to_datetime(df[time_cols]) # 设置时间索引 df = df.set_index('datetime') # 计算相对时间（天） df['days'] = (df.index - df.index[0]).days # 新增：计算时间步长dt（单位：天） df['dt'] = df.index.to_series().diff().dt.total_seconds() / 86400 # 精确到秒级 # 处理时间步长异常值 if len(df) > 1: # 计算有效时间步长（排除<=0的值） valid_dt = df['dt'][df['dt'] > 0] if len(valid_dt) > 0: avg_dt = valid_dt.mean() else: avg_dt = 1.0 else: avg_dt = 1.0 # 替换非正值 df.loc[df['dt'] <= 0, 'dt'] = avg_dt # 填充缺失值 df['dt'] = df['dt'].fillna(avg_dt) # 保存数据 if file_type == "train": self.train_df = df self.train_file_var.set(os.path.basename(file_path)) self.status_var.set(f"已加载训练集: {len(self.train_df)}条数据") else: self.test_df = df self.test_file_var.set(os.path.basename(file_path)) self.status_var.set(f"已加载测试集: {len(self.test_df)}条数据") except Exception as e: error_msg = f"文件读取失败: {str(e)}\n\n请确保：\n1. 文件不是打开状态\n2. 文件格式正确\n3. 包含必需的时间和水位列" messagebox.showerror("文件错误", error_msg) def calculate_metrics(self, y_true, y_pred): """计算评估指标""" from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score mse = mean_squared_error(y_true, y_pred) rmse = np.sqrt(mse) mae = mean_absolute_error(y_true, y_pred) non_zero_idx = np.where(y_true != 0)[0] if len(non_zero_idx) > 0: mape = np.mean(np.abs((y_true[non_zero_idx] - y_pred[non_zero_idx]) / y_true[non_zero_idx])) * 100 else: mape = float('nan') r2 = r2_score(y_true, y_pred) return { 'MSE': mse, 'RMSE': rmse, 'MAE': mae, 'MAPE': mape, # 修正键名 'R2': r2 } def train_model(self): """训练PINNs模型（带早停机制+训练指标监控）""" if self.train_df is None: messagebox.showwarning("警告", "请先选择训练集文件") return try: self.status_var.set("正在预处理数据...") self.root.update() # 从训练集中切分训练子集和验证子集（时间顺序切分） split_ratio = 1 - self.split_ratio_var.get() split_idx = int(len(self.train_df) * split_ratio) train_subset = self.train_df.iloc[:split_idx] valid_subset = self.train_df.iloc[split_idx:] # 检查数据量是否足够 if len(train_subset) < 2 or len(valid_subset) < 2: messagebox.showerror("数据错误", "训练集数据量不足（至少需要2个时间步）") return # 数据预处理 - 分别归一化不同特征 # 归一化时间特征 t_train = train_subset['days'].values[1:].reshape(-1, 1) self.scaler_t.fit(t_train) t_train_scaled = self.scaler_t.transform(t_train).astype(np.float32) # 归一化水位特征 h_train = train_subset['水位'].values[:-1].reshape(-1, 1) self.scaler_h.fit(h_train) h_train_scaled = self.scaler_h.transform(h_train).astype(np.float32) # 归一化时间步长特征 dt_train = train_subset['dt'].values[1:].reshape(-1, 1) self.scaler_dt.fit(dt_train) dt_train_scaled = self.scaler_dt.transform(dt_train).astype(np.float32) # 归一化标签（下一时刻水位） h_next_train = train_subset['水位'].values[1:].reshape(-1, 1) h_next_train_scaled = self.scaler_h.transform(h_next_train).astype(np.float32) # 准备验证数据（同样进行归一化） t_valid = valid_subset['days'].values[1:].reshape(-1, 1) t_valid_scaled = self.scaler_t.transform(t_valid).astype(np.float32) h_valid = valid_subset['水位'].values[:-1].reshape(-1, 1) h_valid_scaled = self.scaler_h.transform(h_valid).astype(np.float32) dt_valid = valid_subset['dt'].values[1:].reshape(-1, 1) dt_valid_scaled = self.scaler_dt.transform(dt_valid).astype(np.float32) h_next_valid_scaled = self.scaler_h.transform( valid_subset['水位'].values[1:].reshape(-1, 1) ).astype(np.float32) # 原始值用于指标计算 h_next_train_true = h_next_train h_next_valid_true = valid_subset['水位'].values[1:].reshape(-1, 1) # 创建模型和优化器 self.model = PINNModel( num_layers=self.num_layers_var.get(), hidden_units=self.hidden_units_var.get() ) optimizer = Adam(learning_rate=0.001) # 在训练循环中，使用归一化后的数据 train_dataset = tf.data.Dataset.from_tensor_slices( ((t_train_scaled, h_train_scaled, dt_train_scaled), h_next_train_scaled) ) train_dataset = train_dataset.shuffle(buffer_size=1024).batch(32) valid_dataset = tf.data.Dataset.from_tensor_slices( ((t_valid_scaled, h_valid_scaled, dt_valid_scaled), h_next_valid_scaled) ) valid_dataset = valid_dataset.batch(32) # 初始化训练历史记录列表 train_data_loss_history = [] physics_loss_history = [] valid_data_loss_history = [] train_metrics_history = [] valid_metrics_history = [] # 早停机制参数 patience = int(self.epochs_var.get() / 3) min_delta = 1e-4 best_valid_loss = float('inf') wait = 0 best_epoch = 0 best_weights = None start_time = time.time() # 自定义训练循环 for epoch in range(self.epochs_var.get()): # 训练阶段 epoch_train_data_loss = [] epoch_physics_loss = [] # 收集训练预测值（归一化后） train_pred_scaled = [] for step, ((t_batch, h_batch, dt_batch), h_next_batch) in enumerate(train_dataset): with tf.GradientTape() as tape: # 预测下一时刻水位 h_pred = self.model([t_batch, h_batch, dt_batch], training=True) # 添加training=True data_loss = tf.reduce_mean(tf.square(h_next_batch - h_pred)) # 动态调整物理损失权重 current_physics_weight = tf.minimum( self.physics_weight_var.get() * (1.0 + epoch / self.epochs_var.get()), 0.8 ) # 计算物理损失（传入时间步长dt） physics_loss = self.model.physics_loss(t_batch, h_batch, dt_batch,training=True) # 添加training=True loss = data_loss + current_physics_weight * physics_loss grads = tape.gradient(loss, self.model.trainable_variables) optimizer.apply_gradients(zip(grads, self.model.trainable_variables)) epoch_train_data_loss.append(data_loss.numpy()) epoch_physics_loss.append(physics_loss.numpy()) train_pred_scaled.append(h_pred.numpy()) # 保存训练预测值（归一化） # 合并训练预测值（归一化后） train_pred_scaled = np.concatenate(train_pred_scaled, axis=0) # 反归一化得到原始预测值 train_pred_true = self.scaler_h.inverse_transform(train_pred_scaled) # 计算训练集指标（使用原始真实值和预测值） train_metrics = self.calculate_metrics( y_true=h_next_train_true.flatten(), y_pred=train_pred_true.flatten() ) train_metrics_history.append(train_metrics) # 验证阶段 epoch_valid_data_loss = [] valid_pred_scaled = [] for ((t_v_batch, h_v_batch, dt_v_batch), h_v_next_batch) in valid_dataset: h_v_pred = self.model([t_v_batch, h_v_batch, dt_v_batch], training=False) # 验证时不启用Dropout valid_data_loss = tf.reduce_mean(tf.square(h_v_next_batch - h_v_pred)) epoch_valid_data_loss.append(valid_data_loss.numpy()) valid_pred_scaled.append(h_v_pred.numpy()) # 保存验证预测值（归一化） # 合并验证预测值（归一化后） valid_pred_scaled = np.concatenate(valid_pred_scaled, axis=0) # 反归一化得到原始预测值 valid_pred_true = self.scaler_h.inverse_transform(valid_pred_scaled) # 计算验证集指标（使用原始真实值和预测值） valid_metrics = self.calculate_metrics( y_true=h_next_valid_true.flatten(), y_pred=valid_pred_true.flatten() ) valid_metrics_history.append(valid_metrics) # 计算平均损失 avg_train_data_loss = np.mean(epoch_train_data_loss) avg_physics_loss = np.mean(epoch_physics_loss) avg_valid_data_loss = np.mean(epoch_valid_data_loss) # 记录损失 train_data_loss_history.append(avg_train_data_loss) physics_loss_history.append(avg_physics_loss) valid_data_loss_history.append(avg_valid_data_loss) # 早停机制逻辑 current_valid_loss = avg_valid_data_loss if current_valid_loss < best_valid_loss - min_delta: best_valid_loss = current_valid_loss best_epoch = epoch + 1 wait = 0 best_weights = self.model.get_weights() else: wait += 1 if wait >= patience: self.status_var.set(f"触发早停！最佳轮次: {best_epoch}，最佳验证损失: {best_valid_loss:.4f}") if best_weights is not None: self.model.set_weights(best_weights) break # 更新状态 if epoch % 1 == 0: # 提取当前训练/验证的关键指标 train_rmse = train_metrics['RMSE'] valid_rmse = valid_metrics['RMSE'] train_r2 = train_metrics['R2'] valid_r2 = valid_metrics['R2'] elapsed = time.time() - start_time self.status_var.set( f"训练中 | 轮次: {epoch + 1}/{self.epochs_var.get()} | " f"训练RMSE: {train_rmse:.4f} | 验证RMSE: {valid_rmse:.4f} | " f"训练R²: {train_r2:.4f} | 验证R²: {valid_r2:.4f} | " f"k1: {self.model.k1.numpy():.6f}, k2: {self.model.k2.numpy():.6f} | 时间: {elapsed:.1f}秒 | 早停等待: {wait}/{patience}" ) self.root.update() # 绘制损失曲线 self.loss_ax.clear() epochs_range = range(1, len(train_data_loss_history) + 1) self.loss_ax.plot(epochs_range, train_data_loss_history, 'b-', label='训练数据损失') self.loss_ax.plot(epochs_range, physics_loss_history, 'r--', label='物理损失') self.loss_ax.plot(epochs_range, valid_data_loss_history, 'g-.', label='验证数据损失') self.loss_ax.set_title('PINNs训练与验证损失') self.loss_ax.set_xlabel('轮次') self.loss_ax.set_ylabel('损失', rotation=0) self.loss_ax.legend() self.loss_ax.grid(True, alpha=0.3) self.loss_ax.set_yscale('log') self.loss_canvas.draw() # 训练完成提示 elapsed = time.time() - start_time if wait >= patience: completion_msg = ( f"早停触发 | 最佳轮次: {best_epoch} | 最佳验证损失: {best_valid_loss:.4f} | " f"最佳验证RMSE: {valid_metrics_history[best_epoch - 1]['RMSE']:.4f} | " f"总时间: {elapsed:.1f}秒" ) else: completion_msg = ( f"训练完成 | 总轮次: {self.epochs_var.get()} | " f"最终训练RMSE: {train_metrics_history[-1]['RMSE']:.4f} | " f"最终验证RMSE: {valid_metrics_history[-1]['RMSE']:.4f} | " f"最终训练R²: {train_metrics_history[-1]['R2']:.4f} | " f"最终验证R²: {valid_metrics_history[-1]['R2']:.4f} | " f"总时间: {elapsed:.1f}秒" ) # 保存训练历史 self.train_history = { 'train_data_loss': train_data_loss_history, 'physics_loss': physics_loss_history, 'valid_data_loss': valid_data_loss_history, 'train_metrics': train_metrics_history, 'valid_metrics': valid_metrics_history } # 保存学习到的物理参数 self.learned_params = { "k1": self.model.k1.numpy(), "k2": self.model.k2.numpy(), "alpha": self.model.alpha.numpy(), "beta": self.model.beta.numpy() } self.status_var.set(completion_msg) messagebox.showinfo("训练完成", f"PINNs模型训练成功完成！\n{completion_msg}") except Exception as e: messagebox.showerror("训练错误", f"模型训练失败:\n{str(e)}") self.status_var.set("训练失败") def predict(self): """使用PINNs模型进行递归预测（带Teacher Forcing和蒙特卡洛Dropout）""" if self.model is None: messagebox.showwarning("警告", "请先训练模型") return if self.test_df is None: messagebox.showwarning("警告", "请先选择测试集文件") return try: self.status_var.set("正在生成预测(使用Teacher Forcing和MC Dropout)...") self.root.update() # 预处理测试数据 - 归一化 t_test = self.test_df['days'].values.reshape(-1, 1) t_test_scaled = self.scaler_t.transform(t_test).astype(np.float32) dt_test = self.test_df['dt'].values.reshape(-1, 1) dt_test_scaled = self.scaler_dt.transform(dt_test).astype(np.float32) h_test = self.test_df['水位'].values.reshape(-1, 1) h_test_scaled = self.scaler_h.transform(h_test).astype(np.float32) # 递归预测参数 n = len(t_test) mc_iterations = 100 # 蒙特卡洛采样次数 teacher_forcing_prob = 0.7 # 使用真实值的概率 # 存储蒙特卡洛采样结果 mc_predictions_scaled = np.zeros((mc_iterations, n, 1), dtype=np.float32) # 进行多次蒙特卡洛采样 for mc_iter in range(mc_iterations): predicted_scaled = np.zeros((n, 1), dtype=np.float32) predicted_scaled[0] = h_test_scaled[0] # 第一个点使用真实值 # 递归预测（带Teacher Forcing） for i in range(1, n): # 决定使用真实值还是预测值（Teacher Forcing） use_actual = np.random.rand() < teacher_forcing_prob if use_actual and i < n - 1: # 不能使用未来值 h_prev = h_test_scaled[i - 1:i] else: h_prev = predicted_scaled[i - 1:i] t_prev = t_test_scaled[i - 1:i] dt_i = dt_test_scaled[i:i + 1] # 启用Dropout进行不确定性估计 h_pred = self.model([t_prev, h_prev, dt_i], training=True) # 启用Dropout进行不确定性估计 predicted_scaled[i] = h_pred.numpy()[0][0] mc_predictions_scaled[mc_iter] = predicted_scaled # 计算预测统计量 mean_pred_scaled = np.mean(mc_predictions_scaled, axis=0) std_pred_scaled = np.std(mc_predictions_scaled, axis=0) # 反归一化结果 predictions = self.scaler_h.inverse_transform(mean_pred_scaled) uncertainty = self.scaler_h.inverse_transform(std_pred_scaled) * 1.96 # 95%置信区间 actual_values = h_test test_time = self.test_df.index # 清除现有图表 self.ax.clear() # 计算合理的y轴范围 - 基于数据集中区域 # 获取实际值和预测值的中位数 median_val = np.median(actual_values) # 计算数据的波动范围（标准差） data_range = np.std(actual_values) * 4 # 4倍标准差覆盖大部分数据 # 设置y轴范围为中心值±数据波动范围 y_center = median_val y_half_range = max(data_range, 10) # 确保最小范围为20个单位 y_min_adjusted = y_center - y_half_range y_max_adjusted = y_center + y_half_range # 确保范围不为零 if y_max_adjusted - y_min_adjusted < 1: y_min_adjusted -= 5 y_max_adjusted += 5 # 绘制结果（带置信区间） self.ax.plot(test_time, actual_values, 'b-', label='真实值', linewidth=2) self.ax.plot(test_time, predictions, 'r--', label='预测均值', linewidth=2) self.ax.fill_between( test_time, (predictions - uncertainty).flatten(), (predictions + uncertainty).flatten(), color='orange', alpha=0.3, label='95%置信区间' ) # 设置自动调整的y轴范围 self.ax.set_ylim(y_min_adjusted, y_max_adjusted) self.ax.set_title('大坝渗流水位预测(PINNs with MC Dropout)') self.ax.set_xlabel('时间') self.ax.set_ylabel('测压管水位', rotation=0) self.ax.legend(loc='best') # 自动选择最佳位置 # 优化时间轴刻度 self.ax.xaxis.set_major_locator(mdates.YearLocator()) self.ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y')) self.ax.xaxis.set_minor_locator(mdates.MonthLocator(interval=2)) self.ax.grid(which='minor', axis='x', linestyle=':', color='gray', alpha=0.3) self.ax.grid(which='major', axis='y', linestyle='-', color='lightgray', alpha=0.5) self.ax.tick_params(axis='x', which='major', rotation=0, labelsize=9) self.ax.tick_params(axis='x', which='minor', length=2) # 计算评估指标（排除第一个点） eval_actual = actual_values[1:].flatten() eval_pred = predictions[1:].flatten() self.evaluation_metrics = self.calculate_metrics(eval_actual, eval_pred) # 添加不确定性指标 avg_uncertainty = np.mean(uncertainty) max_uncertainty = np.max(uncertainty) self.evaluation_metrics['Avg Uncertainty'] = avg_uncertainty self.evaluation_metrics['Max Uncertainty'] = max_uncertainty metrics_text = ( f"MSE: {self.evaluation_metrics['MSE']:.4f} | " f"RMSE: {self.evaluation_metrics['RMSE']:.4f} | " f"MAE: {self.evaluation_metrics['MAE']:.4f} | " f"MAPE: {self.evaluation_metrics['MAPE']:.2f}% | " f"R²: {self.evaluation_metrics['R2']:.4f}\n" f"平均不确定性: {avg_uncertainty:.4f} | 最大不确定性: {max_uncertainty:.4f}" ) self.metrics_var.set(metrics_text) # 在图表上添加指标 self.ax.text( 0.5, 1.05, metrics_text, transform=self.ax.transAxes, ha='center', fontsize=8, bbox=dict(facecolor='white', alpha=0.8) ) params_text = ( f"物理参数: k1={self.learned_params['k1']:.4f}, " f"k2={self.learned_params['k2']:.4f}, " f"alpha={self.learned_params['alpha']:.4f}, " f"beta={self.learned_params['beta']:.4f} | " f"Teacher Forcing概率: {teacher_forcing_prob}" ) self.ax.text( 0.5, 1.12, params_text, transform=self.ax.transAxes, ha='center', fontsize=8, bbox=dict(facecolor='white', alpha=0.8) ) # 调整布局 plt.tight_layout(pad=2.0) self.canvas.draw() # 保存预测结果 self.predictions = predictions self.uncertainty = uncertainty self.actual_values = actual_values self.test_time = test_time self.mc_predictions = mc_predictions_scaled self.status_var.set(f"预测完成(MC Dropout采样{mc_iterations}次)") except Exception as e: messagebox.showerror("预测错误", f"预测失败:\n{str(e)}") self.status_var.set("预测失败") import traceback traceback.print_exc() def save_results(self): """保存预测结果和训练历史数据""" if not hasattr(self, 'predictions') or not hasattr(self, 'train_history'): messagebox.showwarning("警告", "请先生成预测结果并完成训练") return # 选择保存路径 save_path = filedialog.asksaveasfilename( defaultextension=".xlsx", filetypes=[("Excel文件", ".xlsx"), ("所有文件", ".*")], title="保存结果" ) if not save_path: return try: # 1. 创建预测结果DataFrame result_df = pd.DataFrame({ '时间': self.test_time, '实际水位': self.actual_values.flatten(), '预测水位': self.predictions.flatten() }) # 2. 创建评估指标DataFrame metrics_df = pd.DataFrame([self.evaluation_metrics]) # 3. 创建训练历史DataFrame history_data = { '轮次': list(range(1, len(self.train_history['train_data_loss']) + 1)), '训练数据损失': self.train_history['train_data_loss'], '物理损失': self.train_history['physics_loss'], '验证数据损失': self.train_history['valid_data_loss'] } # 添加训练集指标 for metric in ['MSE', 'RMSE', 'MAE', 'MAPE', 'R2']: history_data[f'训练集_{metric}'] = [item[metric] for item in self.train_history['train_metrics']] # 添加验证集指标 for metric in ['MSE', 'RMSE', 'MAE', 'MAPE', 'R2']: history_data[f'验证集_{metric}'] = [item[metric] for item in self.train_history['valid_metrics']] history_df = pd.DataFrame(history_data) # 保存到Excel with pd.ExcelWriter(save_path) as writer: result_df.to_excel(writer, sheet_name='预测结果', index=False) metrics_df.to_excel(writer, sheet_name='评估指标', index=False) history_df.to_excel(writer, sheet_name='训练历史', index=False) # 保存图表 chart_path = os.path.splitext(save_path)[0] + "_chart.png" self.fig.savefig(chart_path, dpi=300) # 保存损失曲线图 loss_path = os.path.splitext(save_path)[0] + "_loss.png" self.loss_fig.savefig(loss_path, dpi=300) self.status_var.set(f"结果已保存至: {os.path.basename(save_path)}") messagebox.showinfo("保存成功", f"预测结果和图表已保存至:\n" f"主文件: {save_path}\n" f"预测图表: {chart_path}\n" f"损失曲线: {loss_path}") except Exception as e: messagebox.showerror("保存错误", f"保存结果失败:\n{str(e)}") def reset(self): # 重置归一化器 self.scaler_t = MinMaxScaler(feature_range=(0, 1)) self.scaler_h = MinMaxScaler(feature_range=(0, 1)) self.scaler_dt = MinMaxScaler(feature_range=(0, 1)) """重置程序状态""" self.train_df = None self.test_df = None self.model = None self.train_file_var.set("") self.test_file_var.set("") # 清除训练历史 if hasattr(self, 'train_history'): del self.train_history # 清除图表 if hasattr(self, 'ax'): self.ax.clear() if hasattr(self, 'loss_ax'): self.loss_ax.clear() # 重绘画布 if hasattr(self, 'canvas'): self.canvas.draw() if hasattr(self, 'loss_canvas'): self.loss_canvas.draw() # 清除状态 self.status_var.set("已重置，请选择新数据") # 清除预测结果 if hasattr(self, 'predictions'): del self.predictions # 清除指标文本 if hasattr(self, 'metrics_var'): self.metrics_var.set("") messagebox.showinfo("重置", "程序已重置，可以开始新的分析") if name == "main": root = tk.Tk() app = DamSeepageModel(root) root.mainloop() 帮我改进一下模型训练时候的学习率使之可以动态变化

f"k1: {self.model.k1.numpy():.6f}, k2: {self.model.k2.numpy():.6f} | 时间: {elapsed:.1f}秒 | 早停等待: {wait}/{patience}" ) self.root.update() # ... 后面的代码保持不变 ... ### 改进说明： 1....

请帮我检查优化代码，重点考虑用历史信息预测形成模型后，开展预测时要还原模型输入的标准要求，才能产生相匹配的模型预测结果：import pandas as pd import numpy as np import matplotlib.pyplot as plt from sklearn.preprocessing import MinMaxScaler from tensorflow.keras.models import Model from tensorflow.keras.layers import Input, LSTM, Dense, Dropout, Concatenate, Multiply, Permute, Reshape from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau from tensorflow.keras.optimizers import Adam from sklearn.model_selection import TimeSeriesSplit import joblib import os from openpyxl.styles import NamedStyle import warnings warnings.filterwarnings('ignore') # ===== 数据预处理函数 ===== def step1_format_data(): """格式化原始数据""" print("===== 步骤1: 格式化原始数据 =====") df = pd.read_excel('01hand.xlsx', sheet_name='Sheet1', header=None) # 提取A列和C列数据 new_df = pd.DataFrame({ 'A': pd.to_numeric(df.iloc[:, 0], errors='coerce'), 'B': pd.to_numeric(df.iloc[:, 2], errors='coerce') }).dropna() # 保存新文件 new_df.to_excel('01hand2.xlsx', index=False, header=False) print(f"新表格 '01hand2.xlsx' 创建成功! 包含 {len(new_df)} 行数据") def step2_process_data(): """数据去重和排序""" print("\n===== 步骤2: 数据去重和排序 =====") input_file = "01hand2.xlsx" output_file1 = "02resultA.xlsx" # 降序输出 output_file2 = "02resultB.xlsx" # 升序输出 # 读取数据并转换为长格式 df = pd.read_excel(input_file, header=None) all_values = df.stack().dropna().astype(str).tolist() # 确保数据长度是8的倍数 valid_length = len(all_values) - (len(all_values) % 8) if len(all_values) != valid_length: print(f"警告: 数据总量 {len(all_values)} 不符合8的倍数, 截断至 {valid_length} 个元素") all_values = all_values[:valid_length] # 转换数据格式 new_data = [] for i in range(0, len(all_values), 8): group = all_values[i:i+8] try: # 转换日期和数字 date = int(group[0]) numbers = [int(float(num)) if '.' in num else int(num) for num in group[1:]] new_data.append([date] + numbers) except: continue # 创建DataFrame并去重 columns = ['日期', '数字1', '数字2', '数字3', '数字4', '数字5', '数字6', '数字7'] df = pd.DataFrame(new_data, columns=columns) df = df.drop_duplicates(subset='日期').dropna() # 保存降序文件 df.sort_values('日期', ascending=False).to_excel(output_file1, index=False) print(f"降序文件保存至: {output_file1}") # 保存升序文件 df.sort_values('日期', ascending=True).to_excel(output_file2, index=False) print(f"升序文件保存至: {output_file2}") print(f"最终数据维度: {df.shape}") return df # ===== 特征工程函数 ===== def create_features(df): """创建模型特征""" print("\n===== 步骤3: 特征工程 =====") features = df[['日期']].copy() red_cols = ['数字1', '数字2', '数字3', '数字4', '数字5', '数字6'] # 基础特征 features['红球和值'] = df[red_cols].sum(axis=1) features['蓝球值'] = df['数字7'] features['奇偶比'] = df[red_cols].applymap(lambda x: x % 2).sum(axis=1) features['大小比'] = df[red_cols].applymap(lambda x: 1 if x > 16 else 0).sum(axis=1) # 窗口特征 (窗口大小10) window_size = 10 for col in ['红球和值', '奇偶比', '大小比']: features[f'{col}_MA{window_size}'] = features[col].rolling(window=window_size).mean() features[f'{col}_STD{window_size}'] = features[col].rolling(window=window_size).std() import pandas as pd import numpy as np from tqdm import tqdm def create_lagged_features(input_file="02resultB.xlsx", window_size=10): """ 构建滞后性模型特征参数: input_file: 输入Excel文件路径 window_size: 滞后窗口大小返回: feature_df: 特征DataFrame next_input: 用于预测下一期的输入特征 """ # 读取数据 df = pd.read_excel(input_file, header=None, names=['日期', '数字1', '数字2', '数字3', '数字4', '数字5', '数字6', '数字7']) # 基础特征工程 red_cols = ['数字1', '数字2', '数字3', '数字4', '数字5', '数字6'] df['红球和值'] = df[red_cols].sum(axis=1) df['奇偶比'] = df[red_cols].apply(lambda x: x % 2).sum(axis=1) df['大小比'] = df[red_cols].apply(lambda x: (x > 16).astype(int)).sum(axis=1) df['区间分布'] = df[red_cols].apply( lambda row: np.histogram(row, bins=[1, 12, 23, 34])[0], axis=1 ) # 拆分区间分布特征 df[['区间1', '区间2', '区间3']] = pd.DataFrame(df['区间分布'].tolist(), index=df.index) # 创建特征容器 feature_data = [] # 创建滞后特征 for i in tqdm(range(window_size, len(df)), desc="构建滞后特征"): current_row = {} current_date = df.iloc[i]['日期'] # 添加日期 current_row['日期'] = current_date # 添加基础特征 current_row['红球和值'] = df.iloc[i]['红球和值'] current_row['奇偶比'] = df.iloc[i]['奇偶比'] current_row['大小比'] = df.iloc[i]['大小比'] current_row['区间1'] = df.iloc[i]['区间1'] current_row['区间2'] = df.iloc[i]['区间2'] current_row['区间3'] = df.iloc[i]['区间3'] # 添加滞后特征 for lag in range(1, window_size + 1): prev_row = df.iloc[i - lag] # 红球滞后特征 for j in range(1, 7): current_row[f'数字{j}_lag{lag}'] = prev_row[f'数字{j}'] # 蓝球滞后特征 current_row[f'数字7_lag{lag}'] = prev_row['数字7'] # 统计特征滞后 current_row[f'红球和值_lag{lag}'] = prev_row['红球和值'] current_row[f'奇偶比_lag{lag}'] = prev_row['奇偶比'] current_row[f'大小比_lag{lag}'] = prev_row['大小比'] # 添加目标变量（下一期的开奖结果） if i < len(df) - 1: next_row = df.iloc[i + 1] for j in range(1, 7): current_row[f'目标_数字{j}'] = next_row[f'数字{j}'] current_row['目标_数字7'] = next_row['数字7'] feature_data.append(current_row) # 创建特征DataFrame feature_df = pd.DataFrame(feature_data) # 创建用于预测下一期的输入 last_row = df.iloc[-1] next_input = {} next_input['日期'] = last_row['日期'] # 使用最后一行日期作为预测基准 # 添加基础特征（最后一行） next_input['红球和值'] = last_row['红球和值'] next_input['奇偶比'] = last_row['奇偶比'] next_input['大小比'] = last_row['大小比'] next_input['区间1'] = last_row['区间1'] next_input['区间2'] = last_row['区间2'] next_input['区间3'] = last_row['区间3'] # 添加滞后特征（最后window_size行） for lag in range(1, window_size + 1): prev_row = df.iloc[-1 - lag] # 红球滞后特征 for j in range(1, 7): next_input[f'数字{j}_lag{lag}'] = prev_row[f'数字{j}'] # 蓝球滞后特征 next_input[f'数字7_lag{lag}'] = prev_row['数字7'] # 统计特征滞后 next_input[f'红球和值_lag{lag}'] = prev_row['红球和值'] next_input[f'奇偶比_lag{lag}'] = prev_row['奇偶比'] next_input[f'大小比_lag{lag}'] = prev_row['大小比'] # 转换为DataFrame next_input_df = pd.DataFrame([next_input]) # 保存结果 feature_df.to_excel('03_feature_target_table.xlsx', index=False) next_input_df.to_excel('04_next_input.xlsx', index=False) print(f"特征宽表已保存，维度：{feature_df.shape}") print(f"下一期预测输入已保存，维度：{next_input_df.shape}") return feature_df, next_input_df # 使用示例 if name == "main": feature_df, next_input = create_lagged_features(window_size=10) # 显示特征表的前几行 print("\n特征宽表示例:") print(feature_df[['日期', '红球和值', '数字1_lag1', '目标_数字1']].head()) # 显示预测输入 print("\n下一期预测输入:") print(next_input[['日期', '红球和值', '数字1_lag1', '数字7_lag1']]) # 滞后特征 (滞后1,2，3,4，5，6,7,8,9期) for lag in [1,2， 3,4， 5，6,7,8,9]: for col in red_cols + ['数字7']: features[f'{col}_lag{lag}'] = df[col].shift(lag) # 目标变量 (下一期开奖结果) red_targets = [] blue_targets = [] for i in range(len(df) - 1): next_row = df.iloc[i + 1] # 红球目标 (33选6) red_target = [1 if num in next_row[red_cols].values else 0 for num in range(1, 34)] # 蓝球目标 (16选1) blue_target = [1 if i == next_row['数字7'] else 0 for i in range(1, 17)] red_targets.append(red_target) blue_targets.append(blue_target) # 转换为numpy数组 red_targets = np.array(red_targets) blue_targets = np.array(blue_targets) # 移除无效数据 (前window_size行和最后一行) features = features.iloc[window_size:-1].reset_index(drop=True) red_targets = red_targets[window_size-1:-1] # 对齐索引 blue_targets = blue_targets[window_size-1:-1] # 保存特征 features.to_excel('04_features.xlsx', index=False) print(f"特征工程完成, 维度: {features.shape}") return features, red_targets, blue_targets # ===== 模型构建函数 ===== def prepare_data(features, red_targets, blue_targets): """准备训练数据""" print("\n===== 步骤4: 数据准备 =====") scaler_X = MinMaxScaler() X_scaled = scaler_X.fit_transform(features.drop(columns=['日期'])) joblib.dump(scaler_X, 'scaler_X.save') # 创建时间序列数据 time_steps = 10 X_seq, y_red_seq, y_blue_seq = [], [], [] for i in range(time_steps, len(X_scaled)): X_seq.append(X_scaled[i-time_steps:i, :]) y_red_seq.append(red_targets[i-1]) # 使用当前时间步的目标 y_blue_seq.append(blue_targets[i-1]) X_seq = np.array(X_seq) y_red_seq = np.array(y_red_seq) y_blue_seq = np.array(y_blue_seq) print(f"时间序列数据形状: X={X_seq.shape}, y_red={y_red_seq.shape}, y_blue={y_blue_seq.shape}") return X_seq, y_red_seq, y_blue_seq def attention_block(inputs, time_steps): """自定义注意力机制块""" # 计算注意力分数 attention = Dense(time_steps, activation='softmax')(inputs) attention = Permute((2, 1))(attention) # 应用注意力 attention_mul = Multiply()([inputs, attention]) return attention_mul def build_attention_lstm_model(input_shape): """构建带注意力机制的LSTM模型""" inputs = Input(shape=input_shape) time_steps = input_shape[0] # LSTM层 lstm_out = LSTM(128, return_sequences=True)(inputs) lstm_out = Dropout(0.3)(lstm_out) # 注意力机制 attention_out = attention_block(lstm_out, time_steps) # 红球分支 red_branch = LSTM(64, return_sequences=False)(attention_out) red_branch = Dropout(0.2)(red_branch) red_output = Dense(33, activation='sigmoid', name='red_output')(red_branch) # 蓝球分支 blue_branch = LSTM(32, return_sequences=False)(attention_out) blue_branch = Dropout(0.2)(blue_branch) blue_output = Dense(16, activation='sigmoid', name='blue_output')(blue_branch) model = Model(inputs=inputs, outputs=[red_output, blue_output]) optimizer = Adam(learning_rate=0.001) model.compile( optimizer=optimizer, loss={'red_output': 'binary_crossentropy', 'blue_output': 'binary_crossentropy'}, metrics={'red_output': 'binary_accuracy', 'blue_output': 'binary_accuracy'}, loss_weights=[0.7, 0.3] # 红球权重更高 ) model.summary() return model # ===== 模型训练函数 ===== def train_model(X, y_red, y_blue): """训练模型""" print("\n===== 步骤5: 模型训练 =====") best_models = [] tscv = TimeSeriesSplit(n_splits=3) for fold, (train_index, val_index) in enumerate(tscv.split(X)): print(f"\n===== 训练 Fold {fold+1}/3 =====") X_train, X_val = X[train_index], X[val_index] y_red_train, y_red_val = y_red[train_index], y_red[val_index] y_blue_train, y_blue_val = y_blue[train_index], y_blue[val_index] model = build_attention_lstm_model((X_train.shape[1], X_train.shape[2])) callbacks = [ EarlyStopping(monitor='val_loss', patience=15, restore_best_weights=True, verbose=1), ReduceLROnPlateau(monitor='val_loss', factor=0.3, patience=7, min_lr=1e-6, verbose=1) ] history = model.fit( X_train, {'red_output': y_red_train, 'blue_output': y_blue_train}, epochs=100, batch_size=32, validation_data=(X_val, {'red_output': y_red_val, 'blue_output': y_blue_val}), callbacks=callbacks, verbose=1 ) model.save(f'best_model_fold{fold+1}.h5') best_models.append(model) # 保存训练历史图 plot_training_history(history, fold+1) return best_models def plot_training_history(history, fold): """绘制训练历史图表""" plt.figure(figsize=(15, 10)) # 损失曲线 plt.subplot(2, 2, 1) plt.plot(history.history['loss'], label='训练损失') plt.plot(history.history['val_loss'], label='验证损失') plt.title(f'Fold {fold} - 总损失曲线') plt.ylabel('损失') plt.xlabel('Epoch') plt.legend() # 红球准确率 plt.subplot(2, 2, 2) plt.plot(history.history['red_output_binary_accuracy'], label='红球训练准确率') plt.plot(history.history['val_red_output_binary_accuracy'], label='红球验证准确率') plt.title(f'Fold {fold} - 红球准确率') plt.ylabel('准确率') plt.xlabel('Epoch') plt.legend() # 蓝球准确率 plt.subplot(2, 2, 3) plt.plot(history.history['blue_output_binary_accuracy'], label='蓝球训练准确率') plt.plot(history.history['val_blue_output_binary_accuracy'], label='蓝球验证准确率') plt.title(f'Fold {fold} - 蓝球准确率') plt.ylabel('准确率') plt.xlabel('Epoch') plt.legend() # 学习率 plt.subplot(2, 2, 4) if 'lr' in history.history: plt.plot(history.history['lr'], label='学习率') plt.title(f'Fold {fold} - 学习率变化') plt.ylabel('学习率') plt.xlabel('Epoch') plt.legend() plt.tight_layout() plt.savefig(f'training_history_fold{fold}.png') plt.close() # ===== 预测函数 ===== def predict_next_period(models, features): """预测下一期开奖结果""" print("\n===== 步骤6: 预测 =====") scaler_X = joblib.load('scaler_X.save') time_steps = 10 # 准备最新数据 recent_data = features.drop(columns=['日期']).iloc[-time_steps:] X_recent = scaler_X.transform(recent_data) X_recent = X_recent.reshape((1, time_steps, X_recent.shape[1])) # 集成模型预测 red_probs = np.zeros((1, 33)) blue_probs = np.zeros((1, 16)) for i, model in enumerate(models): r_prob, b_prob = model.predict(X_recent, verbose=0) # 根据模型性能加权 weight = 0.3 + i * 0.2 # 第一个模型权重0.3, 第二个0.5, 第三个0.7 red_probs += r_prob * weight blue_probs += b_prob * weight # 归一化概率 red_probs /= red_probs.sum() blue_probs /= blue_probs.sum() # 获取红球预测 (前6个) red_indices = np.argsort(red_probs[0])[::-1][:6] red_numbers = [i + 1 for i in red_indices] red_probabilities = [red_probs[0][i] for i in red_indices] # 获取蓝球预测 (前3个) blue_indices = np.argsort(blue_probs[0])[::-1][:3] blue_numbers = [i + 1 for i in blue_indices] blue_probabilities = [blue_probs[0][i] for i in blue_indices] return red_numbers, red_probabilities, blue_numbers, blue_probabilities # ===== 主函数 ===== def main(): # 执行数据处理步骤 step1_format_data() df = step2_process_data() # 特征工程 features, red_targets, blue_targets = create_features(df) # 准备训练数据 X, y_red, y_blue = prepare_data(features, red_targets, blue_targets) # 训练模型 models = train_model(X, y_red, y_blue) # 预测 red_nums, red_probs, blue_nums, blue_probs = predict_next_period(models, features) # 打印结果 print("\n" + "="50) print("双色球下一期预测结果") print("="50) print("\n红球预测 (前6个):") for num, prob in zip(red_nums, red_probs): print(f"号码 {num:2d} : 概率 {prob:.4f}") print("\n蓝球预测 (前3个):") for num, prob in zip(blue_nums, blue_probs): print(f"号码 {num:2d} : 概率 {prob:.4f}") # 保存结果 result_df = pd.DataFrame({ '红球预测': red_nums, '红球概率': red_probs, '蓝球预测': blue_nums, '蓝球概率': blue_probs }) result_df.to_excel('prediction_results.xlsx', index=False) print("\n预测结果已保存至: prediction_results.xlsx") if name == "main": main()

步骤1：修改create_features函数，去掉滞后特征，只保留基础特征和窗口特征。步骤2：在prepare_data函数中，构建时间序列样本时，使用特征宽表的连续10行（每行包含当期的所有特征）作为输入，下一行的目标...

# 基于注意力的多头CNLSTM（AMCNLSTM） # 这是我的AMCNLSTM实现。欢迎尝试调整该架构。 in_layers, out_layers = list(), list() for i in range(nb_features): inputs = Input(shape=(sequence_length, 1)) rnn1 = Conv1D(filters=64, kernel_size=2, strides=1, padding="same")(inputs) lr1 = LeakyReLU()(rnn1) bn1 = BatchNormalization()(lr1) rnn2 = Conv1D(filters=64, kernel_size=2, strides=1, padding="same")(bn1) lr2 = LeakyReLU()(rnn2) bn2 = BatchNormalization()(lr2) rnn3 = LSTM(units=50, return_sequences=True)(bn2) lr3 = LeakyReLU()(rnn3) bn3 = BatchNormalization()(lr3) att = SeqSelfAttention( attention_type=SeqSelfAttention.ATTENTION_TYPE_MUL, kernel_regularizer=keras.regularizers.l2(1e-4), bias_regularizer=keras.regularizers.l1(1e-4), attention_regularizer_weight=1e-4, attention_width=15 )(bn3) pool1 = MaxPooling1D(pool_size=2)(att) flat = Flatten()(pool1) # store layers in_layers.append(inputs) out_layers.append(flat) # merge heads merged = concatenate(out_layers) # interpretation dense1 = Dense(50, activation='relu')(merged) outputs = Dense(nb_out)(dense1) # define model model = Model(inputs=in_layers, outputs=outputs) # replace lr with learning_rate optimizer = keras.optimizers.Adam(learning_rate=3e-5) # compile model model.compile( loss=keras.losses.Huber(), optimizer=optimizer, metrics=[rmse, k_score] )这一段是在TensorFlow中搭建的lstm神经网络框架，请帮我看看有什么可以改进的并且做出修改

dense1 = Dense(256, activation='swish')(dense1) dense1 = Dropout(0.2)(dense1) # 多尺度特征融合 residual = Dense(128, activation='linear')(dense1) dense1 = Add()([dense1, residual]) outputs = Dense...

基于注意力的多头CNLSTM（AMCNLSTM）这是我的AMCNLSTM实现。欢迎尝试调整该架构。 in_layers, out_layers = list(), list() for i in range(nb_features): inputs = Input(shape=(sequence_length, 1)) rnn1 = Conv1D(filters=64, kernel_size=2, strides=1, padding=“same”)(inputs) lr1 = LeakyReLU()(rnn1) bn1 = BatchNormalization()(lr1) rnn2 = Conv1D(filters=64, kernel_size=2, strides=1, padding=“same”)(bn1) lr2 = LeakyReLU()(rnn2) bn2 = BatchNormalization()(lr2) rnn3 = LSTM(units=50, return_sequences=True)(bn2) lr3 = LeakyReLU()(rnn3) bn3 = BatchNormalization()(lr3) att = SeqSelfAttention( attention_type=SeqSelfAttention.ATTENTION_TYPE_MUL, kernel_regularizer=keras.regularizers.l2(1e-4), bias_regularizer=keras.regularizers.l1(1e-4), attention_regularizer_weight=1e-4, attention_width=15 )(bn3) pool1 = MaxPooling1D(pool_size=2)(att) flat = Flatten()(pool1) # store layers in_layers.append(inputs) out_layers.append(flat) merge heads merged = concatenate(out_layers) interpretation dense1 = Dense(50, activation=‘relu’)(merged) outputs = Dense(nb_out)(dense1) define model model = Model(inputs=in_layers, outputs=outputs) replace lr with learning_rate optimizer = keras.optimizers.Adam(learning_rate=3e-5) compile model model.compile( loss=keras.losses.Huber(), optimizer=optimizer, metrics=[rmse, k_score] )这一段是在TensorFlow中搭建的lstm神经网络框架，请帮我看看有什么可以改进的并且做出修改，并且在最后附上修改后的完整代码，还可以加一点注意力机制

attention_type=SeqSelfAttention.ATTENTION_TYPE_MUL, kernel_regularizer=regularizers.l1_l2(1e-5), attention_heads=8, # 8头注意力 attention_width=20 )(lstm) # 通道注意力 se = SEBlock(lstm) #...

input_data = Input(shape=(trainX1.shape[1], trainX1.shape[2],)) timesteps = trainX1.shape[1] features = trainX1.shape[2] # 计算时间步的注意力权重 attention_probs1 = Dense(timesteps, activation='softmax')(input_data) attention_probs1 = Permute((2, 1))(attention_probs1) # 将注意力权重应用于输入数据 attention_mul1 = multiply([input_data, attention_probs]) attention_mul1 = Lambda(lambda x: K.sum(x, axis=1))(attention_mul1) # 计算维的注意力权重 attention_probs2 = Dense(INPUT_DIM, activation='softmax')(input_data) attention_probs2 = Permute((2, 1))(attention_probs2) # 将注意力权重应用于输入数据 attention_mul2 = multiply([input_data, attention_probs2]) attention_mul2 = Lambda(lambda x: K.sum(x, axis=1))(attention_mul2) 如何链接attention_mul1和attention_mul2

可以使用K.concatenate()函数将两个注意力向量连接起来，如下所示： merged_vector = K.concatenate([attention_mul1, attention_mul2]) 这将返回一个连接了两个注意力向量的张量。

def cnn_lstm_attention_model(n_input, n_out, n_features): inputs = Input(shape=(n_input, n_features)) x = Conv1D(filters=64, kernel_size=1, activation='relu')(inputs) # , padding = 'same' x = Dropout(0.3)(x) lstm_out = Bidirectional(LSTM(128, return_sequences=True))(x) lstm_out = Dropout(0.3)(lstm_out) attention_mul = attention_block(lstm_out, n_input) attention_mul = Flatten()(attention_mul)#扁平层，变为一维数据 output = Dense(n_out, activation='sigmoid')(attention_mul) model = Model(inputs=[inputs], outputs=output) model.summary() model.compile(loss="mse", optimizer='adam') return model 什莫意思

- 卷积层：使用卷积运算提取输入数据的空间特征，其中使用了64个过滤器和1维卷积核。 - Dropout层：随机丢弃一定比例的神经元，以防止过拟合。 - 双向LSTM层：使用128个隐藏单元的双向LSTM层，可以学习输入序列的...

torch_sparse-0.6.17+pt113cpu-cp310-cp310-linux_x86_64.whl.zip

2. **转换和操作**：提供了to_dense()方法将稀疏张量转换为稠密张量，add()、mul()等方法实现稀疏张量的加法和乘法，transpose()则用于转置。 3. **索引和切片**：支持稀疏张量的索引和切片操作，如index_...

tf-basics-python：来自PythonProgramming.net的TensorFlow基础

with tf.Session() as sess: print(sess.run(add_tensor)) print(sess.run(mul_tensor)) 在 TensorFlow 2.x 中，会话的概念已被弃用，取而代之的是即时执行模式（Eager Execution）。只需在程序开始时启用： ...

请看看这个代码如何改进： input_data1 = Input(shape=(time_steps1, input_dim1)) #lstm1 = input_data1 # 添加卷积层 conv1d_1 = Conv1D(filters=64, kernel_size=3, activation='relu')(input_data1) # 对于维的注意力机制 #attention_mul1 = attention_3d_block(conv1d_1, 'dim_reduction1' ,'attention_vec1' ) #attention_mul1 = Dense(64, activation='relu')(attention_mul1) context1 = multi_head_attention(conv1d_1,5) # 通过增加层数和隐藏单元的数量，可以增加模型的复杂度和表现能力，但也可能导致过拟合问题 lstm1 = Bidirectional(LSTM(128, return_sequences=True))(context1) # 加入双向LSTM层 lstm1 = Bidirectional(LSTM(64, return_sequences=True))(lstm1) lstm1 = Bidirectional(LSTM(32, return_sequences=True))(lstm1) #lstm1 = LSTM(128, return_sequences=True)(context_dim2) # 加入LSTM层 lstm_out1 = Dense(64, activation='relu')(lstm1) lstm_out1 = Dense(32, activation='tanh')(lstm_out1) lstm_out1 = Dense(16, activation='softmax')(lstm_out1)

1. 对于卷积层和多头注意力机制层，可以尝试使用不同的卷积核大小、卷积核数量和注意力头数，以寻找最佳的超参数组合。 2. 对于LSTM层，可以考虑使用更多或更少的LSTM单元，以及调整LSTM层的堆叠数量，以适应数据集...

我的模型结构： conv1d_1 = Conv1D(filters=64, kernel_size=3, activation='relu')(input_data1) # 对于维的注意力机制 #attention_mul1 = attention_3d_block(conv1d_1, 'dim_reduction1' ,'attention_vec1' ) #attention_mul1 = Dense(64, activation='relu')(attention_mul1) context1 = multi_head_attention(conv1d_1,5) # 通过增加层数和隐藏单元的数量，可以增加模型的复杂度和表现能力，但也可能导致过拟合问题 lstm1 = Bidirectional(LSTM(128, return_sequences=True))(context1) # 加入双向LSTM层 lstm1 = Bidirectional(LSTM(64, return_sequences=True))(lstm1) lstm1 = Bidirectional(LSTM(32))(lstm1) #lstm1 = LSTM(128, return_sequences=True)(context_dim2) # 加入LSTM层 lstm_out1 = Dense(64, activation='relu')(lstm1) lstm_out1 = Dense(32, activation='tanh')(lstm_out1) lstm_out1 = Dense(16, activation='softmax')(lstm_out1) ，如何改进？

1. 添加正则化层：为了避免过拟合问题，您可以添加一些正则化层，例如 Dropout 或 L2 正则化等。这些层可以帮助减少模型中的参数，并从一定程度上防止过拟合。 2. 调整注意力机制：您可以尝试不同的注意力机制，...

def attention_3d_block(inputs,STEPS): # inputs.shape = (batch_size, time_steps, input_dim) input_dim = int(inputs.shape[2]) a = Permute((2, 1))(inputs) a = Reshape((input_dim, STEPS))(a) # this line is not useful. It's just to know which dimension is what. a = Dense(STEPS, activation='softmax')(a) if SINGLE_ATTENTION_VECTOR: a = Lambda(lambda x: K.mean(x, axis=1), name='dim_reduction')(a) a = RepeatVector(input_dim)(a) a_probs = Permute((2, 1))(a) output_attention_mul = Multiply()([inputs, a_probs]) return output_attention_mul

输入为一个三维张量，形状为 (batch_size, time_steps, input_dim)，其中 batch_size 表示批次大小，time_steps 表示时间步，input_dim 表示输入的特征数。该函数可以将注意力应用于输入的每个时间步上，以产生一个...

请加入卷积给我的例子： # 定义输入张量 input_data1 = Input(shape=(time_steps1, input_dim1)) #lstm1 = input_data1 # 对于维的注意力机制 attention_mul1 = attention_3d_block(input_data1, 'dim_reduction1' ,'attention_vec1' ) attention_mul1 = Dense(64, activation='relu')(attention_mul1) # 定义时间步注意力机制 attention_time1 = TimeDistributed(Dense(1, activation='tanh'))(attention_mul1) attention_time1 = Activation('softmax')(attention_time1) context_time1 = Dot(axes=1)([attention_time1, attention_mul1]) # 定义维度注意力机制 attention_dim1 = Dense(input_dim1, activation='tanh')(context_time1) attention_dim1 = Activation('softmax')(attention_dim1) context_dim1 = Dot(axes=1)([attention_dim1, context_time1]) # 定义第二层时间步注意力机制 attention_time2 = TimeDistributed(Dense(1, activation='tanh'))(context_dim1) attention_time2 = Activation('softmax')(attention_time2) context_time2 = Dot(axes=1)([attention_time2, context_dim1]) # 定义第二层维度注意力机制 attention_dim2 = Dense(input_dim1, activation='tanh')(context_time2) attention_dim2 = Activation('softmax')(attention_dim2) context_dim2 = Dot(axes=1)([attention_dim2, context_time2]) lstm1 = LSTM(128, return_sequences=True)(context_dim2) # 加入LSTM层 lstm_out1 = Dense(64, activation='relu')(lstm1) lstm_out1 = Dense(32, activation='tanh')(lstm_out1) lstm_out1 = Dense(16, activation='softmax')(lstm_out1)

attention_time1 = TimeDistributed(Dense(1, activation='tanh'))(attention_mul1) attention_time1 = Activation('softmax')(attention_time1) context_time1 = Dot(axes=1)([attention_time1, attention_mul1]) #...

Vector4d QuaterMul(Vector4d q1, const Vector4d q2) { Matrix4d ps; ps << q1(0), -q1(1), -q1(2), -q1(3), q1(1), q1(0), -q1(3), q1(2), q1(2), q1(3), q1(0), -q1(1), q1(3), -q1(2), q1(1), q1(0); Vector4d mul; mul = ps * q2; return mul; }帮我改正这行代码的错误

#include <Eigen/Dense> using namespace Eigen; Vector4d QuaterMul(const Vector4d& q1, const Vector4d& q2) { Matrix4d ps; ps (0), -q1(1), -q1(2), -q1(3), q1(1), q1(0), -q1(3), q1(2), q1(2), q1(3), ...

tf2.x中tensorboard的使用示例代码

### 回答1： Tensorboard在TensorFlow 2.x中的使用示例代码非常简单，只需要几行代码就可以启动它。下面是一个使用Tensorboard的示例代码：from tensorflow.keras.callbacks import TensorBoardtensorboard_...

self.dense1_mul = dense(self.state, dense1_size, weight_init=tf.random_uniform_initializer((-1/tf.sqrt(tf.to_float(self.state_dims))), 1/tf.sqrt(tf.to_float(self.state_dims))), AttributeError: module 'tensorflow' has no attribute 'to_float'

相关推荐

self.dense1_mul = dense(self.state, dense1_size, weight_init=tf.random_uniform_initializer((-1/tf.sqrt(tf.to_float(self.state_dims))), 1/tf.sqrt(tf.to_float(self.state_dims))), AttributeError: module 'tensorflow' has no attribute 'to_float'

相关推荐

keras中的backend.clip用法

torch_sparse-0.6.9-cp37-cp37m-win_amd64whl.zip

torch_sparse-0.6.10-cp38-cp38-win_amd64whl.zip

torch_sparse-0.6.17+pt113cpu-cp310-cp310-linux_x86_64.whl.zip

tf-basics-python：来自PythonProgramming.net的TensorFlow基础

Vector4d QuaterMul(Vector4d q1, const Vector4d q2) { Matrix4d ps; ps << q1(0), -q1(1), -q1(2), -q1(3), q1(1), q1(0), -q1(3), q1(2), q1(2), q1(3), q1(0), -q1(1), q1(3), -q1(2), q1(1), q1(0); Vector4d mul; mul = ps * q2; return mul; }帮我改正这行代码的错误

tf2.x中tensorboard的使用示例代码

大家在看

Indesign插件合集(支持ID CS6~CC 2021)

爬取招行外汇网站数据.pdf

ORCAD库管理.rar

mapinfo详细教程

.NET frxamework v2.0 64位

最新推荐

造纸机变频分布传动与Modbus RTU通讯技术的应用及其实现

Visual C++.NET编程技术实战指南

HarmonyOS内核深度探秘：优化自由行旅游系统的策略

tkinter模块所有控件

局域网五子棋游戏：娱乐与聊天的完美结合

自由行旅游新篇章：HarmonyOS技术融合与系统架构深度解析

足底支撑相到达73%是什么问题

宾馆预约系统开发与优化建议

HarmonyOS在旅游领域的创新：揭秘最前沿应用实践

数据架构师需要具备什么能力