feature(Inference): Created the inference process, added model saving. (#153)

* feat: Basic scaffolding up for inference process after training.

* feat: Saving and loading models works. Inference works nearly.

* feat: Added inference pipeline.

* feat: Saving model now accoring to date and time; loading models now selects from latest file. Fixed the creation of dictionary of models.

* feat: Added lightweight asset config, but full pipeline.

* feat: Added new naming for dictionary.

* fix: Fixed dictionary naming convention.

* fix: Fixed naming again, now the model structure is good

* fix: Changed the output path and the return values from run_pipeline.

* feat: Added function to make sure folder exists for output models.

Co-authored-by: Daniel Szemerey <szemereydaniel@gmail.com>

.gitignore

@@ -135,3 +135,5 @@ wandb/
 .cachedir/**
 lightning_logs/
 
+results.csv
+models/saved/

config/config.py

@@ -14,7 +14,7 @@ def get_dev_config() -> tuple[dict, dict, dict]:
     )
 
     data_config = dict(
-        assets = ['daily_crypto'],
+        assets = ['daily_only_btc'],
         other_assets = [],
         exogenous_data = [],
         load_non_target_asset= True,
@@ -86,3 +86,48 @@ def get_default_ensemble_config() -> tuple[dict, dict, dict]:
     return model_config, training_config, data_config
 
 
+
+def get_lightweight_ensemble_config() -> tuple[dict, dict, dict]:
+  
+    training_config = dict(
+        primary_models_meta_labeling = True,
+        dimensionality_reduction = True,
+        n_features_to_select = 30,
+        expanding_window_primary = False,
+        expanding_window_meta_labeling = True,
+        sliding_window_size_primary = 380,
+        sliding_window_size_meta_labeling = 240,
+        retrain_every = 20,
+        scaler = 'minmax', # 'normalize' 'minmax' 'standardize' 'none'
+    )
+
+    data_config = dict(
+        assets = ['daily_crypto_lightweight'],
+        other_assets = ['daily_etf'],
+        exogenous_data = ['daily_glassnode'],
+        load_non_target_asset= True,
+        log_returns= True,
+        forecasting_horizon = 1,
+        own_features = ['level_2' ],
+        other_features = ['level_2'],
+        exogenous_features = ['standard_scaling'],
+        index_column= 'int',
+        method= 'classification',
+        no_of_classes= 'two',
+        narrow_format = False,
+    )
+
+    regression_models = ["Lasso", "KNN"]
+    classification_models = ['LR_two_class', 'SVC']
+    meta_labeling_models = ['LR_two_class', 'LGBM']
+    ensemble_model = 'Average'
+
+    model_config = dict(
+        primary_models = regression_models if data_config['method'] == 'regression' else classification_models,
+        meta_labeling_models = meta_labeling_models,
+        ensemble_model = ensemble_model
+    )
+    
+    return model_config, training_config, data_config
+
+

data_loader/collections.py

@@ -8,6 +8,8 @@ __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"]
+
 __hourly_crypto = ["BTC_USD", "DASH_USD", "ETC_USD", "ETH_USD", "LTC_USD", "TRX_USD", "XLM_USD", "XMR_USD", "XRP_USD"]
 
 __daily_glassnode = ['rhodl_ratio',
@@ -41,6 +43,7 @@ __daily_glassnode = ['rhodl_ratio',
 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),
     hourly_crypto = transform_to_data_collection("data/hourly_crypto", __hourly_crypto),
     daily_glassnode =transform_to_data_collection("data/daily_glassnode",  __daily_glassnode),

models/saving.py

@@ -0,0 +1,37 @@
+import pickle
+import datetime
+from typing import Optional, Union
+import os
+import warnings
+
+
+
+def save_models(all_models_for_all_assets: dict, data_config:dict, training_config:dict) -> None:
+    all_models_for_all_assets['training_config'] = training_config
+    all_models_for_all_assets['data_config'] = data_config
+    
+    date_string = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M")
+    
+    if not os.path.exists('output/models'):
+        warnings.warn("No folder exists, creating one.")
+        os.makedirs('output/models')
+        
+    pickle.dump( all_models_for_all_assets, open( "output/models/{}.p".format(date_string), "wb" ) )
+
+
+def load_models(file_name:Union[str, None]) -> tuple[dict, dict, dict]:
+
+    if file_name is None: 
+        warnings.warn("No file name provided, will load latest models and configurations.")
+        files_in_directory:list = os.listdir('output/models')
+        
+        assert len(files_in_directory) > 0, "No models found in output/models."
+        file_name = sorted(files_in_directory)[-1]
+    
+    all_models_for_all_assets = pickle.load( open( "output/models/{}".format(file_name), "rb" ) )
+    
+    data_config = all_models_for_all_assets['data_config']
+    training_config = all_models_for_all_assets['training_config']
+    
+    return all_models_for_all_assets, data_config, training_config
+

run_evaluate_robustness.py

@@ -5,7 +5,7 @@ import pandas as pd
 all_results = []
 all_predictions = []
 for index in range(6):
-    results_1, predictions_1, _ = run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, get_config= get_default_ensemble_config)
+    _, _, _, results_1, predictions_1, _ = run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, get_config= get_default_ensemble_config)
     all_results.append(results_1)
     all_predictions.append(predictions_1)
 

run_inference.py

@@ -0,0 +1,20 @@
+from training.inference import run_inference_pipeline
+from models.saving import load_models
+
+from run_pipeline import run_pipeline
+from config.config import get_dev_config, get_default_ensemble_config, get_lightweight_ensemble_config
+
+from typing import Callable
+
+
+def run_inference(preload_models:bool, get_config:Callable):
+    if preload_models:
+        all_models_all_assets, data_config, training_config = load_models(None)
+    else:
+        all_models_all_assets, data_config, training_config, _, _, _ = run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, get_config=get_config)
+    
+    run_inference_pipeline(data_config, training_config, all_models_all_assets)
+    
+
+if __name__ == '__main__':
+    run_inference(preload_models=False, get_config=get_lightweight_ensemble_config)

run_pipeline.py

@@ -4,23 +4,27 @@ import pandas as pd
 from training.primary_model import train_primary_model
 from reporting.wandb import launch_wandb, register_config_with_wandb
 from models.model_map import default_feature_selector_regression, default_feature_selector_classification
+from models.saving import save_models
 from utils.helpers import has_enough_samples_to_train
 from config.config import get_default_ensemble_config
 from config.preprocess import validate_config, preprocess_config
 from feature_selection.feature_selection import select_features
 from feature_selection.dim_reduction import reduce_dimensionality
 from training.meta_labeling import train_meta_labeling_model
+
 from reporting.reporting import report_results
 from typing import Callable, Optional
 import ray
 ray.init()
 
 
-def run_pipeline(project_name:str, with_wandb: bool, sweep: bool, get_config: Callable) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+def run_pipeline(project_name:str, with_wandb: bool, sweep: bool, get_config: Callable) -> tuple[dict, dict, dict, pd.DataFrame, pd.DataFrame, pd.DataFrame]:
     wandb, model_config, training_config, data_config = __setup_config(project_name, with_wandb, sweep, get_config)
-    results, all_predictions, all_probabilities = __run_training(model_config, training_config, data_config)  
+    results, all_predictions, all_probabilities, all_models_all_assets = __run_training(model_config, training_config, data_config)  
     report_results(results, all_predictions, model_config, wandb, sweep, project_name)
-    return results, all_predictions, all_probabilities
+    save_models(all_models_all_assets, data_config, training_config)
+
+    return all_models_all_assets, data_config, training_config, results, all_predictions, all_probabilities
     
 
 def __setup_config(project_name:str, with_wandb: bool, sweep: bool, get_config: Callable) -> tuple[Optional[object], dict, dict, dict]:
@@ -87,7 +91,7 @@ def __run_training(model_config:dict, training_config:dict, data_config:dict):
         
         all_models_for_all_assets[asset[1]] = dict(
             name=asset[1], 
-            models=all_models_for_single_asset
+            primary_models = all_models_for_single_asset
         )
         
         # 4. Train a Meta-Labeling model for each Primary model and replace their predictions with the meta-labeling predictions
@@ -109,7 +113,7 @@ def __run_training(model_config:dict, training_config:dict, data_config:dict):
                 current_result[model_name] = primary_meta_result
                 current_predictions[model_name] = primary_meta_preds
 
-                all_models_for_all_assets[asset[1]][model_name] = meta_labeling_models
+                all_models_for_all_assets[asset[1]]['primary_models'][model_name]['meta_labeling'] = meta_labeling_models
         
         results = pd.concat([results, current_result], axis=1)
         # With static models, because of the lag in the indicator, the first prediction is NA, so we fill it with zero.
@@ -119,7 +123,7 @@ def __run_training(model_config:dict, training_config:dict, data_config:dict):
         # 5. Ensemble primary model predictions (If Ensemble model is present)
         if model_config['ensemble_model'] is not None:
 
-            ensemble_result, ensemble_predictions, _, _ = train_primary_model(
+            ensemble_result, ensemble_predictions, _, ensemble_models_one_asset = train_primary_model(
                 ticker_to_predict = asset[1],
                 original_X = current_predictions,
                 X = current_predictions,
@@ -136,7 +140,8 @@ def __run_training(model_config:dict, training_config:dict, data_config:dict):
                 print_results= True,
             )
             ensemble_result, ensemble_predictions = ensemble_result.iloc[:,0], ensemble_predictions.iloc[:,0]
-
+            all_models_for_all_assets[asset[1]]['secondary_model'] = ensemble_models_one_asset
+            
             if len(model_config['meta_labeling_models']) > 0: 
 
                 # 3. Train a Meta-labeling model on the averaged level-1 model predictions
@@ -152,12 +157,13 @@ def __run_training(model_config:dict, training_config:dict, data_config:dict):
                     training_config= training_config,
                     model_suffix = 'ensemble'
                 )
-
+                
+                all_models_for_all_assets[asset[1]]['secondary_model'][model_config['ensemble_model']] = dict(meta_labeling=ensemble_meta_labeling_models)
                 results = pd.concat([results, ensemble_meta_result], axis=1)
                 all_predictions = pd.concat([all_predictions, ensemble_meta_predictions], axis=1)
                 all_probabilities = pd.concat([all_probabilities, ensemble_meta_probabilities], axis=1).fillna(0.)
-        
-    return results, all_predictions, all_probabilities
+                
+    return results, all_predictions, all_probabilities, all_models_for_all_assets
     
     
 

training/inference.py

@@ -0,0 +1,79 @@
+
+import pandas as pd
+from data_loader.load_data import load_data
+from typing import Optional, Union
+import warnings
+
+
+def run_inference_pipeline(data_config:dict, training_config:dict, all_models_all_assets:dict):
+    data_params = data_config.copy()
+    data_params['target_asset'] = data_params['assets'][0]
+    
+    X, y, _ = load_data(**data_params)
+    input_features = __select_data(X, training_config)
+    
+    primary_models, secondary_models = __select_models(data_params, all_models_all_assets) 
+    
+    result = __inference(input_features, primary_models, secondary_models)
+    
+    return result
+
+
+def __inference(data:pd.DataFrame, primary_models:Union[dict,None], secondary_models:Union[dict,None]) -> pd.DataFrame:
+    assert primary_models is not None, "No primary models found. Cancelling Inference."
+    
+    data = __primary_models(data, primary_models)
+    
+    if secondary_models is not None:
+        warnings.warn("Secondary models are not specified.")
+        data = __secondary_models(data, secondary_models)
+    
+    return data
+
+
+def __select_models( data_params:dict, all_models_all_assets:dict)-> tuple[Optional[dict],Optional[dict]]:
+    target_asset_name = data_params['target_asset'][1]
+    primary_models, secondary_models = None, None
+
+    if 'primary_models' in all_models_all_assets[target_asset_name]:
+        primary_models = all_models_all_assets[target_asset_name]['primary_models']
+    else: 
+        assert("No primary models found for asset: " + target_asset_name)
+    
+    if 'secondary_model' in all_models_all_assets[target_asset_name]:
+        secondary_models = all_models_all_assets[target_asset_name]['secondary_model']
+    
+    return primary_models, secondary_models
+
+
+def __select_data(X:pd.DataFrame, training_config:dict)-> pd.DataFrame:
+    window_size = training_config['sliding_window_size_primary']
+    num_rows = X.shape[0]
+    
+    if num_rows <= window_size: 
+        return X.copy() 
+    else: 
+        return X.truncate(before=int(num_rows-window_size), after=num_rows, copy=True)
+
+
+def __primary_models(data:pd.DataFrame, models:dict)-> pd.DataFrame:
+    for k, model in models:
+        last_model = model[-1]
+        prediction = last_model.predict(data.to_numpy())
+        
+        # result = evaluate_predictions(
+        #     model_name = model_name,
+        #     target_returns = target_returns,
+        #     y_pred = preds,
+        #     y_true = y,
+        #     method = method,
+        #     no_of_classes=no_of_classes,
+        #     print_results = print_results,
+        #     discretize=True
+        # )
+        
+    return data
+
+def __secondary_models(data:pd.DataFrame, model:dict)-> pd.DataFrame:
+    
+    return data

training/primary_model.py

@@ -54,9 +54,13 @@ def train_primary_model(
             print_results = print_results,
             discretize=True
         )
-        column_name = "model_" + ticker_to_predict + "_" + model_name + "_" + level
+        levelname=("_" + level) if level=='metalabeling' else ""
+        column_name = "model_" + model_name + "_" + ticker_to_predict + levelname 
         results[column_name] = result
-        all_models_single_asset[model_name] = model_over_time
+        all_models_single_asset[column_name]=dict()
+        all_models_single_asset[column_name][level] = model_over_time.tolist()
+        # all_models_single_asset[model_name]=dict()
+        # all_models_single_asset[model_name][level] = 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
         probs_column_name = "probs_" + ticker_to_predict + "_" + model_name + "_" + level