作者：老余捞鱼
原创不易，转载请标明出处及原作者。
**写在前面的话：
**本文手把手教会大家使用 Python 和 AI 进行股票交易预测。首先介绍了不同的预测方法，特别是 LSTM 处理序列预测的能力。然后提供了概念验证步骤，包括安装、创建项目等，还展示代码建立，如导入库、用函数训练测试模型，最后还评估了模型的性能。
** 我们探寻了多种预测股价的方式，像 Facebook 的 Prophet 等预测工具、SARIMA 模型等统计手段、多项式回归等机器学习策略，还有基于人工智能的循环神经网络（RNN）。在众多人工智能模型与技术里，我们发现长短时记忆（LSTM）模型能带来最理想的结果。**
** **** **LSTM 模型是递归神经网络架构的一种变形，擅长处理序列预测难题。它与传统的前馈神经网络不同，具有类似记忆的结构，能在大量序列中保留上下文数据。这一特性使其非常适合时间序列预测、自然语言处理以及其他依赖序列数据的任务。它通过缓解消失和梯度爆炸问题，解决了标准 RNN 的基本缺陷，从而提升了模型识别数据集内长期依赖关系的能力。因此，LSTM 已成为需要长时间深入理解数据的复杂任务的首选。
为了验证其有效性，我们开发了一个概念验证。
一、准备工作
- 你需要在你的计算机中（或选择使用 VSCode 会更加方便）安装最新版本的 Python 和 PIP。
- 创建一个带有 "main.py "文件的 Python 项目。
- 在项目中添加 “data”目录。
- 设置并激活虚拟环境。
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调用和模型训练。
通过可视化实际和预测的收盘价，以及计算相关的误差指标，可以直观地评估模型的预测准确性。
模型的训练和测试应该使用不同的数据集，以确保模型的泛化能力。
调整超参数和使用额外的训练数据可以进一步提高模型的性能。
模型的预测结果可以作为交易决策的参考，但应谨慎使用，因为预测并不总是准确的。
本文内容仅仅是技术探讨和学习，并不构成任何投资建议。
转发请注明原作者和出处。
版权归原作者 老余捞鱼 所有， 如有侵权，请联系我们删除。