From 55f083638fb112f57ae22dd2c46fe42cd0a6a276 Mon Sep 17 00:00:00 2001 From: Daniel Szemerey Date: Tue, 11 Jan 2022 19:15:58 +0100 Subject: [PATCH] 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 --- .gitignore | 2 + config/config.py | 47 ++++++++++++++++++++++- data_loader/collections.py | 3 ++ models/saving.py | 37 ++++++++++++++++++ run_evaluate_robustness.py | 2 +- run_inference.py | 20 ++++++++++ run_pipeline.py | 26 ++++++++----- training/inference.py | 79 ++++++++++++++++++++++++++++++++++++++ training/primary_model.py | 8 +++- 9 files changed, 210 insertions(+), 14 deletions(-) create mode 100644 models/saving.py create mode 100644 run_inference.py create mode 100644 training/inference.py diff --git a/.gitignore b/.gitignore index 7aa671c..e599af5 100644 --- a/.gitignore +++ b/.gitignore @@ -135,3 +135,5 @@ wandb/ .cachedir/** lightning_logs/ +results.csv +models/saved/ diff --git a/config/config.py b/config/config.py index 99c9da8..24512a2 100644 --- a/config/config.py +++ b/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 + + diff --git a/data_loader/collections.py b/data_loader/collections.py index 042da6b..8235d51 100644 --- a/data_loader/collections.py +++ b/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), diff --git a/models/saving.py b/models/saving.py new file mode 100644 index 0000000..1f70a3b --- /dev/null +++ b/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 + diff --git a/run_evaluate_robustness.py b/run_evaluate_robustness.py index 2d1cc37..e2f320d 100644 --- a/run_evaluate_robustness.py +++ b/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) diff --git a/run_inference.py b/run_inference.py new file mode 100644 index 0000000..2678bf1 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/run_pipeline.py b/run_pipeline.py index 3ada071..00e9497 100644 --- a/run_pipeline.py +++ b/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 diff --git a/training/inference.py b/training/inference.py new file mode 100644 index 0000000..efcdbdd --- /dev/null +++ b/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 diff --git a/training/primary_model.py b/training/primary_model.py index d19a8ac..7918f9c 100644 --- a/training/primary_model.py +++ b/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