refactor(Types): added nested types for Reporting (#162)

models/saving.py

@@ -3,11 +3,11 @@ import datetime
 from typing import Optional, Union
 import os
 import warnings
-from utils.encapsulation import Asset
+from reporting.types import Reporting
 
 
 
-def save_models(all_models_for_all_assets: list[Asset], data_config:dict, training_config:dict) -> None:
+def save_models(all_models_for_all_assets: list[Reporting.Asset], data_config:dict, training_config:dict) -> None:
     dict_for_pickle = dict()
     dict_for_pickle['training_config'] = training_config
     dict_for_pickle['data_config'] = data_config

reporting/types.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+import pandas as pd
+from models.base import Model
+
+
+class Reporting:
+    def __init__(self):
+        self.results: pd.DataFrame = pd.DataFrame() 
+        self.all_predictions: pd.DataFrame = pd.DataFrame() 
+        self.all_probabilities: pd.DataFrame = pd.DataFrame() 
+        self.all_assets:list[Reporting.Asset] = []
+    
+    def get_results(self)->tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, list[Reporting.Asset]]:
+        return self.results, self.all_predictions, self.all_probabilities, self.all_assets
+    
+
+    class Single_Model:
+        def __init__(self, model_name: str, model_over_time: list[Model]):
+            self.model_name: str = model_name
+            self.model_over_time: list[Model] = model_over_time
+
+
+    class Training_Step: 
+        def __init__(self, level: str):
+            self.level: str = level
+            self.base: list[Reporting.Single_Model] = []
+            self.metalabeling: list[list[Reporting.Single_Model]] = []
+
+
+    class Asset():
+        def __init__(self, ticker: str, primary: Reporting.Training_Step, secondary: Reporting.Training_Step):
+            self.name: str = ticker
+            self.primary: Reporting.Training_Step = primary
+            self.secondary: Reporting.Training_Step = secondary 

run_pipeline.py

@@ -1,6 +1,5 @@
 import pandas as pd
 from typing import Callable, Optional
-from operator import itemgetter
 
 from data_loader.load_data import load_data
 from data_loader.process_data import process_data, check_data
@@ -15,13 +14,13 @@ from config.preprocess import validate_config, preprocess_config
 
 from training.training_steps import primary_step, secondary_step
 
-from utils.encapsulation import Reporting, Asset, Training_Step
+from reporting.types import Reporting
 
 import ray
 ray.init()
 
 
