feat(Model): added own Model class, SkLearnModel wrapper and StaticMomentumModel (#61)

* feat(Model): added own `Model` class, SkLearnModel wrapper and StaticMomentumModel

* fix(Tests): added missing Model variable

* fix(Tests): added missing clone method()

models/base.py

@@ -0,0 +1,38 @@
+
+from typing import Literal, Optional
+from sklearn.base import clone
+
+class Model:
+
+    # data_format: Literal['dataframe', 'numpy'] 
+    data_scaling: Literal["scaled", "unscaled"]
+    # data_format: Literal["wide", "narrow"]
+    only_column: Optional[str]
+
+    def fit(self, X, y):
+        pass
+
+    def predict(self, X):
+        pass
+
+    def clone(self):
+        pass
+
+
+class SKLearnModel(Model):
+
+    # data_format = 'numpy'
+    data_scaling = 'scaled'
+    only_column = None
+
+    def __init__(self, model):
+        self.model = model
+
+    def fit(self, X, y):
+        self.model.fit(X, y)
+
+    def predict(self, X):
+        return self.model.predict(X)
+    
+    def clone(self):
+        return SKLearnModel(clone(self.model))

models/momentum.py

@@ -0,0 +1,27 @@
+from models.base import Model
+import numpy as np
+
+class StaticMomentumModel(Model):
+    '''
+    Model that uses only one feature: momentum. It's positive if momentum is greater than 0, otherwise it's negative.
+    '''
+
+    data_format = 'dataframe'
+    data_scaling = 'unscaled'
+    only_column = 'mom'
+
+    def __init__(self, allow_short: bool) -> None:
+        super().__init__()
+        self.allow_short = allow_short
+
+    def fit(self, X, y):
+        # This is a static model, it can' learn anything
+        pass
+
+    def predict(self, X):
+        negative_class = -1.0 if self.allow_short == True else 0.0
+        prediction = 1.0 if X[-1][0] > 0 else negative_class
+        return np.array([prediction])
+
+    def clone(self):
+        return self

run_pipeline.py

@@ -13,21 +13,22 @@ from sklearn.svm import SVR
 from sklearn.naive_bayes import GaussianNB
 from sklearn.neural_network import MLPRegressor, MLPClassifier
 from sklearn.ensemble import AdaBoostRegressor, RandomForestRegressor, ExtraTreesRegressor, AdaBoostClassifier, GradientBoostingClassifier, RandomForestClassifier, ExtraTreesClassifier
+from models.base import SKLearnModel
+from models.momentum import StaticMomentumModel
 
 import feature_extractors.feature_extractor_presets as feature_extractor_presets
 from training.pipeline import run_single_asset_trainig_pipeline
 
-from typing import Tuple
 
 
-def get_config()->Tuple[dict, dict, dict]:
+def get_config() -> tuple[dict, dict, dict]:
 
     training_config = dict(
         sliding_window_size = 150,
         retrain_every = 20,
         scaler = 'minmax', # 'normalize' 'minmax' 'standardize' 'none'
         include_original_data_in_ensemble = True,
-        )
+    )
 
     data_config = dict(
         path='data/',
@@ -41,30 +42,30 @@ def get_config()->Tuple[dict, dict, dict]:
         method= 'classification',
     )
 
