Keras Tuner Subclass for Time Series Cross-Validation

Question

Custom Keras Tuner with Time Series Cross-Validation

I have written my own subclass of the default Keras tuner Tune class.

• Objective: I needed a way to incorporate time series cross-validation into the hyperparameter tuning process, which wasn't directly supported by the default Keras tuner.

• Functionality: My TimeSeriesBayesianOptimization subclass integrates time series cross-validation, allowing the model to be evaluated across multiple time-based splits and returning the average performance metrics.

• Use Case: This is particularly useful for my dataset, which involves time series forecasting where traditional random cross-validation can disrupt the temporal structure.

• Feedback Request: I'm looking for feedback on the efficiency of the implementation, potential pitfalls, and any best practices that I might have overlooked. I'm particularly interested in understanding if my method of averaging metrics across time series splits is optimal for guiding the Bayesian optimization process. Additionally, I want to ensure that the logic surrounding how the Oracle interprets the averaged objective over all the folds is sound.

The Tuner

class TimeSeriesBayesianOptimization(BayesianOptimization):
 def __init__(self, time_series_splits=5, *args, **kwargs):
 super(TimeSeriesBayesianOptimization, self).__init__(*args, **kwargs)
 self.time_series_splits = time_series_splits
 self.tscv = TimeSeriesSplit(n_splits=self.time_series_splits)
 def run_trial(self, trial, *args, **kwargs):
 # Extract X_train and y_train without altering original args
 X_train, y_train, *remaining_args = args
 # Callback to save the best epoch
 model_checkpoint = tuner_utils.SaveBestEpoch(
 objective=self.oracle.objective,
 filepath=self._get_checkpoint_fname(trial.trial_id),
 )
 original_callbacks = kwargs.pop("callbacks", [])
 # Track the histories
 histories = []
 for execution in range(self.executions_per_trial):
 total_val_loss = 0.0
 total_loss = 0.0
 total_binary_accuracy = 0.0
 total_val_binary_accuracy = 0.0
 for train_index, val_index in self.tscv.split(X_train):
 X_train_split, X_val_split = X_train[train_index], X_train[val_index]
 y_train_split, y_val_split = y_train[train_index], y_train[val_index]
 # Build the model for this trial's hyperparameters
 model = self.hypermodel.build(trial.hyperparameters)
 # Set up callbacks
 copied_callbacks = self._deepcopy_callbacks(original_callbacks)
 self._configure_tensorboard_dir(copied_callbacks, trial, execution)
 copied_callbacks.append(tuner_utils.TunerCallback(self, trial))
 copied_callbacks.append(model_checkpoint)
 # Train the model for this split
 history = model.fit(
 X_train_split,
 y_train_split,
 validation_data=(X_val_split, y_val_split),
 callbacks=copied_callbacks,
 **kwargs
 )
 # Grab the best values for each metric
 best_val_loss = min(history.history["val_loss"])
 best_loss = min(history.history["loss"])
 best_binary_accuracy = max(history.history["binary_accuracy"])
 best_val_binary_accuracy = max(history.history["val_binary_accuracy"])
 # Accumulate the best values
 total_val_loss += best_val_loss
 total_loss += best_loss
 total_binary_accuracy += best_binary_accuracy
 total_val_binary_accuracy += best_val_binary_accuracy
 # Compute the averages
 avg_val_loss = total_val_loss / self.time_series_splits
 avg_loss = total_loss / self.time_series_splits
 avg_binary_accuracy = total_binary_accuracy / self.time_series_splits
 avg_val_binary_accuracy = (
 total_val_binary_accuracy / self.time_series_splits
 )
 # Store the averages in the histories list
 histories.append(
 {
 "val_loss": avg_val_loss,
 "loss": avg_loss,
 "binary_accuracy": avg_binary_accuracy,
 "val_binary_accuracy": avg_val_binary_accuracy,
 }
 )
 return histories
 def save_model(self, trial_id, model):
 """Save the model for the given trial."""
 fname = os.path.join(self.get_trial_dir(trial_id), "model.keras")
 model.save(fname)

Usage

# Cross Validation Search (WIP)
tuner = TimeSeriesBayesianOptimization(
 hypermodel=search_cnn_lstm_model,
 objective=Objective("val_loss", direction="min"),
 max_trials=6,
 time_series_splits=5, # Number of time series splits for cross-validation
 seed=42,
 executions_per_trial=1,
 directory="tmp/tb",
 project_name="gru_cnn_vl",
)
tuner.search(
 X_train,
 y_train,
 epochs=120,
 shuffle=False,
 batch_size=72,
 callbacks=[
 tf.keras.callbacks.EarlyStopping(monitor="val_loss", patience=10, mode="min")
 ],
 verbose=True,
)

Answer 1

redundant object attribute

 def __init__(self, time_series_splits=5, *args, **kwargs):
 super(TimeSeriesBayesianOptimization, self).__init__(*args, **kwargs)
 self.time_series_splits = time_series_splits
 self.tscv = TimeSeriesSplit(n_splits=self.time_series_splits)

I like the super() call.

I am skeptical that assigning self.time_series_splits is helpful, given that self.tscv.n_splits offers that value. Storing copies of same thing in different places can lead to trouble. Consider writing a @property decorator which will return self.tscv.n_splits.

missing annotation, docstring

 def run_trial(self, trial, *args, **kwargs):

This is not a good signature. It offers little guidance to a maintenance engineer who wants to exercise it with an isolated unit test. It describes neither the type of trial, nor the returned list. It lacks a single English sentence explaining its responsibility.

Now of course, you don't have to do the heavy lifting here. A simple citation of the Keras base run_trial() would concisely answer such questions. The idea is to make it easy for a newly hired engineer to find the spec you're writing against.

names

 # Accumulate the best values
 total_val_loss += best_val_loss
 total_loss += best_loss
 total_binary_accuracy += best_binary_accuracy
 total_val_binary_accuracy += best_val_binary_accuracy

IDK, I guess the LHS identifiers are kind of OK, in the sense that I'm not very keen on longer names like total_best_val_loss. But I'm a little uncomfortable with the truthfulness of a name like total_val_loss, as it handles just a subset of the validation loss figures.

Maybe one way out is to define a 4-attribute stats object, and then a total_stats accompanies it? Sorry, I'm not being very constructive with an alternate solution, I'm mostly just voicing that I'm not yet comfortable with the code I do see.

Hmmm, maybe we should allocate more memory, keep accumulating a list of all stats, and at the end let numpy worry about {min, max, mean}. So we're letting the function name communicate such details, rather than a scalar quantity's name.

Path

In save_model(), consider modeling filespecs with Path rather than str, just because it offers a more convenient / less verbose API to call into. Clearly the current code works just fine as-is.

---

LGTM, ship it!

This codebase achieves its design goals.

I would be willing to delegate or accept maintenance tasks on it.