优化神经网络模型以提升R²值至0.99的全面方案

1. 问题分析与背景

在深度学习项目中，提升模型的R²（决定系数）值至0.99是一个具有挑战性的目标，特别是在处理复杂的时间序列数据时。我们的现有模型结合了LSTM层、自注意力机制和MLP处理标量输入，这种复杂结构虽然强大，但可能存在冗余和效率低下的问题。

R²值是衡量模型对目标变量变异解释程度的指标，值越接近1表示模型拟合效果越好。要达到0.99的R²值，我们需要精心设计模型架构、优化超参数并确保训练过程高效稳定。

2. 环境设置与数据准备

首先，我们需要设置适当的环境并准备数据：

import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.metrics import r2_score
import matplotlib.pyplot as plt
import seaborn as sns
import time
import warnings
warnings.filterwarnings('ignore')# 设置随机种子以确保结果可重现
seed = 42
np.random.seed(seed)
tf.random.set_seed(seed)# 检查GPU可用性
print("GPU可用:", tf.config.list_physical_devices('GPU'))

接下来，我们加载并探索数据集：

# 假设数据已加载为X_seq(序列数据), X_scalar(标量数据), y(目标变量)
# 这里我们创建一个模拟数据集用于演示
def create_synthetic_data(n_samples=10000, seq_length=50, n_scalar_features=10):"""创建合成数据用于演示"""# 创建时间序列数据X_seq = np.random.randn(n_samples, seq_length, 5)# 添加一些模式使预测更有意义for i in range(n_samples):trend = np.linspace(0, 1, seq_length)seasonal = np.sin(2 * np.pi * np.arange(seq_length) / 10)X_seq[i, :, 0] += trend + seasonal# 创建标量数据X_scalar = np.random.randn(n_samples, n_scalar_features)# 创建目标变量 - 结合序列和标量特征的非线性关系y = (np.mean(X_seq[:, :, 0], axis=1) + 0.5 * np.std(X_seq[:, :, 1], axis=1) + 0.3 * X_scalar[:, 0] - 0.7 * X_scalar[:, 1]**2 +np.random.normal(0, 0.1, n_samples))return X_seq, X_scalar, y# 生成数据
X_seq, X_scalar, y = create_synthetic_data()print(f"序列数据形状: {X_seq.shape}")
print(f"标量数据形状: {X_scalar.shape}")
print(f"目标变量形状: {y.shape}")# 数据标准化
seq_scaler = StandardScaler()
scalar_scaler = StandardScaler()
y_scaler = StandardScaler()# 重塑序列数据以适应标准化
X_seq_reshaped = X_seq.reshape(-1, X_seq.shape[-1])
X_seq_reshaped = seq_scaler.fit_transform(X_seq_reshaped)
X_seq = X_seq_reshaped.reshape(X_seq.shape)X_scalar = scalar_scaler.fit_transform(X_scalar)
y = y_scaler.fit_transform(y.reshape(-1, 1)).flatten()# 划分训练集和测试集
X_seq_train, X_seq_test, X_scalar_train, X_scalar_test, y_train, y_test = train_test_split(X_seq, X_scalar, y, test_size=0.2, random_state=seed
)print("数据准备完成")

3. 基准模型评估

在开始优化之前，我们需要建立一个基准模型来评估当前性能：

