Feature: Added sweep functionality (#65)

* feat: Parametricized model selection works now.

* feat: Fixed errors. Sweep generates and you can run it, but it gives an error for model.only_columns attribute.

* feat: Factored the wandb management, default config managment and the model_dictionary out of the run_pipeline to a seperate file.

* fix: Took out prints and fixed the mismatch of ensemble models when classifing.

* fix(Models): added StaticMomentum model to the dictionary, hopefully fixed sklearn-ex RandomForestRegressor problem

* fix(Dependencies): pin scikit-learn-ex's version, moved map_model_name_to_function to `models`

* feat(Sweep): added `run_sweep.py` shortcut

* feat(Pipeline): skip training a meta model if array is empty

Co-authored-by: Mark Aron Szulyovszky <mark.szulyovszky@gmail.com>

default_config.py

@@ -0,0 +1,38 @@
+from utils.load_data import get_crypto_assets
+import feature_extractors.feature_extractor_presets as feature_extractor_presets
+from models.model_map import model_names_classification, model_names_regression
+
+def get_default_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/',
+        all_assets = get_crypto_assets('data/'),
+        load_other_assets= False,
+        log_returns= True,
+        forecasting_horizon = 1,
+        own_features= feature_extractor_presets.date + feature_extractor_presets.level1,
+        other_features= [],
+        index_column= 'int',
+        method= 'classification',
+    )
+
+    # regression_models = ["Lasso", "Ridge", "BayesianRidge", "KNN", "AB", "LR", "MLP", "RF", "SVR"]
+    regression_models = model_names_regression
+    regression_ensemble_models = ['Ensemble_Average']
+    # classification_models = ["LR", "LDA", "KNN", "CART", "NB", "AB", "RF", "StaticMom"]
+    classification_models = model_names_classification        
+    classification_ensemble_models = ['Ensemble_Average']
+
+    model_config = dict(
+        level_1_models = regression_models if data_config['method'] == 'regression' else classification_models,
+        level_2_models = regression_ensemble_models if data_config['method'] == 'regression' else classification_ensemble_models
+    )
+    
+    return model_config, training_config, data_config

environment.yml

@@ -6,7 +6,7 @@ channels:
 dependencies:
   - python=3.9
   - seaborn
-  - scikit-learn-intelex
+  - scikit-learn-intelex=2021.4.0
   - ipython
   - scipy
   - scikit-learn

models/model_map.py

@@ -0,0 +1,58 @@
+from sklearn.linear_model import LinearRegression, Lasso, BayesianRidge, LogisticRegression, Ridge
+from sklearn.tree import DecisionTreeClassifier
+from sklearnex.neighbors import KNeighborsRegressor, KNeighborsClassifier
+from sklearn.discriminant_analysis import LinearDiscriminantAnalysis
+from sklearnex.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, ExtraTreesClassifier
+from sklearnex.ensemble import RandomForestClassifier
+from models.base import SKLearnModel
+from models.momentum import StaticMomentumModel
+from models.average import StaticAverageModel
+from models.naive import StaticNaiveModel
+
+
+model_map = {
+    "regression_models": dict(
+        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)),
+        StaticNaive = StaticNaiveModel(),
+    ),
+    "classification_models": dict(
+        LR= SKLearnModel(LogisticRegression(n_jobs=-1)),
+        LDA= SKLearnModel(LinearDiscriminantAnalysis()),
+        KNN= SKLearnModel(KNeighborsClassifier()),
+        CART= SKLearnModel(DecisionTreeClassifier()),
+        NB= SKLearnModel(GaussianNB()),
+        AB= SKLearnModel(AdaBoostClassifier()),
+        RF= SKLearnModel(RandomForestClassifier(n_jobs=-1)),
+        StaticMom= StaticMomentumModel(allow_short=True),
+    ),     
+    "classification_ensemble_models": dict(
+        Ensemble_CART = SKLearnModel(DecisionTreeClassifier()),
+        Ensemble_Average = StaticAverageModel(),
+    ),
+    "regression_ensemble_models": dict(
+        Ensemble_Ridge = SKLearnModel(Ridge(alpha=0.1)),
+        Ensemble_Average = StaticAverageModel(),
+    )
+}
+
+model_names_classification = model_map["classification_models"].keys()
+model_names_regression = model_map["regression_models"].keys()
+
+
+def map_model_name_to_function(model_config:dict, method:str) -> dict:
+    for level in ['level_1_models', 'level_2_models']:
+        model_category = method + '_models' if level=='level_1_models' else method + '_ensemble_models'
+        model_config[level] = [(model_name, model_map[model_category][model_name]) for model_name in  model_config[level]]
+
+    return model_config

