手把手教会你用 AI 和 Python 进行股票交易预测（完整代码干货）

标签：scaler AI 手把手 Python API test close model data

作者：老余捞鱼

原创不易，转载请标明出处及原作者。

写在前面的话：
       本文手把手教会大家使用 Python 和 AI 进行股票交易预测。首先介绍了不同的预测方法，特别是 LSTM 处理序列预测的能力。然后提供了概念验证步骤，包括安装、创建项目等，还展示代码建立，如导入库、用函数训练测试模型，最后还评估了模型的性能。

       我们探寻了多种预测股价的方式，像 Facebook 的 Prophet 等预测工具、SARIMA 模型等统计手段、多项式回归等机器学习策略，还有基于人工智能的循环神经网络（RNN）。在众多人工智能模型与技术里，我们发现长短时记忆（LSTM）模型能带来最理想的结果。

              LSTM 模型是递归神经网络架构的一种变形，擅长处理序列预测难题。它与传统的前馈神经网络不同，具有类似记忆的结构，能在大量序列中保留上下文数据。这一特性使其非常适合时间序列预测、自然语言处理以及其他依赖序列数据的任务。它通过缓解消失和梯度爆炸问题，解决了标准 RNN 的基本缺陷，从而提升了模型识别数据集内长期依赖关系的能力。因此，LSTM 已成为需要长时间深入理解数据的复杂任务的首选。

为了验证其有效性，我们开发了一个概念验证。

一、准备工作

1. 你需要在你的计算机中（或选择使用 VSCode 会更加方便）安装最新版本的 Python 和 PIP。
2. 创建一个带有 "main.py "文件的 Python 项目。
3. 在项目中添加 “data”目录。
4. 设置并激活虚拟环境。

trading-ai-lstm $ python3 -m venv venv
trading-ai-lstm $ source venv/.bin/activate
(venv) trading-ai-lstm $

​​​​​​​       创建一个 "requirements.txt "文件。

pandas
numpy
scikit-learn
scipy
matplotlib
tensorflow
eodhd
python-dotenv

​​​​​​​       确保已在虚拟环境中升级 PIP 并安装依赖项。

(venv) trading-ai-lstm $ pip install --upgrade pip
(venv) trading-ai-lstm $ python3 -m pip install -r requirements.txt

​​​​​​​       需要在".env "文件中加入了 EODHD API 的 API 密钥。

API_TOKEN=<YOUR_API_KEY_GOES_HERE>

​​​​​​​       一切就绪。如果你正在使用  VSCode ，并希望使用与我们相同的".vscode/settings.json "文件，请点击 Fork 本项目 GitHub 仓库，以备不时之需。

{
  "python.formatting.provider": "none",
  "python.formatting.blackArgs": ["--line-length", "160"],
  "python.linting.flake8Args": [
    "--max-line-length=160",
    "--ignore=E203,E266,E501,W503,F403,F401,C901"
  ],
  "python.analysis.diagnosticSeverityOverrides": {
    "reportUnusedImport": "information",
    "reportMissingImports": "none"
  },
  "[python]": {
    "editor.defaultFormatter": "ms-python.black-formatter"
  }
}

二、代码构建

​​​​​​​       第一步是导入必要的库。

import os
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "1"

import pickle
import pandas as pd
import numpy as np
from dotenv import load_dotenv
from sklearn.metrics import mean_squared_error, mean_absolute_error
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from tensorflow.keras.models import load_model
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt
from eodhd import APIClient

​​​​​​​       TensorFlow 往往会自动生成诸多警告与调试信息。而我们更倾向于简洁明了的输出，故而对这些通知进行了控制。这可以在导入“os”模块后，借助 os.environ 来达成。

​​​​​​​       机器学习和人工智能模型的训练过程需要大量的微调，主要是通过所谓的超参数（hyperparameters）进行管理。这个问题错综复杂，掌握它需要不断学习和耐心，最佳超参数的选择受到各种因素的影响。根据我们通过 EODHD API 获取的标准普尔 500 指数每日数据，我们首先使用了一些广为认可的设置。我们鼓励您修改这些设置以提高结果。目前，建议将序列长度保持在 20。

# Configurable hyperparameters
seq_length = 20
batch_size = 64
lstm_units = 50
epochs = 100

​​​​​​​       下一步是从我们的".env "文件中获取 EODHD API’s 的 API_TOKEN。

# Load environment variables from the .env file
load_dotenv()

# Retrieve the API key
API_TOKEN = os.getenv("API_TOKEN")

if API_TOKEN is not None:
    print(f"API key loaded: {API_TOKEN[:4]}********")
