本文主要介绍循环神经网络(recurrent neural network，RNN)的应用，这种类型的网络专门用于学习序列数据，比如文本或时间序列数据。

首先我们来看两个示例。

1.第一个示例

为了演示RNN的训练和使用，考虑一个基于整数序列的简单示例。首先导入必要的包，并生成测试数据

import osimport randomimport numpy as npimport pandas as pdimport tensorflow as tffrom pprint import pprintimport matplotlib.pyplot as pltplt.rcParams['savefig.dpi'] = 300plt.rcParams['font.family'] = 'serif'pd.set_option('precision', 4)np.set_printoptions(suppress=True, precision=4)os.environ['PYTHONHASHSEED'] = '0'def set_seeds(seed=100):    random.seed(seed)    np.random.seed(seed)    tf.random.set_seed(seed)set_seeds()a = np.arange(100)# 转换成二维的简单数据集a = a.reshape((len(a), -1))a[:5]# Out:# array([[0],#        [1],#        [2],#        [3],#        [4]])

使用TimeseriesGenerator将原始数据转换为适合RNN训练的对象。这样做的目的是利用原始数据中一些滞后数据来训练模型以预测序列中的下一个值，例如，0、1和2是3个滞后值(特征)，用来预测值3(标签)。同理，用 1、2 和 3来预测4。

from keras.preprocessing.sequence import TimeseriesGeneratorlags = 3# 通过 TimeseriesGenerator 创建一批滞后数据g = TimeseriesGenerator(a, a, length=lags, batch_size=5)pprint(list(g)[0])# Out:# (array([[[0],#         [1],#         [2]],# #        [[1],#         [2],#         [3]],# #        [[2],#         [3],#         [4]],# #        [[3],#         [4],#         [5]],# #        [[4],#         [5],#         [6]]]),#  array([[3],#        [4],#        [5],#        [6],#        [7]]))

以下代码使用SimpleRNN类的单个隐藏层。

from keras.models import Sequentialfrom keras.layers import SimpleRNN, LSTM, Densemodel = Sequential()# 用SimpleRNN作为单一隐藏层model.add(SimpleRNN(100, activation='relu', input_shape=(lags, 1)))model.add(Dense(1, activation='linear'))model.compile(optimizer='adagrad', loss='mse', metrics=['mae'])h = model.fit(g, epochs=1000, steps_per_epoch=5, verbose=False)res = pd.DataFrame(h.history)res.plot(figsize=(10, 6), style=['--', '--']);

图1 RNN训练期间的性能指标

基于训练好的RNN，以下代码生成样本内预测和样本外预测。

x = np.array([21, 22, 23]).reshape((1, lags, 1))# 样本内预测y = model.predict(x, verbose=False)result = int(round(y[0, 0]))print(f'IN-SAMPLE: {result}')x = np.array([87, 88, 89]).reshape((1, lags, 1))y = model.predict(x, verbose=False)result = int(round(y[0, 0]))print(f'IN-SAMPLE: {result}')# 样本外预测x = np.array([187, 188, 189]).reshape((1, lags, 1))y = model.predict(x, verbose=False)result = int(round(y[0, 0]))print(f'OUT-OF-SAMPLE: {result}')# 远离样本的预测x = np.array([1187, 1188, 1189]).reshape((1, lags, 1))y = model.predict(x, verbose=False)result = int(round(y[0, 0]))print(f'FAR-OUT-OF-SAMPLE: {result}')# Out:# IN-SAMPLE: 23# IN-SAMPLE: 91# OUT-OF-SAMPLE: 193# FAR-OUT-OF-SAMPLE: 1220

即使对于远离样本的预测，在这种简单的情形下，结果通常也很好。

2.第二个示例

第一个示例说明了针对简单问题的RNN训练，该问题不仅可以通过OLS回归轻松解决， 也可以通过人工检查数据轻松解决。第二个示例更具挑战性，输入数据通过二次项和三角项进行转换，并添加了白噪声。图2显示了区间的序列值。