def create_baseline_model(seq_length=50, n_seq_features=5, n_scalar_features=10):"""创建基准模型"""# 序列输入分支seq_input = layers.Input(shape=(seq_length, n_seq_features), name='seq_input')# LSTM层x = layers.LSTM(64, return_sequences=True)(seq_input)x = layers.LSTM(32, return_sequences=True)(x)# 自注意力机制attention = layers.MultiHeadAttention(num_heads=4, key_dim=32)(x, x)x = layers.Concatenate()([x, attention])x = layers.LSTM(16)(x)# 标量输入分支scalar_input = layers.Input(shape=(n_scalar_features,), name='scalar_input')y = layers.Dense(32, activation='relu')(scalar_input)y = layers.Dense(16, activation='relu')(y)# 合并两个分支combined = layers.Concatenate()([x, y])# MLP层z = layers.Dense(64, activation='relu')(combined)z = layers.Dropout(0.3)(z)z = layers.Dense(32, activation='relu')(z)z = layers.Dropout(0.2)(z)z = layers.Dense(16, activation='relu')(z)# 输出层output = layers.Dense(1, activation='linear')(z)model = keras.Model(inputs=[seq_input, scalar_input], outputs=output)return model# 创建和编译基准模型
baseline_model = create_baseline_model()
baseline_model.compile(optimizer=keras.optimizers.Adam(learning_rate=0.001),loss='mse',metrics=['mae']
)baseline_model.summary()# 训练基准模型
print("训练基准模型...")
baseline_history = baseline_model.fit([X_seq_train, X_scalar_train], y_train,batch_size=32,epochs=50,validation_split=0.2,verbose=1
)# 评估基准模型
baseline_pred = baseline_model.predict([X_seq_test, X_scalar_test]).flatten()
baseline_r2 = r2_score(y_test, baseline_pred)
print(f"基准模型R²值: {baseline_r2:.4f}")

4. 模型优化策略

4.1 超参数优化

我们将使用Keras Tuner进行系统化的超参数搜索：

import kerastuner as ktdef build_model(hp):"""构建可调超参数的模型"""# 序列输入分支seq_input = layers.Input(shape=(X_seq.shape[1], X_seq.shape[2]), name='seq_input')# 可调LSTM层lstm_units_1 = hp.Int('lstm_units_1', min_value=32, max_value=128, step=32)lstm_units_2 = hp.Int('lstm_units_2', min_value=16, max_value=64, step=16)x = layers.LSTM(lstm_units_1, return_sequences=True)(seq_input)x = layers.LSTM(lstm_units_2, return_sequences=True)(x)# 可调自注意力机制num_heads = hp.Int('num_heads', min_value=2, max_value=8, step=2)key_dim = hp.Int('key_dim', min_value=16, max_value=64, step=16)attention = layers.MultiHeadAttention(num_heads=num_heads, key_dim=key_dim)(x, x)x = layers.Concatenate()([x, attention])x = layers.GlobalAveragePooling1D()(x)# 标量输入分支scalar_input = layers.Input(shape=(X_scalar.shape[1],), name='scalar_input')dense_units_1 = hp.Int('dense_units_1', min_value=16, max_value=64, step=16)dense_units_2 = hp.Int('dense_units_2', min_value=8, max_value=32, step=8)y = layers.Dense(dense_units_1, activation='relu')(scalar_input)y = layers.Dense(dense_units_2, activation='relu')(y)# 合并两个分支combined = layers.Concatenate()([x, y])# 可调MLP层mlp_units_1 = hp.Int('mlp_units_1', min_value=32, max_value=128, step=32)mlp_units_2 = hp.Int('mlp_units_2', min_value=16, max_value=64, step=16)mlp_units_3 = hp.Int('mlp_units_3', min_value=8, max_value=32, step=8)dropout_rate_1 = hp.Float('dropout_rate_1', min_value=0.1, max_value=0.5, step=0.1)dropout_rate_2 = hp.Float('dropout_rate_2', min_value=0.1, max_value=0.5, step=0.1)z = layers.Dense(mlp_units_1, activation='relu')(combined)z = layers.Dropout(dropout_rate_1)(z)z = layers.Dense(mlp_units_2, activation='relu')(z)z = layers.Dropout(dropout_rate_2)(z)z = layers.Dense(mlp_units_3, activation='relu')(z)# 输出层output = layers.Dense(1, activation='linear')(z)model = keras.Model(inputs=[seq_input, scalar_input], outputs=output)# 可调学习率learning_rate = hp.Float('learning_rate', min_value=1e-4, max_value=1e-2, sampling='log')model.compile(optimizer=keras.optimizers.Adam(learning_rate=learning_rate),loss='mse',metrics=['mae'])return model# 设置超参数调优
tuner = kt.BayesianOptimization(build_model,objective='val_loss',max_trials=50,executions_per_trial=2,directory='hyperparam_tuning',project_name='r2_optimization'
)# 开始超参数搜索
print("开始超参数搜索...")
tuner.search([X_seq_train, X_scalar_train], y_train,batch_size=64,epochs=30,validation_split=0.2,verbose=1
)# 获取最佳超参数
best_hps = tuner.get_best_hyperparameters(num_trials=1)[0]
print(f"""
最佳超参数:
- LSTM单元1: {best_hps.get('lstm_units_1')}
- LSTM单元2: {best_hps.get('lstm_units_2')}
- 注意力头数: {best_hps.get('num_heads')}
- 键维度: {best_hps.get('key_dim')}
- 标量Dense单元1: {best_hps.get('dense_units_1')}
- 标量Dense单元2: {best_hps.get('dense_units_2')}
- MLP单元1: {best_hps.get('mlp_units_1')}
- MLP单元2: {best_hps.get('mlp_units_2')}
- MLP单元3: {best_hps.get('mlp_units_3')}
- Dropout率1: {best_hps.get('dropout_rate_1')}
- Dropout率2: {best_hps.get('dropout_rate_2')}
- 学习率: {best_hps.get('learning_rate')}
""")

