transformer做时间序列预测python

Transformer模型可以用于时间序列预测，以下是一个使用Python和TensorFlow对时间序列进行预测的示例：

首先，导入必要的库和数据集：

import tensorflow as tf
import numpy as np
import matplotlib.pyplot as plt

# 生成数据集
def generate_time_series():
    frequency1, frequency2, offset1, offset2 = np.random.rand(4) * 0.5
    time = np.linspace(0, 1, 200)
    series = offset1 + np.sin((frequency1 * 10 + time) * 2 * np.pi) * 0.2 + \
             offset2 + np.sin((frequency2 * 20 + time) * 2 * np.pi) * 0.1
    return series[..., np.newaxis].astype(np.float32)

# 划分数据集
data = generate_time_series()
train_data = data[:150]
val_data = data[150:175]
test_data = data[175:]

# 可视化数据集
plt.plot(np.arange(200), data)
plt.show()

接下来，构建Transformer模型：

class Transformer(tf.keras.Model):
    def __init__(self, num_layers, units, d_model, num_heads, dropout, name='transformer'):
        super(Transformer, self).__init__(name=name)

        self.encoder = tf.keras.layers.Dense(units, activation='relu')
        self.decoder = tf.keras.layers.Dense(1)

        self.pos_encoding = positional_encoding(200, d_model)

        self.num_layers = num_layers
        self.d_model = d_model

        self.dec_layers = [DecoderLayer(d_model, num_heads, units, dropout, name='dec_layer_{}'.format(i)) for i in range(num_layers)]

        self.dropout = tf.keras.layers.Dropout(dropout)

    def call(self, inputs, training=False):
        # 编码器
        x = self.encoder(inputs)  # (batch_size, input_seq_len, d_model)
        x *= tf.math.sqrt(tf.cast(self.d_model, tf.float32))
        x += self.pos_encoding[:, :tf.shape(x)[1], :]

        # 解码器
        for i in range(self.num_layers):
            x = self.dec_layers[i](x, training=training)

        x = self.decoder(x)  # (batch_size, input_seq_len, 1)

        return x

# 解码器层
class DecoderLayer(tf.keras.layers.Layer):
    def __init__(self, d_model, num_heads, units, dropout, name='dec_layer'):
        super(DecoderLayer, self).__init__(name=name)

        self.mha1 = MultiHeadAttention(d_model, num_heads)
        self.mha2 = MultiHeadAttention(d_model, num_heads)

        self.ffn = point_wise_feed_forward_network(units, d_model)

        self.layernorm1 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
        self.layernorm2 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
        self.layernorm3 = tf.keras.layers.LayerNormalization(epsilon=1e-6)

        self.dropout1 = tf.keras.layers.Dropout(dropout)
        self.dropout2 = tf.keras.layers.Dropout(dropout)
        self.dropout3 = tf.keras.layers.Dropout(dropout)

    def call(self, inputs, training=False):
        # 第一层多头注意力
        attn1 = self.mha1(inputs, inputs, inputs)
        attn1 = self.dropout1(attn1, training=training)
        out1 = self.layernorm1(inputs + attn1)

        # 第二层多头注意力
        attn2 = self.mha2(out1, out1, out1)
        attn2 = self.dropout2(attn2, training=training)
        out2 = self.layernorm2(out1 + attn2)

        # 前馈神经网络
        ffn_output = self.ffn(out2)
        ffn_output = self.dropout3(ffn_output, training=training)
        out3 = self.layernorm3(out2 + ffn_output)

        return out3

# 多头注意力机制
class MultiHeadAttention(tf.keras.layers.Layer):
    def __init__(self, d_model, num_heads, name='multi_head_attention'):
        super(MultiHeadAttention, self).__init__(name=name)
        self.num_heads = num_heads
        self.d_model = d_model

        assert d_model % self.num_heads == 0

        self.depth = d_model // self.num_heads

        self.wq = tf.keras.layers.Dense(d_model)
        self.wk = tf.keras.layers.Dense(d_model)
        self.wv = tf.keras.layers.Dense(d_model)

        self.dense = tf.keras.layers.Dense(d_model)

    def split_heads(self, x, batch_size):
        x = tf.reshape(x, (batch_size, -1, self.num_heads, self.depth))
        return tf.transpose(x, perm=[0, 2, 1, 3])

    def call(self, v, k, q):
        batch_size = tf.shape(q)[0]

        q = self.wq(q)  # (batch_size, seq_len, d_model)
        k = self.wk(k)  # (batch_size, seq_len, d_model)
        v = self.wv(v)  # (batch_size, seq_len, d_model)

        q = self.split_heads(q, batch_size)  # (batch_size, num_heads, seq_len_q, depth)
        k = self.split_heads(k, batch_size)  # (batch_size, num_heads, seq_len_k, depth)
        v = self.split_heads(v, batch_size)  # (batch_size, num_heads, seq_len_v, depth)

        scaled_attention, attention_weights = scaled_dot_product_attention(q, k, v)

        scaled_attention = tf.transpose(scaled_attention, perm=[0, 2, 1, 3])
        concat_attention = tf.reshape(scaled_attention, (batch_size, -1, self.d_model))

        output = self.dense(concat_attention)

        return output