else:
    raise LookupError("Failed to load API key.")

​​​​​​​       需要确保拥有有效的 EODHD API 的 API_TOKEN 才能成功访问数据。

​​​​​​​       我们已经建立了几个可重复使用的函数，并将在下文中详细介绍它们的功能。我把这些函数进行了代码注释，以说明其操作。

def get_ohlc_data(use_cache: bool = False) -> pd.DataFrame:
    ohlcv_file = "data/ohlcv.csv"

    if use_cache:
        if os.path.exists(ohlcv_file):
            return pd.read_csv(ohlcv_file, index_col=None)
        else:
            api = APIClient(API_TOKEN)
            df = api.get_historical_data(
                symbol="HSPX.LSE",
                interval="d",
                iso8601_start="2010-05-17",
                iso8601_end="2023-10-04",
            )
            df.to_csv(ohlcv_file, index=False)
            return df
    else:
        api = APIClient(API_TOKEN)
        return api.get_historical_data(
            symbol="HSPX.LSE",
            interval="d",
            iso8601_start="2010-05-17",
            iso8601_end="2023-10-04",
        )

def create_sequences(data, seq_length):
    x, y = [], []
    for i in range(len(data) - seq_length):
        x.append(data[i : i + seq_length])
        y.append(data[i + seq_length, 3])  # The prediction target "close" is the 4th column (index 3)
    return np.array(x), np.array(y)

def get_features(df: pd.DataFrame = None, feature_columns: list = ["open", "high", "low", "close", "volume"]) -> list:
    return df[feature_columns].values

def get_target(df: pd.DataFrame = None, target_column: str = "close") -> list:
    return df[target_column].values

def get_scaler(use_cache: bool = True) -> MinMaxScaler:
    scaler_file = "data/scaler.pkl"

    if use_cache:
        if os.path.exists(scaler_file):
            # Load the scaler
            with open(scaler_file, "rb") as f:
                return pickle.load(f)
        else:
            scaler = MinMaxScaler(feature_range=(0, 1))
            with open(scaler_file, "wb") as f:
                pickle.dump(scaler, f)
            return scaler
    else:
        return MinMaxScaler(feature_range=(0, 1))

def scale_features(scaler: MinMaxScaler = None, features: list = []):
    return scaler.fit_transform(features)

def get_lstm_model(use_cache: bool = False) -> Sequential:
    model_file = "data/lstm_model.h5"

    if use_cache:
        if os.path.exists(model_file):
            # Load the model
            return load_model(model_file)
        else:
            # Train the LSTM model and save it
            model = Sequential()
            model.add(LSTM(units=lstm_units, activation='tanh', input_shape=(seq_length, 5)))
            model.add(Dropout(0.2))
            model.add(Dense(units=1))

            model.compile(optimizer="adam", loss="mean_squared_error")
            model.fit(x_train, y_train, epochs=epochs, batch_size=batch_size, validation_data=(x_test, y_test))

            # Save the entire model to a HDF5 file
            model.save(model_file)

            return model

    else:
        # Train the LSTM model
        model = Sequential()
        model.add(LSTM(units=lstm_units, activation='tanh', input_shape=(seq_length, 5)))
        model.add(Dropout(0.2))
        model.add(Dense(units=1))

        model.compile(optimizer="adam", loss="mean_squared_error")
        model.fit(x_train, y_train, epochs=epochs, batch_size=batch_size, validation_data=(x_test, y_test))

        return model

def get_predicted_x_test_prices(x_test: np.ndarray = None):
    predicted = model.predict(x_test)

    # Create a zero-filled matrix to aid in inverse transformation
    zero_filled_matrix = np.zeros((predicted.shape[0], 5))

    # Replace the 'close' column of zero_filled_matrix with the predicted values
    zero_filled_matrix[:, 3] = np.squeeze(predicted)

    # Perform inverse transformation
    return scaler.inverse_transform(zero_filled_matrix)[:, 3]

def plot_x_test_actual_vs_predicted(actual_close_prices: list = [], predicted_x_test_close_prices = []) -> None:
    # Plotting the actual and predicted close prices
    plt.figure(figsize=(14, 7))
    plt.plot(actual_close_prices, label="Actual Close Prices", color="blue")
    plt.plot(predicted_x_test_close_prices, label="Predicted Close Prices", color="red")
    plt.title("Actual vs Predicted Close Prices")
    plt.xlabel("Time")
    plt.ylabel("Price")
    plt.legend()
    plt.show()