4.2 优化模型架构

基于超参数调优结果，我们设计一个更高效的模型架构：

def create_optimized_model(seq_length=50, n_seq_features=5, n_scalar_features=10):"""创建优化后的模型"""# 基于超参数调优结果设置参数lstm_units_1 = 96  # 从调优中获得lstm_units_2 = 48  # 从调优中获得num_heads = 4      # 从调优中获得key_dim = 32       # 从调优中获得dense_units_1 = 48 # 从调优中获得dense_units_2 = 24 # 从调优中获得mlp_units_1 = 96   # 从调优中获得mlp_units_2 = 48   # 从调优中获得mlp_units_3 = 24   # 从调优中获得dropout_rate_1 = 0.2  # 从调优中获得dropout_rate_2 = 0.3  # 从调优中获得learning_rate = 0.0005  # 从调优中获得# 序列输入分支 - 使用更高效的架构seq_input = layers.Input(shape=(seq_length, n_seq_features), name='seq_input')# 使用双向LSTM捕获前后文信息x = layers.Bidirectional(layers.LSTM(lstm_units_1, return_sequences=True))(seq_input)x = layers.Bidirectional(layers.LSTM(lstm_units_2, return_sequences=True))(x)# 简化自注意力机制attention = layers.MultiHeadAttention(num_heads=num_heads, key_dim=key_dim)(x, x)x = layers.Add()([x, attention])  # 使用残差连接而不是拼接x = layers.LayerNormalization()(x)# 使用全局平均池化而不是最后一个LSTM层x = layers.GlobalAveragePooling1D()(x)# 标量输入分支scalar_input = layers.Input(shape=(n_scalar_features,), name='scalar_input')y = layers.Dense(dense_units_1, activation='relu')(scalar_input)y = layers.Dense(dense_units_2, activation='relu')(y)# 合并两个分支combined = layers.Concatenate()([x, y])# 优化MLP层，使用更深的网络但更少的单元z = layers.Dense(mlp_units_1, activation='relu')(combined)z = layers.BatchNormalization()(z)z = layers.Dropout(dropout_rate_1)(z)z = layers.Dense(mlp_units_2, activation='relu')(z)z = layers.BatchNormalization()(z)z = layers.Dropout(dropout_rate_2)(z)z = layers.Dense(mlp_units_3, activation='relu')(z)# 输出层output = layers.Dense(1, activation='linear')(z)model = keras.Model(inputs=[seq_input, scalar_input], outputs=output)# 使用自适应学习率优化器optimizer = keras.optimizers.Adam(learning_rate=learning_rate,beta_1=0.9,beta_2=0.999,epsilon=1e-07,amsgrad=False)model.compile(optimizer=optimizer,loss='mse',metrics=['mae', 'mse'])return model# 创建优化模型
optimized_model = create_optimized_model()
optimized_model.summary()

4.3 高级优化技术