-    classification_models = [
-        ('LR', LogisticRegression(n_jobs=-1)),
-        # ('LDA', LinearDiscriminantAnalysis()),
-        ('KNN', KNeighborsClassifier()),
-        # ('CART', DecisionTreeClassifier()),
-        # ('NB', GaussianNB()),
-        # ('AB', AdaBoostClassifier()),
-        # ('RF', RandomForestClassifier(n_jobs=-1))
-    ]
-
     regression_models = [
-        # ('Lasso', Lasso(alpha=0.1, max_iter=1000)),
-        ('Ridge', Ridge(alpha=0.1)),
-        ('BayesianRidge', BayesianRidge()),
-        # ('KNN', KNeighborsRegressor(n_neighbors=25)),
-        # ('AB', AdaBoostRegressor(random_state=1)),
-        # ('LR', LinearRegression(n_jobs=-1)),
-        # ('MLP', MLPRegressor(hidden_layer_sizes=(100,20), max_iter=1000)),
-        # ('RF', RandomForestRegressor(n_jobs=-1)),
-        # ('SVR', SVR(kernel='rbf', C=1e3, gamma=0.1))
+        # ('Lasso', SKLearnModel(Lasso(alpha=0.1, max_iter=1000))),
+        ('Ridge', SKLearnModel(Ridge(alpha=0.1))),
+        ('BayesianRidge', SKLearnModel(BayesianRidge())),
+        # ('KNN', SKLearnModel(KNeighborsRegressor(n_neighbors=25))),
+        # ('AB', SKLearnModel(AdaBoostRegressor(random_state=1))),
+        # ('LR', SKLearnModel(LinearRegression(n_jobs=-1))),
+        # ('MLP', SKLearnModel(MLPRegressor(hidden_layer_sizes=(100,20), max_iter=1000))),
+        # ('RF', SKLearnModel(RandomForestRegressor(n_jobs=-1))),
+        # ('SVR', SKLearnModel(SVR(kernel='rbf', C=1e3, gamma=0.1)))
     ]
+    regression_ensemble_model = [('Ensemble - Ridge', SKLearnModel(Ridge(alpha=0.1)))]
 