def predict_next_close(df: pd.DataFrame = None, scaler: MinMaxScaler = None) -> float:
    # Take the last X days of data and scale it
    last_x_days = df.iloc[-seq_length:][["open", "high", "low", "close", "volume"]].values
    last_x_days_scaled = scaler.transform(last_x_days)

    # Reshape this data to be a single sequence and make the prediction
    last_x_days_scaled = np.reshape(last_x_days_scaled, (1, seq_length, 5))

    # Predict the future close price
    future_close_price = model.predict(last_x_days_scaled)

    # Create a zero-filled matrix for the inverse transformation
    zero_filled_matrix = np.zeros((1, 5))

    # Put the predicted value in the 'close' column (index 3)
    zero_filled_matrix[0, 3] = np.squeeze(future_close_price)

    # Perform the inverse transformation to get the future price on the original scale
    return scaler.inverse_transform(zero_filled_matrix)[0, 3]

def evaluate_model(x_test: list = []) -> None:
    # Evaluate the model
    y_pred = model.predict(x_test)
    mse = mean_squared_error(y_test, y_pred)
    mae = mean_absolute_error(y_test, y_pred)
    rmse = np.sqrt(mse)

    print(f"Mean Squared Error: {mse}")
    print(f"Mean Absolute Error: {mae}")
    print(f"Root Mean Squared Error: {rmse}")

​​​​​​​       我们需着重指出的是，在各类函数中增添了“use_cache”变量。此策略意在降低对 EODHD 应用程序接口的冗余 API 调用，防止利用相同的每日数据对模型进行重复的重新训练。激活“use_cache”变量会将数据存储至“data/”目录下的文件里。若数据不存在，则会创建；若已存在，则会加载。当多次运行脚本时，此方法能显著提升效率。若要在每次运行时获取新数据，只需在调用函数时禁用“use_cache”选项或清空“data/”目录中的文件，就能得到相同的结果。

​​​​​​​       现在进入代码的核心部分...

if __name__ == "__main__":
    # Retrieve 3369 days of S&P 500 data
    df = get_ohlc_data(use_cache=True)
    print(df)

​​​​​​​       首先，我们从 EODHD API  获取 OHLCV 数据，并将其存入名为 "df "的 Pandas DataFrame。OHLCV 表示开盘价、最高价、最低价、收盘价和成交量，是交易蜡烛图数据的标准属性。如前所述，我们启用了缓存以简化流程。我们还可以选择在屏幕上显示这些数据。

​​​​​​​       我们将一次性介绍以下代码块...

    features = get_features(df)
    target = get_target(df)

    scaler = get_scaler(use_cache=True)
    scaled_features = scale_features(scaler, features)

    x, y = create_sequences(scaled_features, seq_length)

    train_size = int(0.8 * len(x))  # Create a train/test split of 80/20%
    x_train, x_test = x[:train_size], x[train_size:]
    y_train, y_test = y[:train_size], y[train_size:]

    # Re-shape input to fit lstm layer
    x_train = np.reshape(x_train, (x_train.shape[0], seq_length, 5))  # 5 features
    x_test = np.reshape(x_test, (x_test.shape[0], seq_length, 5))  # 5 features

• “features” 包括我们将用来预测目标（即 “close”）的一系列输入。
• “target” 包含一个目标值列表，如 "close"。
• “scaler”代表一种用于将数字标准化的方法，使它们具有可比性。例如，我们的数据集开始时的接近值可能是 784，最后可能是 3538。最后一行的数字越高，并不意味着预测的意义越大。归一化可确保可比性。
• “scaled_features” 是缩放过程的结果，我们将用它来训练人工智能模型。
• “x_train” and “x_test” 分别表示我们将用于训练和测试人工智能模型的数据集，通常的做法是 80/20 分配。这意味着 80% 的交易数据用于训练，20% 用于测试模型。x "表示这些特征或输入。
• “y_train” and “y_test” 的功能类似，但只包含目标值，如 "close"。
• 最后，必须对数据进行重塑，以满足 LSTM 层的要求。

​​​​​​​       我们开发了一种功能，既能对模型进行重新训练，又能载入之前已训练好的模型。

model = get_lstm_model(use_cache=True)

​​​​​​​       从显示的图片中可以一窥训练序列。你会发现，最初， “loss”和 “val_loss” 指标可能并不完全一致。不过，随着训练的进行，这些数据有望趋于一致，这表明训练取得了进展。

• Loss: 这是在训练数据集上计算的均方误差（MSE）。它反映了每个训练期预测标签和真实标签之间的“cost” 或 “error” 。我们的目标是通过连续的历时来减少这一数字。
• Val_loss: 这个均方误差是在验证数据集上确定的，用于衡量模型在训练过程中未遇到的数据上的表现。它是模型泛化到新的未见数据能力的指标。