-def run_pipeline(project_name:str, with_wandb: bool, sweep: bool, get_config: Callable) -> tuple[list[Asset], dict, dict, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+def run_pipeline(project_name:str, with_wandb: bool, sweep: bool, get_config: Callable) -> tuple[list[Reporting.Asset], dict, dict, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
     wandb, model_config, training_config, data_config = __setup_config(project_name, with_wandb, sweep, get_config)
     reporting = __run_training(model_config, training_config, data_config) 
     results, all_predictions, all_probabilities, all_models_all_assets = reporting.get_results()
@@ -66,7 +65,7 @@ def __run_training(model_config:dict, training_config:dict, data_config:dict):
         training_step_secondary = secondary_step(X, y, original_X, X_pca, current_predictions, asset, target_returns, configs, reporting)
         
         # 4. Save the models
-        reporting.all_assets.append(Asset(ticker=asset[1], primary=training_step_primary, secondary=training_step_secondary))
+        reporting.all_assets.append(Reporting.Asset(ticker=asset[1], primary=training_step_primary, secondary=training_step_secondary))
       
     return reporting
     

training/inference.py

@@ -4,9 +4,9 @@ from data_loader.load_data import load_data
 from typing import Optional, Union
 import warnings
 
-from utils.encapsulation import Asset, Single_Model, Training_Step 
+from reporting.types import Reporting 
 
-def run_inference_pipeline(data_config:dict, training_config:dict, all_models_all_assets:list[Asset]):
+def run_inference_pipeline(data_config:dict, training_config:dict, all_models_all_assets:list[Reporting.Asset]):
     data_params = data_config.copy()
     data_params['target_asset'] = data_params['assets'][0]
     
@@ -20,7 +20,7 @@ def run_inference_pipeline(data_config:dict, training_config:dict, all_models_al
     return result
 
 
-def __inference(data:pd.DataFrame, primary_step:Union[Training_Step,None], secondary_step:Union[Training_Step,None]) -> pd.DataFrame:
+def __inference(data:pd.DataFrame, primary_step:Union[Reporting.Training_Step,None], secondary_step:Union[Reporting.Training_Step,None]) -> pd.DataFrame:
     assert primary_step is not None, "No primary models found. Cancelling Inference."
     
     data = __primary_models(data, primary_step)
@@ -32,7 +32,7 @@ def __inference(data:pd.DataFrame, primary_step:Union[Training_Step,None], secon
     return data
 
 
-def __select_models( data_params:dict, all_models_all_assets:list[Asset])-> tuple[Union[Training_Step,None], Union[Training_Step, None]]:
+def __select_models( data_params:dict, all_models_all_assets:list[Reporting.Asset])-> tuple[Union[Reporting.Training_Step,None], Union[Reporting.Training_Step, None]]:
     target_asset_name = data_params['target_asset'][1]
     primary_step, secondary_step, = None, None
     target_asset_models = next((x for x in all_models_all_assets if x.name == target_asset_name), None)

training/meta_labeling.py

@@ -5,7 +5,7 @@ from feature_selection.feature_selection import select_features
 import pandas as pd
 from models.model_map import default_feature_selector_regression, default_feature_selector_classification
 from models.base import Model
-from utils.encapsulation import Single_Model
+from reporting.types import Reporting
 
 
 def train_meta_labeling_model(
@@ -19,7 +19,7 @@ def train_meta_labeling_model(
                             model_config: dict,
                             training_config: dict,
                             model_suffix: str
-                        ) -> tuple[pd.Series, pd.Series, pd.DataFrame, list[Single_Model]]:
+                        ) -> tuple[pd.Series, pd.Series, pd.DataFrame, list[Reporting.Single_Model]]:
 
     
     discretize = discretize_threeway_threshold(0.33)

training/primary_model.py

@@ -5,8 +5,7 @@ from utils.evaluate import evaluate_predictions
 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
+from reporting.types import Reporting
 
 def train_primary_model(
                     ticker_to_predict: str,
@@ -23,12 +22,12 @@ def train_primary_model(
                     no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'],
                     level: str,
                     print_results: bool,
-    ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, list[Single_Model]]:
+    ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, list[Reporting.Single_Model]]:
 
     results = pd.DataFrame()
     predictions = pd.DataFrame(index=y.index)
     probabilities = pd.DataFrame(index=y.index)
-    all_models_single_asset:list[Single_Model] = []
+    all_models_single_asset: list[Reporting.Single_Model] = []
     
     
     
@@ -69,7 +68,7 @@ def train_primary_model(
         results[column_name] = result
         
 
-        all_models_single_asset.append(Single_Model(model_name=column_name, model_over_time=model_over_time.tolist()))
+        all_models_single_asset.append(Reporting.Single_Model(model_name=column_name, model_over_time=model_over_time.tolist()))
   
         # column names for model outputs should be different, so we can differentiate between original data and model predictions later, where necessary
         predictions[column_name] = preds

training/training_steps.py

@@ -4,11 +4,20 @@ from operator import itemgetter
 from training.primary_model import train_primary_model
 from training.meta_labeling import train_meta_labeling_model
 
-from utils.encapsulation import Reporting, Asset, Single_Model, Training_Step
+from reporting.types import Reporting
 
 
-def primary_step(X: pd.DataFrame, y:pd.Series, original_X:pd.DataFrame, X_pca:pd.DataFrame, asset:list, target_returns:pd.Series, configs: dict, reporting: Reporting) -> tuple[Training_Step, pd.DataFrame]:
-    training_step = Training_Step(level='primary')
+def primary_step(
+                X: pd.DataFrame,
+                y:pd.Series,
+                original_X:pd.DataFrame,
+                X_pca:pd.DataFrame,
+                asset:list,
+                target_returns:pd.Series,
+                configs: dict,
+                reporting: Reporting
+                ) -> tuple[Reporting.Training_Step, pd.DataFrame]:
+    training_step = Reporting.Training_Step(level='primary')
     model_config, training_config, data_config = itemgetter('model_config', 'training_config', 'data_config')(configs)
 
     # 3. Train Primary models
@@ -60,8 +69,18 @@ def primary_step(X: pd.DataFrame, y:pd.Series, original_X:pd.DataFrame, X_pca:pd
     return training_step, current_predictions
 
 
-def secondary_step(X:pd.DataFrame, y:pd.Series, original_X:pd.DataFrame, X_pca:pd.DataFrame, current_predictions:pd.DataFrame, asset:list, target_returns:pd.Series, configs: dict, reporting: Reporting) -> Training_Step:
-    training_step = Training_Step(level='secondary')
+def secondary_step(
+                X:pd.DataFrame,
+                y:pd.Series,
+                original_X:pd.DataFrame,
+                X_pca:pd.DataFrame,
+                current_predictions:pd.DataFrame,
+                asset:list,
+                target_returns:pd.Series,
+                configs: dict,
+                reporting: Reporting
+                ) -> Reporting.Training_Step:
+    training_step = Reporting.Training_Step(level='secondary')
     model_config, training_config, data_config = itemgetter('model_config', 'training_config', 'data_config')(configs)
     
     # 5. Ensemble primary model predictions (If Ensemble model is present)
@@ -90,7 +109,6 @@ def secondary_step(X:pd.DataFrame, y:pd.Series, original_X:pd.DataFrame, X_pca:p
         reporting.all_predictions = pd.concat([reporting.all_predictions, ensemble_predictions], axis=1)
     
         
-        
         if len(model_config['meta_labeling_models']) > 0: 
 
             # 3. Train a Meta-labeling model on the averaged level-1 model predictions

utils/encapsulation.py

@@ -1,38 +0,0 @@
-import pandas as pd
-from models.base import Model
-
-# | Reporting
-# |
-
-
-class Single_Model:
-    def __init__(self, model_name:str, model_over_time:list[Model]):
-        self.model_name: str = model_name
-        self.model_over_time: list[Model] = model_over_time
-
-
-class Training_Step: 
-    def __init__(self, level:str):
-        self.level:str = level
-        self.base: list[Single_Model] = []
-        self.metalabeling: list[list[Single_Model]] = []
-
-
-class Asset():
-    def __init__(self, ticker:str, primary: Training_Step, secondary: Training_Step):
-        self.name:str = ticker
-        self.primary:Training_Step = primary
-        self.secondary:Training_Step = secondary 
-
-
-class Reporting:
-    def __init__(self):
-        self.results:pd.DataFrame = pd.DataFrame() 
-        self.all_predictions:pd.DataFrame = pd.DataFrame() 
-        self.all_probabilities:pd.DataFrame = pd.DataFrame() 
-        self.all_assets:list[Asset] = []
-    
-    def get_results(self)->tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, list[Asset]]:
-        return self.results, self.all_predictions, self.all_probabilities, self.all_assets
-    
-