为了实现0.99的R²值，我们需要采用更高级的优化技术：

def create_advanced_model(seq_length=50, n_seq_features=5, n_scalar_features=10):"""创建高级优化模型"""# 序列输入分支seq_input = layers.Input(shape=(seq_length, n_seq_features), name='seq_input')# 使用一维卷积提取局部特征x = layers.Conv1D(64, 3, activation='relu', padding='same')(seq_input)x = layers.BatchNormalization()(x)x = layers.MaxPooling1D(2)(x)x = layers.Conv1D(128, 3, activation='relu', padding='same')(x)x = layers.BatchNormalization()(x)x = layers.MaxPooling1D(2)(x)# 使用门控循环单元(GRU)代替LSTM，计算更高效x = layers.Bidirectional(layers.GRU(64, return_sequences=True))(x)x = layers.Bidirectional(layers.GRU(32, return_sequences=True))(x)# 多头自注意力机制attention = layers.MultiHeadAttention(num_heads=4, key_dim=32)(x, x)# 残差连接和层归一化x = layers.Add()([x, attention])x = layers.LayerNormalization()(x)# 自适应平均池化x = layers.GlobalAveragePooling1D()(x)# 标量输入分支scalar_input = layers.Input(shape=(n_scalar_features,), name='scalar_input')# 使用更深的网络处理标量特征y = layers.Dense(64, activation='relu')(scalar_input)y = layers.BatchNormalization()(y)y = layers.Dropout(0.2)(y)y = layers.Dense(32, activation='relu')(y)y = layers.BatchNormalization()(y)y = layers.Dropout(0.2)(y)# 合并两个分支combined = layers.Concatenate()([x, y])# 深度MLP与残差连接# 第一块z = layers.Dense(128, activation='relu')(combined)z = layers.BatchNormalization()(z)z = layers.Dropout(0.3)(z)# 第二块z1 = layers.Dense(64, activation='relu')(z)z1 = layers.BatchNormalization()(z1)z1 = layers.Dropout(0.3)(z1)# 残差连接z = layers.Add()([z, z1])z = layers.LayerNormalization()(z)# 第三块z2 = layers.Dense(32, activation='relu')(z)z2 = layers.BatchNormalization()(z2)z2 = layers.Dropout(0.2)(z2)# 输出层output = layers.Dense(1, activation='linear')(z2)model = keras.Model(inputs=[seq_input, scalar_input], outputs=output)# 使用带学习率衰减的优化器lr_schedule = keras.optimizers.schedules.ExponentialDecay(initial_learning_rate=0.001,decay_steps=10000,decay_rate=0.9)optimizer = keras.optimizers.Adam(learning_rate=lr_schedule)model.compile(optimizer=optimizer,loss='mse',metrics=['mae'])return model# 创建高级模型
advanced_model = create_advanced_model()
advanced_model.summary()

5. 训练策略与正则化

为了达到0.99的R²值，我们需要精心设计训练策略：

