diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index c1085fd..6680418 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -22,6 +22,7 @@ jobs: - name: Run pipeline shell: bash -l {0} run: | + export RAY_DISABLE_MEMORY_MONITOR=1 python run_pipeline.py - name: Run portfolio reporting shell: bash -l {0} diff --git a/config/config.py b/config/config.py index e339901..911ba3c 100644 --- a/config/config.py +++ b/config/config.py @@ -15,6 +15,7 @@ def get_dev_config() -> tuple[dict, dict, dict]: data_config = dict( assets = ['daily_only_btc'], + target_asset = 'BTC_USD', other_assets = [], exogenous_data = [], load_non_target_asset= True, @@ -52,12 +53,13 @@ def get_default_ensemble_config() -> tuple[dict, dict, dict]: expanding_window_meta_labeling = True, sliding_window_size_primary = 380, sliding_window_size_meta_labeling = 240, - retrain_every = 40, + retrain_every = 10, scaler = 'minmax', # 'normalize' 'minmax' 'standardize' ) data_config = dict( assets = ['daily_crypto'], + target_asset = 'BTC_USD', other_assets = ['daily_etf'], exogenous_data = ['daily_glassnode'], load_non_target_asset= True, @@ -103,6 +105,7 @@ def get_lightweight_ensemble_config() -> tuple[dict, dict, dict]: data_config = dict( assets = ['daily_crypto_lightweight'], + target_asset = 'BTC_USD', other_assets = ['daily_etf'], exogenous_data = ['daily_glassnode'], load_non_target_asset= True, diff --git a/config/preprocess.py b/config/preprocess.py index bee987a..b1797e1 100644 --- a/config/preprocess.py +++ b/config/preprocess.py @@ -36,6 +36,9 @@ def __preprocess_data_collections_config(data_dict: dict) -> dict: for key in keys: preset_names = data_dict[key] data_dict[key] = flatten([data_collections[preset_name] for preset_name in preset_names]) + target_asset = next(iter([asset for asset in data_dict['assets'] if asset[1] == data_dict['target_asset']]), None) + if target_asset is None: raise Exception('Target asset wasnt found in assets') + data_dict['target_asset'] = target_asset return data_dict diff --git a/exploration.ipynb b/exploration.ipynb index e98f412..1cf1574 100644 --- a/exploration.ipynb +++ b/exploration.ipynb @@ -169,7 +169,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.9.7" + "version": "3.9.9" }, "orig_nbformat": 4 }, diff --git a/models/saving.py b/reporting/saving.py similarity index 72% rename from models/saving.py rename to reporting/saving.py index 67148c4..b8725a1 100644 --- a/models/saving.py +++ b/reporting/saving.py @@ -7,12 +7,12 @@ from reporting.types import Reporting -def save_models(all_models_for_all_assets: list[Reporting.Asset], data_config:dict, training_config:dict, model_config:dict) -> None: +def save_models(all_models: Reporting.Asset, data_config:dict, training_config:dict, model_config:dict) -> None: dict_for_pickle = dict() dict_for_pickle['training_config'] = training_config dict_for_pickle['data_config'] = data_config dict_for_pickle['model_config'] = model_config - dict_for_pickle['all_models_for_all_assets'] = all_models_for_all_assets + dict_for_pickle['all_models'] = all_models date_string = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M") @@ -23,7 +23,7 @@ def save_models(all_models_for_all_assets: list[Reporting.Asset], data_config:di pickle.dump( dict_for_pickle, open( "output/models/{}.p".format(date_string), "wb" ) ) -def load_models(file_name:Union[str, None]) -> tuple[list[Reporting.Asset], dict, dict, dict]: +def load_models(file_name:Union[str, None]) -> tuple[Reporting.Asset, dict, dict, dict]: if file_name is None: warnings.warn("No file name provided, will load latest models and configurations.") @@ -37,7 +37,7 @@ def load_models(file_name:Union[str, None]) -> tuple[list[Reporting.Asset], dict data_config = packacked_dict.pop("data_config", None) training_config = packacked_dict.pop("training_config", None) model_config = packacked_dict.pop("model_config", None) - all_models_for_all_assets = packacked_dict.pop("all_models_for_all_assets", None) + all_models = packacked_dict.pop("all_models", None) - return all_models_for_all_assets, data_config, training_config, model_config + return all_models, data_config, training_config, model_config diff --git a/reporting/types.py b/reporting/types.py index a801d8b..85e8e14 100644 --- a/reporting/types.py +++ b/reporting/types.py @@ -1,6 +1,7 @@ from __future__ import annotations import pandas as pd from models.base import Model +from typing import Optional, Union class Reporting: @@ -8,16 +9,17 @@ class Reporting: 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] = [] + self.asset: 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 + def get_results(self) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, Reporting.Asset]: + return self.results, self.all_predictions, self.all_probabilities, self.asset class Single_Model: - def __init__(self, model_name: str, model_over_time: list[Model]): + def __init__(self, model_name: str, model_over_time: pd.Series, transformations_over_time: list[pd.Series]): self.model_name: str = model_name - self.model_over_time: list[Model] = model_over_time + self.model_over_time: pd.Series = model_over_time + self.transformations_over_time: list[pd.Series] = transformations_over_time class Training_Step: @@ -26,14 +28,14 @@ class Reporting: self.base: list[Reporting.Single_Model] = [] self.metalabeling: list[list[Reporting.Single_Model]] = [] - def convert_step_to_tuple(self, step:str)->list[tuple[str, list[Model]]]: - if step == 'base': - return [(x.model_name, x.model_over_time) for x in self.base ] - elif step == 'metalabeling': - return [(x.model_name, x.model_over_time) for sub in self.metalabeling for x in sub] - else: - raise ValueError('Unknown step: {}'.format(step)) + def get_base(self) -> list[tuple[str, pd.Series, list[pd.Series]]]: + return [(x.model_name, x.model_over_time, x.transformations_over_time) for x in self.base ] + def get_metalabeling(self) -> dict: + structured_dict = dict() + for i, model in enumerate(self.base): + structured_dict[model.model_name] = [(x.model_name, x.model_over_time, x.transformations_over_time) for x in self.metalabeling[i]] + return structured_dict class Asset(): def __init__(self, ticker: str, primary: Reporting.Training_Step, secondary: Reporting.Training_Step): diff --git a/run_inference.py b/run_inference.py index 05d32b4..0e27f2c 100644 --- a/run_inference.py +++ b/run_inference.py @@ -1,20 +1,47 @@ -from training.inference import run_inference_pipeline -from models.saving import load_models +from data_loader.load_data import load_data +from data_loader.process_data import check_data +from reporting.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 - +from typing import Callable, Optional +from reporting.types import Reporting +from training.training_steps import primary_step, secondary_step +import warnings def run_inference(preload_models:bool, get_config:Callable): if preload_models: - all_models_all_assets, data_config, training_config, model_config = load_models(None) + all_models, data_config, training_config, model_config = load_models(None) else: - all_models_all_assets, data_config, training_config, model_config, _, _, _ = run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, get_config=get_config) + all_models, data_config, training_config, model_config, _, _, _ = run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, get_config=get_config) - run_inference_pipeline(data_config, training_config, model_config, all_models_all_assets) + configs = dict(model_config=model_config, training_config=training_config, data_config=data_config) + __inference(configs, all_models.primary, all_models.secondary) + + +def __inference(configs: dict, primary_models: Optional[Reporting.Training_Step], secondary_models: Optional[Reporting.Training_Step]): + reporting = Reporting() + asset = configs['data_config']['target_asset'] + # 1. Load data, check for validity and process data + X, y, target_returns = load_data(**configs['data_config']) + assert check_data(X, y, configs['training_config']) == True, "Data is not valid. Cancelling Inference." + + inference_from = X.index.stop - 2 + # 2. Train a Primary model with optional metalabeling for each asset + training_step_primary, current_predictions = primary_step(X, y, target_returns, configs, reporting, from_index = inference_from, preloaded_training_step = primary_models) + + # 3. Train an Ensemble model with optional metalabeling for each asset + if secondary_models is not None: + warnings.warn("Secondary models are not specified.") + training_step_secondary = secondary_step(X, y, current_predictions, target_returns, configs, reporting, from_index = inference_from, preloaded_training_step = secondary_models) + + # 4. Save the models + reporting.asset = Reporting.Asset(ticker=asset, primary=training_step_primary, secondary=training_step_secondary) + + return reporting + if __name__ == '__main__': run_inference(preload_models=True, get_config=get_lightweight_ensemble_config) \ No newline at end of file diff --git a/run_pipeline.py b/run_pipeline.py index 9e2dd88..438f20c 100644 --- a/run_pipeline.py +++ b/run_pipeline.py @@ -7,7 +7,7 @@ from data_loader.process_data import check_data from reporting.wandb import launch_wandb, register_config_with_wandb from reporting.reporting import report_results -from models.saving import save_models +from reporting.saving import save_models from config.config import get_default_ensemble_config from config.preprocess import validate_config, preprocess_config @@ -20,14 +20,14 @@ import ray ray.init() -def run_pipeline(project_name:str, with_wandb: bool, sweep: bool, get_config: Callable) -> tuple[list[Reporting.Asset], dict, dict, dict, pd.DataFrame, pd.DataFrame, pd.DataFrame]: +def run_pipeline(project_name:str, with_wandb: bool, sweep: bool, get_config: Callable) -> tuple[Reporting.Asset, 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) reporting = __run_training(model_config, training_config, data_config) - results, all_predictions, all_probabilities, all_models_all_assets = reporting.get_results() + results, all_predictions, all_probabilities, all_models = reporting.get_results() report_results(results, all_predictions, model_config, wandb, sweep, project_name) - save_models(all_models_all_assets, data_config, training_config, model_config) + save_models(all_models, data_config, training_config, model_config) - return all_models_all_assets, data_config, training_config, model_config, results, all_predictions, all_probabilities + return all_models, data_config, training_config, model_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]: @@ -48,24 +48,19 @@ def __run_training(model_config:dict, training_config:dict, data_config:dict): configs = dict(model_config=model_config, training_config=training_config, data_config=data_config) reporting = Reporting() - for asset in data_config['assets']: - print('--------\nPredicting: ', asset[1]) + # 1. Load data, check for validity + X, y, target_returns = load_data(**configs['data_config']) + assert check_data(X, y, configs['training_config']) == True, "Data is not valid." - configs['data_config']['target_asset'] = asset - - # 1. Load data, check for validity and process data (feature selection, dimensionality reduction, etc.) - X, y, target_returns = load_data(**configs['data_config']) - if check_data(X, y, configs['training_config']) is False: continue - - # 2. Train a Primary model with optional metalabeling for each asset - training_step_primary, current_predictions = primary_step(X, y, asset, target_returns, configs, reporting) - - # 3. Train an Ensemble model with optional metalabeling for each asset - training_step_secondary = secondary_step(X, y, current_predictions, asset, target_returns, configs, reporting) - - # 4. Save the models - reporting.all_assets.append(Reporting.Asset(ticker=asset[1], primary=training_step_primary, secondary=training_step_secondary)) - + # 2. Train a Primary model with optional metalabeling for each asset + training_step_primary, current_predictions = primary_step(X, y, target_returns, configs, reporting, from_index = None) + + # 3. Train an Ensemble model with optional metalabeling for each asset + training_step_secondary = secondary_step(X, y, current_predictions, target_returns, configs, reporting, from_index = None) + + # 4. Save the models + reporting.asset = Reporting.Asset(ticker= data_config['target_asset'][1], primary=training_step_primary, secondary=training_step_secondary) + return reporting diff --git a/tests/test_evaluation.py b/tests/test_evaluation.py index 028c906..93ca444 100644 --- a/tests/test_evaluation.py +++ b/tests/test_evaluation.py @@ -77,6 +77,7 @@ def test_evaluation(): expanding_window=False, window_size=window_length, retrain_every=10, + from_index=None, transformations=[], preloaded_transformations=None) predictions, _ = walk_forward_inference( @@ -86,6 +87,7 @@ def test_evaluation(): X=X, expanding_window=False, window_size=window_length, + from_index=None, ) # verify if predictions are the same as y diff --git a/tests/test_walk_forward.py b/tests/test_walk_forward.py index 0d841cd..ec20b94 100644 --- a/tests/test_walk_forward.py +++ b/tests/test_walk_forward.py @@ -75,6 +75,7 @@ def test_walk_forward_train_test(): expanding_window=False, window_size=window_length, retrain_every=10, + from_index=None, transformations=[], preloaded_transformations=None ) @@ -84,7 +85,8 @@ def test_walk_forward_train_test(): transformations_over_time=transformations_over_time, X=X, expanding_window=False, - window_size=window_length + window_size=window_length, + from_index=None, ) # verify if predictions are the same as y diff --git a/training/inference.py b/training/inference.py deleted file mode 100644 index 92805fc..0000000 --- a/training/inference.py +++ /dev/null @@ -1,75 +0,0 @@ - -import pandas as pd -from typing import Optional, Union -import warnings - -from data_loader.load_data import load_data -from data_loader.process_data import process_data, check_data - -from reporting.types import Reporting -from training.training_steps import primary_step, secondary_step - -def run_inference_pipeline(data_config:dict, training_config:dict, model_config:dict, all_models_all_assets:list[Reporting.Asset]): - configs = dict(model_config=model_config, training_config=training_config, data_config=data_config) - configs['data_config']['target_asset'] = data_config['assets'][0] - - primary_models, secondary_models = __select_models(configs, all_models_all_assets) - - result = __inference(configs, primary_models, secondary_models) - - return result - - -def __inference(configs:dict, primary_models:Union[Reporting.Training_Step,None], secondary_models:Union[Reporting.Training_Step, None]): - reporting = Reporting() - asset = configs['data_config']['target_asset'] - - # 1. Load data, truncate it, check for validity and process data (feature selection, dimensionality reduction, etc.) - X, y, target_returns = load_data(**configs['data_config']) - X, y = __select_data(X, y, configs['training_config']) - assert check_data(X, y, configs['training_config']) == False, "Data is not valid. Cancelling Inference." - X, original_X = process_data(X, y, configs) - - # 2. Train a Primary model with optional metalabeling for each asset - training_step_primary, current_predictions = primary_step(X, y, original_X, asset, target_returns, configs, reporting, primary_models) - - # 3. Train an Ensemble model with optional metalabeling for each asset - if secondary_step is not None: - warnings.warn("Secondary models are not specified.") - training_step_secondary = secondary_step(X, y, original_X, current_predictions, asset, target_returns, configs, reporting, secondary_models) - - # 4. Save the models - reporting.all_assets.append(Reporting.Asset(ticker=asset, primary=training_step_primary, secondary=training_step_secondary)) - - return reporting - - -def __select_models( configs:dict, all_models_all_assets:list[Reporting.Asset])-> tuple[Union[Reporting.Training_Step,None], Union[Reporting.Training_Step, None]]: - target_asset_name = configs['data_config']['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) - - if target_asset_models is not None: - if len(target_asset_models.primary.base)>0: - primary_step = target_asset_models.primary - else: warnings.warn("No primary models found for {}.".format(target_asset_name)) - - if len(target_asset_models.secondary.base)>0: - secondary_step = target_asset_models.secondary - else: warnings.warn("No secondary models found for {}.".format(target_asset_name)) - else: - assert("No models found for asset: " + target_asset_name) - - return primary_step, secondary_step - - -def __select_data(X:pd.DataFrame, y:pd.Series, training_config:dict)-> tuple[pd.DataFrame, pd.Series]: - window_size = training_config['sliding_window_size_primary'] - num_rows = X.shape[0] - - if num_rows <= window_size: - return X.copy(), y.copy() - else: - return X.truncate(before=int(num_rows-window_size), after=num_rows, copy=True), y.truncate(before=int(num_rows-window_size), after=num_rows, copy=True) - - diff --git a/training/meta_labeling.py b/training/meta_labeling.py index 51fa478..40f5d74 100644 --- a/training/meta_labeling.py +++ b/training/meta_labeling.py @@ -5,7 +5,7 @@ import pandas as pd from models.model_map import default_feature_selector_classification, default_feature_selector_regression from models.base import Model from reporting.types import Reporting -from typing import Union +from typing import Union, Optional def train_meta_labeling_model( @@ -18,8 +18,9 @@ def train_meta_labeling_model( data_config: dict, model_config: dict, training_config: dict, - model_suffix: str, - preloaded_models: Union[list[Reporting.Single_Model], None] = None + model_suffix: str, + from_index: Optional[int], + preloaded_models: Optional[list[tuple[str, pd.Series, list[pd.Series]]]] = None ) -> tuple[pd.Series, pd.Series, pd.DataFrame, list[Reporting.Single_Model]]: discretize = discretize_threeway_threshold(0.33) @@ -38,6 +39,7 @@ def train_meta_labeling_model( expanding_window = training_config['expanding_window_meta_labeling'], sliding_window_size = training_config['sliding_window_size_meta_labeling'], retrain_every = training_config['retrain_every'], + from_index = from_index, scaler = training_config['scaler'], no_of_classes = 'two', level = 'meta_labeling', diff --git a/training/primary_model.py b/training/primary_model.py index 533ddcf..b8f61f2 100644 --- a/training/primary_model.py +++ b/training/primary_model.py @@ -1,5 +1,5 @@ import pandas as pd -from typing import Literal, Union +from typing import Literal, Optional, Union from training.walk_forward import walk_forward_train, walk_forward_inference from utils.evaluate import evaluate_predictions from models.base import Model @@ -19,11 +19,12 @@ def train_primary_model( expanding_window: bool, sliding_window_size: int, retrain_every: int, + from_index: Optional[int], scaler: ScalerTypes, no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'], level: str, print_results: bool, - preloaded_models: Union[list[Reporting.Single_Model], None] = None + preloaded_models: Optional[list[tuple[str, pd.Series, list[pd.Series]]]] = None ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, list[Reporting.Single_Model]]: results = pd.DataFrame() @@ -31,13 +32,16 @@ def train_primary_model( probabilities = pd.DataFrame(index=y.index) all_models_single_asset: list[Reporting.Single_Model] = [] + unified_models: list[tuple[str, pd.Series, list[pd.Series]]] = [] if preloaded_models is not None: - models = preloaded_models + unified_models = preloaded_models transformations_over_time = None - for model_name, model in models: - if preloaded_models is None: + + if preloaded_models is None: + train_unified_models=[] + for model_name, model in models: model_over_time, transformations_over_time = walk_forward_train( model_name=model_name, model = model, @@ -47,6 +51,7 @@ def train_primary_model( expanding_window = expanding_window, window_size = sliding_window_size, retrain_every = retrain_every, + from_index = from_index, transformations= [ get_scaler(scaler), PCATransformation(ratio_components_to_keep=0.5, sliding_window_size=sliding_window_size), @@ -54,13 +59,19 @@ def train_primary_model( ], preloaded_transformations=transformations_over_time, ) + train_unified_models.append((model_name, model_over_time, transformations_over_time)) + unified_models = train_unified_models + + for model_tuples in unified_models: + model_name, model_over_time, transformations_over_time = model_tuples[0], model_tuples[1], model_tuples[2] preds, probs = walk_forward_inference( model_name = model_name, - model_over_time= model_over_time if preloaded_models is None else pd.Series(model), + model_over_time= pd.Series(model_over_time), transformations_over_time = transformations_over_time, X = X, expanding_window = expanding_window, - window_size = sliding_window_size + window_size = sliding_window_size, + from_index = from_index, ) assert len(preds) == len(y) @@ -75,11 +86,14 @@ def train_primary_model( discretize=True ) levelname=("_" + level) if level=='metalabeling' else "" - column_name = "model_" + model_name + "_" + ticker_to_predict + levelname + if preloaded_models is None: + column_name = "model_" + model_name + "_" + ticker_to_predict + levelname + else: + column_name = model_name results[column_name] = result - all_models_single_asset.append(Reporting.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, transformations_over_time=transformations_over_time)) # 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 diff --git a/training/training_steps.py b/training/training_steps.py index a4c46c1..58c46c6 100644 --- a/training/training_steps.py +++ b/training/training_steps.py @@ -1,3 +1,4 @@ +from numpy import DataSource import pandas as pd from operator import itemgetter @@ -5,24 +6,24 @@ from training.primary_model import train_primary_model from training.meta_labeling import train_meta_labeling_model from reporting.types import Reporting -from typing import Union +from typing import Union, Optional def primary_step( X: pd.DataFrame, - y:pd.Series, - asset:list, - target_returns:pd.Series, + y: pd.Series, + target_returns: pd.Series, configs: dict, reporting: Reporting, - preloaded_training_step: Union[Reporting.Training_Step, None] = None + from_index: Optional[int], + preloaded_training_step: Optional[Reporting.Training_Step] = None, ) -> 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 current_result, current_predictions, current_probabilities, all_models_for_single_asset = train_primary_model( - ticker_to_predict = asset[1], + ticker_to_predict = data_config['target_asset'][1], X = X, y = y, target_returns = target_returns, @@ -31,11 +32,12 @@ def primary_step( expanding_window = training_config['expanding_window_primary'], sliding_window_size = training_config['sliding_window_size_primary'], retrain_every = training_config['retrain_every'], + from_index = from_index, scaler = training_config['scaler'], no_of_classes = data_config['no_of_classes'], level = 'primary', print_results= True, - preloaded_models = preloaded_training_step.convert_step_to_tuple('base') if preloaded_training_step is not None else None + preloaded_models = preloaded_training_step.get_base() if preloaded_training_step is not None else None ) training_step.base = all_models_for_single_asset @@ -45,7 +47,7 @@ def primary_step( for model_name in current_result.columns: primary_model_predictions = current_predictions[model_name] primary_meta_result, primary_meta_preds, primary_meta_probabilities, meta_labeling_models = train_meta_labeling_model( - target_asset=asset[1], + target_asset = data_config['target_asset'][1], X = X, input_predictions= primary_model_predictions, y = y, @@ -55,13 +57,15 @@ def primary_step( model_config= model_config, training_config= training_config, model_suffix = 'meta', - preloaded_models =preloaded_training_step.convert_step_to_tuple('metalabeling') if preloaded_training_step is not None else None + from_index = from_index, + preloaded_models = preloaded_training_step.get_metalabeling()[model_name] if preloaded_training_step is not None else None ) current_result[model_name] = primary_meta_result current_predictions[model_name] = primary_meta_preds training_step.metalabeling.append(meta_labeling_models) - + + reporting.results = pd.concat([reporting.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. reporting.all_predictions = pd.concat([reporting.all_predictions, current_predictions], axis=1).fillna(0.) @@ -74,10 +78,11 @@ def secondary_step( X:pd.DataFrame, y:pd.Series, current_predictions:pd.DataFrame, - asset:list, target_returns:pd.Series, configs: dict, - reporting: Reporting + reporting: Reporting, + from_index: Optional[int], + preloaded_training_step: Optional[Reporting.Training_Step] = None, ) -> Reporting.Training_Step: training_step = Reporting.Training_Step(level='secondary') model_config, training_config, data_config = itemgetter('model_config', 'training_config', 'data_config')(configs) @@ -85,7 +90,7 @@ def secondary_step( # 5. Ensemble primary model predictions (If Ensemble model is present) if model_config['ensemble_model'] is not None: ensemble_result, ensemble_predictions, _, ensemble_models_one_asset = train_primary_model( - ticker_to_predict = asset[1], + ticker_to_predict = data_config['target_asset'][1], X = current_predictions, y = y, target_returns = target_returns, @@ -94,10 +99,12 @@ def secondary_step( expanding_window = False, sliding_window_size = 1, retrain_every = training_config['retrain_every'], + from_index = from_index, scaler = training_config['scaler'], no_of_classes = data_config['no_of_classes'], level = 'ensemble', print_results= True, + preloaded_models = preloaded_training_step.get_base() if preloaded_training_step is not None else None ) ensemble_result, ensemble_predictions = ensemble_result.iloc[:,0], ensemble_predictions.iloc[:,0] @@ -111,7 +118,7 @@ def secondary_step( # 3. Train a Meta-labeling model on the averaged level-1 model predictions ensemble_meta_result, ensemble_meta_predictions, ensemble_meta_probabilities, ensemble_meta_labeling_models = train_meta_labeling_model( - target_asset=asset[1], + target_asset = data_config['target_asset'][1], X = X, input_predictions= ensemble_predictions, y = y, @@ -120,7 +127,9 @@ def secondary_step( data_config= data_config, model_config= model_config, training_config= training_config, - model_suffix = 'ensemble' + model_suffix = 'ensemble', + from_index = from_index, + preloaded_models = preloaded_training_step.get_metalabeling()[ensemble_predictions.name] if preloaded_training_step is not None else None ) training_step.metalabeling.append(ensemble_meta_labeling_models) diff --git a/training/walk_forward.py b/training/walk_forward.py index c53bd9b..8501562 100644 --- a/training/walk_forward.py +++ b/training/walk_forward.py @@ -15,6 +15,7 @@ def walk_forward_train( expanding_window: bool, window_size: int, retrain_every: int, + from_index: Optional[int], transformations: list[Transformation], preloaded_transformations: Optional[list[pd.Series]], ) -> tuple[pd.Series, list[pd.Series]]: @@ -23,7 +24,7 @@ def walk_forward_train( transformations_over_time = [pd.Series(index=y.index).rename(t.get_name()) for t in transformations] first_nonzero_return = max(get_first_valid_return_index(target_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y)) - train_from = first_nonzero_return + window_size + 1 + train_from = first_nonzero_return + window_size + 1 if from_index is None else from_index train_till = len(y) iterations_before_retrain = 0 @@ -80,11 +81,12 @@ def walk_forward_inference( X: pd.DataFrame, expanding_window: bool, window_size: int, + from_index: Optional[int], ) -> tuple[pd.Series, pd.DataFrame]: predictions = pd.Series(index=X.index).rename(model_name) probabilities = pd.DataFrame(index=X.index) - inference_from = get_first_valid_return_index(model_over_time) + inference_from = max(get_first_valid_return_index(model_over_time), get_first_valid_return_index(X.iloc[:,0])) if from_index is None else from_index inference_till = X.shape[0] first_model = model_over_time[inference_from] diff --git a/utils/evaluate.py b/utils/evaluate.py index c8673bb..40adfee 100644 --- a/utils/evaluate.py +++ b/utils/evaluate.py @@ -6,7 +6,7 @@ from utils.helpers import get_first_valid_return_index import pandas as pd import numpy as np -def backtest(returns: pd.Series, signal: pd.Series, transaction_cost = 0.003) -> pd.Series: +def backtest(returns: pd.Series, signal: pd.Series, transaction_cost = 0.002) -> pd.Series: delta_pos = signal.diff(1).abs().fillna(0.) costs = transaction_cost * delta_pos return (signal * returns) - costs