優化神經網絡模型以提升R2值至0.99的全面方案

1. 問題分析與背景

在深度學習項目中，提升模型的R2（決定系數）值至0.99是一個具有挑戰性的目標，特別是在處理復雜的時間序列數據時。我們的現有模型結合了LSTM層、自注意力機制和MLP處理標量輸入，這種復雜結構雖然強大，但可能存在冗余和效率低下的問題。

R2值是衡量模型對目標變量變異解釋程度的指標，值越接近1表示模型擬合效果越好。要達到0.99的R2值，我們需要精心設計模型架構、優化超參數并確保訓練過程高效穩定。

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"基準模型R2值: {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的R2值，我們需要采用更高級的優化技術：

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的R2值，我們需要精心設計訓練策略：

# 定義回調函數
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):"""自定義R2指標"""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):"""評估模型并返回R2值"""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"R2值: {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))# 繪制R2值比較
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('R2值')
plt.title('模型R2值比較')
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('重要性 (R2下降)')
plt.title('序列特征重要性')plt.subplot(1, 2, 2)
plt.bar(range(len(scalar_importance)), scalar_importance)
plt.xlabel('標量特征索引')
plt.ylabel('重要性 (R2下降)')
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模型R2值: {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"最終模型R2值: {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)# 如果R2值未達到0.99，嘗試集成學習
if final_r2 < 0.99:print("R2值未達到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} R2值: {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"集成模型R2值: {ensemble_r2:.6f}")if ensemble_r2 >= 0.99:print("成功達到目標R2值0.99!")else:print("未能達到目標R2值0.99，可能需要更多數據或更復雜的模型架構")
else:print("成功達到目標R2值0.99!")