feat(Pipeline): added multi-asset pipeline, ensembling_method to config

config/preprocess.py

@@ -10,10 +10,11 @@ from labeling.labellers_map import labellers_map
 from models.sklearn import SKLearnModel
 from sklearn.ensemble import VotingClassifier
 from transformations.retrieve import get_pca, get_rfe, get_scaler
+from copy import deepcopy
 
 
 def preprocess_config(raw_config: RawConfig) -> Config:
-    config_dict = vars(raw_config)
+    config_dict = vars(deepcopy(raw_config))
     config_dict = __preprocess_model_config(config_dict)
     config_dict = __preprocess_feature_extractors_config(config_dict)
     config_dict = __preprocess_data_collections_config(config_dict)
@@ -39,21 +40,36 @@ def __preprocess_feature_extractors_config(data_dict: dict) -> dict:
 
 
 def __preprocess_model_config(model_config: dict) -> dict:
+    def map_ensembling_method(method: str) -> str:
+        if method == "voting_soft":
+            return "soft"
+        elif method == "voing_hard":
+            return "hard"
+        else:
+            raise Exception(f"Unknown ensembling method: {method}")
+
     directional_models = [
         get_model(model_name) for model_name in model_config["directional_models"]
     ]
     model_config.pop("directional_models")
     model_config["directional_model"] = SKLearnModel(
-        VotingClassifier([(m.name, m) for m in directional_models], voting="soft")
+        VotingClassifier(
+            [(m.name, m) for m in directional_models],
+            voting=map_ensembling_method(model_config["ensembling_method"]),
+        )
     )
     if len(model_config["meta_models"]) > 0:
         meta_models = [
             get_model(model_name) for model_name in model_config["meta_models"]
         ]
         model_config["meta_model"] = SKLearnModel(
-            VotingClassifier([(m.name, m) for m in meta_models], voting="soft")
+            VotingClassifier(
+                [(m.name, m) for m in meta_models],
+                voting=map_ensembling_method(model_config["ensembling_method"]),
+            )
         )
     model_config.pop("meta_models")
+    model_config.pop("ensembling_method")
 
     return model_config
 

config/presets.py

@@ -33,4 +33,5 @@ def get_default_config() -> RawConfig:
         labeling="two_class",
         forecasting_horizon=50,
         save_models=False,
+        ensembling_method="voting_soft",
     )

config/types.py

@@ -29,6 +29,7 @@ class RawConfig(BaseModel):
     labeling: Literal["two_class", "three_class_balanced", "three_class_imbalanced"]
     forecasting_horizon: int
     save_models: bool
+    ensembling_method: Literal["voting_soft", "voting_hard"]
 
     directional_models: list[str]
     meta_models: list[str]

data_loader/collections.py

@@ -8,30 +8,8 @@ def transform_to_data_collection(path: str, file_names: list[str]) -> DataCollec
 
 __daily_etf = ["GLD", "IEF", "QQQ", "SPY", "TLT"]
 
-__daily_crypto = [
-    "ADA_USD",
-    "BCH_USD",
-    "BNB_USD",
-    "BTC_USD",
-    "DOT_USD",
-    "ETC_USD",
-    "ETH_USD",
-    "FIL_USD",
-    "LTC_USD",
-    "SOL_USD",
-    "THETA_USD",
-    "TRX_USD",
-    "UNI_USD",
-    "XLM_USD",
-    "XRP_USD",
-    "XTZ_USD",
-]
-
-__daily_crypto_lightweight = ["ADA_USD", "BCH_USD"]
-
 __5min_crypto = [
     "BTC_USD",
-    "DASH_USD",
     "ETC_USD",
     "ETH_USD",
     "LTC_USD",
@@ -70,11 +48,6 @@ __daily_glassnode = [
 
 
 data_collections = dict(
-    daily_only_btc=transform_to_data_collection("data/daily_crypto", ["BTC_USD"]),
-    daily_crypto=transform_to_data_collection("data/daily_crypto", __daily_crypto),
-    daily_crypto_lightweight=transform_to_data_collection(
-        "data/daily_crypto", __daily_crypto_lightweight
-    ),
     daily_etf=transform_to_data_collection("data/daily_etf", __daily_etf),
     fivemin_crypto=transform_to_data_collection("data/5min_crypto", __5min_crypto),
     daily_glassnode=transform_to_data_collection(

run_multi_asset_pipeline.py

@@ -0,0 +1,33 @@
+from run_pipeline import run_training, setup_config
+from config.types import RawConfig
+from data_loader.collections import data_collections
+from reporting.reporting import report_results
+from config.presets import get_default_config
+
+
+def run_multi_asset_pipeline(
+    project_name: str, with_wandb: bool, sweep: bool, raw_config: RawConfig
+):
+    collection = data_collections["fivemin_crypto"]
+    for asset in collection:
+        print(f"# Predicting asset: {asset[1]}\n")
+        raw_config.target_asset = asset[1]
+        wandb, config = setup_config(project_name, with_wandb, sweep, raw_config)
+        pipeline_outcome = run_training(config)
+        report_results(
+            pipeline_outcome.directional_training.training.stats,
+            pipeline_outcome.get_output_stats(),
+            pipeline_outcome.get_output_weights(),
+            config,
+            wandb,
+            sweep,
+        )
+
+
+if __name__ == "__main__":
+    run_multi_asset_pipeline(
+        project_name="price-prediction",
+        with_wandb=False,
+        sweep=False,
+        raw_config=get_default_config(),
+    )

run_pipeline.py

@@ -25,8 +25,8 @@ ray.init()
 def run_pipeline(
     project_name: str, with_wandb: bool, sweep: bool, raw_config: RawConfig
 ) -> tuple[PipelineOutcome, Config]:
-    wandb, config = __setup_config(project_name, with_wandb, sweep, raw_config)
-    pipeline_outcome = __run_training(config)
+    wandb, config = setup_config(project_name, with_wandb, sweep, raw_config)
+    pipeline_outcome = run_training(config)
     report_results(
         pipeline_outcome.directional_training.training.stats,
         pipeline_outcome.get_output_stats(),
@@ -40,7 +40,7 @@ def run_pipeline(
     return pipeline_outcome, config
 
 
-def __setup_config(
+def setup_config(
     project_name: str, with_wandb: bool, sweep: bool, raw_config: RawConfig
 ) -> tuple[Optional[object], Config]:
     wandb = None
@@ -54,7 +54,7 @@ def __setup_config(
     return wandb, config
 
 
-def __run_training(config: Config) -> PipelineOutcome:
+def run_training(config: Config) -> PipelineOutcome:
 
     print("---> Load data, check for validity")
     X, returns = load_data(