def transform(x):    # 确定性变换    y = 0.05 * x ** 2 + 0.2 * x + np.sin(x) + 5    # 随机变换    y += np.random.standard_normal(len(x)) * 0.2    return yx = np.linspace(-2 * np.pi, 2 * np.pi, 500)a = transform(x)plt.figure(figsize=(10, 6))plt.plot(x, a);

图2 样本序列数据

和之前一样，应用 TimeseriesGenerator 对原始数据进行变换，并训练具有单个隐藏层的 RNN

a = a.reshape((len(a), -1))lags = 5g = TimeseriesGenerator(a, a, length=lags, batch_size=5)model = Sequential()model.add(SimpleRNN(500, activation='relu', input_shape=(lags, 1)))model.add(Dense(1, activation='linear'))model.compile(optimizer='rmsprop', loss='mse', metrics=['mae'])h = model.fit(g, epochs=500, steps_per_epoch=10, verbose=False)

以下代码会预测区间的序列值，这个区间是训练区间的3倍，并且在训练区间的左侧和右侧都包含样本外预测。图3显示了即使在样本外该模型表现也非常好。

a = a.reshape((len(a), -1))lags = 5g = TimeseriesGenerator(a, a, length=lags, batch_size=5)model = Sequential()model.add(SimpleRNN(500, activation='relu', input_shape=(lags, 1)))model.add(Dense(1, activation='linear'))model.compile(optimizer='rmsprop', loss='mse', metrics=['mae'])h = model.fit(g, epochs=500, steps_per_epoch=10, verbose=False)# 扩大样本数据集。x = np.linspace(-6 * np.pi, 6 * np.pi, 1000)d = transform(x)g_ = TimeseriesGenerator(d, d, length=lags, batch_size=len(d))f = list(g_)[0][0].reshape((len(d) - lags, lags, 1))# 样本内预测和样本外预测y = model.predict(f, verbose=False)plt.figure(figsize=(10, 6))plt.plot(x[lags:], d[lags:], label='data', alpha=0.75)plt.plot(x[lags:], y, 'r.', label='pred', ms=3)plt.axvline(-2 * np.pi, c='g', ls='--')plt.axvline(2 * np.pi, c='g', ls='--')plt.text(-15, 22, 'out-of-sample')plt.text(-2, 22, 'in-sample')plt.text(10, 22, 'out-of-sample')plt.legend();

图3 RNN的样本内预测和样本外预测

上述两个示例是有意简化过的，示例中提出的两个问题都可以使用OLS回归更有效地解决，例如，通过允许在第二个示例中使用三角函数。然而，对不规则的序列数据(如金融时间序列数据)的处理，OLS回归的性能通常无法与RNN相比。

3.金融价格序列

这里考虑日内欧元/美元报价。使用前介绍的方法，RNN在金融时间序列上做简单的训练。首先，导入数据并重新采样。

url = ';symbol = 'EUR_USD'raw = pd.read_csv(url, index_col=0, parse_dates=True)def generate_data():    # 选择收盘价    data = pd.DataFrame(raw['CLOSE'])    # 将该列重命名    data.columns = [symbol]    # 按30分钟重新采样数据    data = data.resample('30min', label='right').last().ffill()    return datadata = generate_data()# 应用高斯归一化data = (data - data.mean()) / data.std()# 将数据集重塑为二维p = data[symbol].valuesp = p.reshape((len(p), -1))

其次，基于生成器对象对 RNN 模型进行训练。函数 create_rnn_model() 允许使用SimpleRNN层或 LSTM(长短期记忆)层创建 RNN。

lags = 5g = TimeseriesGenerator(p, p, length=lags, batch_size=5)def create_rnn_model(hu=100, lags=lags, layer='SimpleRNN', features=1, algorithm='estimation'):    model = Sequential()    if layer == 'SimpleRNN':        # 增加一个SimpleRNN层或LSTM层        model.add(SimpleRNN(hu, activation='relu', input_shape=(lags, features)))    else:        model.add(LSTM(hu, activation='relu',    input_shape=(lags, features)))    if algorithm == 'estimation':        # 输出层        model.add(Dense(1, activation='linear'))        model.compile(optimizer='adam', loss='mse', metrics=['mae'])    else:        # 输出层        model.add(Dense(1, activation='sigmoid'))        model.compile(optimizer='adam', loss='mse', metrics=['accuracy'])    return modelmodel = create_rnn_model()model.fit(g, epochs=500, steps_per_epoch=10, verbose=False)              