# 定义回调函数
def get_callbacks():"""返回训练回调列表"""callbacks = [# 早停法防止过拟合keras.callbacks.EarlyStopping(monitor='val_loss',patience=15,restore_best_weights=True,verbose=1),# 动态学习率调整keras.callbacks.ReduceLROnPlateau(monitor='val_loss',factor=0.5,patience=5,min_lr=1e-7,verbose=1),# 模型检查点keras.callbacks.ModelCheckpoint('best_model.h5',monitor='val_loss',save_best_only=True,verbose=1),# TensorBoard日志keras.callbacks.TensorBoard(log_dir='./logs',histogram_freq=1)]return callbacks# 自定义损失函数和指标
def r_squared(y_true, y_pred):"""自定义R²指标"""ss_res = tf.reduce_sum(tf.square(y_true - y_pred))ss_tot = tf.reduce_sum(tf.square(y_true - tf.reduce_mean(y_true)))return 1 - ss_res / (ss_tot + tf.keras.backend.epsilon())# 数据增强函数
def augment_sequence_data(X_seq, X_scalar, y, noise_level=0.01):"""对序列数据进行增强"""X_seq_aug = X_seq + np.random.normal(0, noise_level, X_seq.shape)X_scalar_aug = X_scalar + np.random.normal(0, noise_level, X_scalar.shape)return X_seq_aug, X_scalar_aug, y# 创建数据生成器
class DataGenerator(keras.utils.Sequence):"""自定义数据生成器，支持实时数据增强"""def __init__(self, X_seq, X_scalar, y, batch_size=32, shuffle=True, augment=False):self.X_seq = X_seqself.X_scalar = X_scalarself.y = yself.batch_size = batch_sizeself.shuffle = shuffleself.augment = augmentself.indexes = np.arange(len(X_seq))if self.shuffle:np.random.shuffle(self.indexes)def __len__(self):return int(np.ceil(len(self.X_seq) / self.batch_size))def __getitem__(self, index):start_idx = index * self.batch_sizeend_idx = min((index + 1) * self.batch_size, len(self.X_seq))batch_idx = self.indexes[start_idx:end_idx]X_seq_batch = self.X_seq[batch_idx]X_scalar_batch = self.X_scalar[batch_idx]y_batch = self.y[batch_idx]if self.augment:X_seq_batch = X_seq_batch + np.random.normal(0, 0.01, X_seq_batch.shape)X_scalar_batch = X_scalar_batch + np.random.normal(0, 0.01, X_scalar_batch.shape)return [X_seq_batch, X_scalar_batch], y_batchdef on_epoch_end(self):if self.shuffle:np.random.shuffle(self.indexes)# 创建训练和验证数据生成器
train_generator = DataGenerator(X_seq_train, X_scalar_train, y_train, batch_size=64, shuffle=True, augment=True
)val_generator = DataGenerator(X_seq_test, X_scalar_test, y_test,batch_size=64, shuffle=False, augment=False
)

6. 模型训练与评估

现在我们将训练优化后的模型并评估其性能：

# 训练优化模型
print("训练优化模型...")
optimized_history = optimized_model.fit(train_generator,epochs=100,validation_data=val_generator,callbacks=get_callbacks(),verbose=1
)# 训练高级模型
print("训练高级模型...")
advanced_history = advanced_model.fit(train_generator,epochs=100,validation_data=val_generator,callbacks=get_callbacks(),verbose=1
)# 评估模型性能
def evaluate_model(model, X_seq_test, X_scalar_test, y_test, model_name):"""评估模型并返回R²值"""start_time = time.time()y_pred = model.predict([X_seq_test, X_scalar_test]).flatten()inference_time = time.time() - start_timer2 = r2_score(y_test, y_pred)mse = np.mean((y_test - y_pred) ** 2)mae = np.mean(np.abs(y_test - y_pred))print(f"{model_name} 评估结果:")print(f"R²值: {r2:.6f}")print(f"MSE: {mse:.6f}")print(f"MAE: {mae:.6f}")print(f"推理时间: {inference_time:.4f}秒")print()return r2, y_pred# 评估所有模型
baseline_r2, baseline_pred = evaluate_model(baseline_model, X_seq_test, X_scalar_test, y_test, "基准模型"
)optimized_r2, optimized_pred = evaluate_model(optimized_model, X_seq_test, X_scalar_test, y_test, "优化模型"
)advanced_r2, advanced_pred = evaluate_model(advanced_model, X_seq_test, X_scalar_test, y_test, "高级模型"
)# 可视化结果
plt.figure(figsize=(15, 10))# 绘制R²值比较
plt.subplot(2, 2, 1)
models = ['基准模型', '优化模型', '高级模型']
r2_scores = [baseline_r2, optimized_r2, advanced_r2]
colors = ['red', 'orange', 'green']
bars = plt.bar(models, r2_scores, color=colors)
plt.ylabel('R²值')
plt.title('模型R²值比较')
plt.ylim(0.9, 1.0)# 在柱状图上添加数值标签
for bar, score in zip(bars, r2_scores):height = bar.get_height()plt.text(bar.get_x() + bar.get_width()/2., height + 0.005,f'{score:.4f}', ha='center', va='bottom')# 绘制预测值与真实值对比
plt.subplot(2, 2, 2)
plt.scatter(y_test, advanced_pred, alpha=0.5)
plt.plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()], 'r--')
plt.xlabel('真实值')
plt.ylabel('预测值')
plt.title('高级模型: 预测值 vs 真实值')# 绘制残差图
plt.subplot(2, 2, 3)
residuals = y_test - advanced_pred
plt.scatter(advanced_pred, residuals, alpha=0.5)
plt.axhline(y=0, color='r', linestyle='--')
plt.xlabel('预测值')
plt.ylabel('残差')
plt.title('高级模型残差图')# 绘制训练历史
plt.subplot(2, 2, 4)
plt.plot(advanced_history.history['loss'], label='训练损失')
plt.plot(advanced_history.history['val_loss'], label='验证损失')
plt.xlabel('Epoch')
plt.ylabel('损失')
plt.legend()
plt.title('训练和验证损失')plt.tight_layout()
plt.savefig('model_comparison.png', dpi=300, bbox_inches='tight')
plt.show()

