用TF和Keras在Python中预测股票价格

预测股价一直是吸引投资者和研究人员的话题。投资者总是猜测股票的价格是否会上涨，因为有许多复杂的财务指标，只有投资者和具有良好财务知识的专业人员才能理解，所以股市的走势对普通百姓来说非常难以琢磨。

但是随着AI人工智能技术的兴起，它可以帮助我们进行股票价格预测并获得稳定的财富，并且可以帮助专家获得最有用的指标并做出更好的预测。

本教程的目的是在TensorFlow 2和Keras中构建一个预测股市价格的神经网络。更具体地说，我们将使用LSTM单元构建循环神经网络，因为这个神经网络是时间序列预测的最新技术。

▊ 安装环境

好吧，让我们开始吧。首先，您需要安装Tensorflow 2和其他库：

pip3 install tensorflow pandas numpy matplotlib yahoo_fin sklearn

完成所有设置后，打开一个新的Python文件（或bfwstudio）并导入以下库：

import tensorflow as tffrom tensorflow.keras.models import Sequentialfrom tensorflow.keras.layers import LSTM, Dense, Dropout, Bidirectionalfrom tensorflow.keras.callbacks import ModelCheckpoint, TensorBoardfrom sklearn import preprocessingfrom sklearn.model_selection import train_test_splitfrom sklearn.metrics import accuracy_scorefrom yahoo_fin import stock_info as sifrom collections import dequeimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport timeimport osimport random

我们正在使用yahoo_fin模块，它实际上是一个Python抓取工具，可从Yahoo Finance平台提取财务数据，因此它不是可靠的API，请随时使用Alpha Vantage等其他数据源。

另外，我们需要确保在进行训练/测试后能获得稳定的结果，可以设置种子：

np.random.seed(314)tf.random.set_seed(314)random.seed(314)

▊ 准备数据集

第一步，我们需要编写一个函数，该函数从Internet下载数据集并对其进行预处理：

def load_data(ticker, n_steps=50, scale=True, shuffle=True, lookup_step=1,                 test_size=0.2, feature_columns=['adjclose', 'volume', 'open', 'high', 'low']):    '''    Loads data from Yahoo Finance source, as well as scaling, shuffling, normalizing and splitting.    Params:        ticker (str/pd.DataFrame): the ticker you want to load, examples include AAPL, TESL, etc.        n_steps (int): the historical sequence length (i.e window size) used to predict, default is 50        scale (bool): whether to scale prices from 0 to 1, default is True        shuffle (bool): whether to shuffle the data, default is True        lookup_step (int): the future lookup step to predict, default is 1 (e.g next day)        test_size (float): ratio for test data, default is 0.2 (20% testing data)        feature_columns (list): the list of features to use to feed into the model, default is everything grabbed from yahoo_fin    '''    # see if ticker is already a loaded stock from yahoo finance    if isinstance(ticker, str):        # load it from yahoo_fin library        df = si.get_data(ticker)    elif isinstance(ticker, pd.DataFrame):        # already loaded, use it directly        df = ticker    else:        raise TypeError('ticker can be either a str or a `pd.DataFrame` instances')    # this will contain all the elements we want to return from this function    result = {}    # we will also return the original dataframe itself    result['df'] = df.copy()    # make sure that the passed feature_columns exist in the dataframe    for col in feature_columns:        assert col in df.columns, f''{col}' does not exist in the dataframe.'    if scale:        column_scaler = {}        # scale the data (prices) from 0 to 1        for column in feature_columns:            scaler = preprocessing.MinMaxScaler()            df[column] = scaler.fit_transform(np.expand_dims(df[column].values, axis=1))            column_scaler[column] = scaler        # add the MinMaxScaler instances to the result returned        result['column_scaler'] = column_scaler    # add the target column (label) by shifting by `lookup_step`    df['future'] = df['adjclose'].shift(-lookup_step)    # last `lookup_step` columns contains NaN in future column    # get them before droping NaNs    last_sequence = np.array(df[feature_columns].tail(lookup_step))    # drop NaNs    df.dropna(inplace=True)    sequence_data = []    sequences = deque(maxlen=n_steps)    for entry, target in zip(df[feature_columns].values, df['future'].values):        sequences.append(entry)        if len(sequences) == n_steps:            sequence_data.append([np.array(sequences), target])    # get the last sequence by appending the last `n_step` sequence with `lookup_step` sequence    # for instance, if n_steps=50 and lookup_step=10, last_sequence should be of 60 (that is 50+10) length    # this last_sequence will be used to predict future stock prices not available in the dataset    last_sequence = list(sequences) + list(last_sequence)    last_sequence = np.array(last_sequence)    # add to result    result['last_sequence'] = last_sequence    # construct the X's and y's    X, y = [], []    for seq, target in sequence_data:        X.append(seq)        y.append(target)    # convert to numpy arrays    X = np.array(X)    y = np.array(y)    # reshape X to fit the neural network    X = X.reshape((X.shape[0], X.shape[2], X.shape[1]))    # split the dataset    result['X_train'], result['X_test'], result['y_train'], result['y_test'] = train_test_split(X, y,                                                                                test_size=test_size, shuffle=shuffle)    # return the result    return result