-    regression_ensemble_model = [('Ensemble - Ridge', Ridge(alpha=0.1))]
-    classification_ensemble_model = [('Ensemble - CART', DecisionTreeClassifier())]
+    classification_models = [
+        ('LR', SKLearnModel(LogisticRegression(n_jobs=-1))),
+        # ('LDA', SKLearnModel(LinearDiscriminantAnalysis())),
+        # ('KNN', SKLearnModel(KNeighborsClassifier())),
+        # ('CART', SKLearnModel(DecisionTreeClassifier())),
+        ('StaticMomentum', StaticMomentumModel(allow_short=True))
+        # ('NB', SKLearnModel(GaussianNB())),
+        # ('AB', SKLearnModel(AdaBoostClassifier())),
+        # ('RF', SKLearnModel(RandomForestClassifier(n_jobs=-1)))
+    ]
+    classification_ensemble_model = [('Ensemble - CART', SKLearnModel(DecisionTreeClassifier()))]
 
     model_config = dict(
         level_1_models = regression_models if data_config['method'] == 'regression' else classification_models,

tests/test_evaluation.py

@@ -2,7 +2,7 @@
 import numpy as np
 import pandas as pd
 from training.walk_forward import walk_forward_train_test
-from sklearn.base import BaseEstimator
+from models.base import Model
 from utils.evaluate import evaluate_predictions
 
 no_of_rows = 100
@@ -29,12 +29,15 @@ def __generate_even_odd_test_data(no_of_rows) -> tuple[pd.DataFrame, pd.Series]:
 
     return X, y
 
-class EvenOddStubModel(BaseEstimator):
+class EvenOddStubModel(Model):
     '''
     A deteministic model that can predict the future with 100% accuracy
     It verifies that the X[n][any_column] == 1 if n is even,
     '''
 
+    data_scaling = "unscaled"
+    only_column = None
+
     def __init__(self, window_length) -> None:
         super().__init__()
         self.window_length = window_length
@@ -47,6 +50,9 @@ class EvenOddStubModel(BaseEstimator):
     def predict(self, X):
         return np.array([-1 if X[0][0] == 1 else 1])
 
+    def clone(self):
+        return self
+
 
 def test_evaluation():
     X, y = __generate_even_odd_test_data(no_of_rows)

tests/test_walk_forward.py

@@ -1,7 +1,7 @@
 import numpy as np
 import pandas as pd
 from training.walk_forward import walk_forward_train_test
-from sklearn.base import BaseEstimator
+from models.base import Model
 
 no_of_rows = 100
 
@@ -27,12 +27,15 @@ def __generate_incremental_test_data(no_of_rows) -> tuple[pd.DataFrame, pd.Serie
 
 
 
-class IncrementingStubModel(BaseEstimator):
+class IncrementingStubModel(Model):
     '''
     A deteministic model that can predict the future with 100% accuracy
     It verifies that the X[n][any_column]+1 == y[n]
     '''
 
+    data_scaling = "unscaled"
+    only_column = None
+
     def __init__(self, window_length) -> None:
         super().__init__()
         self.window_length = window_length
@@ -45,6 +48,8 @@ class IncrementingStubModel(BaseEstimator):
     def predict(self, X):
         return np.array([X[0][0] + 1])
 
+    def clone(self):
+        return self
 
 def test_walk_forward_train_test():
     X, y = __generate_incremental_test_data(no_of_rows)

training/pipeline.py

@@ -3,7 +3,7 @@ from typing import Literal
 from training.walk_forward import walk_forward_train_test
 from sklearn.preprocessing import MinMaxScaler, Normalizer, StandardScaler
 from utils.evaluate import evaluate_predictions
-from utils.typing import SKLearnModel
+from models.base import Model
 
 def __get_scaler(type: Literal['normalize', 'minmax', 'standardize', 'none']):
     if type == 'normalize':
@@ -20,7 +20,7 @@ def run_single_asset_trainig_pipeline(
                     X: pd.DataFrame,
                     y: pd.Series,
                     target_returns: pd.Series,
-                    models: list[tuple[str, SKLearnModel]],
+                    models: list[tuple[str, Model]],
                     method: Literal['regression', 'classification'],
                     sliding_window_size: int,
                     retrain_every: int,

training/walk_forward.py

@@ -1,12 +1,13 @@
 import pandas as pd
-from sklearn.base import clone
-from utils.typing import SKLearnModel
+from models.base import Model
 import numpy as np
 from utils.helpers import get_first_valid_return_index
 
+from sklearn.base import clone
+
 def walk_forward_train_test(
                             model_name: str,
-                            model: SKLearnModel,
+                            model: Model,
                             X: pd.DataFrame,
                             y: pd.Series,
                             target_returns: pd.Series,
@@ -23,7 +24,12 @@ def walk_forward_train_test(
     train_till = len(y)
     iterations_before_retrain = 0
 
-    if scaler is not None:
+    if model.only_column is not None:
+        X = X[[column for column in X.columns if model.only_column in column]]
+
+    is_scaling_on = scaler is not None and model.data_scaling == 'scaled'
+
+    if is_scaling_on:
         scaler = clone(scaler)
 
     for index in range(train_from, train_till):
@@ -32,7 +38,7 @@ def walk_forward_train_test(
             train_window_start = index - window_size - 1
             train_window_end = index - 1
             
-            if scaler is not None:
+            if is_scaling_on:
                 # First we need to fit on the expanding window data slice
                 # This is our only way to avoid lookahead bia
                 X_expanding_window = X[first_nonzero_return:train_window_end]
@@ -41,12 +47,12 @@ def walk_forward_train_test(
             X_slice = X[train_window_start:train_window_end]
             y_slice = y[train_window_start:train_window_end]
 
-            if scaler is not None:
+            if is_scaling_on:
                 X_slice = scaler.transform(X_slice.values)
             else:
                 X_slice = X_slice.to_numpy()
 
-            current_model = clone(model)
+            current_model = model.clone()
             current_model.fit(X_slice, y_slice.to_numpy())
             iterations_before_retrain = retrain_every
         else:
@@ -55,7 +61,7 @@ def walk_forward_train_test(
         models[index] = current_model
 
         next_timestep = X.iloc[index].to_numpy().reshape(1, -1)
-        if scaler is not None:
+        if is_scaling_on:
             next_timestep = scaler.transform(next_timestep)
 
         prediction = current_model.predict(next_timestep).item()

utils/typing.py

@@ -1,12 +1,6 @@
-from typing import Protocol, Callable, Union
+from typing import Callable, Union
 import pandas as pd
 
-class SKLearnModel(Protocol):
-    def fit(self, X, y, sample_weight=None): ...
-    def predict(self, X): ...
-    def score(self, X, y, sample_weight=None): ...
-    def set_params(self, **params): ...
-
 
 Period = int
 IsLogReturn = bool