7. 模型解释性与特征重要性

为了理解模型的决策过程，我们需要分析特征重要性：

# 特征重要性分析
def analyze_feature_importance(model, X_seq_test, X_scalar_test, y_test):"""分析特征重要性"""# 计算基准性能baseline_pred = model.predict([X_seq_test, X_scalar_test])baseline_r2 = r2_score(y_test, baseline_pred)# 分析序列特征重要性seq_importance = np.zeros(X_seq_test.shape[2])for i in range(X_seq_test.shape[2]):X_seq_perturbed = X_seq_test.copy()np.random.shuffle(X_seq_perturbed[:, :, i])  # 打乱第i个特征perturbed_pred = model.predict([X_seq_perturbed, X_scalar_test])perturbed_r2 = r2_score(y_test, perturbed_pred)seq_importance[i] = baseline_r2 - perturbed_r2# 分析标量特征重要性scalar_importance = np.zeros(X_scalar_test.shape[1])for i in range(X_scalar_test.shape[1]):X_scalar_perturbed = X_scalar_test.copy()np.random.shuffle(X_scalar_perturbed[:, i])  # 打乱第i个特征perturbed_pred = model.predict([X_seq_test, X_scalar_perturbed])perturbed_r2 = r2_score(y_test, perturbed_pred)scalar_importance[i] = baseline_r2 - perturbed_r2return seq_importance, scalar_importance# 计算特征重要性
seq_importance, scalar_importance = analyze_feature_importance(advanced_model, X_seq_test, X_scalar_test, y_test
)# 可视化特征重要性
plt.figure(figsize=(15, 6))plt.subplot(1, 2, 1)
plt.bar(range(len(seq_importance)), seq_importance)
plt.xlabel('序列特征索引')
plt.ylabel('重要性 (R²下降)')
plt.title('序列特征重要性')plt.subplot(1, 2, 2)
plt.bar(range(len(scalar_importance)), scalar_importance)
plt.xlabel('标量特征索引')
plt.ylabel('重要性 (R²下降)')
plt.title('标量特征重要性')plt.tight_layout()
plt.savefig('feature_importance.png', dpi=300, bbox_inches='tight')
plt.show()# 使用SHAP进行更详细的解释
try:import shap# 创建背景数据集background_seq = X_seq_train[np.random.choice(X_seq_train.shape[0], 100, replace=False)]background_scalar = X_scalar_train[np.random.choice(X_scalar_train.shape[0], 100, replace=False)]# 创建解释器def model_predict(data):"""模型预测函数"""if isinstance(data, tuple):seq_data, scalar_data = dataelse:# 假设数据是拼接的seq_len = X_seq_test.shape[1] * X_seq_test.shape[2]seq_data = data[:, :seq_len].reshape(-1, X_seq_test.shape[1], X_seq_test.shape[2])scalar_data = data[:, seq_len:]return advanced_model.predict([seq_data, scalar_data]).flatten()# 创建SHAP解释器explainer = shap.KernelExplainer(model_predict,shap.sample(np.hstack([background_seq.reshape(background_seq.shape[0], -1),background_scalar]), 50))# 计算SHAP值sample_idx = np.random.randint(0, len(X_seq_test))shap_values = explainer.shap_values(np.hstack([X_seq_test[sample_idx].flatten().reshape(1, -1),X_scalar_test[sample_idx].reshape(1, -1)]))# 可视化SHAP值shap.initjs()shap.force_plot(explainer.expected_value, shap_values[0], feature_names=[f"seq_{i}" for i in range(X_seq_test.shape[1]*X_seq_test.shape[2])] + [f"scalar_{i}" for i in range(X_scalar_test.shape[1])])except ImportError:print("SHAP库未安装，跳过SHAP分析")