# 缩放点积注意力机制
def scaled_dot_product_attention(q, k, v):
    matmul_qk = tf.matmul(q, k, transpose_b=True)

    dk = tf.cast(tf.shape(k)[-1], tf.float32)
    scaled_attention_logits = matmul_qk / tf.math.sqrt(dk)

    # 掩码
    mask = tf.linalg.band_part(tf.ones_like(scaled_attention_logits), -1, 0)
    scaled_attention_logits += (mask * -1e9)

    attention_weights = tf.nn.softmax(scaled_attention_logits, axis=-1)

    output = tf.matmul(attention_weights, v)

    return output, attention_weights

# 位置编码
def get_angles(pos, i, d_model):
    angle_rates = 1 / np.power(10000, (2 * (i // 2)) / np.float32(d_model))
    return pos * angle_rates

def positional_encoding(position, d_model):
    angle_rads = get_angles(np.arange(position)[:, np.newaxis],
                            np.arange(d_model)[np.newaxis, :],
                            d_model)

    # 将 sin 应用于数组中的偶数索引（indices）；2i
    angle_rads[:, 0::2] = np.sin(angle_rads[:, 0::2])

    # 将 cos 应用于数组中的奇数索引；2i+1
    angle_rads[:, 1::2] = np.cos(angle_rads[:, 1::2])

    pos_encoding = angle_rads[np.newaxis, ...]

    return tf.cast(pos_encoding, dtype=tf.float32)

# 前馈神经网络
def point_wise_feed_forward_network(d_model, diff):
    return tf.keras.Sequential([
        tf.keras.layers.Dense(diff, activation='relu'),
        tf.keras.layers.Dense(d_model)
    ])

接下来，定义损失函数和优化器，并进行模型训练：

# 损失函数
loss_object = tf.keras.losses.MeanSquaredError()

# 优化器
optimizer = tf.keras.optimizers.Adam(learning_rate=1e-4)

# 指标
train_loss = tf.keras.metrics.Mean(name='train_loss')
val_loss = tf.keras.metrics.Mean(name='val_loss')

# 训练步骤
@tf.function
def train_step(inputs, targets):
    with tf.GradientTape() as tape:
        predictions = transformer(inputs, training=True)
        loss = loss_object(targets, predictions)
    gradients = tape.gradient(loss, transformer.trainable_variables)
    optimizer.apply_gradients(zip(gradients, transformer.trainable_variables))
    train_loss(loss)

# 验证步骤
@tf.function
def val_step(inputs, targets):
    predictions = transformer(inputs, training=False)
    loss = loss_object(targets, predictions)
    val_loss(loss)

# 训练模型
num_epochs = 2000

transformer = Transformer(num_layers=4, units=64, d_model=128, num_heads=8, dropout=0.1)

for epoch in range(num_epochs):
    train_loss.reset_states()
    val_loss.reset_states()

    for input_series in range(len(train_data) - 10):
        inputs = train_data[input_series:input_series + 10]
        targets = train_data[input_series + 10]

        train_step(inputs, targets)

    for input_series in range(len(val_data) - 10):
        inputs = val_data[input_series:input_series + 10]
        targets = val_data[input_series + 10]

        val_step(inputs, targets)

    if epoch % 100 == 0:
        print('Epoch {}, Train Loss: {:.4f}, Val Loss: {:.4f}'.format(epoch, train_loss.result(), val_loss.result()))

最后，使用训练好的模型进行预测并可视化：

# 预测
predictions = []
for input_series in range(len(test_data) - 10):
    inputs = test_data[input_series:input_series + 10]
    prediction = transformer(inputs[np.newaxis, ...], training=False)[0, -1].numpy()
    predictions.append(prediction)

# 可视化预测结果
plt.plot(np.arange(175, 200), np.array(predictions), label='Predictions')
plt.plot(np.arange(175, 200), test_data[10:], label='Real Data')
plt.legend()
plt.show()

这里的示例是基于一个简单的时间序列预测问题，你可以根据自己的数据集和需求进行调整。