最后，生成样本内预测。如图4所示，RNN能够捕获归一化金融时间序列数据的结构。 从可视化结果可以看出，预测准确率似乎相当不错。

y = model.predict(g, verbose=False)data['pred'] = np.nandata['pred'].iloc[lags:] = y.flatten()data[[symbol, 'pred']].plot(figsize=(10, 6), style=['b', 'r-.'], alpha=0.75);

图4 RNN对金融价格序列的样本内预测(数据全集)

然而，可视化显示的结果经不起仔细检查，将图5放大，仅显示来自原始数据集和预测的50个数据点。很明显，来自RNN的预测值基本上只是最近的滞后值，并移动了一个时间间隔。直观上看，预测的价格序列是金融时间序列本身，向右移动了一个时间间隔。

data[[symbol, 'pred']].iloc[50:100].plot(figsize=(10, 6), style=['b', 'r-.'], alpha=0.75);

图5 RNN 对金融价格序列的样本内预测(数据子集)

4.金融收益率序列

正如之前的分析所表明的那样，预测收益率可能比预测价格更容易。因此，下面的代码根据对数收益率重复前面的分析。

data = generate_data()data['r'] = np.log(data / data.shift(1))data.dropna(inplace=True)data = (data - data.mean()) / data.std()r = data['r'].valuesr = r.reshape((len(r), -1))g = TimeseriesGenerator(r, r, length=lags, batch_size=5)model = create_rnn_model()model.fit(g, epochs=500, steps_per_epoch=10, verbose=False) 

如图6所示，RNN的预测绝对值不太好。然而，它们似乎以某种方式正确地了解了市场方向(收益率的迹象)。

y = model.predict(g, verbose=False)data['pred'] = np.nandata['pred'].iloc[lags:] = y.flatten()data.dropna(inplace=True)data[['r', 'pred']].iloc[50:100].plot(figsize=(10, 6), style=['b', 'r-.'], alpha=0.75);plt.axhline(0, c='grey', ls='--');

图 8-6:RNN 对金融收益率序列的样本内预测(数据子集)

虽然图6只提供了一个指示信息，但以相对较高的准确率验证了RNN在收益率序列上的预测表现可能比在价格序列上的预测表现效果更好的假设。

from sklearn.metrics import accuracy_scoreaccuracy_score(np.sign(data['r']), np.sign(data['pred']))# Out:# 0.6686323429349059

然而，为了获得真实的效果，需要对数据进行训练−测试拆分。样本外的准确率得分没有样本内的准确率高，但对我们关心的问题来说这个准确率仍然很高。

# 将数据拆分为训练数据子集和测试数据子集split = int(len(r) * 0.8)train = r[:split]test = r[split:]# 在训练数据上拟合模型g = TimeseriesGenerator(train, train, length=lags, batch_size=5)set_seeds()model = create_rnn_model(hu=100)h = model.fit(g, epochs=100, steps_per_epoch=10, verbose=False)# 在测试数据上测试模型g_ = TimeseriesGenerator(test, test, length=lags, batch_size=5)y = model.predict(g_)accuracy_score(np.sign(test[lags:]), np.sign(y))# Out:# 0.6662870159453302

RNN的应用不仅限于原始价格或收益率数据，还可以包括附加特征以改进它的预测性能。以下代码向数据集中添加了典型的金融特征。

data = generate_data()data['r'] = np.log(data / data.shift(1))window = 20# 增加时间序列动量特征data['mom'] = data['r'].rolling(window).mean()# 增加滚动波动率特征data['vol'] = data['r'].rolling(window).std()data.dropna(inplace=True)

在预测任务中，样本外准确率可能会显著下降，这有些出人意料。换句话说，在这种特殊情况下添加金融特征并没有观察到任何改进。