​​​​​​​       查看测试集的预测收盘价列表，可以使用此代码。

    predicted_x_test_close_prices = get_predicted_x_test_prices(x_test)
    print("Predicted close prices:", predicted_x_test_close_prices)

​​​​​​​       单看这些数据，可能并不特别具有启发性或直观。不过，通过绘制实际收盘价与预测收盘价的对比图（请注意，这只占整个数据集的 20%），我们可以得到更清晰的图像，如下图所示。

    # Plot the actual and predicted close prices for the test data
    plot_x_test_actual_vs_predicted(df["close"].tail(len(predicted_x_test_close_prices)).values, predicted_x_test_close_prices)

​​​​​​​       结果表明，在测试阶段，该模型在预测收盘价方面表现出色。

​​​​​​​       现在，我们来看看最令人期待的方面：我们能确定明天的预测收盘价吗？

   # Predict the next close price
    predicted_next_close =  predict_next_close(df, scaler)
    print("Predicted next close price:", predicted_next_close)

Predicted next close price: 3536.906685638428

​​​​​​​       这是一个用于教育目的的基本示例，仅仅是一个开始。从这里开始，您可以考虑加入更多的训练数据，调整超参数，或将模型应用于不同的市场和时间区间。

​​​​​​​       如果您想对模型进行评估，可以将其包括在内。

 # Evaluate the model
    evaluate_model(x_test)

​​​​​​​       在我们的方案中的输出情况是:

Mean Squared Error: 0.00021641664334765608
Mean Absolute Error: 0.01157513692221611
Root Mean Squared Error: 0.014711106122506767

​​​​​​​       "平均平方误差"（mean_squared_error）和 "平均绝对误差"（mean_absolute_error）函数来自 scikit-learn 的度量模块，分别用于计算平均平方误差（MSE）和平均绝对误差（MAE）。均方根误差 (RMSE) 是通过对 MSE 取平方根得出的。

​​​​​​​       这些指标为模型的准确性提供了数字化的评估，也为模型的性能进行了定量的分析，而图形化的展示则更有利于直观地对比预测值与实际数值，以及直观地比较预测值和实际值。

三、总结

​​​​​​​       在本文中我详细介绍了用 Python 和 AI 做交易预测的流程。首先是各种预测办法，像 Facebook 的 Prophet、SARIMA 模型、多项式回归，还有基于人工智能的循环神经网络（RNN），这里面我觉得 LSTM 模型最厉害。LSTM 模型是种特殊的递归神经网络，能处理序列预测问题，还解决了标准 RNN 的消失和梯度爆炸问题，适合时间序列预测和自然语言处理这些任务。

​​​​​​​       接下来，我给大家提供了一个概念验证的准备步骤，包括安装Python和PIP、创建项目和文件、设置虚拟环境以及创建requirements.txt文件。还包括 VSCode的设置文件示例，以及本项目的 GitHub 代码仓库。

​​​​​​​       而在建立代码的部分，我详细说明了如何导入必要的库和调用 EODHD API’s，并介绍了一系列可重用的函数，这些函数用于获取数据、创建序列、获取特征和目标值、缩放特征、获取LSTM模型、进行预测以及评估模型。此外，我们还讨论了如何使用缓存来减少不必要的API调用和数据重复加载。

​​​​​​​       最后，本文展示了如何使用这些函数来训练和测试LSTM模型，并展示了如何预测下一个交易日的收盘价。通过比较实际收盘价和预测收盘价的图表，以及计算均方误差（MSE）、均方根误差（RMSE）和均绝对误差（MAE）等指标，来评估模型的性能。简单总结起来就是下面6句话：

LSTM模型在交易预测中的效果优于其他方法，因为它能够更好地处理长期依赖问题。

使用缓存机制可以提高数据处理的效率，避免重复的API调用和模型训练。

通过可视化实际和预测的收盘价，以及计算相关的误差指标，可以直观地评估模型的预测准确性。

模型的训练和测试应该使用不同的数据集，以确保模型的泛化能力。

调整超参数和使用额外的训练数据可以进一步提高模型的性能。

模型的预测结果可以作为交易决策的参考，但应谨慎使用，因为预测并不总是准确的。

---

本文内容仅仅是技术探讨和学习，并不构成任何投资建议。

转发请注明原作者和出处。

标签：scaler,AI,手把手,Python,API,test,close,model,data
From： https://blog.csdn.net/weixin_70955880/article/details/142328182