此函数很长但很方便，它接受几个参数以使其尽可能灵活。

第一个参数ticker表示我们要加载的股票代码，例如，你可以使用TSLA特斯拉股市，AAPL苹果等。

n_steps整数表示我们要使用的历史序列长度，有人称它为窗口大小，回想一下我们将要使用递归神经网络，因此我们需要将序列数据输入网络，选择50表示我们将使用50天的股价来预测第二天。

scale是一个布尔变量，指示是否将价格从0缩放到1，我们将其设置为True，因为将高值从0缩放到1将帮助神经网络更快，更有效地学习。

lookup_step是要预测的将来的查找步骤，默认设置为1（例如，第二天）。

我们将使用此数据集中所有可用的功能，即开盘价，最高价，最低价，成交量和调整后的收盘价。

上面的函数执行以下操作：

首先，它使用yahoo_fin模块中的stock_info.get_data（）函数加载数据集。

如果将scale参数作为True传递，它将使用sklearn的MinMaxScaler类将所有价格从0缩放到1（包括volume）。请注意，每列都有自己的缩放器。

然后，通过将调整后的关闭列移动lookup_step，添加指示目标值（用于预测的标签或y的目标）的Future列。

之后，它会重新整理和拆分数据并返回结果。

为了更好地理解代码，我强烈建议您手动打印输出变量（result），并查看功能和标签的制作方式。

▊ 模型制作

现在我们有了一个适当的函数来加载和准备数据集，我们需要另一个核心函数来构建模型：

def create_model(sequence_length, units=256, cell=LSTM, n_layers=2, dropout=0.3, loss='mean_absolute_error', optimizer='rmsprop', bidirectional=False): model = Sequential() for i in range(n_layers): if i == 0: # first layer if bidirectional: model.add(Bidirectional(cell(units, return_sequences=True), input_shape=(None, sequence_length))) else: model.add(cell(units, return_sequences=True, input_shape=(None, sequence_length))) elif i == n_layers - 1: # last layer if bidirectional: model.add(Bidirectional(cell(units, return_sequences=False))) else: model.add(cell(units, return_sequences=False)) else: # hidden layers if bidirectional: model.add(Bidirectional(cell(units, return_sequences=True))) else: model.add(cell(units, return_sequences=True)) # add dropout after each layer model.add(Dropout(dropout)) model.add(Dense(1, activation='linear')) model.compile(loss=loss, metrics=['mean_absolute_error'], optimizer=optimizer) return model

同样，此功能也很灵活，您可以更改层数，丢失率，RNN单元，损耗和用于编译模型的优化器。

上面的函数构造了一个具有密集层的RNN作为带有1个神经元的输出层，该模型需要一系列具有sequence_length特征的序列（在这种情况下，我们将传递50 或 100个）连续的时间步长（在此数据集中为天）并输出一个指示下一个时间价格的单一值。

您可以根据需要调整默认参数，n_layers是要堆叠的RNN层数，dropout是每个RNN层后的丢包率，units是RNN cell单位数（无论是LSTM，SimpleRNN还是GRU），bidirectional是一个布尔值指示是否使用双向RNN，可以尝试一下！

▊ 训练模型

现在我们已经准备好所有核心功能，让我们训练模型，但是在执行此操作之前，让我们初始化所有参数（以便以后可以根据需要编辑它们）：