8. 模型部署与优化

为了实际应用，我们需要优化模型以便部署：

# 模型量化与优化
def optimize_model_for_deployment(model):"""优化模型以便部署"""# 转换为TensorFlow Lite格式converter = tf.lite.TFLiteConverter.from_keras_model(model)# 设置优化选项converter.optimizations = [tf.lite.Optimize.DEFAULT]# 尝试量化到INT8try:def representative_dataset():for i in range(100):yield [X_seq_test[i:i+1].astype(np.float32),X_scalar_test[i:i+1].astype(np.float32)]converter.representative_dataset = representative_datasetconverter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]converter.inference_input_type = tf.int8converter.inference_output_type = tf.int8except Exception as e:print(f"INT8量化失败: {e}, 使用默认量化")# 转换模型tflite_model = converter.convert()# 保存模型with open('optimized_model.tflite', 'wb') as f:f.write(tflite_model)print("模型已优化并保存为TFLite格式")return tflite_model# 优化模型
optimized_tflite_model = optimize_model_for_deployment(advanced_model)# 测试优化后模型性能
interpreter = tf.lite.Interpreter(model_content=optimized_tflite_model)
interpreter.allocate_tensors()input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()# 准备输入数据
tflite_predictions = []
for i in range(len(X_seq_test)):# 设置输入interpreter.set_tensor(input_details[0]['index'], X_seq_test[i:i+1].astype(np.float32))interpreter.set_tensor(input_details[1]['index'], X_scalar_test[i:i+1].astype(np.float32))# 运行推理interpreter.invoke()# 获取输出tflite_pred = interpreter.get_tensor(output_details[0]['index'])tflite_predictions.append(tflite_pred[0][0])tflite_predictions = np.array(tflite_predictions)
tflite_r2 = r2_score(y_test, tflite_predictions)print(f"优化后TFLite模型R²值: {tflite_r2:.6f}")# 性能比较
original_inference_time = []
for i in range(100):start_time = time.time()advanced_model.predict([X_seq_test[i:i+1], X_scalar_test[i:i+1]], verbose=0)original_inference_time.append(time.time() - start_time)tflite_inference_time = []
for i in range(100):start_time = time.time()interpreter.set_tensor(input_details[0]['index'], X_seq_test[i:i+1].astype(np.float32))interpreter.set_tensor(input_details[1]['index'], X_scalar_test[i:i+1].astype(np.float32))interpreter.invoke()interpreter.get_tensor(output_details[0]['index'])tflite_inference_time.append(time.time() - start_time)print(f"原始模型平均推理时间: {np.mean(original_inference_time)*1000:.2f}ms")
print(f"TFLite模型平均推理时间: {np.mean(tflite_inference_time)*1000:.2f}ms")
print(f"加速比: {np.mean(original_inference_time)/np.mean(tflite_inference_time):.2f}x")

9. 完整训练流程与自动化

为了确保实验的可重复性和自动化，我们创建一个完整的训练管道：

