apply revisions + add typeguard

apply revisions

Author: Francisco Santos

examples/timeseries/tscwgan_example.py

@@ -1,10 +1,9 @@
-from pandas import DataFrame
 from numpy import squeeze
 
-from ydata_synthetic.postprocessing.timeseries.inverse_preprocesser import inverse_transform
 from ydata_synthetic.preprocessing.timeseries import processed_stock
 from ydata_synthetic.synthesizers.timeseries import TSCWGAN
 from ydata_synthetic.synthesizers import ModelParameters, TrainParameters
+from ydata_synthetic.postprocessing.regular.inverse_preprocesser import inverse_transform
 
 model = TSCWGAN
 
@@ -24,7 +23,7 @@
 critic_iter = 5
 
 # Get transformed data stock - Univariate
-data, processed_data, scaler = processed_stock(path='./data/stock_data.csv', seq_len=seq_len, cols = 'Open')
+data, processed_data, scaler = processed_stock(path='./data/stock_data.csv', seq_len=seq_len, cols = ['Open'])
 data_sample = processed_data[0]
 
 model_parameters = ModelParameters(batch_size=batch_size,
@@ -51,10 +50,10 @@
 
 #Sampling the data
 #Note that the data returned is not inverse processed.
-step = int(len(processed_data)/(5-1))
-cond_array = DataFrame(data=[squeeze(processed_data[i][:cond_dim], axis=1) for i in range(0, len(processed_data), step)])
+cond_index = 100  # Arbitrary sequence for conditioning
+cond_array = squeeze(processed_data[cond_index][:cond_dim], axis=1)
 
-data_sample = synth.sample(cond_array, 200)
+data_sample = synth.sample(cond_array, 1000)
 
 # Inverting the scaling of the synthetic samples
 data_sample = inverse_transform(data_sample, scaler)

src/ydata_synthetic/postprocessing/timeseries/inverse_preprocesser.py

@@ -2,21 +2,21 @@
     Get the stock data from Yahoo finance data
     Data from the period 01 January 2017 - 24 January 2021
 """
-from typing import Union, List
+from typing import Optional, List
 
 import pandas as pd
+from typeguard import typechecked
 
 from ydata_synthetic.preprocessing.timeseries.utils import real_data_loading
 
-def transformations(path, seq_len: int, cols: Union[str, List] = None):
+@typechecked
+def transformations(path, seq_len: int, cols: Optional[List] = None):
     """Apply min max scaling and roll windows of a temporal dataset.
 
     Args:
         path(str): path to a csv temporal dataframe
         seq_len(int): length of the rolled sequences
         cols (Union[str, List]): Column or list of columns to be used"""
-    if isinstance(cols, str):
-        cols = [cols]
     if isinstance(cols, list):
         stock_df = pd.read_csv(path)[cols]
     else:

src/ydata_synthetic/preprocessing/timeseries/stock.py

@@ -187,7 +187,7 @@ def save(self, path):
         super().save(path)
 
 
-class Generator(Model):
+class Generator(tf.keras.Model):
     def __init__(self, batch_size):
         """Simple generator with dense feedforward layers."""
         self.batch_size = batch_size
@@ -202,7 +202,7 @@ def build_model(self, input_shape, dim, data_dim, activation_info: Optional[Name
             x = GumbelSoftmaxActivation(activation_info)(x)
         return Model(inputs=input_, outputs=x)
 
-class Critic(Model):
+class Critic(tf.keras.Model):
     def __init__(self, batch_size):
         """Simple critic with dense feedforward and dropout layers."""
         self.batch_size = batch_size

src/ydata_synthetic/synthesizers/regular/cramergan/model.py

@@ -4,16 +4,15 @@
 And on: https://github.com/CasperHogenboom/WGAN_financial_time-series
 """
 from tqdm import trange
+from numpy import array, vstack
 from numpy.random import normal
-from pandas import DataFrame
 
 from tensorflow import concat, float32, convert_to_tensor, reshape, GradientTape, reduce_mean, make_ndarray, make_tensor_proto, tile, expand_dims
 from tensorflow import data as tfdata
 from tensorflow.keras import Model, Sequential
 from tensorflow.keras.optimizers import Adam
 from tensorflow.keras.layers import Input, Conv1D, Dense, LeakyReLU, Flatten, Add
 
-
 from ydata_synthetic.synthesizers.gan import BaseModel
 from ydata_synthetic.synthesizers import TrainParameters
 from ydata_synthetic.synthesizers.loss import Mode, gradient_penalty
@@ -61,10 +60,7 @@ def train(self, data, train_arguments: TrainParameters):
 
             g_loss = self.update_generator(real_batch, noise_batch)
 
-            print(
-                "Epoch: {} | critic_loss: {} | gen_loss: {}".format(
-                    epoch, c_loss, g_loss
-                ))
+            print(f"Epoch: {epoch} | critic_loss: {c_loss} | gen_loss: {g_loss}")
 
         self.g_optimizer = self.g_optimizer.get_config()
         self.c_optimizer = self.c_optimizer.get_config()
@@ -148,21 +144,19 @@ def get_batch_data(self, data, n_windows= None):
 
     def sample(self, cond_array, n_samples):
         """Provided that cond_array is passed, produce n_samples for each condition vector in cond_array."""
-        assert len(cond_array.shape) == 2, "Condition array should have 2 dimensions."
-        assert cond_array.shape[1] == self.cond_dim, \
-            f"Each sequence in the condition array should have a {self.cond_dim} length."
-        n_conds = cond_array.shape[0]
+        assert len(cond_array.shape) == 1, "Condition array should be one-dimensional."
+        assert cond_array.shape[0] == self.cond_dim, \
+            f"The condition sequence should have a {self.cond_dim} length."
         steps = n_samples // self.batch_size + 1
         data = []
         z_dist = self.get_batch_noise()
-        for seq in range(n_conds):
-            cond_seq = expand_dims(convert_to_tensor(cond_array.iloc[seq], float32), axis=0)
-            cond_seq = tile(cond_seq, multiples=[self.batch_size, 1])
-            for step in trange(steps, desc=f'Synthetic data generation - Condition {seq+1}/{n_conds}'):
-                gen_input = concat([cond_seq, next(z_dist)], axis=1)
-                records = make_ndarray(make_tensor_proto(self.generator(gen_input, training=False)))
-                data.append(records)
-        return DataFrame(concat(data, axis=0))
+        cond_seq = expand_dims(convert_to_tensor(cond_array, float32), axis=0)
+        cond_seq = tile(cond_seq, multiples=[self.batch_size, 1])
+        for step in trange(steps, desc=f'Synthetic data generation'):
+            gen_input = concat([cond_seq, next(z_dist)], axis=1)
+            records = make_ndarray(make_tensor_proto(self.generator(gen_input, training=False)))
+            data.append(records)
+        return array(vstack(data))
 
 
 class Generator(Model):