# Window size or the sequence lengthN_STEPS = 70# Lookup step, 1 is the next dayLOOKUP_STEP = 1# test ratio size, 0.2 is 20%TEST_SIZE = 0.2# features to useFEATURE_COLUMNS = ['adjclose', 'volume', 'open', 'high', 'low']# date nowdate_now = time.strftime('%Y-%m-%d')### model parametersN_LAYERS = 3# LSTM cellCELL = LSTM# 256 LSTM neuronsUNITS = 256# 40% dropoutDROPOUT = 0.4# whether to use bidirectional RNNsBIDIRECTIONAL = False### training parameters# mean absolute error loss# LOSS = 'mae'# huber lossLOSS = 'huber_loss'OPTIMIZER = 'adam'BATCH_SIZE = 64EPOCHS = 400# Tesla stock marketticker = 'TSLA'ticker_data_filename = os.path.join('data', f'{ticker}_{date_now}.csv')# model name to save, making it as unique as possible based on parametersmodel_name = f'{date_now}_{ticker}-{LOSS}-{OPTIMIZER}-{CELL.__name__}-seq-{N_STEPS}-step-{LOOKUP_STEP}-layers-{N_LAYERS}-units-{UNITS}'if BIDIRECTIONAL:    model_name += '-b'

因此，以上代码都是关于定义我们将要使用的所有超级参数的，我们解释了其中一些参数：

TEST_SIZE：测试采样率。例如0.2表示总数据集的20％。

FEATURE_COLUMNS：我们将用来预测下一个价格值的功能。

N_LAYERS：要使用的RNN层数。

CELL：要使用的RNN单元，默认值为LSTM。

UNITS：cell单位数量。

BIDIRECTIONAL：是否使用双向递归神经网络。

LOSS：用于此回归问题的损失函数，我们使用的是Huber损失，也可以使用平均绝对误差（mae）或均方误差（mse）。

OPTIMIZER：要使用的优化算法，默认为Adam。

BATCH_SIZE：每次训练迭代中使用的数据样本数。

EPOCHS：学习算法将遍历整个训练数据集的次数，我们在这里使用了400次，但尝试进一步增加它。

随意尝试这些值以获得比我的更好的结果。

好吧，让我们确保在训练之前，结果，日志和数据文件夹存在：

if not os.path.isdir('results'): os.mkdir('results')if not os.path.isdir('logs'): os.mkdir('logs')if not os.path.isdir('data'): os.mkdir('data')

最后，让我们训练模型：

# load the datadata = load_data(ticker, N_STEPS, lookup_step=LOOKUP_STEP, test_size=TEST_SIZE, feature_columns=FEATURE_COLUMNS)# save the dataframedata['df'].to_csv(ticker_data_filename)# construct the modelmodel = create_model(N_STEPS, loss=LOSS, units=UNITS, cell=CELL, n_layers=N_LAYERS,                    dropout=DROPOUT, optimizer=OPTIMIZER, bidirectional=BIDIRECTIONAL)# some tensorflow callbackscheckpointer = ModelCheckpoint(os.path.join('results', model_name + '.h5'), save_weights_only=True, save_best_only=True, verbose=1)tensorboard = TensorBoard(log_dir=os.path.join('logs', model_name))history = model.fit(data['X_train'], data['y_train'],                    batch_size=BATCH_SIZE,                    epochs=EPOCHS,                    validation_data=(data['X_test'], data['y_test']),                    callbacks=[checkpointer, tensorboard],                    verbose=1)model.save(os.path.join('results', model_name) + '.h5')

我们使用ModelCheckpoint在训练期间将模型保存在每个时期。我们还使用TensorBoard在训练过程中可视化模型的性能。

运行上面的代码块之后，它将训练模型300个神经元，因此将需要一些时间，这是第一行输出：

