From 3eb3ea94e311c57715d2eacfcda50fe2514d32b5 Mon Sep 17 00:00:00 2001 From: Mark Aron Szulyovszky Date: Sat, 29 Jan 2022 06:41:40 +0100 Subject: [PATCH] Refactor(Training): new outcome types, representative pipeline steps, bet-sizing (#187) * refactor(Training): added InferenceResult & TrainedModel types * refactor(Pipeline): introduced TrainingOutcome, BetSizingWithMetaOutcome, etc. * fix(Pipeline): getting it to compile * refactor(WalkForward): separate preprocessing step * feat(Pipeline): separate out transformations processing step * refactor(Pipeline): use the Directional model terminology, put bet_sizing into pipeline instead of hiding it in a step * refactor(WalkForward): moved functions to separate folder * fix(WalkForward): use sparse array to store models, process transformations in parallel (lot faster) * fix(Tests): and evaluation * fix(Tests): for realz * fix(Inference): preloading everything now, renamed primary models to directional models * fix(BetSizing): was running transformations on the wrong data, oops * fix(BetSizing): concatenated on the wrong axis accidentally * fix(Reporting): able to use the new Stats type * fix(BetSizing): renamed int column names * fix(Portfolio): name the column properly * fix(Reporting): rename the correct Series, lol * fix(Inference): walk_forwad_inference() can deal with models not being aligned with the starting index * fix(WalkForward): accidentally using the wrong index * fix(WalkForward): use the correct indicies to fetch last model/transformations * fix(CI): changed the name of the results --- .github/workflows/test.yml | 2 +- config/preprocess.py | 14 +- config/presets.py | 39 +++-- config/sweep_ensemble.yaml | 10 +- config/sweep_meta_labeling.yaml | 10 +- config/sweep_primary_models.yaml | 10 +- config/types.py | 22 ++- data_loader/process.py | 2 +- models/average.py | 32 ----- models/base.py | 5 +- models/model_map.py | 55 +++---- models/momentum.py | 3 - models/naive.py | 3 - models/neural.py | 4 - models/sklearn.py | 3 - models/statsmodels.py | 3 - models/xgboost.py | 3 - reporting/reporting.py | 38 ++--- reporting/saving.py | 16 +-- reporting/types.py | 43 ------ reporting/wandb.py | 6 +- run_inference.py | 45 +++--- run_pipeline.py | 47 +++--- tests/test_evaluation.py | 17 +-- tests/test_walk_forward.py | 19 ++- training/bet_sizing.py | 88 ++++++++++++ training/directional_training.py | 62 ++++++++ training/ensemble.py | 22 +++ training/meta_labeling.py | 65 --------- training/primary_model.py | 103 ------------- training/train_model.py | 85 +++++++++++ training/training_steps.py | 135 ------------------ training/types.py | 51 +++++++ training/walk_forward.py | 123 ---------------- training/walk_forward/__init__.py | 3 + training/walk_forward/inference.py | 58 ++++++++ training/walk_forward/inference_parallel.py | 62 ++++++++ .../walk_forward/process_transformations.py | 48 +++++++ .../process_transformations_parallel.py | 46 ++++++ training/walk_forward/train.py | 55 +++++++ utils/evaluate.py | 48 +++---- utils/helpers.py | 3 + 42 files changed, 772 insertions(+), 736 deletions(-) delete mode 100644 models/average.py delete mode 100644 reporting/types.py create mode 100644 training/bet_sizing.py create mode 100644 training/directional_training.py create mode 100644 training/ensemble.py delete mode 100644 training/meta_labeling.py delete mode 100644 training/primary_model.py create mode 100644 training/train_model.py delete mode 100644 training/training_steps.py create mode 100644 training/types.py delete mode 100644 training/walk_forward.py create mode 100644 training/walk_forward/__init__.py create mode 100644 training/walk_forward/inference.py create mode 100644 training/walk_forward/inference_parallel.py create mode 100644 training/walk_forward/process_transformations.py create mode 100644 training/walk_forward/process_transformations_parallel.py create mode 100644 training/walk_forward/train.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index aa38086..8047ae9 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -56,5 +56,5 @@ jobs: env: REPO_TOKEN: ${{ secrets.GITHUB_TOKEN }} run: | - cat output/results_level2.csv >> report.md + cat output/results.csv >> report.md cml-send-comment report.md diff --git a/config/preprocess.py b/config/preprocess.py index a8bc386..5f1ce63 100644 --- a/config/preprocess.py +++ b/config/preprocess.py @@ -28,11 +28,9 @@ def __preprocess_feature_extractors_config(data_dict: dict) -> dict: return data_dict def __preprocess_model_config(model_config:dict) -> dict: - model_config['primary_models'] = [(model_name, get_model(model_name)) for model_name in model_config['primary_models']] - if len(model_config['meta_labeling_models']) > 0: - model_config['meta_labeling_models'] = [(model_name, get_model(model_name)) for model_name in model_config['meta_labeling_models']] - if model_config['ensemble_model'] is not None: - model_config['ensemble_model'] = (model_config['ensemble_model'], get_model(model_config['ensemble_model'])) + model_config['directional_models'] = [get_model(model_name) for model_name in model_config['directional_models']] + if len(model_config['meta_models']) > 0: + model_config['meta_models'] = [get_model(model_name) for model_name in model_config['meta_models']] return model_config @@ -57,9 +55,9 @@ def __preprocess_event_labeller_config(data_dict: dict) -> dict: def validate_config(config: Config): # We need to make sure there's only one output from the pipeline - # If level-2 model is there, we need more than one level-1 models to train - if len(config.meta_labeling_models) > 1: assert len(config.primary_models) > 0 + # If meta model is there, we need more than one directional models to train + if len(config.meta_models) > 1: assert len(config.directional_models) > 0 # If there's no level-2 model, we need to have only one level-1 model - if len(config.meta_labeling_models) == 0: assert len(config.primary_models) == 1 + if len(config.meta_models) == 0: assert len(config.directional_models) == 1 diff --git a/config/presets.py b/config/presets.py index d3e9643..c32e5fd 100644 --- a/config/presets.py +++ b/config/presets.py @@ -6,13 +6,13 @@ def get_dev_config() -> RawConfig: classification_models = ["LogisticRegression_two_class"] return RawConfig( - primary_models_meta_labeling = False, + directional_models_meta = False, dimensionality_reduction = False, n_features_to_select = 30, expanding_window_base = False, - expanding_window_meta_labeling = False, + expanding_window_meta = False, sliding_window_size_base = 380, - sliding_window_size_meta_labeling = 1, + sliding_window_size_meta = 1, retrain_every = 20, scaler = 'minmax', # 'normalize' 'minmax' 'standardize' @@ -25,9 +25,8 @@ def get_dev_config() -> RawConfig: other_features = ['single_mom'], exogenous_features = ['z_score'], - primary_models = classification_models, - meta_labeling_models = [], - ensemble_model = None, + directional_models = classification_models, + meta_models = [], event_filter = 'none', labeling = 'two_class' @@ -38,17 +37,16 @@ def get_default_ensemble_config() -> RawConfig: regression_models = ["Lasso", "KNN", "RFR"] classification_models = ["LogisticRegression_two_class", "LDA", "NB", "RFC", "XGB_two_class", "LGBM", "StaticMom"] - meta_labeling_models = ['LogisticRegression_two_class', 'LGBM'] - ensemble_model = 'Average' + meta_models = ['LogisticRegression_two_class', 'LGBM'] return RawConfig( - primary_models_meta_labeling = True, + directional_models_meta = True, dimensionality_reduction = False, n_features_to_select = 30, expanding_window_base = False, - expanding_window_meta_labeling = True, + expanding_window_meta = True, sliding_window_size_base = 380, - sliding_window_size_meta_labeling = 240, + sliding_window_size_meta = 240, retrain_every = 10, scaler = 'minmax', # 'normalize' 'minmax' 'standardize' @@ -61,9 +59,8 @@ def get_default_ensemble_config() -> RawConfig: other_features = ['level_2', 'lags_up_to_5'], exogenous_features = ['z_score'], - primary_models = classification_models, - meta_labeling_models = meta_labeling_models, - ensemble_model = ensemble_model, + directional_models = classification_models, + meta_models = meta_models, event_filter = 'cusum_vol', labeling = 'two_class' @@ -75,17 +72,16 @@ def get_lightweight_ensemble_config() -> RawConfig: regression_models = ["Lasso", "KNN"] classification_models = ['LogisticRegression_two_class', 'SVC'] - meta_labeling_models = ['LogisticRegression_two_class', 'LGBM'] - ensemble_model = 'Average' + meta_models = ['LogisticRegression_two_class', 'LGBM'] return RawConfig( - primary_models_meta_labeling = True, + directional_models_meta = True, dimensionality_reduction = True, n_features_to_select = 30, expanding_window_base = False, - expanding_window_meta_labeling = True, + expanding_window_meta = True, sliding_window_size_base = 380, - sliding_window_size_meta_labeling = 240, + sliding_window_size_meta = 240, retrain_every = 40, scaler = 'minmax', # 'normalize' 'minmax' 'standardize' @@ -98,9 +94,8 @@ def get_lightweight_ensemble_config() -> RawConfig: other_features = ['level_2'], exogenous_features = ['z_score'], - primary_models = classification_models, - meta_labeling_models = meta_labeling_models, - ensemble_model = ensemble_model, + directional_models = classification_models, + meta_models = meta_models, event_filter = 'none', labeling = 'two_class' diff --git a/config/sweep_ensemble.yaml b/config/sweep_ensemble.yaml index 5e426d9..fe6cabf 100644 --- a/config/sweep_ensemble.yaml +++ b/config/sweep_ensemble.yaml @@ -6,7 +6,7 @@ metric: goal: maximize name: sharpe parameters: - primary_models_meta_labeling: + directional_models_meta: value: True assets: value: ['daily_crypto'] @@ -16,11 +16,11 @@ parameters: value: ['daily_glassnode'] expanding_window_base: value: True - expanding_window_meta_labeling: + expanding_window_meta: value: True sliding_window_size_base: value: 380 - sliding_window_size_meta_labeling: + sliding_window_size_meta: values: [250, 300, 380] distribution: categorical n_features_to_select: @@ -36,13 +36,13 @@ parameters: value: 'two' load_non_target_asset: value: True - primary_models: + directional_models: distribution: categorical values: - ["LDA", "LogisticRegression_two_class", "KNN", "SVC", "CART", "NB", "AB", "RFC", "XGB_two_class", "LGBM", "StaticMom"] - ["LogisticRegression_two_class", "LDA", "NB", "RFC", "XGB_two_class", "LGBM", "StaticMom"] - ["LogisticRegression_two_class", "LDA", "LGBM", "RFC", "XGB_two_class"] - meta_labeling_models: + meta_models: value: ["LGBM", "LogisticRegression_two_class"] own_features: value: ['date_days', 'level_2', 'lags_up_to_5'] diff --git a/config/sweep_meta_labeling.yaml b/config/sweep_meta_labeling.yaml index 259b570..712a5ca 100644 --- a/config/sweep_meta_labeling.yaml +++ b/config/sweep_meta_labeling.yaml @@ -6,7 +6,7 @@ metric: goal: maximize name: sharpe parameters: - primary_models_meta_labeling: + directional_models_meta: value: True assets: value: ['daily_crypto'] @@ -17,7 +17,7 @@ parameters: expanding_window_base: values: [True, False] distribution: categorical - expanding_window_meta_labeling: + expanding_window_meta: values: [True, False] distribution: categorical n_features_to_select: @@ -28,7 +28,7 @@ parameters: sliding_window_size_base: values: [180, 280, 380] distribution: categorical - sliding_window_size_meta_labeling: + sliding_window_size_meta: values: [180, 280, 380] distribution: categorical retrain_every: @@ -42,9 +42,9 @@ parameters: load_non_target_asset: values: [True, False] distribution: categorical - primary_models: + directional_models: value: ["LogisticRegression_two_class", "LDA", "KNN", "CART", "NB", "AB", "RFC", "StaticMom"] - meta_labeling_models: + meta_models: values: ["LogisticRegression_two_class", "LDA", "KNN", "CART", "NB", "AB", "RFC"] distribution: categorical own_features: diff --git a/config/sweep_primary_models.yaml b/config/sweep_primary_models.yaml index 0be3722..5c30137 100644 --- a/config/sweep_primary_models.yaml +++ b/config/sweep_primary_models.yaml @@ -6,7 +6,7 @@ metric: goal: maximize name: sharpe parameters: - primary_models_meta_labeling: + directional_models_meta: value: True assets: value: ['daily_crypto'] @@ -17,7 +17,7 @@ parameters: expanding_window_base: values: [True, False] distribution: categorical - expanding_window_meta_labeling: + expanding_window_meta: value: False n_features_to_select: value: 50 @@ -25,7 +25,7 @@ parameters: value: True sliding_window_size_base: value: 380 - sliding_window_size_meta_labeling: + sliding_window_size_meta: value: 380 retrain_every: values: [10, 20, 30] @@ -36,10 +36,10 @@ parameters: value: 'two' load_non_target_asset: value: True - primary_models: + directional_models: values: [['LogisticRegression_two_class'], ['SVC'], ['LDA'], ['KNN'], ['CART'], ['MNB'], ['NB'], ['AB'], ['RFC'], ['XGB_two_class'], ['LGBM']] distribution: categorical - meta_labeling_models: + meta_models: value: ["LGBM", "LogisticRegression_two_class"] own_features: value: ['date_days', 'level_2', 'lags_up_to_5'] diff --git a/config/types.py b/config/types.py index 2f2fb01..77c12c7 100644 --- a/config/types.py +++ b/config/types.py @@ -9,13 +9,13 @@ from labeling.types import EventFilter, EventLabeller # RawConfig is needed to ensure we can declare config presets here with static typing, we then convert it to Config class RawConfig(BaseModel): - primary_models_meta_labeling: bool + directional_models_meta: bool dimensionality_reduction: bool n_features_to_select: int expanding_window_base: bool - expanding_window_meta_labeling: bool + expanding_window_meta: bool sliding_window_size_base: int - sliding_window_size_meta_labeling: int + sliding_window_size_meta: int retrain_every: int scaler: Literal['normalize', 'minmax', 'standardize'] @@ -30,19 +30,18 @@ class RawConfig(BaseModel): event_filter: Literal['none', 'cusum_vol', 'cusum_fixed'] labeling: Literal['two_class', 'three_class_balanced', 'three_class_imbalanced'] - primary_models: list[str] - meta_labeling_models: list[str] - ensemble_model: Optional[str] + directional_models: list[str] + meta_models: list[str] class Config(BaseModel): - primary_models_meta_labeling: bool + directional_models_meta: bool dimensionality_reduction: bool n_features_to_select: int expanding_window_base: bool - expanding_window_meta_labeling: bool + expanding_window_meta: bool sliding_window_size_base: int - sliding_window_size_meta_labeling: int + sliding_window_size_meta: int retrain_every: int scaler: Literal['normalize', 'minmax', 'standardize'] @@ -58,9 +57,8 @@ class Config(BaseModel): labeling: EventLabeller no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'] - primary_models: list[tuple[str, Model]] - meta_labeling_models: list[tuple[str, Model]] - ensemble_model: Optional[tuple[str, Model]] + directional_models: list[Model] + meta_models: list[Model] class Config: arbitrary_types_allowed = True diff --git a/data_loader/process.py b/data_loader/process.py index 94dcb4a..01bac1c 100644 --- a/data_loader/process.py +++ b/data_loader/process.py @@ -16,4 +16,4 @@ def check_data(X: XDataFrame, config: Config) -> bool: def has_enough_samples_to_train(X: XDataFrame, config: Config) -> bool: first_valid_index = get_first_valid_return_index(X.iloc[:,0]) samples_to_train = len(X) - first_valid_index - return samples_to_train > config.sliding_window_size_base + config.sliding_window_size_meta_labeling + 100 \ No newline at end of file + return samples_to_train > config.sliding_window_size_base + config.sliding_window_size_meta + 100 \ No newline at end of file diff --git a/models/average.py b/models/average.py deleted file mode 100644 index d72bc5a..0000000 --- a/models/average.py +++ /dev/null @@ -1,32 +0,0 @@ -from __future__ import annotations -from models.base import Model -import numpy as np - -class StaticAverageModel(Model): - ''' - Model that averages . - ''' - - data_transformation = 'original' - only_column = 'model_' - model_type = 'static' - predict_window_size = 'single_timestamp' - - def fit(self, X: np.ndarray, y: np.ndarray) -> None: - # This is a static model, it can' learn anything - pass - - def predict(self, X) -> tuple[float, np.ndarray]: - # Make sure there's data to average - assert X.shape[1] > 0 - prediction = np.average(X[-1]) - return (prediction, np.array([])) - - def clone(self) -> StaticAverageModel: - return self - - def get_name(self) -> str: - return 'static_average' - - def initialize_network(self, input_dim:int, output_dim:int): - pass \ No newline at end of file diff --git a/models/base.py b/models/base.py index 4dc22d8..5eea960 100644 --- a/models/base.py +++ b/models/base.py @@ -5,6 +5,7 @@ import numpy as np class Model(ABC): + name: str = "" method: Literal["regression", "classification"] data_transformation: Literal["transformed", "original"] only_column: Optional[str] @@ -22,10 +23,6 @@ class Model(ABC): @abstractmethod def clone(self) -> Model: raise NotImplementedError - - @abstractmethod - def get_name(self) -> str: - raise NotImplementedError @abstractmethod def initialize_network(self, input_dim:int, output_dim:int): diff --git a/models/model_map.py b/models/model_map.py index 3850ba1..37bdc73 100644 --- a/models/model_map.py +++ b/models/model_map.py @@ -4,89 +4,92 @@ from sklearnex.ensemble import RandomForestRegressor from .base import Model default_feature_selector_classification = SKLearnModel(RandomForestClassifier(n_jobs=-1, max_depth=20, random_state=1), 'classification') -default_feature_selector_regression = SKLearnModel(RandomForestRegressor(n_jobs=-1, max_depth=20, random_state=1), 'regression') def get_model(model_name: str) -> Model: + def set_name(model: Model) -> Model: + model.name = model_name + return model + if model_name == 'LinearRegression': from sklearn.linear_model import LinearRegression - return SKLearnModel(LinearRegression(n_jobs=-1), 'regression') + return set_name(SKLearnModel(LinearRegression(n_jobs=-1), 'regression')) elif model_name == 'Lasso': from sklearn.linear_model import Lasso - return SKLearnModel(Lasso(alpha=100, random_state=1), 'regression') + return set_name(SKLearnModel(Lasso(alpha=100, random_state=1), 'regression')) elif model_name == 'Ridge': from sklearn.linear_model import Ridge - return SKLearnModel(Ridge(alpha=0.1), 'regression') + return set_name(SKLearnModel(Ridge(alpha=0.1), 'regression')) elif model_name == 'BayesianRidge': from sklearn.linear_model import BayesianRidge - return SKLearnModel(BayesianRidge(), 'regression') + return set_name(SKLearnModel(BayesianRidge(), 'regression')) elif model_name == 'KNN': from sklearnex.neighbors import KNeighborsRegressor - return SKLearnModel(KNeighborsRegressor(n_neighbors=25), 'regression') + return set_name(SKLearnModel(KNeighborsRegressor(n_neighbors=25), 'regression')) elif model_name == 'AB': from sklearn.ensemble import AdaBoostRegressor - return SKLearnModel(AdaBoostRegressor(random_state=1), 'regression') + return set_name(SKLearnModel(AdaBoostRegressor(random_state=1), 'regression')) elif model_name == 'MLP': from sklearn.neural_network import MLPRegressor - return SKLearnModel(MLPRegressor(hidden_layer_sizes=(100,20), max_iter=1000), 'regression') + return set_name(SKLearnModel(MLPRegressor(hidden_layer_sizes=(100,20), max_iter=1000), 'regression')) elif model_name == 'RFR': - return SKLearnModel(RandomForestRegressor(n_jobs=-1, max_depth=20, random_state=1), 'regression') + return set_name(SKLearnModel(RandomForestRegressor(n_jobs=-1, max_depth=20, random_state=1), 'regression')) elif model_name == 'SVR': from sklearnex.svm import SVR - return SKLearnModel(SVR(kernel='rbf', C=1e3, gamma=0.1), 'regression') + return set_name(SKLearnModel(SVR(kernel='rbf', C=1e3, gamma=0.1), 'regression')) elif model_name == 'StaticNaive': from models.naive import StaticNaiveModel - return StaticNaiveModel() + return set_name(StaticNaiveModel()) elif model_name == 'DNN': from models.neural import LightningNeuralNetModel from models.pytorch.neural_nets import MultiLayerPerceptron import torch.nn.functional as F - return LightningNeuralNetModel( + return set_name(LightningNeuralNetModel( MultiLayerPerceptron( hidden_layers_ratio = [1.0], probabilities = False, loss_function = F.mse_loss), max_epochs=15 - ) + )) elif model_name == 'LogisticRegression_two_class': from sklearn.linear_model import LogisticRegression - return SKLearnModel(LogisticRegression(C=10, random_state=1, solver='liblinear', max_iter=1000), 'classification') + return set_name(SKLearnModel(LogisticRegression(C=10, random_state=1, solver='liblinear', max_iter=1000), 'classification')) elif model_name == 'LogisticRegression_three_class': from sklearnex.linear_model import LogisticRegression as LogisticRegression_EX - return SKLearnModel(LogisticRegression_EX(C=10, random_state=1, max_iter=1000, n_jobs=-1), 'classification') + return set_name(SKLearnModel(LogisticRegression_EX(C=10, random_state=1, max_iter=1000, n_jobs=-1), 'classification')) elif model_name == 'LDA': from sklearn.discriminant_analysis import LinearDiscriminantAnalysis - return SKLearnModel(LinearDiscriminantAnalysis(), 'classification') + return set_name(SKLearnModel(LinearDiscriminantAnalysis(), 'classification')) elif model_name == 'KNN': from sklearn.neighbors import KNeighborsClassifier - return SKLearnModel(KNeighborsClassifier(), 'classification') + return set_name(SKLearnModel(KNeighborsClassifier(), 'classification')) elif model_name == 'CART': from sklearn.tree import DecisionTreeClassifier - return SKLearnModel(DecisionTreeClassifier(max_depth=15, random_state=1), 'classification') + return set_name(SKLearnModel(DecisionTreeClassifier(max_depth=15, random_state=1), 'classification')) elif model_name == 'NB': from sklearn.naive_bayes import GaussianNB - return SKLearnModel(GaussianNB(), 'classification') + return set_name(SKLearnModel(GaussianNB(), 'classification')) elif model_name == 'AB': from sklearn.ensemble import AdaBoostClassifier - return SKLearnModel(AdaBoostClassifier(n_estimators=15), 'classification') + return set_name(SKLearnModel(AdaBoostClassifier(n_estimators=15), 'classification')) elif model_name == 'RFC': - return SKLearnModel(RandomForestClassifier(n_jobs=-1, max_depth=20, random_state=1), 'classification') + return set_name(SKLearnModel(RandomForestClassifier(n_jobs=-1, max_depth=20, random_state=1), 'classification')) elif model_name == 'SVC': from sklearn.svm import SVC - return SKLearnModel(SVC(kernel='rbf', C=1e3, probability=True, random_state=1), 'classification') + return set_name(SKLearnModel(SVC(kernel='rbf', C=1e3, probability=True, random_state=1), 'classification')) elif model_name == 'XGB_two_class': from xgboost import XGBClassifier from models.xgboost import XGBoostModel - return XGBoostModel(XGBClassifier(n_jobs=-1, max_depth = 20, random_state=1, objective='binary:logistic', use_label_encoder= False, eval_metric='mlogloss')) + return set_name(XGBoostModel(XGBClassifier(n_jobs=-1, max_depth = 20, random_state=1, objective='binary:logistic', use_label_encoder= False, eval_metric='mlogloss'))) elif model_name == 'LGBM': from lightgbm import LGBMClassifier - return SKLearnModel(LGBMClassifier(n_jobs=-1, max_depth=20, random_state=1), 'classification') + return set_name(SKLearnModel(LGBMClassifier(n_jobs=-1, max_depth=20, random_state=1), 'classification')) elif model_name == 'StaticMom': from models.momentum import StaticMomentumModel - return StaticMomentumModel(allow_short=True) + return set_name(StaticMomentumModel(allow_short=True)) elif model_name == 'Average': from models.average import StaticAverageModel - return StaticAverageModel() + return set_name(StaticAverageModel()) else: raise Exception(f'Model {model_name} not found') \ No newline at end of file diff --git a/models/momentum.py b/models/momentum.py index b93e93e..c969fa7 100644 --- a/models/momentum.py +++ b/models/momentum.py @@ -28,9 +28,6 @@ class StaticMomentumModel(Model): def clone(self) -> StaticMomentumModel: return self - - def get_name(self) -> str: - return 'static_mom' def initialize_network(self, input_dim:int, output_dim:int): pass \ No newline at end of file diff --git a/models/naive.py b/models/naive.py index 1efa317..81a9033 100644 --- a/models/naive.py +++ b/models/naive.py @@ -23,8 +23,5 @@ class StaticNaiveModel(Model): def clone(self) -> StaticNaiveModel: return self - def get_name(self) -> str: - return 'static_naive' - def initialize_network(self, input_dim:int, output_dim:int): pass \ No newline at end of file diff --git a/models/neural.py b/models/neural.py index 2c78809..114bbc5 100644 --- a/models/neural.py +++ b/models/neural.py @@ -36,7 +36,3 @@ class LightningNeuralNetModel(Model): def initialize_network(self, input_dim:int, output_dim:int): self.model.initialize_network(input_dim, output_dim) - - def get_name(self) -> str: - return self.model.__class__.__name__ - \ No newline at end of file diff --git a/models/sklearn.py b/models/sklearn.py index 2da08e7..8b681ff 100644 --- a/models/sklearn.py +++ b/models/sklearn.py @@ -27,9 +27,6 @@ class SKLearnModel(Model): def clone(self) -> SKLearnModel: return SKLearnModel(clone(self.model), self.method) - - def get_name(self) -> str: - return self.model.__class__.__name__ def initialize_network(self, input_dim:int, output_dim:int): pass diff --git a/models/statsmodels.py b/models/statsmodels.py index d67086e..9e43c32 100644 --- a/models/statsmodels.py +++ b/models/statsmodels.py @@ -25,9 +25,6 @@ class StatsModel(Model): def clone(self) -> StatsModel: return StatsModel(deepcopy(self.model)) - - def get_name(self) -> str: - return self.model.__class__.__name__ def initialize_network(self, input_dim:int, output_dim:int): pass diff --git a/models/xgboost.py b/models/xgboost.py index a7c0ab9..3473615 100644 --- a/models/xgboost.py +++ b/models/xgboost.py @@ -27,9 +27,6 @@ class XGBoostModel(Model): def clone(self) -> XGBoostModel: return XGBoostModel(clone(self.model)) - - def get_name(self) -> str: - return self.model.__class__.__name__ def initialize_network(self, input_dim:int, output_dim:int): pass diff --git a/reporting/reporting.py b/reporting/reporting.py index 3280943..4989b4f 100644 --- a/reporting/reporting.py +++ b/reporting/reporting.py @@ -2,39 +2,29 @@ from reporting.wandb import send_report_to_wandb import pandas as pd from utils.helpers import weighted_average from config.types import Config +from training.types import WeightsSeries, Stats -def report_results(results:pd.DataFrame, all_predictions:pd.DataFrame, config: Config, wandb, sweep: bool, project_name:str): - - primary_results = results[[column for column in results.columns if 'ensemble' not in column]] - ensemble_results = results[[column for column in results.columns if 'ensemble' in column]] +def report_results(directional_stats: list[Stats], output_stats: Stats, output_weights: WeightsSeries, config: Config, wandb, sweep: bool): # Only send the results of the final model to wandb - results_to_send = ensemble_results if ensemble_results.shape[1] > 0 else primary_results - send_report_to_wandb(results_to_send, wandb) - results.to_csv('output/results.csv') + send_report_to_wandb(output_stats, wandb) + pd.Series(output_stats).to_csv('output/results.csv') - primary_weights = all_predictions[[column for column in all_predictions.columns if 'ensemble' not in column]] - ensemble_weights = all_predictions[[column for column in all_predictions.columns if 'ensemble' in column]] - predictions_to_save = ensemble_weights if ensemble_weights.shape[1] > 0 else primary_weights - predictions_to_save.to_csv('output/predictions.csv') + output_weights.rename(config.target_asset[1]).to_csv('output/predictions.csv') print("\n--------\n") - all_avg_results = weighted_average(results, 'no_of_samples') - primary_avg_results = weighted_average(primary_results, 'no_of_samples') - ensemble_avg_results = weighted_average(ensemble_results, 'no_of_samples') + directional_avg_stats = weighted_average(pd.concat([pd.Series(stat) for stat in directional_stats], axis = 1), 'no_of_samples') - print("Benchmark buy-and-hold sharpe: ", round(all_avg_results.loc['benchmark_sharpe'], 3)) + print("Benchmark buy-and-hold sharpe: ", output_stats['benchmark_sharpe']) - print("Level-1: Number of samples evaluated: ", primary_results.loc['no_of_samples'].sum()) - print("Mean Sharpe ratio for Level-1 models: ", round(primary_avg_results.loc['sharpe'], 3)) - print("Mean Probabilistic Sharpe ratio for Level-1 models: ", round(primary_avg_results.loc['prob_sharpe'].mean(), 3)) + print("Level-1: Number of samples evaluated: ", directional_avg_stats.loc['no_of_samples'].sum()) + print("Mean Sharpe ratio for Level-1 models: ", round(directional_avg_stats.loc['sharpe'], 3)) + print("Mean Probabilistic Sharpe ratio for Level-1 models: ", round(directional_avg_stats.loc['prob_sharpe'].mean(), 3)) - if len(config.meta_labeling_models) > 0: - print("Level-2 (Ensemble): Number of samples evaluated: ", ensemble_results.loc['no_of_samples'].sum()) - print("Mean Sharpe ratio for Level-2 (Ensemble) models: ", round(ensemble_avg_results.loc['sharpe'].mean(), 3)) - print("Mean Probabilistic Sharpe ratio for Level-2 (Ensemble) models: ", round(ensemble_avg_results.loc['prob_sharpe'].mean(), 3)) - - ensemble_avg_results.to_csv('output/results_level2.csv') + if len(config.meta_models) > 0: + print("Level-2 (Ensemble): Number of samples evaluated: ", output_stats['no_of_samples']) + print("Mean Sharpe ratio for Level-2 (Ensemble) models: ", output_stats['sharpe']) + print("Mean Probabilistic Sharpe ratio for Level-2 (Ensemble) models: ", output_stats['prob_sharpe']) if sweep: if wandb.run is not None: diff --git a/reporting/saving.py b/reporting/saving.py index ebe038c..a561977 100644 --- a/reporting/saving.py +++ b/reporting/saving.py @@ -1,17 +1,15 @@ import pickle import datetime from config.types import Config -from typing import Optional, Union +from typing import Optional import os import warnings -from reporting.types import Reporting +from training.types import PipelineOutcome - - -def save_models(all_models: Reporting.Asset, config: Config) -> None: +def save_models(pipeline_outcome: PipelineOutcome, config: Config) -> None: dict_for_pickle = dict() dict_for_pickle['config'] = config - dict_for_pickle['all_models'] = all_models + dict_for_pickle['pipeline_outcome'] = pipeline_outcome date_string = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M") @@ -22,7 +20,7 @@ def save_models(all_models: Reporting.Asset, config: Config) -> None: pickle.dump( dict_for_pickle, open( "output/models/{}.p".format(date_string), "wb" ) ) -def load_models(file_name:Union[str, None]) -> tuple[Reporting.Asset, Config]: +def load_models(file_name: Optional[str]) -> tuple[PipelineOutcome, Config]: if file_name is None: warnings.warn("No file name provided, will load latest models and configurations.") @@ -34,7 +32,7 @@ def load_models(file_name:Union[str, None]) -> tuple[Reporting.Asset, Config]: packacked_dict = pickle.load( open( "output/models/{}".format(file_name), "rb" ) ) config = packacked_dict.pop("config", None) - all_models = packacked_dict.pop("all_models", None) + pipeline_outcome = packacked_dict.pop("pipeline_outcome", None) - return all_models, config + return pipeline_outcome, config diff --git a/reporting/types.py b/reporting/types.py deleted file mode 100644 index ce43fdb..0000000 --- a/reporting/types.py +++ /dev/null @@ -1,43 +0,0 @@ -from __future__ import annotations -from dataclasses import dataclass -import pandas as pd - - -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.asset: Reporting.Asset - - def get_results(self) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, Reporting.Asset]: - return self.results, self.all_predictions, self.all_probabilities, self.asset - - @dataclass - class Single_Model: - model_name: str - model_over_time: pd.Series - transformations_over_time: list[pd.Series] - - - 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]] = [] - - 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 - - @dataclass - class Asset(): - name: str - primary: Reporting.Training_Step - secondary: Reporting.Training_Step - diff --git a/reporting/wandb.py b/reporting/wandb.py index 03b75e2..2a848db 100644 --- a/reporting/wandb.py +++ b/reporting/wandb.py @@ -2,6 +2,7 @@ import pandas as pd from config.types import RawConfig from typing import Optional from utils.helpers import weighted_average +from training.types import Stats def launch_wandb(project_name:str, default_config: RawConfig, sweep:bool=False) -> Optional[object]: from wandb_setup import get_wandb @@ -28,14 +29,13 @@ def override_config_with_wandb_values(wandb: Optional[object], raw_config: RawCo return RawConfig(**config_dict) -def send_report_to_wandb(results: pd.DataFrame, wandb:Optional[object]): +def send_report_to_wandb(stats: Stats, wandb:Optional[object]): if wandb is None: return run = wandb.run run.save() - mean_results = weighted_average(results, 'no_of_samples') - for key, value in mean_results.iteritems(): + for key, value in stats.items(): run.log({ key: value }) run.finish() diff --git a/run_inference.py b/run_inference.py index 6a63926..4851153 100644 --- a/run_inference.py +++ b/run_inference.py @@ -6,24 +6,23 @@ from run_pipeline import run_pipeline from config.types import Config, RawConfig from config.presets import get_dev_config, get_default_ensemble_config, get_lightweight_ensemble_config from labeling.process import label_data -from typing import Callable, Optional -from reporting.types import Reporting -from training.training_steps import primary_step, secondary_step import pandas as pd -import warnings -def run_inference(preload_models:bool, raw_config: RawConfig): +from training.directional_training import train_directional_models +from training.bet_sizing import bet_sizing_with_meta_models +from training.ensemble import ensemble_weights +from training.types import PipelineOutcome + +def run_inference(preload_models:bool, fallback_raw_config: RawConfig): if preload_models: - all_models, config = load_models(None) + pipeline_outcome, config = load_models(None) else: - all_models, config, _, _, _ = run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, raw_config=raw_config) + pipeline_outcome, config = run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, raw_config=fallback_raw_config) - __inference(config, all_models.primary, all_models.secondary) + __inference(config, pipeline_outcome) -def __inference(config: Config, primary_models: Optional[Reporting.Training_Step], secondary_models: Optional[Reporting.Training_Step]): - reporting = Reporting() - asset = config.target_asset +def __inference(config: Config, pipeline_outcome: PipelineOutcome): # 1. Load data, check for validity and process data X, returns, forward_returns = load_data( @@ -42,19 +41,23 @@ def __inference(config: Config, primary_models: Optional[Reporting.Training_Step inference_from: pd.Timestamp = X.index[len(X.index) - 2] - # 2. Train a Primary model with optional metalabeling for each asset - training_step_primary, current_predictions = primary_step(X, y, forward_returns, config, reporting, from_index = inference_from, preloaded_training_step = primary_models) + # 2. Filter for significant events when we want to trade, and label data + events, X, y, forward_returns = label_data(config.event_filter, config.labeling, X, returns, forward_returns) - # 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, forward_returns, config, reporting, from_index = inference_from, preloaded_training_step = secondary_models) + # 3. Train directional models + directional_training_outcome = train_directional_models(X, y, forward_returns, config, config.directional_models, from_index = inference_from, preloaded_training_step = pipeline_outcome.directional_training) + + # 4. Run bet sizing on primary model's output + bet_sizing_outcomes = [bet_sizing_with_meta_models(X, training_outcome.predictions, y, forward_returns, config.meta_models, config, 'meta', from_index = inference_from, transformations_over_time = preloaded_outcome.meta_transformations, preloaded_models = [b.model_over_time for b in preloaded_outcome.meta_training]) for training_outcome, preloaded_outcome in zip(directional_training_outcome.training, pipeline_outcome.bet_sizing)] - # 4. Save the models - reporting.asset = Reporting.Asset(name=asset[1], primary=training_step_primary, secondary=training_step_secondary) + # 4. Ensemble weights + ensemble_outcome = ensemble_weights([o.weights for o in bet_sizing_outcomes], forward_returns, y, config.no_of_classes) - return reporting + # 5. (Optional) Additional bet sizing on top of the ensembled weights + ensemble_bet_sizing_outcome = bet_sizing_with_meta_models(X, ensemble_outcome.weights, y, forward_returns, config.meta_models, config, 'ensemble', from_index = inference_from, transformations_over_time = pipeline_outcome.secondary_bet_sizing.meta_transformations, preloaded_models= [b.model_over_time for b in pipeline_outcome.secondary_bet_sizing.meta_training]) if len(config.meta_models) > 0 else None + + return PipelineOutcome(directional_training_outcome, bet_sizing_outcomes, ensemble_outcome, ensemble_bet_sizing_outcome) if __name__ == '__main__': - run_inference(preload_models=True, raw_config=get_lightweight_ensemble_config()) \ No newline at end of file + run_inference(preload_models=True, fallback_raw_config=get_lightweight_ensemble_config()) \ No newline at end of file diff --git a/run_pipeline.py b/run_pipeline.py index eb5afac..1aa947a 100644 --- a/run_pipeline.py +++ b/run_pipeline.py @@ -1,5 +1,4 @@ -import pandas as pd -from typing import Callable, Optional +from typing import Optional from config.types import Config, RawConfig from config.preprocess import preprocess_config, validate_config @@ -12,24 +11,23 @@ from labeling.process import label_data from reporting.wandb import launch_wandb, override_config_with_wandb_values from reporting.reporting import report_results - from reporting.saving import save_models -from reporting.types import Reporting -from training.training_steps import primary_step, secondary_step +from training.directional_training import train_directional_models +from training.bet_sizing import bet_sizing_with_meta_models +from training.ensemble import ensemble_weights +from training.types import PipelineOutcome import ray ray.init() -def run_pipeline(project_name:str, with_wandb: bool, sweep: bool, raw_config: RawConfig) -> tuple[Reporting.Asset, Config, pd.DataFrame, pd.DataFrame, pd.DataFrame]: +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) - reporting = __run_training(config) - results, all_predictions, all_probabilities, all_models = reporting.get_results() - report_results(results, all_predictions, config, wandb, sweep, project_name) - save_models(all_models, config) - - return all_models, config, results, all_predictions, all_probabilities + pipeline_outcome = __run_training(config) + report_results([s.stats for s in pipeline_outcome.directional_training.training], pipeline_outcome.get_output_stats(), pipeline_outcome.get_output_weights(), config, wandb, sweep) + save_models(pipeline_outcome, config) + return pipeline_outcome, config def __setup_config(project_name:str, with_wandb: bool, sweep: bool, raw_config: RawConfig) -> tuple[Optional[object], Config]: @@ -43,10 +41,9 @@ def __setup_config(project_name:str, with_wandb: bool, sweep: bool, raw_config: -def __run_training(config: Config): +def __run_training(config: Config) -> PipelineOutcome: validate_config(config) - reporting = Reporting() # 1. Load data, check for validity X, returns, forward_returns = load_data( @@ -65,19 +62,19 @@ def __run_training(config: Config): # 2. Filter for significant events when we want to trade, and label data events, X, y, forward_returns = label_data(config.event_filter, config.labeling, X, returns, forward_returns) - # 3. Train a Primary model with optional metalabeling for each asset - training_step_primary, current_predictions = primary_step(X, y, forward_returns, config, reporting, from_index = None) - - # 4. Train an Ensemble model with optional metalabeling for each asset - training_step_secondary = secondary_step(X, y, current_predictions, forward_returns, config, reporting, from_index = None) - - # 5. Save the models - reporting.asset = Reporting.Asset(name = config.target_asset[1], primary = training_step_primary, secondary = training_step_secondary) - - return reporting - + # 3. Train directional models + directional_training_outcome = train_directional_models(X, y, forward_returns, config, config.directional_models, from_index = None, preloaded_training_step = None) + # 4. Run bet sizing on primary model's output + bet_sizing_outcomes = [bet_sizing_with_meta_models(X, outcome.predictions, y, forward_returns, config.meta_models, config, 'meta', None, None, None) for outcome in directional_training_outcome.training] + # 4. Ensemble weights + ensemble_outcome = ensemble_weights([o.weights for o in bet_sizing_outcomes], forward_returns, y, config.no_of_classes) + + # 5. (Optional) Additional bet sizing on top of the ensembled weights + ensemble_bet_sizing_outcome = bet_sizing_with_meta_models(X, ensemble_outcome.weights, y, forward_returns, config.meta_models, config, 'ensemble', None, None, None) if len(config.meta_models) > 0 else None + + return PipelineOutcome(directional_training_outcome, bet_sizing_outcomes, ensemble_outcome, ensemble_bet_sizing_outcome) if __name__ == '__main__': run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, raw_config=get_default_ensemble_config()) \ No newline at end of file diff --git a/tests/test_evaluation.py b/tests/test_evaluation.py index 1b12009..67734ee 100644 --- a/tests/test_evaluation.py +++ b/tests/test_evaluation.py @@ -53,9 +53,7 @@ class EvenOddStubModel(Model): def clone(self): return self - - def get_name(self) -> str: - return 'test' + def initialize_network(self, input_dim: int, output_dim: int): pass @@ -65,28 +63,28 @@ def test_evaluation(): X, y = __generate_even_odd_test_data(no_of_rows) window_length = 10 + retrain_every = 10 model = EvenOddStubModel(window_length = window_length) - model_over_time, transformations_over_time = walk_forward_train( - model_name='test', + model_over_time = walk_forward_train( model=model, X=X, y=y, forward_returns=y, expanding_window=False, window_size=window_length, - retrain_every=10, + retrain_every=retrain_every, from_index=None, - transformations=[], - preloaded_transformations=None) + transformations_over_time=[]) predictions, _ = walk_forward_inference( model_name='test', model_over_time=model_over_time, - transformations_over_time=transformations_over_time, + transformations_over_time=[], X=X, expanding_window=False, window_size=window_length, + retrain_every = retrain_every, from_index=None, ) @@ -98,7 +96,6 @@ def test_evaluation(): processed_predictions_to_match_returns = predictions * 0.1 result = evaluate_predictions( - model_name='test', forward_returns=fake_forward_returns, y_pred=processed_predictions_to_match_returns, y_true=y, diff --git a/tests/test_walk_forward.py b/tests/test_walk_forward.py index 4a307a4..0347efc 100644 --- a/tests/test_walk_forward.py +++ b/tests/test_walk_forward.py @@ -11,16 +11,16 @@ def __generate_incremental_test_data(no_of_rows) -> tuple[pd.DataFrame, pd.Serie no_columns = 6 X = [[row] * no_columns for row in range(no_of_rows)] - assert X[0][0] == 0 assert X[1][0] == 1 assert X[2][0] == 2 assert X[3][0] == 3 + assert X[4][0] == 4 X = pd.DataFrame(X) y = [row+1 for row in range(no_of_rows)] - assert y[0] == 1 assert y[1] == 2 assert y[2] == 3 + assert y[3] == 4 y = pd.Series(y) return X, y @@ -52,9 +52,6 @@ class IncrementingStubModel(Model): def clone(self): return self - - def get_name(self) -> str: - return 'test' def initialize_network(self, input_dim: int, output_dim: int): pass @@ -63,29 +60,29 @@ def test_walk_forward_train_test(): X, y = __generate_incremental_test_data(no_of_rows) window_length = 10 + retrain_every = 10 model = IncrementingStubModel(window_length = window_length) - model_over_time, transformations_over_time = walk_forward_train( - model_name='test', + model_over_time = walk_forward_train( model=model, X=X, y=y, forward_returns=y, expanding_window=False, window_size=window_length, - retrain_every=10, + retrain_every=retrain_every, from_index=None, - transformations=[], - preloaded_transformations=None + transformations_over_time=[], ) predictions, _ = walk_forward_inference( model_name='test', model_over_time=model_over_time, - transformations_over_time=transformations_over_time, + transformations_over_time=[], X=X, expanding_window=False, window_size=window_length, + retrain_every=retrain_every, from_index=None, ) diff --git a/training/bet_sizing.py b/training/bet_sizing.py new file mode 100644 index 0000000..d911ae2 --- /dev/null +++ b/training/bet_sizing.py @@ -0,0 +1,88 @@ +from data_loader.types import ForwardReturnSeries, XDataFrame, ySeries +from utils.evaluate import discretize_threeway_threshold, evaluate_predictions +from utils.helpers import equal_except_nan +from .train_model import train_models +import pandas as pd +from models.base import Model +from models.model_map import default_feature_selector_classification +from typing import Optional +from config.types import Config +from .types import BetSizingWithMetaOutcome, ModelOverTime, TransformationsOverTime +from training.walk_forward import walk_forward_process_transformations +from transformations.scaler import get_scaler +from transformations.rfe import RFETransformation +from transformations.pca import PCATransformation + +def bet_sizing_with_meta_models( + X: XDataFrame, + input_predictions: pd.Series, + y: ySeries, + forward_returns: ForwardReturnSeries, + models: list[Model], + config: Config, + model_suffix: str, + from_index: Optional[pd.Timestamp], + transformations_over_time: Optional[TransformationsOverTime] = None, + preloaded_models: Optional[list[ModelOverTime]] = None + ) -> BetSizingWithMetaOutcome: + + input_predictions.name = "model_predictions" + discretized_predictions = input_predictions.apply(discretize_threeway_threshold(0.33)) + discretized_predictions.name = "model_discretized_predictions" + + meta_y: pd.Series = pd.concat([discretized_predictions, y], axis=1).apply(equal_except_nan, axis = 1) + meta_X = pd.concat([X, input_predictions, discretized_predictions], axis = 1) + + if transformations_over_time is None: + print("Preprocess transformations") + transformations_over_time = walk_forward_process_transformations( + X = meta_X, + y = meta_y, + forward_returns = forward_returns, + expanding_window = config.expanding_window_meta, + window_size = config.sliding_window_size_meta, + retrain_every = config.retrain_every, + from_index = from_index, + transformations= [ + get_scaler(config.scaler), + PCATransformation(ratio_components_to_keep=0.5, sliding_window_size=config.sliding_window_size_meta), + RFETransformation(n_feature_to_select=40, model=default_feature_selector_classification) + ], + ) + + meta_outcomes = train_models( + ticker_to_predict = "prediction_correct", + X = meta_X, + y = meta_y, + forward_returns = forward_returns, + models = models, + expanding_window = config.expanding_window_meta, + sliding_window_size = config.sliding_window_size_meta, + retrain_every = config.retrain_every, + from_index = from_index, + no_of_classes = 'two', + level = 'meta', + print_results = False, + transformations_over_time = transformations_over_time, + models_over_time = preloaded_models, + ) + + # Ensemble predictions if necessary + if len(models) > 1: + # meta_predictions = pd.concat([outcome.predictions for outcome in meta_outcomes]).mean(axis = 1) + bet_size = pd.concat([outcome.probabilities[outcome.probabilities.columns[1::2]] for outcome in meta_outcomes], axis = 1).mean(axis = 1) + else: + bet_size = meta_outcomes[0].probabilities.iloc[:,1] + avg_predictions_with_sizing = input_predictions * bet_size + + stats = evaluate_predictions( + forward_returns = forward_returns, + y_pred = avg_predictions_with_sizing, + y_true = y, + no_of_classes = 'two', + print_results = True, + discretize=False + ) + model_id = "model_" + config.target_asset[1] + "_" + model_suffix + + return BetSizingWithMetaOutcome(model_id, meta_outcomes, transformations_over_time, avg_predictions_with_sizing, stats) diff --git a/training/directional_training.py b/training/directional_training.py new file mode 100644 index 0000000..42df5bd --- /dev/null +++ b/training/directional_training.py @@ -0,0 +1,62 @@ +import pandas as pd + +from .types import DirectionalTrainingOutcome +from training.train_model import train_model +from training.walk_forward import walk_forward_process_transformations + +from typing import Optional +from config.types import Config +from models.base import Model +from models.model_map import default_feature_selector_classification + +from transformations.scaler import get_scaler +from transformations.rfe import RFETransformation +from transformations.pca import PCATransformation + +def train_directional_models( + X: pd.DataFrame, + y: pd.Series, + forward_returns: pd.Series, + config: Config, + models: list[Model], + from_index: Optional[pd.Timestamp], + preloaded_training_step: Optional[DirectionalTrainingOutcome] = None, + ) -> DirectionalTrainingOutcome: + + if preloaded_training_step is None: + print("Preprocess transformations") + transformations_over_time = walk_forward_process_transformations( + X = X, + y = y, + forward_returns = forward_returns, + expanding_window = config.expanding_window_base, + window_size = config.sliding_window_size_base, + retrain_every = config.retrain_every, + from_index = from_index, + transformations= [ + get_scaler(config.scaler), + PCATransformation(ratio_components_to_keep=0.5, sliding_window_size=config.sliding_window_size_base), + RFETransformation(n_feature_to_select=40, model=default_feature_selector_classification) + ], + ) + else: + transformations_over_time = preloaded_training_step.transformations + + training_outcomes = [train_model( + ticker_to_predict = config.target_asset[1], + X = X, + y = y, + forward_returns = forward_returns, + model = model, + expanding_window = config.expanding_window_base, + sliding_window_size = config.sliding_window_size_base, + retrain_every = config.retrain_every, + from_index = from_index, + no_of_classes = config.no_of_classes, + level = 'primary', + print_results= True, + transformations_over_time = transformations_over_time, + model_over_time = preloaded_training_step.training[index].model_over_time if preloaded_training_step else None + ) for index, model in enumerate(models)] + return DirectionalTrainingOutcome(training_outcomes, transformations_over_time) + diff --git a/training/ensemble.py b/training/ensemble.py new file mode 100644 index 0000000..7eaada6 --- /dev/null +++ b/training/ensemble.py @@ -0,0 +1,22 @@ +from .types import WeightsSeries, EnsembleOutcome +import pandas as pd +from utils.evaluate import evaluate_predictions +from data_loader.types import ForwardReturnSeries, ySeries +from typing import Literal + +def ensemble_weights( + input_weights: list[WeightsSeries], + forward_returns: ForwardReturnSeries, + y: ySeries, + no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'], + ) -> EnsembleOutcome: + weights = pd.concat(input_weights, axis=1).mean(axis=1) + stats = evaluate_predictions( + forward_returns = forward_returns, + y_pred = weights, + y_true = y, + no_of_classes = no_of_classes, + print_results = False, + discretize = True, + ) + return EnsembleOutcome(weights, stats) diff --git a/training/meta_labeling.py b/training/meta_labeling.py deleted file mode 100644 index a214458..0000000 --- a/training/meta_labeling.py +++ /dev/null @@ -1,65 +0,0 @@ -from utils.evaluate import discretize_threeway_threshold, evaluate_predictions -from utils.helpers import equal_except_nan -from training.primary_model import train_primary_model -import pandas as pd -from models.base import Model -from reporting.types import Reporting -from typing import Union, Optional -from config.types import Config - -def train_meta_labeling_model( - target_asset: str, - X: pd.DataFrame, - input_predictions: pd.Series, - y: pd.Series, - forward_returns: pd.Series, - models: list[tuple[str, Model]], - config: Config, - model_suffix: str, - from_index: Optional[pd.Timestamp], - 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) - discretized_predictions = input_predictions.apply(discretize) - meta_y: pd.Series = pd.concat([discretized_predictions, y], axis=1).apply(equal_except_nan, axis = 1) - - meta_X = pd.concat([X, input_predictions, discretized_predictions], axis = 1) - - _, meta_preds, meta_probabilities, all_models_single_asset = train_primary_model( - ticker_to_predict = "prediction_correct", - X = meta_X, - y = meta_y, - forward_returns = forward_returns, - models = models, - expanding_window = config.expanding_window_meta_labeling, - sliding_window_size = config.sliding_window_size_meta_labeling, - retrain_every = config.retrain_every, - from_index = from_index, - scaler = config.scaler, - no_of_classes = 'two', - level = 'meta_labeling', - print_results = False, - preloaded_models = preloaded_models - ) - if len(models) > 1: - meta_preds = meta_preds.mean(axis = 1) - bet_size = meta_probabilities[meta_probabilities.columns[1::2]].mean(axis = 1) - else: - bet_size = meta_probabilities.iloc[:,1] - avg_predictions_with_sizing = input_predictions * bet_size - avg_predictions_with_sizing.rename("model_" + target_asset + "_" + model_suffix, inplace=True) - - meta_result = evaluate_predictions( - model_name = "Meta", - forward_returns = forward_returns, - y_pred = avg_predictions_with_sizing, - y_true = y, - no_of_classes = 'two', - print_results = True, - discretize=False - ) - meta_result.rename("model_" + target_asset + "_" + model_suffix, inplace=True) - - - return meta_result, avg_predictions_with_sizing, meta_probabilities, all_models_single_asset diff --git a/training/primary_model.py b/training/primary_model.py deleted file mode 100644 index 211e723..0000000 --- a/training/primary_model.py +++ /dev/null @@ -1,103 +0,0 @@ -import pandas as pd -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 -from transformations.scaler import get_scaler, ScalerTypes -from reporting.types import Reporting -from transformations.rfe import RFETransformation -from transformations.pca import PCATransformation - -def train_primary_model( - ticker_to_predict: str, - X: pd.DataFrame, - y: pd.Series, - forward_returns: pd.Series, - models: list[tuple[str, Model]], - expanding_window: bool, - sliding_window_size: int, - retrain_every: int, - from_index: Optional[pd.Timestamp], - scaler: ScalerTypes, - no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'], - level: str, - print_results: bool, - 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() - predictions = pd.DataFrame(index=y.index) - 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: - unified_models = preloaded_models - - transformations_over_time = 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, - X = X, - y = y, - forward_returns = forward_returns, - 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), - RFETransformation(n_feature_to_select=40, model=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= pd.Series(model_over_time), - transformations_over_time = transformations_over_time, - X = X, - expanding_window = expanding_window, - window_size = sliding_window_size, - from_index = from_index, - ) - - assert len(preds) == len(y) - result = evaluate_predictions( - model_name = model_name, - forward_returns = forward_returns, - y_pred = preds, - y_true = y, - no_of_classes=no_of_classes, - print_results = print_results, - discretize=True - ) - levelname=("_" + level) if level=='metalabeling' else "" - 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, 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 - probs_column_name = "probs_" + ticker_to_predict + "_" + model_name + "_" + level - probs.columns = [probs_column_name + "_" + c for c in probs.columns] - probabilities = pd.concat([probabilities, probs], axis=1) - - - - return results, predictions, probabilities, all_models_single_asset \ No newline at end of file diff --git a/training/train_model.py b/training/train_model.py new file mode 100644 index 0000000..e74b7ee --- /dev/null +++ b/training/train_model.py @@ -0,0 +1,85 @@ +import pandas as pd +from typing import Literal, Optional +from training.walk_forward import walk_forward_train, walk_forward_inference +from utils.evaluate import evaluate_predictions +from models.base import Model +from .types import ModelOverTime, TransformationsOverTime, TrainingOutcome + +def train_models( + ticker_to_predict: str, + X: pd.DataFrame, + y: pd.Series, + forward_returns: pd.Series, + models: list[Model], + expanding_window: bool, + sliding_window_size: int, + retrain_every: int, + from_index: Optional[pd.Timestamp], + no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'], + level: str, + print_results: bool, + transformations_over_time: TransformationsOverTime, + models_over_time: Optional[list[ModelOverTime]] + ) -> list[TrainingOutcome]: + return [train_model(ticker_to_predict, X, y, forward_returns, model, expanding_window, sliding_window_size, retrain_every, from_index, no_of_classes, level, print_results, transformations_over_time, models_over_time[index] if models_over_time else None) for index, model in enumerate(models)] + + +def train_model( + ticker_to_predict: str, + X: pd.DataFrame, + y: pd.Series, + forward_returns: pd.Series, + model: Model, + expanding_window: bool, + sliding_window_size: int, + retrain_every: int, + from_index: Optional[pd.Timestamp], + no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'], + level: str, + print_results: bool, + transformations_over_time: TransformationsOverTime, + model_over_time: Optional[ModelOverTime] + ) -> TrainingOutcome: + + if model_over_time is None: + print("Train model") + model_over_time = walk_forward_train( + model = model, + X = X, + y = y, + forward_returns = forward_returns, + expanding_window = expanding_window, + window_size = sliding_window_size, + retrain_every = retrain_every, + from_index = from_index, + transformations_over_time = transformations_over_time, + ) + + levelname = ("_" + level) if level == 'meta' else "" + if model_over_time is None: + model_id = "model_" + model.name + "_" + ticker_to_predict + levelname + else: + model_id = model_over_time.name + + predictions, probabilities = walk_forward_inference( + model_name = model_id, + model_over_time= model_over_time, + transformations_over_time = transformations_over_time, + X = X, + expanding_window = expanding_window, + window_size = sliding_window_size, + retrain_every = retrain_every, + from_index = from_index, + ) + + assert len(predictions) == len(y) + stats = evaluate_predictions( + forward_returns = forward_returns, + y_pred = predictions, + y_true = y, + no_of_classes=no_of_classes, + print_results = print_results, + discretize=True + ) + + return TrainingOutcome(model_id, predictions, probabilities, stats, model_over_time) \ No newline at end of file diff --git a/training/training_steps.py b/training/training_steps.py deleted file mode 100644 index ee621ba..0000000 --- a/training/training_steps.py +++ /dev/null @@ -1,135 +0,0 @@ -from numpy import DataSource -import pandas as pd -from operator import itemgetter - -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, Optional -from config.types import Config - - -def primary_step( - X: pd.DataFrame, - y: pd.Series, - forward_returns: pd.Series, - config: Config, - reporting: Reporting, - from_index: Optional[pd.Timestamp], - preloaded_training_step: Optional[Reporting.Training_Step] = None, - ) -> tuple[Reporting.Training_Step, pd.DataFrame]: - training_step = Reporting.Training_Step(level='primary') - - # 3. Train Primary models - current_result, current_predictions, current_probabilities, all_models_for_single_asset = train_primary_model( - ticker_to_predict = config.target_asset[1], - X = X, - y = y, - forward_returns = forward_returns, - models = config.primary_models, - expanding_window = config.expanding_window_base, - sliding_window_size = config.sliding_window_size_base, - retrain_every = config.retrain_every, - from_index = from_index, - scaler = config.scaler, - no_of_classes = config.no_of_classes, - level = 'primary', - print_results= True, - preloaded_models = preloaded_training_step.get_base() if preloaded_training_step is not None else None - ) - - training_step.base = all_models_for_single_asset - - # 4. Train a Meta-Labeling model for each Primary model and replace their predictions with the meta-labeling predictions - if config.primary_models_meta_labeling == True: - 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 = config.target_asset[1], - X = X, - input_predictions= primary_model_predictions, - y = y, - forward_returns = forward_returns, - model_suffix = 'meta', - models = config.meta_labeling_models, - config = config, - 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.) - reporting.all_probabilities = pd.concat([reporting.all_probabilities, current_probabilities], axis=1).fillna(0.) - - return training_step, current_predictions - - -def secondary_step( - X:pd.DataFrame, - y:pd.Series, - current_predictions:pd.DataFrame, - forward_returns:pd.Series, - config: Config, - reporting: Reporting, - from_index: Optional[pd.Timestamp], - preloaded_training_step: Optional[Reporting.Training_Step] = None, - ) -> Reporting.Training_Step: - training_step = Reporting.Training_Step(level='secondary') - - # 5. Ensemble primary model predictions (If Ensemble model is present) - if config.ensemble_model is not None: - ensemble_result, ensemble_predictions, _, ensemble_models_one_asset = train_primary_model( - ticker_to_predict = config.target_asset[1], - X = current_predictions, - y = y, - forward_returns = forward_returns, - models = [config.ensemble_model], - expanding_window = False, - sliding_window_size = 1, - retrain_every = config.retrain_every, - from_index = from_index, - scaler = config.scaler, - no_of_classes = 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] - - training_step.base = ensemble_models_one_asset - - reporting.results = pd.concat([reporting.results, ensemble_result], axis=1) - reporting.all_predictions = pd.concat([reporting.all_predictions, ensemble_predictions], axis=1) - - - if len(config.meta_labeling_models) > 0: - - # 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 = config.target_asset[1], - X = X, - input_predictions= ensemble_predictions, - y = y, - forward_returns = forward_returns, - models = config.meta_labeling_models, - config = config, - 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) - - - reporting.results = pd.concat([reporting.results, ensemble_meta_result], axis=1) - reporting.all_predictions = pd.concat([reporting.all_predictions, ensemble_meta_predictions], axis=1) - reporting.all_probabilities = pd.concat([reporting.all_probabilities, ensemble_meta_probabilities], axis=1).fillna(0.) - - return training_step diff --git a/training/types.py b/training/types.py new file mode 100644 index 0000000..cd1f791 --- /dev/null +++ b/training/types.py @@ -0,0 +1,51 @@ +import pandas as pd +from dataclasses import dataclass +from typing import Optional, Dict + +PredictionsSeries = pd.Series +WeightsSeries = pd.Series +ProbabilitiesDataFrame = pd.DataFrame +Stats = Dict[str, float] + +ModelOverTime = pd.Series +TransformationsOverTime = list[pd.Series] + + +@dataclass +class TrainingOutcome: + model_id: str + predictions: PredictionsSeries + probabilities: ProbabilitiesDataFrame + stats: Stats + model_over_time: ModelOverTime + +@dataclass +class EnsembleOutcome: + weights: WeightsSeries + stats: Stats + +@dataclass +class BetSizingWithMetaOutcome: + model_id: str + meta_training: list[TrainingOutcome] + meta_transformations: TransformationsOverTime + weights: WeightsSeries + stats: Stats + +@dataclass +class DirectionalTrainingOutcome: + training: list[TrainingOutcome] + transformations: TransformationsOverTime + +@dataclass +class PipelineOutcome: + directional_training: DirectionalTrainingOutcome + bet_sizing: list[BetSizingWithMetaOutcome] + ensemble: EnsembleOutcome + secondary_bet_sizing: Optional[BetSizingWithMetaOutcome] + + def get_output_weights(self) -> WeightsSeries: + return self.secondary_bet_sizing.weights if self.secondary_bet_sizing else self.ensemble.weights + + def get_output_stats(self) -> Stats: + return self.secondary_bet_sizing.stats if self.secondary_bet_sizing else self.ensemble.stats \ No newline at end of file diff --git a/training/walk_forward.py b/training/walk_forward.py deleted file mode 100644 index da90aee..0000000 --- a/training/walk_forward.py +++ /dev/null @@ -1,123 +0,0 @@ -import pandas as pd -from models.base import Model -import numpy as np -from utils.helpers import get_first_valid_return_index -from tqdm import tqdm -from transformations.base import Transformation -from typing import Optional - -def walk_forward_train( - model_name: str, - model: Model, - X: pd.DataFrame, - y: pd.Series, - forward_returns: pd.Series, - expanding_window: bool, - window_size: int, - retrain_every: int, - from_index: Optional[pd.Timestamp], - transformations: list[Transformation], - preloaded_transformations: Optional[list[pd.Series]], - ) -> tuple[pd.Series, list[pd.Series]]: - assert len(X) == len(y) - models_over_time = pd.Series(index=y.index).rename(model_name) - 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(forward_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y)) - train_from = first_nonzero_return + window_size + 1 if from_index is None else X.index.to_list().index(from_index) - train_till = len(y) - iterations_before_retrain = 0 - - if model.only_column is not None: - X = X[[column for column in X.columns if model.only_column in column]] - - if model.data_transformation == 'original': - transformations = [] - - for index in tqdm(range(train_from, train_till)): - train_window_start = X.index[first_nonzero_return] if expanding_window else X.index[index - window_size - 1] - - if iterations_before_retrain <= 0 or pd.isna(models_over_time[index-1]): - - train_window_end = X.index[index - 1] - - X_expanding_window = X[X.index[first_nonzero_return]:train_window_end] - y_expanding_window = y[X.index[first_nonzero_return]:train_window_end] - - if preloaded_transformations is not None and len(transformations) > 0: - current_transformations = [transformation_over_time[index] for transformation_over_time in preloaded_transformations] - else: - current_transformations = [t.clone() for t in transformations] - for transformation_index, transformation in enumerate(current_transformations): - X_expanding_window = transformation.fit_transform(X_expanding_window, y_expanding_window) - - X_slice = X[train_window_start:train_window_end] - - for transformation in current_transformations: - X_slice = transformation.transform(X_slice) - - X_slice = X_slice.to_numpy() - y_slice = y[train_window_start:train_window_end].to_numpy() - - current_model = model.clone() - - current_model.initialize_network(input_dim = len(X_slice[0]), output_dim=1) - current_model.fit(X_slice, y_slice) - - iterations_before_retrain = retrain_every - - models_over_time[X.index[index]] = current_model - for transformation_index, transformation in enumerate(current_transformations): - transformations_over_time[transformation_index][X.index[index]] = transformation - - iterations_before_retrain -= 1 - - return models_over_time, transformations_over_time - -def walk_forward_inference( - model_name: str, - model_over_time: pd.Series, - transformations_over_time: list[pd.Series], - X: pd.DataFrame, - expanding_window: bool, - window_size: int, - from_index: Optional[pd.Timestamp], - ) -> tuple[pd.Series, pd.DataFrame]: - predictions = pd.Series(index=X.index).rename(model_name) - probabilities = pd.DataFrame(index=X.index) - - 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 X.index.to_list().index(from_index) - inference_till = X.shape[0] - first_model = model_over_time[inference_from] - - if first_model.only_column is not None: - X = X[[column for column in X.columns if first_model.only_column in column]] - - if first_model.data_transformation == 'original': - transformations_over_time = [] - - for index in tqdm(range(inference_from, inference_till)): - - train_window_start = X.index[inference_from] if expanding_window else X.index[index - window_size - 1] - - current_model = model_over_time[X.index[index]] - current_transformations = [transformation_over_time[X.index[index]] for transformation_over_time in transformations_over_time] - - if current_model.predict_window_size == 'window_size': - next_timestep = X.loc[train_window_start:X.index[index]] - else: - # we need to get a Dataframe out of it, since the transformation step always expects a 2D array, but it's equivalent to X.iloc[index] - next_timestep = X.loc[X.index[index]:X.index[index]] - - for transformation in current_transformations: - next_timestep = transformation.transform(next_timestep) - - next_timestep = next_timestep.to_numpy() - - prediction, probs = current_model.predict(next_timestep) - predictions[X.index[index]] = prediction - if len(probabilities.columns) != len(probs): - probabilities = probabilities.reindex(columns = ["prob_" + str(num) for num in range(0, len(probs.T))]) - probabilities.loc[X.index[index]] = probs - - return predictions, probabilities diff --git a/training/walk_forward/__init__.py b/training/walk_forward/__init__.py new file mode 100644 index 0000000..f3865ed --- /dev/null +++ b/training/walk_forward/__init__.py @@ -0,0 +1,3 @@ +from .inference import walk_forward_inference +from .train import walk_forward_train +from .process_transformations_parallel import walk_forward_process_transformations \ No newline at end of file diff --git a/training/walk_forward/inference.py b/training/walk_forward/inference.py new file mode 100644 index 0000000..05465b5 --- /dev/null +++ b/training/walk_forward/inference.py @@ -0,0 +1,58 @@ +import pandas as pd +from training.types import ModelOverTime, TransformationsOverTime, PredictionsSeries, ProbabilitiesDataFrame +from utils.helpers import get_first_valid_return_index +from tqdm import tqdm +from typing import Optional +from data_loader.types import XDataFrame +from utils.helpers import get_last_non_na_index + +def walk_forward_inference( + model_name: str, + model_over_time: ModelOverTime, + transformations_over_time: TransformationsOverTime, + X: XDataFrame, + expanding_window: bool, + window_size: int, + retrain_every: int, + from_index: Optional[pd.Timestamp], + ) -> tuple[PredictionsSeries, ProbabilitiesDataFrame]: + predictions = pd.Series(index=X.index).rename(model_name) + probabilities = pd.DataFrame(index=X.index) + + 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 X.index.to_list().index(from_index) + inference_till = X.shape[0] + model_index_offset = get_last_non_na_index(model_over_time, inference_from) if pd.isna(model_over_time[inference_from]) else 0 + first_model = model_over_time[inference_from - model_index_offset] if pd.isna(model_over_time[inference_from]) else model_over_time[inference_from] + + if first_model.only_column is not None: + X = X[[column for column in X.columns if first_model.only_column in column]] + + if first_model.data_transformation == 'original': + transformations_over_time = [] + + for index in tqdm(range(inference_from, inference_till)): + + last_model_index = index - ((index - inference_from) % retrain_every) - model_index_offset + train_window_start = X.index[inference_from] if expanding_window else X.index[index - window_size - 1] + + current_model = model_over_time[X.index[last_model_index]] + current_transformations = [transformation_over_time[X.index[last_model_index]] for transformation_over_time in transformations_over_time] + + if current_model.predict_window_size == 'window_size': + next_timestep = X.loc[train_window_start:X.index[index]] + else: + # we need to get a Dataframe out of it, since the transformation step always expects a 2D array, but it's equivalent to X.iloc[index] + next_timestep = X.loc[X.index[index]:X.index[index]] + + for transformation in current_transformations: + next_timestep = transformation.transform(next_timestep) + + next_timestep = next_timestep.to_numpy() + + prediction, probs = current_model.predict(next_timestep) + predictions[X.index[index]] = prediction + if inference_from == index and len(probabilities.columns) != len(probs): + probabilities = probabilities.reindex(columns = ["prob_" + str(num) for num in range(0, len(probs.T))]) + probabilities.loc[X.index[index]] = probs + + return predictions, probabilities diff --git a/training/walk_forward/inference_parallel.py b/training/walk_forward/inference_parallel.py new file mode 100644 index 0000000..326f660 --- /dev/null +++ b/training/walk_forward/inference_parallel.py @@ -0,0 +1,62 @@ +import pandas as pd +from models.base import Model +from training.types import ModelOverTime, TransformationsOverTime, PredictionsSeries, ProbabilitiesDataFrame +from transformations.base import Transformation +from utils.helpers import get_first_valid_return_index +from tqdm import tqdm +from typing import Optional +from data_loader.types import XDataFrame +import ray + +def walk_forward_inference( + model_name: str, + model_over_time: ModelOverTime, + transformations_over_time: TransformationsOverTime, + X: XDataFrame, + expanding_window: bool, + window_size: int, + retrain_every: int, + from_index: Optional[pd.Timestamp], + ) -> tuple[PredictionsSeries, ProbabilitiesDataFrame]: + predictions = pd.Series(index=X.index).rename(model_name) + probabilities = pd.DataFrame(index=X.index) + + 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 X.index.to_list().index(from_index) + inference_till = X.shape[0] + first_model = model_over_time[inference_from] + + if first_model.only_column is not None: + X = X[[column for column in X.columns if first_model.only_column in column]] + + if first_model.data_transformation == 'original': + transformations_over_time = [] + + results = ray.get([__inference_from_window.remote(index, inference_from, retrain_every, X, model_over_time, transformations_over_time, expanding_window, window_size) for index in range(inference_from, inference_till)]) + for index, prediction, probs in results: + predictions[X.index[index]] = prediction + probabilities.loc[X.index[index]] = probs + + return predictions, probabilities + +@ray.remote +def __inference_from_window(index: int, inference_from: int, retrain_every: int, X: XDataFrame, model_over_time: ModelOverTime, transformations_over_time: TransformationsOverTime, expanding_window: bool, window_size: int) -> tuple[int, float, pd.Series]: + + last_model_index = index - ((index - inference_from) % retrain_every) + train_window_start = X.index[inference_from] if expanding_window else X.index[index - window_size - 1] + + current_model = model_over_time[X.index[last_model_index]] + current_transformations = [transformation_over_time[X.index[last_model_index]] for transformation_over_time in transformations_over_time] + + if current_model.predict_window_size == 'window_size': + next_timestep = X.loc[train_window_start:X.index[index]] + else: + # we need to get a Dataframe out of it, since the transformation step always expects a 2D array, but it's equivalent to X.iloc[index] + next_timestep = X.loc[X.index[index]:X.index[index]] + + for transformation in current_transformations: + next_timestep = transformation.transform(next_timestep) + + next_timestep = next_timestep.to_numpy() + + prediction, probs = current_model.predict(next_timestep) + return index, prediction, probs \ No newline at end of file diff --git a/training/walk_forward/process_transformations.py b/training/walk_forward/process_transformations.py new file mode 100644 index 0000000..c0da95a --- /dev/null +++ b/training/walk_forward/process_transformations.py @@ -0,0 +1,48 @@ +import pandas as pd +from training.types import TransformationsOverTime +from utils.helpers import get_first_valid_return_index +from tqdm import tqdm +from transformations.base import Transformation +from typing import Optional +from data_loader.types import ForwardReturnSeries, XDataFrame, ySeries + + +def walk_forward_process_transformations( + X: XDataFrame, + y: ySeries, + forward_returns: ForwardReturnSeries, + expanding_window: bool, + window_size: int, + retrain_every: int, + from_index: Optional[pd.Timestamp], + transformations: list[Transformation], + ) -> TransformationsOverTime: + 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(forward_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y)) + train_from = first_nonzero_return + window_size + 1 if from_index is None else X.index.to_list().index(from_index) + train_till = len(y) + iterations_before_retrain = 0 + + for index in tqdm(range(train_from, train_till)): + train_window_start = X.index[first_nonzero_return] if expanding_window else X.index[index - window_size - 1] + + if iterations_before_retrain <= 0 or pd.isna(transformations_over_time[0][index-1]): + + train_window_end = X.index[index - 1] + + X_expanding_window = X[train_window_start:train_window_end] + y_expanding_window = y[train_window_start:train_window_end] + + current_transformations = [t.clone() for t in transformations] + for transformation_index, transformation in enumerate(current_transformations): + X_expanding_window = transformation.fit_transform(X_expanding_window, y_expanding_window) + + iterations_before_retrain = retrain_every + + for transformation_index, transformation in enumerate(current_transformations): + transformations_over_time[transformation_index][X.index[index]] = transformation + + iterations_before_retrain -= 1 + + return transformations_over_time diff --git a/training/walk_forward/process_transformations_parallel.py b/training/walk_forward/process_transformations_parallel.py new file mode 100644 index 0000000..269b2bc --- /dev/null +++ b/training/walk_forward/process_transformations_parallel.py @@ -0,0 +1,46 @@ +import pandas as pd +from training.types import TransformationsOverTime +from utils.helpers import get_first_valid_return_index +from tqdm import tqdm +from transformations.base import Transformation +from typing import Optional +from data_loader.types import ForwardReturnSeries, XDataFrame, ySeries +import ray + +def walk_forward_process_transformations( + X: XDataFrame, + y: ySeries, + forward_returns: ForwardReturnSeries, + expanding_window: bool, + window_size: int, + retrain_every: int, + from_index: Optional[pd.Timestamp], + transformations: list[Transformation], + ) -> TransformationsOverTime: + 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(forward_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y)) + train_from = first_nonzero_return + window_size + 1 if from_index is None else X.index.to_list().index(from_index) + train_till = len(y) + + processed_transformations = ray.get([preprocess_transformations_window.remote(X, y, expanding_window, window_size, transformations, first_nonzero_return, index) for index in range(train_from, train_till, retrain_every)]) + + for transformation, index_time in processed_transformations: + for transformation_index, transformation in enumerate(transformation): + transformations_over_time[transformation_index][X.index[index_time]] = transformation + + return transformations_over_time + +@ray.remote +def preprocess_transformations_window(X: XDataFrame, y: ySeries, expanding_window: bool, window_size: int, transformations: list[Transformation], first_nonzero_return: int, index: int) -> tuple[list[Transformation], int]: + train_window_start = X.index[first_nonzero_return] if expanding_window else X.index[index - window_size - 1] + train_window_end = X.index[index - 1] + + X_expanding_window = X[train_window_start:train_window_end] + y_expanding_window = y[train_window_start:train_window_end] + + current_transformations = [t.clone() for t in transformations] + for transformation in current_transformations: + X_expanding_window = transformation.fit_transform(X_expanding_window, y_expanding_window) + + return (current_transformations, index) \ No newline at end of file diff --git a/training/walk_forward/train.py b/training/walk_forward/train.py new file mode 100644 index 0000000..f2a772f --- /dev/null +++ b/training/walk_forward/train.py @@ -0,0 +1,55 @@ +import pandas as pd +from models.base import Model +from training.types import ModelOverTime, TransformationsOverTime +from utils.helpers import get_first_valid_return_index +from tqdm import tqdm +from typing import Optional +from data_loader.types import ForwardReturnSeries, XDataFrame, ySeries + + +def walk_forward_train( + model: Model, + X: XDataFrame, + y: ySeries, + forward_returns: ForwardReturnSeries, + expanding_window: bool, + window_size: int, + retrain_every: int, + from_index: Optional[pd.Timestamp], + transformations_over_time: TransformationsOverTime, + ) -> ModelOverTime: + models_over_time = pd.Series(index=y.index).rename(model.name) + + first_nonzero_return = max(get_first_valid_return_index(forward_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y)) + train_from = first_nonzero_return + window_size + 1 if from_index is None else X.index.to_list().index(from_index) + train_till = len(y) + + if model.only_column is not None: + X = X[[column for column in X.columns if model.only_column in column]] + + if model.data_transformation == 'original': + transformations_over_time = [] + + for index in tqdm(range(train_from, train_till, retrain_every)): + train_window_start = X.index[first_nonzero_return] if expanding_window else X.index[index - window_size - 1] + + train_window_end = X.index[index - 1] + current_transformations = [transformation_over_time[index] for transformation_over_time in transformations_over_time] + X_slice = X[train_window_start:train_window_end] + + for transformation in current_transformations: + X_slice = transformation.transform(X_slice) + + X_slice = X_slice.to_numpy() + y_slice = y[train_window_start:train_window_end].to_numpy() + + current_model = model.clone() + + current_model.initialize_network(input_dim = len(X_slice[0]), output_dim=1) + current_model.fit(X_slice, y_slice) + + models_over_time[X.index[index]] = current_model + for transformation_index, transformation in enumerate(current_transformations): + transformations_over_time[transformation_index][X.index[index]] = transformation + + return models_over_time \ No newline at end of file diff --git a/utils/evaluate.py b/utils/evaluate.py index 12f4722..fdc3e56 100644 --- a/utils/evaluate.py +++ b/utils/evaluate.py @@ -5,6 +5,8 @@ from utils.metrics import probabilistic_sharpe_ratio, sharpe_ratio from utils.helpers import get_first_valid_return_index import pandas as pd import numpy as np +from data_loader.types import ForwardReturnSeries, ySeries +from training.types import Stats, WeightsSeries def backtest(returns: pd.Series, signal: pd.Series, transaction_cost = 0.002) -> pd.Series: delta_pos = signal.diff(1).abs().fillna(0.) @@ -12,7 +14,7 @@ def backtest(returns: pd.Series, signal: pd.Series, transaction_cost = 0.002) -> return (signal * returns) - costs -def __preprocess(forward_returns: pd.Series, y_pred: pd.Series, y_true: pd.Series, no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'], discretize: bool) -> pd.DataFrame: +def __preprocess(forward_returns: ForwardReturnSeries, y_pred: pd.Series, y_true: pd.Series, no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'], discretize: bool) -> pd.DataFrame: y_pred.name = 'y_pred' forward_returns.name = 'forward_returns' df = pd.concat([y_pred, forward_returns],axis=1).dropna() @@ -27,14 +29,13 @@ def __preprocess(forward_returns: pd.Series, y_pred: pd.Series, y_true: pd.Serie return df def evaluate_predictions( - model_name: str, - forward_returns: pd.Series, - y_pred: pd.Series, - y_true: pd.Series, + forward_returns: ForwardReturnSeries, + y_pred: WeightsSeries, + y_true: ySeries, no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'], print_results: bool, discretize: bool = False, - ) -> pd.Series: + ) -> Stats: # ignore the predictions until we see a non-zero returns (and definitely skip the first sliding_window_size) evaluate_from = max(get_first_valid_return_index(forward_returns), get_first_valid_return_index(y_pred)) @@ -43,42 +44,41 @@ def evaluate_predictions( df = __preprocess(forward_returns, y_pred, y_true, no_of_classes, discretize) - scorecard = pd.Series() + scorecard = dict() def count_non_zero(series: pd.Series) -> int: return len(series[series != 0]) no_of_samples = count_non_zero(df.y_pred) - scorecard.loc['no_of_samples'] = no_of_samples + scorecard['no_of_samples'] = no_of_samples sharpe = sharpe_ratio(df.result) - scorecard.loc['sharpe'] = sharpe + scorecard['sharpe'] = sharpe benchmark_sharpe = sharpe_ratio(df.forward_returns) - scorecard.loc['benchmark_sharpe'] = benchmark_sharpe - scorecard.loc['prob_sharpe'] = probabilistic_sharpe_ratio(sharpe, benchmark_sharpe, no_of_samples) - scorecard.loc['sortino'] = sortino(df.result) - scorecard.loc['skew'] = skew(df.result) + scorecard['benchmark_sharpe'] = benchmark_sharpe + scorecard['prob_sharpe'] = probabilistic_sharpe_ratio(sharpe, benchmark_sharpe, no_of_samples) + scorecard['sortino'] = sortino(df.result) + scorecard['skew'] = skew(df.result) labels = [1, -1] if no_of_classes == 'two' else [1, -1, 0] avg_type = 'weighted' if no_of_classes == 'two' else 'macro' if discretize == True: - scorecard.loc['accuracy'] = accuracy_score(df.sign_true, df.sign_pred) * 100 - scorecard.loc['recall'] = recall_score(df.sign_true, df.sign_pred, labels = labels, average=avg_type) - scorecard.loc['precision'] = precision_score(df.sign_true, df.sign_pred, labels = labels, average=avg_type) - scorecard.loc['f1_score'] = f1_score(df.sign_true, df.sign_pred, labels = labels, average=avg_type) - scorecard.loc['edge'] = df.result.mean() - scorecard.loc['noise'] = df.y_pred.diff().abs().mean() - scorecard.loc['edge_to_noise'] = scorecard.loc['edge'] / scorecard.loc['noise'] + scorecard['accuracy'] = accuracy_score(df.sign_true, df.sign_pred) * 100 + scorecard['recall'] = recall_score(df.sign_true, df.sign_pred, labels = labels, average=avg_type) + scorecard['precision'] = precision_score(df.sign_true, df.sign_pred, labels = labels, average=avg_type) + scorecard['f1_score'] = f1_score(df.sign_true, df.sign_pred, labels = labels, average=avg_type) + scorecard['edge'] = df.result.mean() + scorecard['noise'] = df.y_pred.diff().abs().mean() + scorecard['edge_to_noise'] = scorecard['edge'] / (scorecard['noise'] + 0.00001) if discretize == True: for index, row in df.sign_true.value_counts().iteritems(): - scorecard.loc['sign_true_ratio_' + str(index)] = row / len(df.sign_true) + scorecard['sign_true_ratio_' + str(index)] = row / len(df.sign_true) for index, row in df.sign_pred.value_counts().iteritems(): - scorecard.loc['sign_pred_ratio_' + str(index)] = row / len(df.sign_pred) + scorecard['sign_pred_ratio_' + str(index)] = row / len(df.sign_pred) - scorecard = scorecard.round(3) + scorecard = {k: round(float(v), 3) for k, v in scorecard.items()} if print_results: - print("Model name: ", model_name) print(scorecard) return scorecard diff --git a/utils/helpers.py b/utils/helpers.py index 549e21f..6a639a8 100644 --- a/utils/helpers.py +++ b/utils/helpers.py @@ -3,6 +3,7 @@ import numpy as np import os import string import random +from itertools import dropwhile def get_files_from_dir(path: str) -> list[str]: return [f for f in os.listdir(path) if os.path.isfile(os.path.join(path,f)) and not f.startswith('.')] @@ -16,6 +17,8 @@ def get_first_valid_return_index(series: pd.Series) -> int: return 0 return nested_result[0] +def get_last_non_na_index(series: pd.Series, index: int) -> int: + return next(dropwhile(lambda x: pd.isna(x[1]), enumerate(reversed(series[:index+1]))))[0] def flatten(list_of_lists: list) -> list: