diff --git a/run_pipeline.py b/run_pipeline.py index bb4fe89..4f82d0e 100644 --- a/run_pipeline.py +++ b/run_pipeline.py @@ -17,140 +17,152 @@ from sklearn.ensemble import AdaBoostRegressor, RandomForestRegressor, ExtraTree import feature_extractors.feature_extractor_presets as feature_extractor_presets from training.pipeline import run_single_asset_trainig_pipeline - -WANDB=True +from typing import Tuple -# Parameters -model_config = dict( - regression_models = [ - # ('Lasso', Lasso(alpha=0.1, max_iter=1000)), - ('Ridge', Ridge(alpha=0.1)), - ('BayesianRidge', BayesianRidge()), - ('KNN', KNeighborsRegressor(n_neighbors=25)), - # ('AB', AdaBoostRegressor(random_state=1)), - # ('LR', LinearRegression(n_jobs=-1)), - # ('MLP', MLPRegressor(hidden_layer_sizes=(100,20), max_iter=1000)), - # ('RF', RandomForestRegressor(n_jobs=-1)), - # ('SVR', SVR(kernel='rbf', C=1e3, gamma=0.1)) - ], - regression_ensemble_model = [('Ensemble - Ridge', Ridge(alpha=0.1))], +def get_config()->Tuple[dict, dict, dict]: + # Parameters + model_config = dict( + regression_models = [ + # ('Lasso', Lasso(alpha=0.1, max_iter=1000)), + ('Ridge', Ridge(alpha=0.1)), + ('BayesianRidge', BayesianRidge()), + # ('KNN', KNeighborsRegressor(n_neighbors=25)), + # ('AB', AdaBoostRegressor(random_state=1)), + # ('LR', LinearRegression(n_jobs=-1)), + # ('MLP', MLPRegressor(hidden_layer_sizes=(100,20), max_iter=1000)), + # ('RF', RandomForestRegressor(n_jobs=-1)), + # ('SVR', SVR(kernel='rbf', C=1e3, gamma=0.1)) + ], + regression_ensemble_model = [('Ensemble - Ridge', Ridge(alpha=0.1))], - classification_models = [ - ('LR', LogisticRegression(n_jobs=-1)), - ('LDA', LinearDiscriminantAnalysis()), - ('KNN', KNeighborsClassifier()), - ('CART', DecisionTreeClassifier()), - ('NB', GaussianNB()), - # ('AB', AdaBoostClassifier()), - # ('RF', RandomForestClassifier(n_jobs=-1)) - ], - classification_ensemble_model = [('Ensemble - CART', DecisionTreeClassifier())] -) + classification_models = [ + ('LR', LogisticRegression(n_jobs=-1)), + # ('LDA', LinearDiscriminantAnalysis()), + # ('KNN', KNeighborsClassifier()), + # ('CART', DecisionTreeClassifier()), + # ('NB', GaussianNB()), + # ('AB', AdaBoostClassifier()), + # ('RF', RandomForestClassifier(n_jobs=-1)) + ], + classification_ensemble_model = [('Ensemble - CART', DecisionTreeClassifier())] + ) -training_config = dict( - path = 'data/', - sliding_window_size = 150, - retrain_every = 20, - scaler = 'minmax', # 'normalize' 'minmax' 'standardize' 'none' - include_original_data_in_ensemble = True, - method = 'classification', - forecasting_horizon = 1) + training_config = dict( + # path = 'data/', + sliding_window_size = 150, + retrain_every = 20, + scaler = 'minmax', # 'normalize' 'minmax' 'standardize' 'none' + include_original_data_in_ensemble = True, + # method = 'regression', + # forecasting_horizon = 1 + ) -feature_extractors = feature_extractor_presets.date + feature_extractor_presets.level1 -data_config = dict( - path=training_config['path'], - all_assets = get_crypto_assets(training_config['path']), - load_other_assets= False, - log_returns= True, - forecasting_horizon = training_config['forecasting_horizon'], - own_features= feature_extractors, - other_features= [], - index_column= 'int', - method= training_config['method'], -) - - -if WANDB: - from wandb_setup import get_wandb - wandb = get_wandb() + data_config = dict( + path='data/', + all_assets = get_crypto_assets('data/'), + load_other_assets= False, + log_returns= True, + forecasting_horizon = 1, + own_features= feature_extractor_presets.date + feature_extractor_presets.level1, + other_features= [], + index_column= 'int', + method= 'regression', + ) - if type(wandb) == type(None): - WANDB = False - else: - ''' 3. Initialize Weights and Biases with default values, then grab the config file (necessary for sweep) ''' - wandb.init(project="price-forecasting", - config={"data_config":data_config, "training_config":training_config, "model_config": model_config}) # default config - - training_config = wandb.config['training_config'] - # vvv this doesnt work, wandb casts the functions to strings vvv - # data_config = wandb.config['data_config'] - # model_config = wandb.config['model_config'] + return model_config, training_config, data_config +def launch_wandb(config, sweep=False): + from wandb_setup import get_wandb + wandb = get_wandb() + if type(wandb) == type(None): + return None + elif sweep: + wandb.init(project="price-forecasting", config = config) + return wandb + else: + wandb.init(project="price-forecasting", config=config, reinit=True) + return wandb + +def run_pipeline(with_wandb, sweep): + model_config, training_config, data_config = get_config() + + wandb = None + if with_wandb: + wandb = launch_wandb(dict(**model_config, **training_config, **data_config), sweep) + + if type(wandb) is not type(None): + for k in training_config: training_config[k] = wandb.config[k] + # for k in model_config: model_config[k] = wandb.config[k] + # for k in data_config: data_config[k] = wandb.config[k] + + pipeline(model_config, training_config, data_config, wandb) + # Run pipeline -results = pd.DataFrame() +def pipeline(model_config:dict, training_config:dict, data_config:dict, wandb): + results = pd.DataFrame() -for asset in data_config['all_assets']: - print('--------\nPredicting: ', asset) - all_predictions = pd.DataFrame() + for asset in data_config['all_assets']: + print('--------\nPredicting: ', asset) + all_predictions = pd.DataFrame() - # 1. Load data - data_params = data_config.copy() - data_params['target_asset'] = asset + # 1. Load data + data_params = data_config.copy() + data_params['target_asset'] = asset - X, y, target_returns = load_data(**data_params) + X, y, target_returns = load_data(**data_params) - # 2. Train Level-1 models - current_result, current_predictions = run_single_asset_trainig_pipeline( - ticker_to_predict = asset, - X = X, - y = y, - target_returns = target_returns, - models = model_config['regression_models'] if training_config['method'] == 'regression' else model_config['classification_models'], - method = training_config['method'], - sliding_window_size = training_config['sliding_window_size'], - retrain_every = training_config['retrain_every'], - scaler = training_config['scaler'] - ) - results = pd.concat([results, current_result], axis=1) - all_predictions = pd.concat([all_predictions, current_predictions], axis=1) + # 2. Train Level-1 models + current_result, current_predictions = run_single_asset_trainig_pipeline( + ticker_to_predict = asset, + X = X, + y = y, + target_returns = target_returns, + models = model_config['regression_models'] if data_config['method'] == 'regression' else model_config['classification_models'], + method = data_config['method'], + sliding_window_size = training_config['sliding_window_size'], + retrain_every = training_config['retrain_every'], + scaler = training_config['scaler'], + wandb = wandb + ) + results = pd.concat([results, current_result], axis=1) + all_predictions = pd.concat([all_predictions, current_predictions], axis=1) - # 3. Train Level-2 (Ensemble) model - ensemble_X = all_predictions - if training_config['include_original_data_in_ensemble']: - ensemble_X = pd.concat([ensemble_X, X], axis=1) + # 3. Train Level-2 (Ensemble) model + ensemble_X = all_predictions + if training_config['include_original_data_in_ensemble']: + ensemble_X = pd.concat([ensemble_X, X], axis=1) - ensemble_result, ensemble_preds = run_single_asset_trainig_pipeline( - ticker_to_predict = asset, - X = ensemble_X, - y = y, - target_returns = target_returns, - models = model_config['regression_ensemble_model'] if training_config['method'] == 'regression' else model_config['classification_ensemble_model'], - method = training_config['method'], - sliding_window_size = training_config['sliding_window_size'], - retrain_every = training_config['retrain_every'], - scaler = training_config['scaler'] - ) + ensemble_result, ensemble_preds = run_single_asset_trainig_pipeline( + ticker_to_predict = asset, + X = ensemble_X, + y = y, + target_returns = target_returns, + models = model_config['regression_ensemble_model'] if data_config['method'] == 'regression' else model_config['classification_ensemble_model'], + method = data_config['method'], + sliding_window_size = training_config['sliding_window_size'], + retrain_every = training_config['retrain_every'], + scaler = training_config['scaler'], + wandb = wandb + ) - results = pd.concat([results, ensemble_result], axis=1) - all_predictions = pd.concat([all_predictions, ensemble_preds], axis=1) + results = pd.concat([results, ensemble_result], axis=1) + all_predictions = pd.concat([all_predictions, ensemble_preds], axis=1) + + + results.to_csv('results.csv') + + level1_columns = results[[column for column in results.columns if 'Ensemble' not in column]] + ensemble_columns = results[[column for column in results.columns if 'Ensemble' in column]] + + print("Mean Sharpe ratio for Level-1 models: ", level1_columns.loc['sharpe'].mean()) + print("Mean Sharpe ratio for Level-2 (Ensemble) models: ", ensemble_columns.loc['sharpe'].mean()) -if WANDB: - combined_metrics = results.mean(axis=1) - wandb.log({'results': results}) - if wandb.run is not None: - wandb.finish() - -results.to_csv('results.csv') - -level1_columns = results[[column for column in results.columns if 'Ensemble' not in column]] -ensemble_columns = results[[column for column in results.columns if 'Ensemble' in column]] - -print("Mean Sharpe ratio for Level-1 models: ", level1_columns.loc['sharpe'].mean()) -print("Mean Sharpe ratio for Level-2 (Ensemble) models: ", ensemble_columns.loc['sharpe'].mean()) \ No newline at end of file +if __name__ == '__main__': + run_pipeline(False, False) \ No newline at end of file diff --git a/training/pipeline.py b/training/pipeline.py index 26a1826..4158def 100644 --- a/training/pipeline.py +++ b/training/pipeline.py @@ -25,6 +25,7 @@ def run_single_asset_trainig_pipeline( sliding_window_size: int, retrain_every: int, scaler: Literal['normalize', 'minmax', 'standardize', 'none'], + wandb ) -> tuple[pd.DataFrame, pd.DataFrame]: @@ -32,9 +33,10 @@ def run_single_asset_trainig_pipeline( results = pd.DataFrame() predictions = pd.DataFrame() + + wandb_active = type(wandb) is not type(None) for model_name, model in models: - model_over_time, preds = walk_forward_train_test( model_name=model_name, model = model, @@ -55,5 +57,15 @@ def run_single_asset_trainig_pipeline( column_name = ticker_to_predict + "_" + model_name results[column_name] = result predictions[column_name] = preds + + if wandb_active: + run = wandb.init(project="price-forecasting", config={"model_type": model_name, "ticker": ticker_to_predict}, reinit=True) + wandb.run.name = ticker_to_predict + "-" + model_name+ "-" + wandb.run.id + wandb.run.save() + + for rownum,(indx,val) in enumerate(result.iteritems()): + run.log({"model_type": model_name, indx:val }) + + run.finish() return results, predictions \ No newline at end of file