run_pipeline.py

@@ -1,115 +1,23 @@
-from sklearnex import patch_sklearn
+from utils.load_data import load_data
-patch_sklearn()
-
-from utils.load_data import get_crypto_assets, get_etf_assets, load_data
-
 import pandas as pd
-import numpy as np
+from training.training import run_single_asset_trainig
-from sklearn.linear_model import LinearRegression, Lasso, BayesianRidge, LogisticRegression, Ridge
+from utils.launch_wandb import launch_wandb, seperate_configs
-from sklearn.tree import DecisionTreeClassifier
+from models.model_map import map_model_name_to_function
-from sklearn.neighbors import KNeighborsRegressor, KNeighborsClassifier
+from default_config import get_default_config
-from sklearn.discriminant_analysis import LinearDiscriminantAnalysis
-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
-from models.average import StaticAverageModel
-from models.naive import StaticNaiveModel
 
-import feature_extractors.feature_extractor_presets as feature_extractor_presets
+def setup_pipeline(project_name:str, with_wandb: bool, sweep: bool):
-from training.pipeline import run_single_asset_trainig_pipeline
+    model_config, training_config, data_config = get_default_config()
-
-
-
-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/',
-        all_assets = get_crypto_assets('data/'),
-        load_other_assets= False,
-        log_returns= True,
-        forecasting_horizon = 1,
-        own_features= feature_extractor_presets.date + feature_extractor_presets.level1,
-        other_features= [],
-        index_column= 'int',
-        method= 'classification',
-    )
-
-    regression_models = [
-        # ('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 - Average', StaticAverageModel())]
-    # regression_ensemble_model = [('Ensemble - Ridge', SKLearnModel(Ridge(alpha=0.1)))]
-
-    classification_models = [
-        ('LR', SKLearnModel(LogisticRegression(n_jobs=-1))),
-        ('LDA', SKLearnModel(LinearDiscriminantAnalysis())),
-        ('KNN', SKLearnModel(KNeighborsClassifier())),
-        ('CART', SKLearnModel(DecisionTreeClassifier())),
-        ('StaticMomentum', StaticMomentumModel(allow_short=True)),
-        # ('StaticNaive', StaticNaiveModel()),
-        # ('NB', SKLearnModel(GaussianNB())),
-        # ('AB', SKLearnModel(AdaBoostClassifier())),
-        # ('RF', SKLearnModel(RandomForestClassifier(n_jobs=-1)))
-    ]
-    classification_ensemble_model = [('Ensemble - Average', StaticAverageModel())]
-    # classification_ensemble_model = [('Ensemble - CART', SKLearnModel(DecisionTreeClassifier()))]
-
-    model_config = dict(
-        level_1_models = regression_models if data_config['method'] == 'regression' else classification_models,
-        level_2_model = regression_ensemble_model if data_config['method'] == 'regression' else classification_ensemble_model,
-    )
-    return model_config, training_config, data_config
-
-def launch_wandb(config, sweep=False):
-        from wandb_setup import get_wandb
-        wandb = get_wandb()   
-        
-        if type(wandb) == type(None): 
-            return None
-        elif sweep:
-            wandb.init(project="price-forecasting",  config = config)             
-            return wandb
-        else:
-            wandb.init(project="price-forecasting", config=config, reinit=True)
-            return wandb
-    
-
-def run_pipeline(with_wandb: bool, sweep: bool):
-    model_config, training_config, data_config = get_config()
     
     wandb = None
     if with_wandb: 
-        wandb = launch_wandb(dict(**model_config, **training_config, **data_config), sweep)
+        wandb = launch_wandb(project_name=project_name, default_config=dict(**model_config, **training_config, **data_config), sweep=sweep)
-        
+        model_config, training_config, data_config = seperate_configs(wandb, model_config, training_config, data_config)
-        if type(wandb) is not type(None):
+
-            for k in training_config: training_config[k] = wandb.config[k]
+    model_config = map_model_name_to_function(model_config, data_config['method'])
-        # for k in model_config: model_config[k] = wandb.config[k]
+    pipeline(project_name, wandb, sweep, model_config, training_config, data_config)  
-        # for k in data_config: data_config[k] = wandb.config[k]
-            
-    pipeline(model_config, training_config, data_config, wandb)  
     
 
-# Run pipeline
+def pipeline(project_name:str, wandb, sweep:bool, model_config:dict, training_config:dict, data_config:dict ):
-
-def pipeline(model_config:dict, training_config:dict, data_config:dict, wandb):
     results = pd.DataFrame()
 
     for asset in data_config['all_assets']:
@@ -123,7 +31,7 @@ def pipeline(model_config:dict, training_config:dict, data_config:dict, wandb):
         X, y, target_returns = load_data(**data_params)
 
         # 2. Train Level-1 models
-        current_result, current_predictions = run_single_asset_trainig_pipeline(
+        current_result, current_predictions = run_single_asset_trainig(
             ticker_to_predict = asset,
             X = X,
             y = y,
@@ -133,32 +41,37 @@ def pipeline(model_config:dict, training_config:dict, data_config:dict, wandb):
             sliding_window_size = training_config['sliding_window_size'],
             retrain_every =  training_config['retrain_every'],
             scaler =  training_config['scaler'],
-            wandb = wandb
+            wandb = wandb,
+            project_name=project_name,
+            sweep=sweep
         )
         results = pd.concat([results, current_result], axis=1)
         all_predictions = pd.concat([all_predictions, current_predictions], axis=1)
 
-        # 3. Train Level-2 (Ensemble) model
+        if len(model_config['level_2_models']) > 0: 
-        ensemble_X = all_predictions
+            # 3. Train Level-2 (Ensemble) model
-        if training_config['include_original_data_in_ensemble']:
+            
-            ensemble_X = pd.concat([ensemble_X, X], axis=1)
+            ensemble_X = all_predictions
+            if training_config['include_original_data_in_ensemble']:
+                ensemble_X = pd.concat([ensemble_X, X], axis=1)
 
-        ensemble_result, ensemble_preds = run_single_asset_trainig_pipeline(
+            ensemble_result, ensemble_preds = run_single_asset_trainig(
-            ticker_to_predict = asset,
+                ticker_to_predict = asset,
-            X = ensemble_X,
+                X = ensemble_X,
-            y = y,
+                y = y,
-            target_returns = target_returns,
+                target_returns = target_returns,
-            models = model_config['level_2_model'],
+                models = model_config['level_2_models'],
-            method = data_config['method'],
+                method = data_config['method'],
-            sliding_window_size = training_config['sliding_window_size'],
+                sliding_window_size = training_config['sliding_window_size'],
-            retrain_every = training_config['retrain_every'],
+                retrain_every = training_config['retrain_every'],
-            scaler = training_config['scaler'],
+                scaler = training_config['scaler'],
-            wandb = wandb
+                wandb = wandb,
-        )
+                project_name=project_name,
-
+                sweep=sweep
-        results = pd.concat([results, ensemble_result], axis=1)
+            )
-        all_predictions = pd.concat([all_predictions, ensemble_preds], axis=1)
 
+            results = pd.concat([results, ensemble_result], axis=1)
+            all_predictions = pd.concat([all_predictions, ensemble_preds], axis=1)
 
     results.to_csv('results.csv')
 
@@ -168,7 +81,9 @@ def pipeline(model_config:dict, training_config:dict, data_config:dict, wandb):
     print("Mean Sharpe ratio for Level-1 models: ", level1_columns.loc['sharpe'].mean())
     print("Mean Sharpe ratio for Level-2 (Ensemble) models: ", ensemble_columns.loc['sharpe'].mean())
 
-    
+    if sweep:
+        if wandb.run is not None:
+            wandb.finish()
     
 if __name__ == '__main__':
-    run_pipeline(with_wandb = False, sweep = False)
+    setup_pipeline(project_name='price-prediction', with_wandb = False, sweep = False)

run_sweep.py

@@ -0,0 +1,3 @@
+from run_pipeline import setup_pipeline
+
+setup_pipeline(project_name='price-prediction', with_wandb = True, sweep = True)

sweep.yaml

@@ -1,28 +1,49 @@
-program: rnn_sweep.py
+program: run_pipeline.py
-method: bayes
+method: grid
-project: integer-sequence
+project: price-forecasting
 name: Finding best hyperparameters for price prediction
-early_terminate:
+# early_terminate:
-  type: hyperband
+#   type: hyperband
-  min_iter: 2000
+#   min_iter: 2000
-metric:
+# metric:
-  goal: maximize
+#   goal: maximize
-  name: sharpe
+#   name: sharpe
 parameters:
-    path : 'data/'
+    path : 
+      value: 'data/'
     sliding_window_size: 
       values: [50, 90, 130, 160, 180, 280, 380, 500]
       distribution: categorical
     retrain_every: 
       values: [7, 14, 30, 60, 100]
+      distribution: categorical
     scaler:
       values: ['minmax', 'normalize', 'minmax', 'standardize', 'none']
+      distribution: categorical
     include_original_data_in_ensemble:
-      values: [True, False]
+      value: True
     method: 
-      values: ['classification', 'regression']
+      value: 'classification'
     forecasting_horizon: 
       values: [1,2,3,4,5,6,7,8,9,10]
+      distribution: categorical
+    load_other_assets: 
+      value: False
+    log_returns: 
+      value: True
+    own_features: 
+      value: []
+    other_features: 
+      value: []
+    index_column: 
+      value: 'int'
+    level_1_models: 
+      value: ["LR", "LDA", "KNN", "CART", "NB", "AB", "RF"]
+      distribution: constant
+    level_2_models: 
+      value: ['Ensemble_CART']
+      distribution: constant
+
 
 
 

training/training.py

@@ -15,7 +15,7 @@ def __get_scaler(type: Literal['normalize', 'minmax', 'standardize', 'none']):
     else:
         return None
 
-def run_single_asset_trainig_pipeline(
+def run_single_asset_trainig(
                     ticker_to_predict: str,
                     X: pd.DataFrame,
                     y: pd.Series,
@@ -25,7 +25,9 @@ def run_single_asset_trainig_pipeline(
                     sliding_window_size: int,
                     retrain_every: int,
                     scaler: Literal['normalize', 'minmax', 'standardize', 'none'],
-                    wandb
+                    wandb,
+                    project_name:str,
+                    sweep:bool
                     ) -> tuple[pd.DataFrame, pd.DataFrame]:
 
 
@@ -59,14 +61,21 @@ def run_single_asset_trainig_pipeline(
         # column names for model outputs should be different, so we can differentiate between original data and model predictions later, where necessary
         predictions["model_" + column_name] = preds
         
-        if wandb_active:
+        if wandb_active and not sweep:
-            run = wandb.init(project="price-forecasting", config={"model_type": model_name, "ticker": ticker_to_predict}, reinit=True)
+                run = wandb.init(project=project_name, config={"model_type": model_name, "ticker": ticker_to_predict}, reinit=True)
-            wandb.run.name = ticker_to_predict + "-" + model_name+ "-" + wandb.run.id
+                wandb.run.name = ticker_to_predict + "-" + model_name+ "-" + wandb.run.id
-            wandb.run.save()
+                wandb.run.save()
             
-            for rownum,(indx,val) in enumerate(result.iteritems()):
+                for rownum,(indx,val) in enumerate(result.iteritems()):
-                run.log({"model_type": model_name, indx:val })
+                    run.log({"model_type": model_name, indx:val })
-                
+                    
-            run.finish()
+                run.finish()
+    
+    if wandb_active and sweep:
+        mean_results = results.mean()
+         
+        wandb.log({"model_type": 'avarage_model', 'results':results })
+        for rownum,(indx,val) in enumerate(mean_results.iteritems()):
+            wandb.log({"model_type": 'avarage_model', indx:val })
 
     return results, predictions

utils/launch_wandb.py

@@ -0,0 +1,27 @@
+
+
+def launch_wandb(project_name:str, default_config:dict, sweep:bool=False):
+    from wandb_setup import get_wandb
+    wandb = get_wandb()   
+    
+    if type(wandb) == type(None): 
+        return None
+    elif sweep:
+        wandb.init(project=project_name,  config = default_config)             
+        return wandb
+    else:
+        wandb.init(project=project_name, config = default_config, reinit=True)
+        return wandb
+    
+    
+def seperate_configs(wandb, model_config:dict, training_config:dict, data_config:dict) -> tuple[dict,dict,dict]:
+    config:dict = wandb.config
+
+    if type(wandb) is not type(None):
+        for k in training_config: training_config[k] = config[k]
+        for k in model_config: model_config[k] = config[k]
+        # for k in data_config: data_config[k] = config[k]
+
+    return model_config, training_config, data_config
+
+