i am trying to validate a tensorflow code for simple timeseries prediction in where:
X = np.arange(0, 2000, 0.5)
y = 2 * np.sin(X) + 0.8 * np.random.rand(X.shape[0])
and the following parameters for the timeseries:
LOOK_BACK = 100
FORECAST_HORIZON = 100
any model I use or try gets very bad performance, LSTM, MLP, CNN... loss is not improved in any epoch.
def create_dataset(X_data, y_data, shuffle=False, repeat=False):
# Ensure input and output are tensors
X_data = tf.convert_to_tensor(X_data, dtype=tf.float32)
y_data = tf.convert_to_tensor(y_data, dtype=tf.float32)
# Create windowed dataset
dataset = tf.keras.preprocessing.timeseries_dataset_from_array(
data=tf.concat([X_data, y_data], axis=-1), # combine to keep alignment
targets=None,
sequence_length=look_back + forecast,
sequence_stride=1,
shuffle=shuffle,
batch_size=1,
)
def split_input_target(sequence):
input_seq = sequence[:, :look_back, : X_data.shape[-1]]
target_seq = sequence[:, look_back:, X_data.shape[-1] :]
return input_seq, target_seq
dataset = dataset.map(split_input_target)
return dataset
train_dataset = create_dataset(X_train, y_train, shuffle=True, repeat=True)
val_dataset = create_dataset(X_val, y_val)
test_dataset = create_dataset(X_test, y_test)
return train_dataset, val_dataset, test_dataset
Model is as follows:
def build_and_compile_model(num_features, out_steps, targets):
model = tf.keras.Sequential(
[
tf.keras.layers.Input(shape=(LOOK_BACK, num_features)),
tf.keras.layers.LSTM(128),
tf.keras.layers.Dense(out_steps * targets),
tf.keras.layers.Reshape((out_steps, targets)),
]
)
model.compile(
loss="mse", optimizer=tf.keras.optimizers.RMSprop(0.001), metrics=["mse"]
)
return model
The following code is what i am currently trying:
import matplotlib.pyplot as plt
from keras.callbacks import CSVLogger
import tensorflow as tf
import numpy as np
BATCH_SIZE = 16
LOOK_BACK = 100
FORECAST_HORIZON = 100
seed = 101
tf.keras.utils.set_random_seed(seed)
X = (np.arange(0, 2000, 0.5)).reshape(-1, 1)
y = 20 * np.sin(X)
#### DEF FUNCTIONS ####
def manual_train_test_split(X, y, train_size=0.7, val_size=0.1):
total_len = len(X)
train_end = int(total_len * train_size)
val_end = train_end + int(total_len * val_size)
X_train, y_train = X[:train_end], y[:train_end]
X_val, y_val = X[train_end:val_end], y[train_end:val_end]
X_test, y_test = X[val_end:], y[val_end:]
return X_train, X_val, X_test, y_train, y_val, y_test
def make_time_series_datasets(
X_train, y_train, X_val, y_val, X_test, y_test, look_back, forecast, batch_size
):
def create_dataset(X_data, y_data, shuffle=False, repeat=False):
# Ensure input and output are tensors
X_data = tf.convert_to_tensor(X_data, dtype=tf.float32)
y_data = tf.convert_to_tensor(y_data, dtype=tf.float32)
# Create windowed dataset
dataset = tf.keras.preprocessing.timeseries_dataset_from_array(
data=tf.concat([X_data, y_data], axis=-1), # combine to keep alignment
targets=None,
sequence_length=look_back + forecast,
sequence_stride=1,
shuffle=shuffle,
batch_size=1,
)
def split_input_target(sequence):
input_seq = sequence[:, :look_back, : X_data.shape[-1]]
target_seq = sequence[:, look_back:, X_data.shape[-1] :]
return input_seq, target_seq
dataset = dataset.map(split_input_target)
return dataset
train_dataset = create_dataset(X_train, y_train, shuffle=True, repeat=True)
val_dataset = create_dataset(X_val, y_val)
test_dataset = create_dataset(X_test, y_test)
return train_dataset, val_dataset, test_dataset
def build_and_compile_model(num_features, out_steps, targets):
model = tf.keras.Sequential(
[
tf.keras.layers.Input(shape=(LOOK_BACK, num_features)),
tf.keras.layers.LSTM(128),
tf.keras.layers.Dense(out_steps * targets),
tf.keras.layers.Reshape((out_steps, targets)),
]
)
model.compile(
loss="mse", optimizer=tf.keras.optimizers.RMSprop(0.001), metrics=["mse"]
)
return model
def build_and_compile_mlp_model(input_dim, output_dim):
model = tf.keras.Sequential(
[
tf.keras.layers.Input(shape=(input_dim,)),
tf.keras.layers.Dense(128),
tf.keras.layers.Activation("relu"),
tf.keras.layers.Dense(
64,
),
tf.keras.layers.Activation("relu"),
tf.keras.layers.Dense(output_dim),
]
)
model.compile(optimizer="adam", loss="mae", metrics=["mape"])
model.summary()
return model
#### START ####
X_train, X_val, X_test, y_train, y_val, y_test = manual_train_test_split(X, y)
train_dataset, val_dataset, test_dataset = make_time_series_datasets(
X_train,
y_train,
X_val,
y_val,
X_test,
y_test,
LOOK_BACK,
FORECAST_HORIZON,
BATCH_SIZE,
)
csv_logger = CSVLogger("training.log", separator=",", append=False)
early_stopping = tf.keras.callbacks.EarlyStopping(
monitor="val_loss", patience=100000, restore_best_weights=True
)
model = build_and_compile_model(
num_features=X_train.shape[-1], out_steps=FORECAST_HORIZON, targets=1
)
history = model.fit(
train_dataset,
epochs=20,
callbacks=[csv_logger, early_stopping],
validation_data=val_dataset,
).history
# Helper function to plot input sequence, true future, and predicted future
def multi_step_output_plot(true_future, prediction):
plt.figure(figsize=(18, 6))
print(np.shape(true_future))
plt.plot(true_future, label="True Future")
plt.plot(prediction, label="Predicted Future")
plt.legend(loc="upper left")
plt.xlabel("Time Step")
plt.ylabel("Value")
plt.title("Multi-Step Time Series Forecasting")
plt.grid(True)
plt.show()
# Take one batch from validation dataset
for x_batch, y_batch in val_dataset.take(1):
input_seq = x_batch # First sequence in batch
true_future = y_batch[0] # Corresponding true future values
prediction = model.predict(input_seq)[0] # Shape: (output steps,)
print(y_batch[0])
print(prediction)
multi_step_output_plot(np.squeeze(true_future), np.squeeze(prediction))
The original approach attempts to predict a sine wave by feeding the model a sequence of time values (x), which is a linearly increasing array. The model is then asked to predict the corresponding future values of sin(x). Notwhithstanding, Neural Networks do not perform well with linear and non-bounded inputs. In the case of series forecasting, you want to use previous values of the predicting variable (y) to forecast the next output. Additionally, you may need to normalize the amplitude value, as the output performs well with values between -1 and 1 (not 20).
Let sin_generator be a sin-wave data generator for a given configuration, in your case, you can use the following config:
def sin_generator(look_back: int,
forecast_horizon: int,
num_samples: int = 1000,
x_initial_value: float = 0.0,
x_end_value: float = 2000.0,
y_amplitude: float = 20.0,
dtype: str = 'float32'):
"""
Create a generator for the time series data.
y = y_amplitude * np.sin(x)
:param look_back: The number of previous time steps to use as input.
:param forecast_horizon: The number of time steps to predict.
:param num_samples: The number of samples to generate.
:param x_initial_value: The initial value of x.
:param x_end_value: The end value of x.
:param y_amplitude: The amplitude of the sine wave.
:param dtype: The data type of the output.
:return: A generator that yields (x, y) pairs.
"""
for _ in range(num_samples):
# Random loopback:
initial_pos = np.random.randint(low=x_initial_value, high=x_end_value - look_back)
# Create x values:
x = np.arange(initial_pos, initial_pos + look_back + forecast_horizon, dtype=dtype)
# Create y values:
y = (y_amplitude * np.sin(x)).astype(dtype)
# Create the generator:
# yield x[:look_back], y[look_back:] -> wrong
yield y[:look_back], y[look_back:]
You can build a tf_dataset as follows:
def sin_tf_dataset(generator, look_back: int, forecast_horizon: int, batch_size: int):
"""
Create a TensorFlow dataset from the generator.
:param generator: The generator to use.
:param look_back: The number of previous time steps to use as input.
:param forecast_horizon: The number of time steps to predict.
:param batch_size: The batch size for the dataset.
:return: A TensorFlow dataset.
"""
# Create the dataset:
dataset = tf.data.Dataset.from_generator(
generator,
output_signature=(
tf.TensorSpec(shape=(look_back,), dtype=tf.float32),
tf.TensorSpec(shape=(forecast_horizon,), dtype=tf.float32)
)
)
# Batch the dataset:
dataset = dataset.batch(batch_size)
return dataset
And, finally let plot_dataset be an utility function to show the generated data:
def plot_dataset(tf_dataset, n_examples=3):
"""
Plot a few examples from a time series dataset.
:param tf_dataset: The tf.data.Dataset object (should yield (input, target)).
:param n_examples: How many examples to plot.
"""
for i, (x, y) in enumerate(tf_dataset.unbatch().take(n_examples)):
plt.figure(figsize=(10, 4))
look_back = x.shape[0]
forecast_horizon = y.shape[0]
# Plot input (past)
plt.plot(range(look_back), x.numpy(), label="Input (Past)", color='blue')
# Plot target (future)
plt.plot(range(look_back, look_back + forecast_horizon), y.numpy(), label="Target (Future)", color='orange')
# Vertical line to mark the split
plt.axvline(x=look_back - 1, color="gray", linestyle="--")
plt.title(f"Sample {i+1}: Forecasting {forecast_horizon} steps from {look_back} past values")
plt.xlabel("Timestep")
plt.ylabel("Value")
plt.grid(True)
plt.legend()
plt.tight_layout()
plt.show()
Then, a simple LSTM model can predict the time series with the following snipet:
def create_model(input_shape, output_shape):
"""
Create a simple LSTM model for time series forecasting.
:param input_shape: The shape of the input data.
:param output_shape: The shape of the output data.
:return: A compiled Keras model.
"""
model = tf.keras.Sequential([
tf.keras.layers.LSTM(128, activation='tanh', input_shape=input_shape),
tf.keras.layers.Dense(output_shape)
])
model.compile(optimizer='adam', loss='mse')
return model
if __name__ == "__main__":
# Set parameters:
batch_size = 32
config = {
'look_back': 100,
'forecast_horizon': 100,
'num_samples': 1000,
'x_initial_value': 0.0,
'x_end_value': 2000.0,
'y_amplitude': 1.0,
'dtype': 'float32',
}
# Create generator:
dataset = sin_tf_dataset(
lambda: sin_generator(**config),
look_back=config['look_back'],
forecast_horizon=config['forecast_horizon'],
batch_size=batch_size
)
# Plot the generator:
plot_dataset(dataset, n_examples=3)
# Create model:
input_shape = (config['look_back'], 1)
output_shape = config['forecast_horizon']
model = create_model(input_shape, output_shape)
model.summary()
# Train model:
model.fit(dataset, epochs=10)
In this case, you need to normalize the y_amplitude, you can lately multiply by the value you need as part of pos-processing. This model reports the following log:
Model: "sequential"
┌─────────────────────────────────┬────────────────────────┬───────────────┐
│ Layer (type) │ Output Shape │ Param # │
├─────────────────────────────────┼────────────────────────┼───────────────┤
│ lstm (LSTM) │ (None, 128) │ 66,560 │
├─────────────────────────────────┼────────────────────────┼───────────────┤
│ dense (Dense) │ (None, 100) │ 12,900 │
└─────────────────────────────────┴────────────────────────┴───────────────┘
Total params: 79,460 (310.39 KB)
Trainable params: 79,460 (310.39 KB)
Non-trainable params: 0 (0.00 B)
Epoch 1/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 3s 45ms/step - loss: 0.4757
Epoch 2/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 1s 44ms/step - loss: 0.1000
Epoch 3/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 1s 43ms/step - loss: 0.0011
Epoch 4/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 1s 43ms/step - loss: 1.5433e-04
Epoch 5/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 1s 45ms/step - loss: 8.2073e-05
Epoch 6/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 2s 47ms/step - loss: 6.9301e-05
Epoch 7/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 2s 47ms/step - loss: 6.1493e-05
Epoch 8/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 1s 45ms/step - loss: 5.0608e-05
Epoch 9/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 2s 51ms/step - loss: 4.3731e-05
Epoch 10/10
32/32 ━━━━━━━━━━━━━━━━━━━━ 2s 48ms/step - loss: 3.9795e-05
Here, you have transformed the task from a regression over raw time to a standard sequence-to-sequence forecasting problem, which is what recurrent models like LSTM are designed for. This structure makes it easier for the model to capture periodicity, trends, and normalized local dynamics in the signal.