Epoch 1/3003510/3510 [==============================] - 21s 6ms/sample - loss: 0.0117 - mean_absolute_error: 0.0515 - val_loss: 0.0065 - val_mean_absolute_error: 0.0487Epoch 2/3003264/3510 [==========================>...] - ETA: 0s - loss: 0.0049 - mean_absolute_error: 0.0352Epoch 00002: val_loss did not improve from 0.006503510/3510 [==============================] - 1s 309us/sample - loss: 0.0051 - mean_absolute_error: 0.0357 - val_loss: 0.0082 - val_mean_absolute_error: 0.0494Epoch 3/3003456/3510 [============================>.] - ETA: 0s - loss: 0.0039 - mean_absolute_error: 0.0329Epoch 00003: val_loss improved from 0.00650 to 0.00095, saving model to results\2020-01-08_NFLX-mse-LSTM-seq-50-step-1-layers-3-units-2563510/3510 [==============================] - 14s 4ms/sample - loss: 0.0039 - mean_absolute_error: 0.0328 - val_loss: 9.5337e-04 - val_mean_absolute_error: 0.0150Epoch 4/3003264/3510 [==========================>...] - ETA: 0s - loss: 0.0034 - mean_absolute_error: 0.0304Epoch 00004: val_loss did not improve from 0.000953510/3510 [==============================] - 1s 222us/sample - loss: 0.0037 - mean_absolute_error: 0.0316 - val_loss: 0.0034 - val_mean_absolute_error: 0.0300

训练结束后（或在训练期间），请尝试使用以下命令运行tensorboard：

tensorboard --logdir='logs'

现在，这将启动本地HTTP服务器“ localhost：6006”，进入浏览器后，您将看到类似以下内容：

损失是在LOSS参数中指定的Huber损失（您可以随时将其更改为平均误差或均方误差），橙色曲线是训练损失，而蓝色曲线是我们最关心的是验证损失。如您所见，它随着时间的推移而显着减少，因此这是可行的！

▊ 测试模型

在测试模型之前，我们将需要重新加载数据而无须改组，因为我们将以正确的顺序绘制股价曲线：

data = load_data(ticker, N_STEPS, lookup_step=LOOKUP_STEP, test_size=TEST_SIZE, feature_columns=FEATURE_COLUMNS, shuffle=False)# construct the modelmodel = create_model(N_STEPS, loss=LOSS, units=UNITS, cell=CELL, n_layers=N_LAYERS, dropout=DROPOUT, optimizer=OPTIMIZER, bidirectional=BIDIRECTIONAL)model_path = os.path.join('results', model_name) + '.h5'model.load_weights(model_path)

如果要随身携带笔记本，则不应重构模型并加载权重，因此需要将其注释掉。但是，如果您使用另一个Python文件进行测试，则应该这样做。

现在让我们测试模型：

# evaluate the modelmse, mae = model.evaluate(data['X_test'], data['y_test'], verbose=0)# calculate the mean absolute error (inverse scaling)mean_absolute_error = data['column_scaler']['adjclose'].inverse_transform([[mae]])[0][0]print('Mean Absolute Error:', mean_absolute_error)

请记住，输出将是介于0到1之间的值，因此我们需要将其恢复为实际价格值，这是输出：

Mean Absolute Error: 6.516846878481972

不错，平均而言，预测价格仅比实际价格高出6.52 $。

好吧，让我们尝试预测苹果股票市场的未来价格：

def predict(model, data):    # retrieve the last sequence from data    last_sequence = data['last_sequence'][-N_STEPS:]    # retrieve the column scalers    column_scaler = data['column_scaler']    # reshape the last sequence    last_sequence = last_sequence.reshape((last_sequence.shape[1], last_sequence.shape[0]))    # expand dimension    last_sequence = np.expand_dims(last_sequence, axis=0)    # get the prediction (scaled from 0 to 1)    prediction = model.predict(last_sequence)    # get the price (by inverting the scaling)    predicted_price = column_scaler['adjclose'].inverse_transform(prediction)[0][0]    return predicted_price

此函数使用我们保存在load_data（）函数中的last_sequence变量，该变量基本上是价格的最后一个序列，我们用它来预测下一个价格，我们称之为：

future_price = predict(model, data)print(f'Future price after {LOOKUP_STEP} days is {future_price:.2f}$')

输出：

Future price after 1 days is 404.78$

听起来不错 ！前两天，价格为447.37 $，昨天为416.43 $ ，该模型表示第二天价格为404.78 $ 。趋势下降。该模型仅使用了70 天的功能就能够获得该价值，让我们绘制价格并查看：

