feat(Transformations): added Transformations abstraction & handling in walk_forward_train() & inference() (#161)

* feat(Transformations): added Transformations abstraction & handling in walk_forward_train() & inference()

* fix(WalkForward): use Dataframes to call Transformation.fit_transform()

* feat(WalkForward): restored option for models to recieve unscaled data

* fix(Transformations): output DataFrame as expected

* fix(Tests): missing new property

feature_selection/feature_selection.py

@@ -25,7 +25,7 @@ def select_features(X: pd.DataFrame, y: pd.Series, model: Model, n_features_to_s
     # 2. Recursive feature selection
     cv = TimeSeriesSplit(n_splits=5)
     scaler = get_scaler(scaling)
-    X_scaled = scaler.fit_transform(X)
+    X_scaled = scaler.fit_transform(X, y)
 
     feat_selector_model = model.model
     if hasattr(feat_selector_model, 'feature_importances_') == False and hasattr(feat_selector_model, 'coef_') == False:

models/average.py

@@ -7,7 +7,7 @@ class StaticAverageModel(Model):
     Model that averages .
     '''
 
-    data_scaling = 'unscaled'
+    data_transformation = 'original'
     only_column = 'model_'
     feature_selection = 'off'
     model_type = 'static'

models/base.py

@@ -7,7 +7,7 @@ import numpy as np
 
 class Model(ABC):
 
-    data_scaling: Literal["scaled", "unscaled"]
+    data_transformation: Literal["transformed", "original"]
     feature_selection: Literal["on", "off"]
     # data_format: Literal["wide", "narrow"]
     only_column: Optional[str]

models/momentum.py

@@ -7,7 +7,7 @@ class StaticMomentumModel(Model):
     Model that uses only one feature: momentum. It's positive if momentum is greater than 0, otherwise it's negative.
     '''
 
-    data_scaling = 'unscaled'
+    data_transformation = 'original'
     only_column = 'mom'
     feature_selection = 'off'
     model_type = 'static'

models/naive.py

@@ -7,7 +7,7 @@ class StaticNaiveModel(Model):
     Model that carries the last observation (from returns) to the next one, naively.
     '''
 
-    data_scaling = 'unscaled'
+    data_transformation = 'original'
     only_column = None
     feature_selection = 'off'
     model_type = 'static'

models/neural.py

@@ -7,7 +7,7 @@ import pytorch_lightning as pl
 
 class LightningNeuralNetModel(Model):
 
-    data_scaling = 'scaled'
+    data_transformation = 'transformed'
     only_column = None
     feature_selection = 'off'
     model_type = 'ml'

models/sklearn.py

@@ -6,7 +6,7 @@ from sklearn.base import clone
 
 class SKLearnModel(Model):
 
-    data_scaling = 'scaled'
+    data_transformation = 'transformed'
     only_column = None
     feature_selection = 'on'
     model_type = 'ml'

models/statsmodels.py

@@ -8,7 +8,7 @@ from copy import deepcopy
 class StatsModel(Model):
 
     # This is work in progress
-    data_scaling = 'scaled'
+    data_transformation = 'transformed'
     only_column = None
     feature_selection = 'on'
     model_type = 'ml'

models/xgboost.py

@@ -6,7 +6,7 @@ from sklearn.base import clone
 
 class XGBoostModel(Model):
 
-    data_scaling = 'scaled'
+    data_transformation = 'transformed'
     only_column = None
     feature_selection = 'on'
     model_type = 'ml'

tests/test_evaluation.py

@@ -4,7 +4,6 @@ import pandas as pd
 from training.walk_forward import walk_forward_train, walk_forward_inference
 from models.base import Model
 from utils.evaluate import evaluate_predictions
-from sklearn.preprocessing import MinMaxScaler
 
 no_of_rows = 100
 
@@ -36,7 +35,7 @@ class EvenOddStubModel(Model):
     It verifies that the X[n][any_column] == 1 if n is even,
     '''
 
-    data_scaling = "unscaled"
+    data_transformation = "original"
     only_column = None
     predict_window_size = 'single_timestamp'
 
@@ -68,9 +67,8 @@ def test_evaluation():
     window_length = 10
 
     model = EvenOddStubModel(window_length = window_length)
-    scaler = MinMaxScaler()
     
-    models, scalers = walk_forward_train(
+    model_over_time, transformations_over_time = walk_forward_train(
         model_name='test',
         model=model,
         X=X,
@@ -79,11 +77,11 @@ def test_evaluation():
         expanding_window=False,
         window_size=window_length,
         retrain_every=10,
-        scaler=scaler)
+        transformations=[])
-    predictions, probs = walk_forward_inference(
+    predictions, _ = walk_forward_inference(
         model_name='test',
-        models=models,
+        model_over_time=model_over_time,
-        scalers=scalers,
+        transformations_over_time=transformations_over_time,
         X=X,
         expanding_window=False,
         window_size=window_length

tests/test_walk_forward.py

@@ -2,7 +2,6 @@ import numpy as np
 import pandas as pd
 from training.walk_forward import walk_forward_train, walk_forward_inference
 from models.base import Model
-from sklearn.preprocessing import MinMaxScaler
 
 no_of_rows = 100
 
@@ -34,7 +33,7 @@ class IncrementingStubModel(Model):
     It verifies that the X[n][any_column]+1 == y[n]
     '''
 
-    data_scaling = "unscaled"
+    data_transformation = "original"
     only_column = None
     predict_window_size = 'single_timestamp'
 
@@ -66,9 +65,8 @@ def test_walk_forward_train_test():
     window_length = 10
 
     model = IncrementingStubModel(window_length = window_length)
-    scaler = MinMaxScaler()
 
-    models, scalers = walk_forward_train(
+    model_over_time, transformations_over_time = walk_forward_train(
         model_name='test',
         model=model,
         X=X,
@@ -77,11 +75,11 @@ def test_walk_forward_train_test():
         expanding_window=False,
         window_size=window_length,
         retrain_every=10,
-        scaler=scaler)
+        transformations=[])
-    predictions, probs = walk_forward_inference(
+    predictions, _ = walk_forward_inference(
         model_name='test',
-        models=models,
+        model_over_time=model_over_time,
-        scalers=scalers,
+        transformations_over_time=transformations_over_time,
         X=X,
         expanding_window=False,
         window_size=window_length

training/primary_model.py

@@ -6,6 +6,7 @@ from models.base import Model
 from utils.scaler import get_scaler
 from utils.types import ScalerTypes
 from utils.encapsulation import Training_Step, Single_Model, Asset
+from transformations.sklearn import SKLearnTransformation
 
 def train_primary_model(
                     ticker_to_predict: str,
@@ -24,8 +25,6 @@ def train_primary_model(
                     print_results: bool,
     ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, list[Single_Model]]:
 
-    scaler = get_scaler(scaler)
-
     results = pd.DataFrame()
     predictions = pd.DataFrame(index=y.index)
     probabilities = pd.DataFrame(index=y.index)
@@ -34,7 +33,7 @@ def train_primary_model(
     
     
     for model_name, model in models:
-        model_over_time, scaler_over_time = walk_forward_train(
+        model_over_time, transformations_over_time = walk_forward_train(
             model_name=model_name,
             model = model,
             X = X if model.feature_selection == 'on' else original_X,
@@ -43,15 +42,15 @@ def train_primary_model(
             expanding_window = expanding_window,
             window_size = sliding_window_size,
             retrain_every = retrain_every,
-            scaler = scaler
+            transformations= [get_scaler(scaler)],
         )
         preds, probs = walk_forward_inference(
             model_name = model_name,
-            models = model_over_time,
+            model_over_time= model_over_time,
+            transformations_over_time = transformations_over_time,
             X = X if model.feature_selection == 'on' else original_X,
             expanding_window = expanding_window,
-            window_size = sliding_window_size,
+            window_size = sliding_window_size
-            scalers = scaler_over_time
         )
         
         assert len(preds) == len(y)

training/walk_forward.py

@@ -6,6 +6,7 @@ from tqdm import tqdm
 from sklearn.preprocessing import MinMaxScaler, Normalizer, StandardScaler
 from typing import Union
 from sklearn.base import clone
+from transformations.base import Transformation
 
 def walk_forward_train(
                         model_name: str,
@@ -16,11 +17,11 @@ def walk_forward_train(
                         expanding_window: bool,
                         window_size: int,
                         retrain_every: int,
-                        scaler: Union[MinMaxScaler, Normalizer, StandardScaler],
+                        transformations: list[Transformation],
-                    ) -> tuple[pd.Series, pd.Series]:
+                    ) -> tuple[pd.Series, list[pd.Series]]:
     assert len(X) == len(y)
-    models = pd.Series(index=y.index).rename(model_name)
+    models_over_time = pd.Series(index=y.index).rename(model_name)
-    scalers = pd.Series(index=y.index).rename("scaler_" + model_name)
+    transformations_over_time = [pd.Series(index=y.index).rename(t.get_name()) for t in transformations]
 
     first_nonzero_return = max(get_first_valid_return_index(target_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y))
     train_from = first_nonzero_return + window_size + 1
@@ -29,36 +30,32 @@ def walk_forward_train(
 
     if model.only_column is not None:
         X = X[[column for column in X.columns if model.only_column in column]]
-        
+    
-    is_scaling_on = model.data_scaling == 'scaled'
+    if model.data_transformation == 'original':
-
+        transformations = []
-    if is_scaling_on:
+    
-        scaler = clone(scaler)
-
     for index in tqdm(range(train_from, train_till)):
-        if expanding_window:
+        train_window_start = first_nonzero_return if expanding_window else index - window_size - 1
-            train_window_start = first_nonzero_return
-        else:
-            train_window_start = index - window_size - 1
 
-        if iterations_before_retrain <= 0 or pd.isna(models[index-1]):
+        if iterations_before_retrain <= 0 or pd.isna(models_over_time[index-1]):
 
             train_window_end = index - 1
             
-            current_scaler = None
+            X_expanding_window = X[first_nonzero_return:train_window_end]
-            if is_scaling_on:
+            y_expanding_window = y[first_nonzero_return:train_window_end]
-                # We need to fit on the expanding window data slice
-                # This is our only way to avoid lookahead bias
-                current_scaler = clone(scaler)
-                X_expanding_window = X[first_nonzero_return:train_window_end]
-                current_scaler.fit(X_expanding_window.values)
 
-            X_slice = X[train_window_start:train_window_end].to_numpy()
+            current_transformations = [t.clone() for t in transformations]
+            for transformation_index, transformation in enumerate(current_transformations):
+                transformation.fit_transform(X_expanding_window, y_expanding_window)
+
+            X_slice = X[train_window_start:train_window_end]
+
+            for transformation in current_transformations:
+                X_slice = transformation.transform(X_slice)
+            
+            X_slice = X_slice.to_numpy()
             y_slice = y[train_window_start:train_window_end].to_numpy()
 
-            if is_scaling_on:
-                X_slice = current_scaler.transform(X_slice)
-            
             current_model = model.clone()
 
             current_model.initialize_network(input_dim = len(X_slice[0]), output_dim=1) 
@@ -66,17 +63,18 @@ def walk_forward_train(
 
             iterations_before_retrain = retrain_every
 
-        models[index] = current_model
+        models_over_time[index] = current_model
-        scalers[index] = current_scaler
+        for transformation_index, transformation in enumerate(current_transformations):
+            transformations_over_time[transformation_index][index] = transformation
 
         iterations_before_retrain -= 1
 
-    return models, scalers
+    return models_over_time, transformations_over_time
 
 def walk_forward_inference(
                     model_name: str,
-                    models: pd.Series,
+                    model_over_time: pd.Series,
-                    scalers: pd.Series,
+                    transformations_over_time: list[pd.Series],
                     X: pd.DataFrame,
                     expanding_window: bool,
                     window_size: int,
@@ -84,32 +82,33 @@ def walk_forward_inference(
     predictions = pd.Series(index=X.index).rename(model_name)
     probabilities = pd.DataFrame(index=X.index)
 
-    first_nonzero_return = get_first_valid_return_index(models)
+    inference_from = get_first_valid_return_index(model_over_time)
-    train_from = first_nonzero_return
+    inference_till = X.shape[0]
-    train_till = X.shape[0]
+    first_model = model_over_time[inference_from]
-    first_model = models[first_nonzero_return]
 
     if first_model.only_column is not None:
         X = X[[column for column in X.columns if first_model.only_column in column]]
+    
+    if first_model.data_transformation == 'original':
+        transformations_over_time = []
 
-    is_scaling_on = first_model.data_scaling == 'scaled'
+    for index in tqdm(range(inference_from, inference_till)):
+        
+        train_window_start = inference_from if expanding_window else index - window_size - 1
 
-    for index in tqdm(range(train_from, train_till)):
+        current_model = model_over_time[index]
-        if expanding_window:
+        current_transformations = [transformation_over_time[index] for transformation_over_time in transformations_over_time]
-            train_window_start = first_nonzero_return
-        else:
-            train_window_start = index - window_size - 1
-
-        current_model = models[index]
-        curren_scaler = scalers[index]
 
         if current_model.predict_window_size == 'window_size': 
-            next_timestep = X.iloc[train_window_start:index].to_numpy()#.reshape(1, -1)
+            next_timestep = X.iloc[train_window_start:index]
         else: 
-            next_timestep = X.iloc[index].to_numpy().reshape(1, -1)
+            # we need to get a Dataframe out of it, since the transformation step always expects a 2D array, but it's equivalent to X.iloc[index]
+            next_timestep = X.iloc[index:index+1]
         
-        if is_scaling_on:
+        for transformation in current_transformations:
-            next_timestep = curren_scaler.transform(next_timestep)
+            next_timestep = transformation.transform(next_timestep)
+
+        next_timestep = next_timestep.to_numpy()
 
         prediction, probs = current_model.predict(next_timestep)
         predictions[index] = prediction

transformations/base.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+from typing import Literal, Optional, Union
+from abc import ABC, abstractmethod
+import pandas as pd
+
+class Transformation(ABC):
+
+    @abstractmethod
+    def fit(self, X: pd.DataFrame, y: Optional[pd.Series]) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def fit_transform(self, X: pd.DataFrame, y: Optional[pd.Series] = None) -> pd.DataFrame:
+        raise NotImplementedError
+
+    @abstractmethod
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        raise NotImplementedError
+
+    @abstractmethod
+    def clone(self) -> Transformation:
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_name(self) -> str:
+        raise NotImplementedError
+    
+
+
+    
+    

transformations/sklearn.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+from transformations.base import Transformation
+from typing import Literal, Optional, Union
+from sklearn.base import clone, BaseEstimator
+import pandas as pd
+
+class SKLearnTransformation(Transformation):
+
+    transformer: BaseEstimator
+
+    def __init__(self, transformer: BaseEstimator):
+        self.transformer = transformer
+
+    def fit(self, X: pd.DataFrame, y: Optional[pd.Series] = None) -> None:
+        self.transformer.fit(X, y)
+    
+    def fit_transform(self, X: pd.DataFrame, y: Optional[pd.Series]) -> pd.DataFrame:
+        self.fit(X, y)
+        return self.transform(X)
+
+    def transform(self, X: pd.DataFrame) -> pd.DataFrame:
+        return pd.DataFrame(self.transformer.transform(X), index = X.index, columns = X.columns)
+
+    def clone(self) -> SKLearnTransformation:
+        return SKLearnTransformation(clone(self.transformer))
+
+    def get_name(self) -> str:
+        return self.transformer.__class__.__name__
+
+    
+
+
+    
+    

utils/scaler.py

@@ -1,13 +1,13 @@
 from sklearn.preprocessing import MinMaxScaler, Normalizer, StandardScaler
-from typing import Union
+from transformations.sklearn import SKLearnTransformation
 from utils.types import ScalerTypes
 
-def get_scaler(type: ScalerTypes) -> Union[MinMaxScaler, Normalizer, StandardScaler]:
+def get_scaler(type: ScalerTypes) -> SKLearnTransformation:
     if type == 'normalize':
-        return Normalizer()
+        return SKLearnTransformation(Normalizer())
     elif type == 'minmax':
-        return MinMaxScaler(feature_range= (-1, 1))
+        return SKLearnTransformation(MinMaxScaler(feature_range= (-1, 1)))
     elif type == 'standardize':
-        return StandardScaler()
+        return SKLearnTransformation(StandardScaler())
     else:
         raise Exception("Scaler type not supported")