def complete_training_pipeline(X_seq, X_scalar, y, test_size=0.2, random_state=42):"""完整的训练管道"""print("开始完整训练管道...")# 数据预处理print("1. 数据预处理...")seq_scaler = StandardScaler()scalar_scaler = StandardScaler()y_scaler = StandardScaler()X_seq_reshaped = X_seq.reshape(-1, X_seq.shape[-1])X_seq_reshaped = seq_scaler.fit_transform(X_seq_reshaped)X_seq = X_seq_reshaped.reshape(X_seq.shape)X_scalar = scalar_scaler.fit_transform(X_scalar)y = y_scaler.fit_transform(y.reshape(-1, 1)).flatten()# 划分训练测试集X_seq_train, X_seq_test, X_scalar_train, X_scalar_test, y_train, y_test = train_test_split(X_seq, X_scalar, y, test_size=test_size, random_state=random_state)# 超参数调优print("2. 超参数调优...")tuner = kt.BayesianOptimization(build_model,objective='val_loss',max_trials=20,executions_per_trial=1,directory='hyperparam_tuning',project_name='final_tuning')tuner.search([X_seq_train, X_scalar_train], y_train,batch_size=64,epochs=20,validation_split=0.2,verbose=0)best_hps = tuner.get_best_hyperparameters(num_trials=1)[0]# 构建最终模型print("3. 构建最终模型...")final_model = tuner.hypermodel.build(best_hps)# 添加自定义指标final_model.compile(optimizer=final_model.optimizer,loss='mse',metrics=['mae', r_squared])# 训练最终模型print("4. 训练最终模型...")final_history = final_model.fit([X_seq_train, X_scalar_train], y_train,batch_size=64,epochs=100,validation_split=0.2,callbacks=get_callbacks(),verbose=1)# 评估最终模型print("5. 评估最终模型...")final_pred = final_model.predict([X_seq_test, X_scalar_test]).flatten()final_r2 = r2_score(y_test, final_pred)print(f"最终模型R²值: {final_r2:.6f}")# 保存最终模型print("6. 保存模型...")final_model.save('final_model.h5')# 保存缩放器import joblibjoblib.dump(seq_scaler, 'seq_scaler.pkl')joblib.dump(scalar_scaler, 'scalar_scaler.pkl')joblib.dump(y_scaler, 'y_scaler.pkl')return final_model, final_r2, final_history# 运行完整管道
final_model, final_r2, final_history = complete_training_pipeline(X_seq, X_scalar, y)# 如果R²值未达到0.99，尝试集成学习
if final_r2 < 0.99:print("R²值未达到0.99，尝试集成学习...")# 创建模型集成def create_ensemble(models, weights=None):"""创建模型集成"""if weights is None:weights = [1/len(models)] * len(models)  # 平均权重def ensemble_predict(X_seq, X_scalar):predictions = []for model in models:pred = model.predict([X_seq, X_scalar], verbose=0).flatten()predictions.append(pred)# 加权平均weighted_avg = np.zeros_like(predictions[0])for i, pred in enumerate(predictions):weighted_avg += weights[i] * predreturn weighted_avgreturn ensemble_predict# 训练多个模型进行集成n_models = 5models = []r2_scores = []for i in range(n_models):print(f"训练集成模型 {i+1}/{n_models}...")model = create_optimized_model()model.compile(optimizer=keras.optimizers.Adam(learning_rate=0.0005),loss='mse',metrics=['mae'])history = model.fit([X_seq_train, X_scalar_train], y_train,batch_size=64,epochs=50,validation_split=0.2,verbose=0)pred = model.predict([X_seq_test, X_scalar_test], verbose=0).flatten()r2 = r2_score(y_test, pred)models.append(model)r2_scores.append(r2)print(f"模型 {i+1} R²值: {r2:.6f}")# 根据性能分配权重weights = np.array(r2_scores) / np.sum(r2_scores)print(f"模型权重: {weights}")# 创建集成预测ensemble_predict = create_ensemble(models, weights)ensemble_pred = ensemble_predict(X_seq_test, X_scalar_test)ensemble_r2 = r2_score(y_test, ensemble_pred)print(f"集成模型R²值: {ensemble_r2:.6f}")if ensemble_r2 >= 0.99:print("成功达到目标R²值0.99!")else:print("未能达到目标R²值0.99，可能需要更多数据或更复杂的模型架构")
else:print("成功达到目标R²值0.99!")