def plot_graph(model, data): y_test = data['y_test'] X_test = data['X_test'] y_pred = model.predict(X_test) y_test = np.squeeze(data['column_scaler']['adjclose'].inverse_transform(np.expand_dims(y_test, axis=0))) y_pred = np.squeeze(data['column_scaler']['adjclose'].inverse_transform(y_pred)) # last 200 days, feel free to edit that plt.plot(y_test[-200:], c='b') plt.plot(y_pred[-200:], c='r') plt.xlabel('Days') plt.ylabel('Price') plt.legend(['Actual Price', 'Predicted Price']) plt.show()

此函数绘制测试集的最后200天（您可以根据需要对其进行编辑）以及预测价格的图表，我们对其进行调用并查看其外观：

plot_graph(model, data)

结果：

如您所见，蓝色曲线是实际测试集，红色曲线是预测价格，太好了！请注意，正如我们预测的那样，最近股价正在下跌。

如果您的LOOKUP_STEP数据更高，这仍然会起作用，但是它将使用较旧的数据（按LOOKUP_STEP天数表示）以绘制红线。

到目前为止，我们仅习惯于预测第二天，我尝试构建使用不同lookup_steps的其他模型，这是张量板中的一个有趣结果：

有趣的是，蓝色曲线是我们在本教程中使用的模型，该模型使用下一时间步股票价格作为标签，而绿色和橙色曲线分别使用了10和30个查找步骤，例如，在本示例中，橙色模型可以预测30天后的股价，这是进行长期投资的理想模型（通常是这种情况）。

现在您可能会认为，但是，如果我们只是想预测价格是上涨还是下跌，而不是像我们在这里所做的那样，那么您可以使用以下两种方法之一来进行预测吗？将预测价格与当前价格一起做出决定，或者构建一个完整的模型并将最后输出的激活函数更改为sigmoid，同时将损失和度量标准更改为Sigmoid。

以下函数通过将预测价格转换为0或1（0表示价格下跌，而1表示价格上涨）来计算准确性得分：

def get_accuracy(model, data): y_test = data['y_test'] X_test = data['X_test'] y_pred = model.predict(X_test) y_test = np.squeeze(data['column_scaler']['adjclose'].inverse_transform(np.expand_dims(y_test, axis=0))) y_pred = np.squeeze(data['column_scaler']['adjclose'].inverse_transform(y_pred)) y_pred = list(map(lambda current, future: int(float(future) > float(current)), y_test[:-LOOKUP_STEP], y_pred[LOOKUP_STEP:])) y_test = list(map(lambda current, future: int(float(future) > float(current)), y_test[:-LOOKUP_STEP], y_test[LOOKUP_STEP:])) return accuracy_score(y_test, y_pred)

现在让我们调用该函数：

print(str(LOOKUP_STEP) + ':', 'Accuracy Score:', get_accuracy(model, data))

这是我为不同的人训练3个模型时的结果LOOKUP_STEPS：

1: Accuracy Score: 0.564257028112449810: Accuracy Score: 0.719262295081967330: Accuracy Score: 0.8318965517241379

您可能会注意到，该模型对长期价格的预测更为准确，当我们训练模型以预测未来10天的价格时，该模型的准确性约为71.9％，而使用30个LOOKUP_STEP，该模型的准确性约为83.2％。

▊ 总结

好了，就是本教程的内容了，您可以调整参数并查看如何提高模型性能，尝试训练更多的时期（例如500甚至更多），增加或减少BATCH_SIZE，看看是否确实变好了，或与N_STEPS和LOOKUP_STEPS一起玩，看看哪种组合效果最好。

您还可以更改模型参数，例如增加层数或LSTM单位数，甚至尝试使用GRU单元代替LSTM。

请注意，还有其他功能和指标要使用，为了改进预测，通常会使用一些其他信息作为功能，例如技术指标，公司产品创新，利率，汇率，公共政策，网络和财经新闻，甚至还有雇员人数！

我鼓励您更改模型架构，尝试使用CNN或Seq2Seq模型，甚至向该现有模型添加双向LSTM，以查看是否可以改进它！

另外，使用不同的股票市场，查看Yahoo Finance页面，然后查看您真正想要的页面！

好了，关注我，每天更新一篇技术好文。