TensorFlow PyTorch 股票預測效能比較

# TensorFlow 版本
import tensorflow as tf

model = tf.keras.models.Sequential([...])
model.compile(optimizer='adam', loss='mean_squared_error')
model.fit(x_train, y_train, epochs=100)

# PyTorch 版本
import torch
import torch.nn as nn

class Model(nn.Module):
    def __init__(self):
        super(Model, self).__init__()
        self.layers = nn.Sequential([...])

    def forward(self, x):
        return self.layers(x)

model = Model()
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
for epoch in range(100):
    optimizer.zero_grad()
    outputs = model(inputs)
    loss = criterion(outputs, labels)
    loss.backward()
    optimizer.step()

name: tsla_regression
channels:
- defaults
dependencies:
- python=3.12.3
- numpy=1.26.4
- tensorflow=2.17.0
- pyspark=3.5.2
- scikit-learn=1.5.1
- matplotlib=3.9.2

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:
    """
    用於TSLA股票價格迴歸的類別。
    """
    def __init__(self):
        """
        初始化TSLARegressor物件。
        """
        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):
        """
        預處理資料。
        
        引數：
        - file_path (str)：CSV檔案的路徑。
        - feature_cols (list)：特徵欄位的列表。
        - train_ratio (float)：訓練資料的比例。
        - seed (int)：用於分割資料的隨機種子。
        
        傳回：
        - tuple：包含訓練和測試DataFrame的元組。
        """
        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 DataFrame轉換為numpy陣列。
        
        引數：
        - train_df (DataFrame)：包含訓練資料的Spark DataFrame。
        - test_df (DataFrame)：包含測試資料的Spark DataFrame。
        
        傳回：
        - tuple：包含訓練和測試特徵及標籤的numpy陣列的元組。
        """
        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

    def create_and_train_model(self, train_features, train_labels, epochs=100, batch_size=32):
        """
        建立和訓練神經網路模型。
        
        引數：
        - train_features (ndarray)：訓練特徵。
        - train_labels (ndarray)：訓練標籤。
        - epochs (int)：訓練的輪數。
        - batch_size (int)：訓練的批次大小。
        
        傳回：
        - None
        """
        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')
        self.model.fit(train_features, train_labels, epochs=epochs, batch_size=batch_size)

    def evaluate_model(self, test_features, test_labels):
        """
        評估訓練好的模型。
        
        引數：
        - test_features (ndarray)：測試特徵。
        - test_labels (ndarray)：測試標籤。
        
        傳回：
        - float：測試損失。
        """
        test_loss = self.model.evaluate(test_features, test_labels)
        return test_loss

    def predict_and_evaluate(self, test_features, test_labels):
        """
        預測和評估神經網路模型。
        
        引數：
        - test_features (ndarray)：測試特徵。
        - test_labels (ndarray)：測試標籤。
        
        傳回：
        - float：R2分數。
        """
        test_predictions = self.model.predict(test_features)
        r2_score_value = r2_score(test_labels, test_predictions)
        return r2_score_value

    def plot_actual_vs_predicted(self, test_features, test_labels):
        """
       繪製實際值與預測值。
        
       引數：
       - test_features (ndarray)：測試特徵。
       - test_labels (ndarray)：測試標籤。
        
       傳回：
       - None
       """
       test_predictions = self.model.predict(test_features)
       print("Actual Predicted")
       for actual, predicted in zip(test_labels, test_predictions):
           print(f"{actual} {predicted}")