Θα προχωρήσουμε σε πρόβλεψη των επόμενων 3 αριθμών με την τεχνική seq2seq.
import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
import numpy as np
import tensorflow as tf
import matplotlib.pyplot as plt
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error, mean_absolute_error, median_absolute_error
from sklearn.metrics import mean_absolute_percentage_error, r2_score
from tensorflow.keras.models import Model
from tensorflow.keras.layers import LSTM, Dense, Input, Bidirectional, BatchNormalization, Concatenate, Add, Activation
from tensorflow.keras import regularizers
from tensorflow.keras.constraints import UnitNorm
from tensorflow.keras.initializers import RandomUniform
from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau, ModelCheckpoint, LearningRateScheduler
from typing import Tuple, List
seed_value = 1
tf.random.set_seed(seed_value)
np.random.seed(seed_value)
def create_dataset() -> np.ndarray:
return np.array([
[0.1, 0.8, 0.9],
[0.4, 0.6, 0.7],
[0.0, 0.5, 0.6],
[0.4, 0.5, 0.9],
[0.1, 0.6, 0.8],
[0.0, 0.3, 0.4],
[0.3, 0.4, 0.6],
[0.1, 0.3, 0.7],
[0.0, 0.1, 0.6],
[0.1, 0.2, 0.4],
[0.7, 0.7, 0.8],
[0.1, 0.2, 0.8],
[0.3, 0.5, 0.8],
[0.1, 0.4, 0.9],
[0.4, 0.5, 0.8],
[0.0, 0.2, 0.9],
[0.0, 0.2, 0.5],
[0.1, 0.7, 0.9],
[0.6, 0.7, 0.9],
[0.0, 0.2, 0.8],
[0.5, 0.6, 0.7],
[0.0, 0.3, 0.9],
[0.1, 0.4, 0.6],
[0.3, 0.4, 0.4],
[0.0, 0.5, 0.9],
[0.0, 0.7, 0.7],
[0.2, 0.4, 0.6],
[0.1, 0.4, 0.5],
[0.2, 0.3, 0.5],
[0.4, 0.7, 0.9],
[0.3, 0.5, 0.5],
[0.1, 0.2, 0.7],
[0.2, 0.5, 0.6],
[0.7, 0.7, 0.7],
[0.2, 0.2, 0.3],
[0.0, 0.4, 0.6],
[0.3, 0.5, 0.9],
[0.7, 0.8, 0.8],
[0.2, 0.3, 0.8],
[0.4, 0.7, 0.7],
[0.0, 0.1, 0.6],
[0.1, 0.2, 0.4],
[0.6, 0.7, 0.7],
[0.3, 0.5, 0.7],
[0.4, 0.6, 0.9],
[0.5, 0.6, 0.9],
[0.0, 0.2, 0.7],
[0.2, 0.4, 0.4],
[0.2, 0.3, 0.6],
[0.0, 0.0, 0.4],
[0.0, 0.2, 0.7],
[0.3, 0.5, 0.8],
[0.0, 0.3, 0.7],
[0.5, 0.8, 0.9],
[0.6, 0.9, 0.9],
[0.7, 0.7, 0.9],
[0.3, 0.8, 0.8],
[0.2, 0.6, 0.8],
[0.4, 0.4, 0.5],
[0.4, 0.6, 0.9],
[0.1, 0.5, 0.6],
[0.1, 0.2, 0.5],
[0.3, 0.3, 0.5],])
def create_sequences(data: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
input_data, output_data = [], []
for i in range(len(data) - 1):
input_data.append(data[i])
output_data.append(data[i + 1])
return np.array(input_data), np.array(output_data)
def split_data(input_data: np.ndarray, output_data: np.ndarray, split_ratio: float = 0.75) -> Tuple[
np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
split_point = int(len(input_data) * split_ratio)
return input_data[:split_point], input_data[split_point:], output_data[:split_point], output_data[split_point:]
class CustomAttention(tf.keras.layers.Layer):
def __init__(self, num_heads, key_dim):
super(CustomAttention, self).__init__()
self.num_heads = num_heads
self.key_dim = key_dim
self.query_dense = Dense(key_dim)
self.key_dense = Dense(key_dim)
self.value_dense = Dense(key_dim)
def call(self, inputs):
query, key, value = inputs, inputs, inputs
query = self.query_dense(query)
key = self.key_dense(key)
value = self.value_dense(value)
scores = tf.matmul(query, key, transpose_b=True)
scores = Activation('tanh')(scores)
attention_weights = tf.nn.softmax(scores, axis=-1)
context = tf.matmul(attention_weights, value)
return context
def build_model(input_shape: Tuple[int, int], num_units: List[int], num_heads: int) -> Model:
encoder_inputs = Input(shape=input_shape)
encoder_lstm1 = Bidirectional(LSTM(num_units[0],
activation='tanh',
recurrent_activation='sigmoid',
return_sequences=True,
return_state=True,
unroll=True,
unit_forget_bias=True,
recurrent_dropout=0.1,
stateful=False,
go_backwards=False,
use_bias=True,
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
recurrent_constraint= UnitNorm(axis=0),
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
recurrent_initializer= RandomUniform(minval=-1, maxval=1),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
recurrent_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001)))
encoder_outputs1, forward_h1, forward_c1, backward_h1, backward_c1 = encoder_lstm1(encoder_inputs)
encoder_outputs1 = BatchNormalization()(encoder_outputs1)
state_h1 = Concatenate()([forward_h1, backward_h1])
state_c1 = Concatenate()([forward_c1, backward_c1])
attention1 = CustomAttention(num_heads=num_heads, key_dim=num_units[0])
attention_output1 = attention1(encoder_outputs1)
encoder_lstm2 = Bidirectional(LSTM(num_units[1],
activation='tanh',
recurrent_activation='sigmoid',
return_sequences=True,
return_state=True,
unroll=True,
unit_forget_bias=True,
recurrent_dropout=0.1,
stateful=False,
go_backwards=False,
use_bias=True,
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
recurrent_constraint= UnitNorm(axis=0),
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
recurrent_initializer= RandomUniform(minval=-1, maxval=1),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
recurrent_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001)))
encoder_outputs2, forward_h2, forward_c2, backward_h2, backward_c2 = encoder_lstm2(attention_output1)
encoder_outputs2 = BatchNormalization()(encoder_outputs2)
state_h2 = Concatenate()([forward_h2, backward_h2])
state_c2 = Concatenate()([forward_c2, backward_c2])
attention2 = CustomAttention(num_heads=num_heads, key_dim=num_units[1])
attention_output2 = attention2(encoder_outputs2)
encoder_lstm3 = Bidirectional(LSTM(num_units[2],
activation='tanh',
recurrent_activation='sigmoid',
return_sequences=True,
return_state=True,
unroll=True,
unit_forget_bias=True,
recurrent_dropout=0.1,
stateful=False,
go_backwards=False,
use_bias=True,
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
recurrent_constraint= UnitNorm(axis=0),
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
recurrent_initializer= RandomUniform(minval=-1, maxval=1),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
recurrent_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001)))
encoder_outputs3, forward_h3, forward_c3, backward_h3, backward_c3 = encoder_lstm3(attention_output2)
encoder_outputs3 = BatchNormalization()(encoder_outputs3)
state_h3 = Concatenate()([forward_h3, backward_h3])
state_c3 = Concatenate()([forward_c3, backward_c3])
attention3 = CustomAttention(num_heads=num_heads, key_dim=num_units[2])
attention_output3 = attention3(encoder_outputs3)
encoder_states = [state_h3, state_c3]
decoder_inputs = Input(shape=input_shape)
decoder_lstm1 = LSTM(num_units[0],
activation='tanh',
recurrent_activation='sigmoid',
return_sequences=True,
return_state=True,
unroll=True,
unit_forget_bias=True,
recurrent_dropout=0.1,
stateful=False,
go_backwards=False,
use_bias=True,
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
recurrent_constraint= UnitNorm(axis=0),
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
recurrent_initializer= RandomUniform(minval=-1, maxval=1),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
recurrent_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))
state_h1_proj = Dense(num_units[0],
activation='tanh',
use_bias=True,
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))(state_h1)
state_c1_proj = Dense(num_units[0],
activation='tanh',
use_bias=True,
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))(state_c1)
decoder_outputs1, state_h1_dec, state_c1_dec = decoder_lstm1(decoder_inputs, initial_state=[state_h1_proj, state_c1_proj])
decoder_outputs1 = BatchNormalization()(decoder_outputs1)
attention4 = CustomAttention(num_heads=num_heads, key_dim=num_units[0])
attention_output4 = attention4(decoder_outputs1)
decoder_lstm2 = LSTM(num_units[1],
activation='tanh',
recurrent_activation='sigmoid',
return_sequences=True,
return_state=True,
unroll=True,
unit_forget_bias=True,
recurrent_dropout=0.1,
stateful=False,
go_backwards=False,
use_bias=True,
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
recurrent_constraint= UnitNorm(axis=0),
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
recurrent_initializer= RandomUniform(minval=-1, maxval=1),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
recurrent_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))
state_h2_proj = Dense(num_units[1],
activation='tanh',
use_bias=True,
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))(state_h2)
state_c2_proj = Dense(num_units[1],
activation='tanh',
use_bias=True,
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))(state_c2)
decoder_outputs2, state_h2_dec, state_c2_dec = decoder_lstm2(attention_output4, initial_state=[state_h2_proj, state_c2_proj])
decoder_outputs2 = BatchNormalization()(decoder_outputs2)
attention5 = CustomAttention(num_heads=num_heads, key_dim=num_units[1])
attention_output5 = attention5(decoder_outputs2)
decoder_lstm3 = LSTM(num_units[2],
activation='tanh',
recurrent_activation='sigmoid',
return_sequences=True,
return_state=True,
unroll=True,
unit_forget_bias=True,
recurrent_dropout=0.1,
stateful=False,
go_backwards=False,
use_bias=True,
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
recurrent_constraint= UnitNorm(axis=0),
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
recurrent_initializer= RandomUniform(minval=-1, maxval=1),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
recurrent_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))
state_h3_proj = Dense(num_units[2],
activation='tanh',
use_bias=True,
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))(state_h3)
state_c3_proj = Dense(num_units[2],
activation='tanh',
use_bias=True,
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))(state_c3)
decoder_outputs3, state_h3_dec, state_c3_dec = decoder_lstm3(attention_output5, initial_state=[state_h3_proj, state_c3_proj])
decoder_outputs3 = BatchNormalization()(decoder_outputs3)
attention6 = CustomAttention(num_heads=num_heads, key_dim=num_units[2])
attention_output6 = attention6(decoder_outputs3)
decoder_outputs3 = Add()([decoder_outputs3, attention_output6])
combined_decoder_outputs = Concatenate()([decoder_outputs1, decoder_outputs2, decoder_outputs3])
combined_attention_outputs = Concatenate()([attention_output4, attention_output5, attention_output6])
decoder_outputs_combined = Add()([combined_decoder_outputs, combined_attention_outputs])
decoder_dense = Dense(input_shape[1],
activation='tanh',
use_bias=True,
bias_initializer= RandomUniform(minval=-1, maxval=1),
kernel_initializer= RandomUniform(minval=-1, maxval=1),
bias_constraint= UnitNorm(axis=0),
kernel_constraint= UnitNorm(axis=0),
bias_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
activity_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001),
kernel_regularizer= regularizers.l1_l2(l1=0.00000001, l2=0.00000001))
decoder_outputs = decoder_dense(decoder_outputs_combined)
model = Model([encoder_inputs, decoder_inputs], decoder_outputs)
return model
def scheduler(epoch, lr):
if epoch < 30:
return lr
else:
return float(lr * tf.math.exp(-0.01))
def train_model(model: Model, train_X: np.ndarray, train_Y: np.ndarray, val_X: np.ndarray, val_Y: np.ndarray,
batch_size: int = 10, epochs: int = 130) -> tf.keras.callbacks.History:
early_stopping = EarlyStopping(monitor='val_loss', patience=90, verbose=2, mode='min', restore_best_weights=True)
reduce_lr = ReduceLROnPlateau( monitor='val_loss', patience=30, verbose=2, mode='min', factor=0.001, min_lr=1e-8)
model_checkpoint = ModelCheckpoint('best_model.keras', monitor='val_loss', save_best_only=True, mode='min', verbose=2)
lr_scheduler = LearningRateScheduler(scheduler)
model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
loss=tf.keras.losses.MeanSquaredError(),
metrics=[tf.keras.metrics.MeanSquaredError(),
tf.keras.metrics.MeanAbsoluteError(),
tf.keras.metrics.MeanAbsolutePercentageError()])
history = model.fit([train_X, train_X], train_Y,
epochs=epochs,
batch_size=batch_size,
validation_data=([val_X, val_X], val_Y),
callbacks=[early_stopping, reduce_lr, model_checkpoint, lr_scheduler])
plot_training_history(history)
return history
def plot_training_history(history):
plt.figure(figsize=(16, 8))
plt.subplot(1, 2, 1)
plt.plot(history.history['loss'], label='Training Loss')
plt.plot(history.history['val_loss'], label='Validation Loss')
plt.title('Training and Validation Loss')
plt.xlabel('Epochs')
plt.ylabel('Loss')
plt.legend()
plt.subplot(1, 2, 2)
plt.plot(history.history['mean_squared_error'], label='Training MSE')
plt.plot(history.history['val_mean_squared_error'], label='Validation MSE')
plt.plot(history.history['mean_absolute_error'], label='Training MAE')
plt.plot(history.history['val_mean_absolute_error'], label='Validation MAE')
plt.plot(history.history['mean_absolute_percentage_error'], label='Training MAPE')
plt.plot(history.history['val_mean_absolute_percentage_error'], label='Validation MAPE')
plt.title('Training and Validation Metrics')
plt.xlabel('Epochs')
plt.ylabel('Metrics')
plt.legend()
plt.show()
def make_predictions(model: Model, val_X: np.ndarray) -> np.ndarray:
return model.predict([val_X, val_X])
def display_results(predicted_next_line: np.ndarray, dataset: np.ndarray) -> None:
predictions = np.round(predicted_next_line * 10).astype(int)
predicted_values = np.round(predicted_next_line * 10).astype(int)
actual_values = np.round(dataset[-1] * 10).astype(int)
predicted_set = set(predicted_values)
actual_set = set(actual_values)
common_values = predicted_set.intersection(actual_set)
common_values_sorted = sorted(common_values)
sorted_predictions = sorted(predictions)
print(f"\nΟΙ ΠΡΟΒΛΕΠΟΜΕΝΟΙ ΑΡΙΘΜΟΙ: {', '.join(map(str, sorted_predictions))}")
temperatures = ', '.join(f"{temp * 10:.0f}" for temp in dataset[-1])
print(f"Η ΤΕΛΕΥΤΑΙΑ ΚΛΗΡΩΣΗ: {temperatures}")
print(f"\nΚΟΙΝΕΣ ΤΙΜΕΣ: {common_values_sorted}")
plt.figure(figsize=(12, 6))
plt.plot(predicted_values, label='Predicted Values')
plt.plot(actual_values, label='Actual Values')
plt.title('Predicted vs Actual Values')
plt.xlabel('Index')
plt.ylabel('Value')
plt.legend()
plt.show()
def main():
dataset = create_dataset()
input_data, output_data = create_sequences(dataset)
input_data = input_data.reshape(input_data.shape[0], 1, input_data.shape[1])
output_data = output_data.reshape(output_data.shape[0], 1, output_data.shape[1])
train_X, val_X, train_Y, val_Y = split_data(input_data, output_data)
num_units = [128, 128, 128]
num_heads = 3
model = build_model((1, dataset.shape[1]), num_units, num_heads)
history = train_model(model, train_X, train_Y, val_X, val_Y)
next_line_input = dataset[-1].reshape(1, 1, dataset.shape[1])
predicted_next_line = make_predictions(model, next_line_input).reshape(-1)
display_results(predicted_next_line, dataset)
tf.keras.backend.clear_session()
if __name__ == "__main__":
main()
ΑΠΟΤΕΛΕΣΜΑΤΑΟΙ ΠΡΟΒΛΕΠΟΜΕΝΟΙ ΑΡΙΘΜΟΙ: 2, 4, 9
Η ΤΕΛΕΥΤΑΙΑ ΚΛΗΡΩΣΗ: 3, 3, 5
ΚΟΙΝΕΣ ΤΙΜΕΣ: []