根據 TensorFlow 和 Keras 構建特斯拉股價預測模型，首先需對歷史股價資料進行預處理，包含特徵縮放、資料分割等步驟，以確保模型輸入資料的品質。接著，利用 Python 建立多層感知器模型，並使用 Adam 最佳化器與均方誤差作為損失函式進行訓練。訓練過程中，我們可以透過繪製損失函式圖表來監控模型的收斂情況，並藉由調整超引數、網路架構以及應用 L1/L2 正規化等技巧來提升模型效能。此外，為了避免過擬合，我們引入早停機制，及時停止訓練以保留最佳模型狀態。最後，我們使用測試集評估模型的預測能力，並利用 R2 分數等指標來驗證模型的泛化能力。
提升模型效能的技術：正規化與超引數調校
在機器學習領域，模型的效能最佳化是至關重要的。本章將探討多種提升模型效能的技術，包括正規化、提前停止、超引數調校等，並結合實務經驗與具體案例進行分析。
基線模型的建立
首先，我們建立一個基線神經網路模型，用於預測Tesla的股價。該模型由多個密集連線層組成，採用ReLU啟用函式，最後是一個輸出層。這個基線模型將作為我們進行正規化和超引數調校的基準。
程式碼實作：基線模型的建立
import logging
import numpy as np
import tensorflow as tf
from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler, StandardScaler
from sklearn.metrics import r2_score
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense
import matplotlib.pyplot as plt
class TSLARegressor:
def __init__(self):
self.logger = logging.getLogger(__name__)
self.spark = SparkSession.builder.appName("TSLA_Regression").getOrCreate()
self.input_shape = None
self.model = None
def preprocess_data(self, file_path, feature_cols=['Open', 'High', 'Low', 'Volume'], train_ratio=0.8, seed=42):
# 讀取CSV檔案並進行資料預處理
df = self.spark.read.csv(file_path, header=True, inferSchema=True)
assembler = VectorAssembler(inputCols=feature_cols, outputCol='features')
df = assembler.transform(df)
scaler = StandardScaler(inputCol='features', outputCol='scaled_features', withMean=True, withStd=True)
scaler_model = scaler.fit(df)
df = scaler_model.transform(df)
train_df, test_df = df.randomSplit([train_ratio, 1-train_ratio], seed=seed)
return train_df, test_df
def convert_to_numpy(self, train_df, test_df):
# 將Spark DataFrames轉換為NumPy陣列
train_features = train_df.select('scaled_features').collect()
train_labels = train_df.select('Close').collect()
test_features = test_df.select('scaled_features').collect()
test_labels = test_df.select('Close').collect()
train_features = np.array([x['scaled_features'] for x in train_features])
train_labels = np.array([x['Close'] for x in train_labels])
test_features = np.array([x['scaled_features'] for x in test_features])
test_labels = np.array([x['Close'] for x in test_labels])
return train_features, train_labels, test_features, test_labels
內容解密：
- 類別初始化：
TSLARegressor類別的初始化方法設定了記錄器、Spark Session以及模型的初始狀態。 - 資料預處理：
preprocess_data方法讀取CSV檔案，使用VectorAssembler組合特徵欄位，並透過StandardScaler進行特徵縮放，最後將資料分割為訓練集和測試集。 - 轉換為NumPy陣列：
convert_to_numpy方法將Spark DataFrames轉換為NumPy陣列，以便於後續的神經網路訓練。
正規化技術的應用
本章將介紹多種正規化技術，包括Dropout、提前停止、L1和L2正規化等，以提升模型的泛化能力。
L1和L2正規化
L1正規化（Lasso）透過在損失函式中新增權重的絕對值，鼓勵模型權重的稀疏性，從而減少模型複雜度。L2正規化（Ridge）則是新增權重的平方值，鼓勵權重整體變小，防止過擬合。
超引數調校
超引數調校是最佳化模型效能的關鍵步驟，包括學習率的調整、網路架構的修改等。本章將介紹手動和自動化的超引數調校方法，並使用Keras Tuner進行自動化調校。
建構與訓練特斯拉股價預測模型
本章節將詳細介紹如何使用神經網路模型預測特斯拉（TSLA）股價，涵蓋資料預處理、模型建立與訓練、模型評估等步驟。
資料預處理與轉換
首先，我們需要對原始資料進行預處理，包括特徵縮放和資料分割。TSLARegressor 類別中的 preprocess_data 方法負責讀取 CSV 檔案、縮放特徵以及將資料分割為訓練集和測試集。
程式碼實作：
def preprocess_data(self, file_path):
spark_df = spark.read.csv(file_path, header=True, inferSchema=True)
feature_columns = spark_df.columns[:-1]
assembler = VectorAssembler(inputCols=feature_columns, outputCol='features')
scaler = StandardScaler(inputCol='features', outputCol='scaled_features')
pipeline = Pipeline(stages=[assembler, scaler])
model = pipeline.fit(spark_df)
scaled_df = model.transform(spark_df)
train_df, test_df = scaled_df.randomSplit([0.8, 0.2], seed=42)
return train_df, test_df
內容解密：
- 使用
VectorAssembler將多個特徵欄位合併為單一的features欄位。 - 透過
StandardScaler對特徵進行標準化縮放，產生scaled_features欄位。 - 將資料隨機分割為訓練集（80%）和測試集（20%）。
資料轉換為 NumPy 陣列
由於 TensorFlow 模型需要 NumPy 陣列作為輸入，因此我們需要將 Spark DataFrame 轉換為 NumPy 陣列。convert_to_numpy 方法實作了這一轉換。
程式碼實作：
def convert_to_numpy(self, train_df, test_df):
train_features = np.array(train_df.select('scaled_features').rdd.map(lambda x: x.scaled_features.toArray()).collect())
train_labels = np.array(train_df.select('Close').rdd.map(lambda x: x.Close).collect())
test_features = np.array(test_df.select('scaled_features').rdd.map(lambda x: x.scaled_features.toArray()).collect())
test_labels = np.array(test_df.select('Close').rdd.map(lambda x: x.Close).collect())
return train_features, train_labels, test_features, test_labels
內容解密：
- 使用
select和rdd.map將 DataFrame 中的scaled_features和Close欄位轉換為 RDD，並進一步轉換為 NumPy 陣列。 - 透過
collect()方法將分散式資料收集到驅動程式節點，並轉換為 NumPy 陣列。
建立與訓練神經網路模型
create_and_train_model 方法負責建立和訓練神經網路模型。該模型採用序列式（Sequential）架構，並使用 Adam 最佳化器和均方誤差（MSE）作為損失函式。
程式碼實作：
def create_and_train_model(self, train_features, train_labels, epochs=100, batch_size=32):
self.input_shape = (train_features.shape[1],)
self.model = Sequential([
Dense(64, activation='relu', input_shape=self.input_shape),
Dense(32, activation='relu'),
Dense(16, activation='relu'),
Dense(1)
])
self.model.compile(optimizer='adam', loss='mse')
history = self.model.fit(train_features, train_labels, epochs=epochs, batch_size=batch_size, verbose=0)
return history
內容解密：
- 定義一個具有多層密集連線（Dense）層的神經網路模型，採用 ReLU 作為啟用函式。
- 使用 Adam 最佳化器和均方誤差（MSE）作為損失函式來編譯模型。
- 透過
fit方法訓練模型，設定訓練週期（epochs）和批次大小（batch_size）。
模型評估
模型訓練完成後，我們需要對其進行評估。evaluate_model 方法用於計算測試資料上的損失，而 predict_and_evaluate 方法則用於計算 R2 分數，以評估模型的預測效能。
程式碼實作：
def evaluate_model(self, test_features, test_labels):
test_loss = self.model.evaluate(test_features, test_labels)
return test_loss
def predict_and_evaluate(self, test_features, test_labels):
test_predictions = self.model.predict(test_features)
r2_score_value = r2_score(test_labels, test_predictions)
return r2_score_value
內容解密：
- 使用
evaluate方法計算測試資料上的損失。 - 使用
predict方法進行預測，並透過 R2 分數評估模型的預測能力。
主函式與執行流程
主函式 main 負責協調整個工作流程，包括資料預處理、模型訓練、模型評估和結果記錄。
程式碼實作：
def main(file_path):
try:
tsla_regressor = TSLARegressor()
train_df, test_df = tsla_regressor.preprocess_data(file_path)
train_features, train_labels, test_features, test_labels = tsla_regressor.convert_to_numpy(train_df, test_df)
history = tsla_regressor.create_and_train_model(train_features, train_labels)
test_loss = tsla_regressor.evaluate_model(test_features, test_labels)
r2_score_value = tsla_regressor.predict_and_evaluate(test_features, test_labels)
logging.info("Test Loss: {}".format(test_loss))
logging.info("R2 Score: {}".format(r2_score_value))
plt.plot(history.history['loss'])
plt.title('Model Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.show()
except Exception as e:
logging.error(f"An error occurred: {e}")
內容解密：
- 建立
TSLARegressor物件並呼叫各個方法完成資料預處理、模型訓練和評估。 - 紀錄測試損失和 R2 分數，並繪製訓練過程中的損失變化圖。
提升模型效能的關鍵技術：以特斯拉股票價格預測為例
在機器學習領域，模型的效能直接影響到預測的準確性和可靠性。以特斯拉（TSLA）股票價格預測為例，透過使用TensorFlow框架，我們可以構建一個強大的迴歸模型來預測股票價格。在本章中，我們將探討如何提升模型的效能，包括資料預處理、模型架構調整、以及各種正則化技術的應用。
模型訓練與評估
首先，我們使用predict_tsla_stock.py指令碼來訓練和評估我們的模型。該指令碼使用Python 3在/home/ubuntu/airflow/dags目錄下執行。模型的評估指標包括測試損失（Test Loss）和R2分數（R2 Score），這些指標對於評估模型的效能至關重要。
程式碼範例：模型評估
test_loss = tsla_regressor.evaluate_model(test_features, test_labels)
logging.info("Test Loss: {}".format(test_loss))
r2_score_value = tsla_regressor.predict_and_evaluate(test_features, test_labels)
logging.info("R2 Score: {}".format(r2_score_value))
內容解密：
evaluate_model方法用於計算測試損失，評估模型在測試資料上的表現。predict_and_evaluate方法用於計算R2分數，衡量模型對資料變異性的解釋能力。- 低測試損失（15.5924）和高R2分數（0.9987）表明模型在測試資料上表現出色。
訓練損失視覺化
模型的訓練過程可以透過繪製訓練損失圖來視覺化。這有助於我們瞭解模型的收斂情況和是否存在過擬合或欠擬合的問題。
程式碼範例：訓練損失視覺化
history = tsla_regressor.create_and_train_model(train_features, train_labels)
plt.plot(history.history['loss'])
plt.title('Model Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.axvline(x=10, color='r', linestyle='--', label='Epoch 10 plateau')
plt.savefig('model_loss_plot.png')
plt.show()
內容解密：
create_and_train_model方法傳回訓練歷史，包括每個epoch的損失值。- 使用
matplotlib繪製訓練損失圖，並在第10個epoch處新增垂直線標記。 - 圖表顯示訓練損失在第10個epoch後趨於平穩，表明模型可能已經收斂。
模型收斂與過擬合分析
當模型的訓練損失趨於平穩時，可能意味著模型已經收斂或過擬合。我們需要進一步分析模型的表現，以確定最佳的處理方案。
- 如果模型在驗證集上的表現良好，則平穩的訓練損失可能表明模型已經成功學習了訓練資料的模式。
- 如果懷疑模型過擬合，可以採取以下措施：
- 調整超引數（學習率、批次大小等）
- 簡化或複雜化模型架構
- 使用資料增強技術增加訓練資料的多樣性
- 實施正則化技術（如Dropout、L1/L2正則化、早停法）以防止過擬合
早停法：防止過擬合的有效技術
早停法是一種簡單而有效的正則化技術，透過監控模型在驗證集上的表現，在效能開始下降時停止訓練，從而防止過擬合。
程式碼範例：早停法實作
from tensorflow.keras.callbacks import EarlyStopping
def create_and_train_model(self, train_features, train_labels, epochs=100, batch_size=32):
# ... 模型定義 ...
early_stopping = EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)
self.model.compile(optimizer='adam', loss='mse')
history = self.model.fit(train_features, train_labels, epochs=epochs, batch_size=batch_size,
verbose=0, validation_data=(val_features, val_labels),
callbacks=[early_stopping])
內容解密：
EarlyStopping回撥函式用於監控驗證損失，並在損失連續10個epoch沒有改善時停止訓練。restore_best_weights=True確保模型還原到驗證損失最低時的權重。- 透過早停法，可以有效防止模型過擬合，提高模型的泛化能力。
總之，提升模型效能需要綜合考慮資料預處理、模型架構、超引數調整和正則化技術等多個方面。透過視覺化和分析訓練過程，我們可以更好地理解模型的表現，並採取相應的最佳化措施。早停法作為一種簡單有效的正則化技術，可以幫助我們防止過擬合，提高模型的泛化能力。