split = int(len(data) * 0.8)train = data.iloc[:split].copy()# 计算训练数据的一阶矩和二阶矩mu, std = train.mean(), train.std()# 对训练数据应用高斯归一化train = (train - mu) / stdtest = data.iloc[split:].copy()# 对测试数据应用高斯归一化(基于来自训练数据的统计数据)test = (test - mu) / std# 在训练数据上拟合模型g = TimeseriesGenerator(train.values, train['r'].values, length=lags, batch_size=5)set_seeds()model = create_rnn_model(hu=100, features=len(data.columns), layer='SimpleRNN')h = model.fit(g, epochs=100, steps_per_epoch=10, verbose=False)# 在测试数据上测试模型g_ = TimeseriesGenerator(test.values, test['r'].values, length=lags, batch_size=5)y = model.predict(g_).flatten()accuracy_score(np.sign(test['r'].iloc[lags:]), np.sign(y))# Out:# 0.6784897025171625

迄今为止的分析都在使用 Keras 中的RNN模型进行估计，以预测金融工具价格的未来方向。我们所关心的问题可能会更好地直接转换为分类问题。以下代码会处理二进制标签数据并直接预测价格变动的方向。这次我们使用LSTM层，即使对于相对少量的隐藏单元和有限的几个训练轮数，样本外的准确率也相当高。该方法通过适当调整类权重来解决 类别不平衡的问题。在这种情况下，预测准确率非常高，约为 68.8%。

set_seeds()model = create_rnn_model(hu=50,            features=len(data.columns),            layer='LSTM',            algorithm='classification')  # 分类的RNN模型# 二元训练标签train_y = np.where(train['r'] > 0, 1, 0)#  训练标签的组频率# np.bincount(train_y)def cw(a):    c0, c1 = np.bincount(a)    w0 = (1 / c0) * (len(a)) / 2    w1 = (1 / c1) * (len(a)) / 2    return {0: w0, 1: w1}g = TimeseriesGenerator(train.values, train_y,                        length=lags, batch_size=5)model.fit(g, epochs=5, steps_per_epoch=10,          verbose=False, class_weight=cw(train_y))# 二进制测试标签test_y = np.where(test['r'] > 0, 1, 0)g_ = TimeseriesGenerator(test.values, test_y,                         length=lags, batch_size=5)y = np.where(model.predict(g_, batch_size=None) > 0.5,             1, 0).flatten()accuracy_score(test_y[lags:], y)# Out:# 0.6876430205949656

最后，我们尝试添加多个隐藏层的RNN。对于非最终隐藏层，参数return_sequences需要被设置为True。以下用于创建深度 RNN的函数还添加了 Dropout 层以避免潜在的过拟合，预测准确率与之前的预测准确率相当。

from keras.layers import Dropoutdef create_deep_rnn_model(hl=2, hu=100, layer='SimpleRNN',                          optimizer='rmsprop', features=1,                          dropout=False, rate=0.3, seed=100):    # 保证最少有两个隐藏层    if hl <= 2:        hl = 2    if layer == 'SimpleRNN':        layer = SimpleRNN    else:        layer = LSTM    model = Sequential()    # 第一个隐藏层    model.add(layer(hu, input_shape=(lags, features), return_sequences=True))    if dropout:        model.add(Dropout(rate, seed=seed))    for _ in range(2, hl):        model.add(layer(hu, return_sequences=True))        if dropout:            model.add(Dropout(rate, seed=seed))    # 最终隐藏层    model.add(layer(hu))    # 建立分类模型    model.add(Dense(1, activation='sigmoid'))    model.compile(optimizer=optimizer,                  loss='binary_crossentropy',                  metrics=['accuracy'])    return modelset_seeds()model = create_deep_rnn_model(hl=2, hu=50, layer='SimpleRNN',                              features=len(data.columns),                              dropout=True, rate=0.3)h = model.fit(g, epochs=50, steps_per_epoch=100, verbose=False, class_weight=cw(train_y))y = np.where(model.predict(g_, batch_size=None) > 0.5, 1, 0).flatten()accuracy_score(test_y[lags:], y)# Out:# 0.6521739130434783