From 9d47ee942d0cd3dff679bb106ea4be0645e680a5 Mon Sep 17 00:00:00 2001 From: Mark Aron Szulyovszky Date: Thu, 17 Feb 2022 16:36:35 +0100 Subject: [PATCH] feat(Project): use SKLearn models directly, removed custom ensembling, use 5 minute data, batch inference, numba cusum filter (#192) * feat(Project): use 5 minute data, running training in parallel, sped up cusum filter by 10x with numba * fix(WalkForward): inference mini-batch parallelization * fix(WalkForward): don't use the parallel version of any of the functions * feat(CI): download the data required * fix(Project): 5min_crypto folder added * fix(Evaluate): make sure we have numerical stability in returns * feat(Models): use SKLearn models directly to enable composability * feat(Inference): batched inference now working, added forecasting_horizon * fix(Inference): works again * fix(Inference) * chore(Models): remove unused Ensemble model * fix(Labeller): don't just forward shift returns, also take the sum of the data happened until then * Update test.yml --- .github/workflows/test.yml | 12 ++- .gitignore | 2 +- config/preprocess.py | 26 +++---- config/presets.py | 40 +++++----- config/types.py | 32 ++++++-- data/5min_crypto/.gitkeep | 0 data_loader/collections.py | 4 +- data_loader/load.py | 15 +--- environment.yml | 3 +- labeling/event_filters/cusum.py | 43 ++++++----- labeling/eventfilters_map.py | 2 +- .../fixed_time_three_class_balanced.py | 14 ++-- .../fixed_time_three_class_imbalanced.py | 14 ++-- labeling/labellers/fixed_time_two_class.py | 16 ++-- labeling/labellers/utils.py | 9 +++ labeling/labellers_map.py | 6 +- labeling/process.py | 26 ++++--- labeling/types.py | 2 +- models/base.py | 13 +--- models/model_map.py | 76 ++++--------------- models/momentum.py | 20 ++--- models/naive.py | 27 ------- models/neural.py | 38 ---------- models/pytorch/neural_nets.py | 61 --------------- models/pytorch/pytorch_dataset.py | 26 ------- models/sklearn.py | 36 ++++----- models/statsmodels.py | 31 -------- models/xgboost.py | 47 +++++------- mycodemod/replace_functions.py | 2 +- reporting/reporting.py | 16 ++-- ...h_data_minute.py => run_fetch_data_5min.py | 9 ++- run_inference.py | 25 ++---- run_pipeline.py | 37 ++++----- tests/test_evaluation.py | 14 ++-- tests/test_walk_forward.py | 12 ++- training/bet_sizing.py | 25 +++--- training/directional_training.py | 22 +++--- training/ensemble.py | 26 ------- training/train_model.py | 24 +----- training/types.py | 19 ++--- training/walk_forward/__init__.py | 3 +- training/walk_forward/inference.py | 6 +- training/walk_forward/inference_batched.py | 56 ++++++++++++++ training/walk_forward/inference_parallel.py | 52 +++++++------ .../walk_forward/process_transformations.py | 2 +- .../process_transformations_parallel.py | 3 +- training/walk_forward/train.py | 10 +-- training/walk_forward/train_parallel.py | 58 ++++++++++++++ transformations/rfe.py | 10 +-- utils/evaluate.py | 2 +- utils/parallel.py | 15 ++++ utils/resample.py | 9 +++ 52 files changed, 470 insertions(+), 628 deletions(-) create mode 100644 data/5min_crypto/.gitkeep create mode 100644 labeling/labellers/utils.py delete mode 100644 models/naive.py delete mode 100644 models/neural.py delete mode 100644 models/pytorch/neural_nets.py delete mode 100644 models/pytorch/pytorch_dataset.py delete mode 100644 models/statsmodels.py rename run_fetch_data_minute.py => run_fetch_data_5min.py (83%) delete mode 100644 training/ensemble.py create mode 100644 training/walk_forward/inference_batched.py create mode 100644 training/walk_forward/train_parallel.py create mode 100644 utils/parallel.py create mode 100644 utils/resample.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 8047ae9..cbe4b39 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -19,15 +19,19 @@ jobs: shell: bash -l {0} run: | pytest --junit-xml pytest.xml + - name: Download data + shell: bash -l {0} + run: | + python run_fetch_data_5min.py - name: Run pipeline shell: bash -l {0} run: | export RAY_DISABLE_MEMORY_MONITOR=1 python run_pipeline.py - - name: Run portfolio reporting - shell: bash -l {0} - run: | - python run_portfolio_reporting.py +# - name: Run portfolio reporting +# shell: bash -l {0} +# run: | +# python run_portfolio_reporting.py - name: Run inference shell: bash -l {0} run: | diff --git a/.gitignore b/.gitignore index 3e3175d..599b596 100644 --- a/.gitignore +++ b/.gitignore @@ -138,4 +138,4 @@ lightning_logs/ results.csv models/saved/ -data/minute_crypto/** \ No newline at end of file +data/5min_crypto/** \ No newline at end of file diff --git a/config/preprocess.py b/config/preprocess.py index d327c61..efd4482 100644 --- a/config/preprocess.py +++ b/config/preprocess.py @@ -5,6 +5,8 @@ from models.model_map import get_model from data_loader.collections import data_collections from labeling.eventfilters_map import eventfilters_map from labeling.labellers_map import labellers_map +from models.sklearn import SKLearnModel +from sklearn.ensemble import VotingClassifier def preprocess_config(raw_config: RawConfig) -> Config: config_dict = vars(raw_config) @@ -17,7 +19,6 @@ def preprocess_config(raw_config: RawConfig) -> Config: config_dict['no_of_classes'] = 'two' config_dict['mode'] = 'training' config = Config(**config_dict) - validate_config(config) return config def __preprocess_feature_extractors_config(data_dict: dict) -> dict: @@ -28,10 +29,14 @@ def __preprocess_feature_extractors_config(data_dict: dict) -> dict: data_dict[key] = flatten([feature_extractor_presets[preset_name] for preset_name in preset_names]) return data_dict -def __preprocess_model_config(model_config:dict) -> dict: - model_config['directional_models'] = [get_model(model_name) for model_name in model_config['directional_models']] +def __preprocess_model_config(model_config: dict) -> dict: + directional_models = [get_model(model_name) for model_name in model_config['directional_models']] + model_config.pop('directional_models') + model_config['directional_model'] = SKLearnModel(VotingClassifier([(m.name, m)for m in directional_models], voting ='soft')) if len(model_config['meta_models']) > 0: - model_config['meta_models'] = [get_model(model_name) for model_name in model_config['meta_models']] + meta_models = [get_model(model_name) for model_name in model_config['meta_models']] + model_config['meta_model'] = SKLearnModel(VotingClassifier([(m.name, m)for m in meta_models], voting ='soft')) + model_config.pop('meta_models') return model_config @@ -49,16 +54,9 @@ def __preprocess_event_filter_config(data_dict: dict) -> dict: data_dict['event_filter'] = eventfilters_map[data_dict['event_filter']] return data_dict -def __preprocess_event_labeller_config(data_dict: dict) -> dict: - data_dict['labeling'] = labellers_map[data_dict['labeling']] - return data_dict +def __preprocess_event_labeller_config(config_dict: dict) -> dict: + config_dict['labeling'] = labellers_map[config_dict['labeling']](config_dict['forecasting_horizon']) + return config_dict -def validate_config(config: Config): - # We need to make sure there's only one output from the pipeline - # 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_models) == 0: assert len(config.directional_models) == 1 - diff --git a/config/presets.py b/config/presets.py index c32e5fd..a75ec94 100644 --- a/config/presets.py +++ b/config/presets.py @@ -2,7 +2,6 @@ from .types import RawConfig, Config def get_dev_config() -> RawConfig: - regression_models = ["Lasso"] classification_models = ["LogisticRegression_two_class"] return RawConfig( @@ -29,13 +28,13 @@ def get_dev_config() -> RawConfig: meta_models = [], event_filter = 'none', - labeling = 'two_class' + labeling = 'two_class', + forecasting_horizon = 100, ) 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_models = ['LogisticRegression_two_class', 'LGBM'] @@ -63,42 +62,43 @@ def get_default_ensemble_config() -> RawConfig: meta_models = meta_models, event_filter = 'cusum_vol', - labeling = 'two_class' + labeling = 'two_class', + forecasting_horizon = 100, ) def get_lightweight_ensemble_config() -> RawConfig: - regression_models = ["Lasso", "KNN"] - classification_models = ['LogisticRegression_two_class', 'SVC'] + classification_models = ['LogisticRegression_two_class', 'LGBM'] meta_models = ['LogisticRegression_two_class', 'LGBM'] return RawConfig( directional_models_meta = True, dimensionality_reduction = True, n_features_to_select = 30, - expanding_window_base = False, + expanding_window_base = True, expanding_window_meta = True, - sliding_window_size_base = 380, - sliding_window_size_meta = 240, - retrain_every = 40, + sliding_window_size_base = 3800, + sliding_window_size_meta = 2400, + retrain_every = 1000, scaler = 'minmax', # 'normalize' 'minmax' 'standardize' - assets = ['daily_crypto_lightweight'], - target_asset = 'BCH_USD', - other_assets = ['daily_etf'], - exogenous_data = ['daily_glassnode'], - load_non_target_asset= True, - own_features = ['level_2' ], - other_features = ['level_2'], - exogenous_features = ['z_score'], + assets = ['fivemin_crypto'], + target_asset = 'BTC_USD', + other_assets = [], + exogenous_data = [], + load_non_target_asset= False, + own_features = ['level_1'], + other_features = [], + exogenous_features = [], directional_models = classification_models, meta_models = meta_models, - event_filter = 'none', - labeling = 'two_class' + event_filter = 'cusum_fixed', + labeling = 'two_class', + forecasting_horizon = 50, ) diff --git a/config/types.py b/config/types.py index b6dbed3..e6cd2d4 100644 --- a/config/types.py +++ b/config/types.py @@ -1,11 +1,12 @@ -from pydantic import BaseModel +from pydantic import BaseModel, validator from typing import Literal, Optional from labeling.types import EventFilter from models.base import Model from data_loader.types import DataCollection, DataSource from feature_extractors.types import FeatureExtractor, ScalerTypes from labeling.types import EventFilter, EventLabeller - +from sklearn.base import BaseEstimator +from dataclasses import dataclass # RawConfig is needed to ensure we can declare config presets here with static typing, we then convert it to Config class RawConfig(BaseModel): @@ -29,12 +30,14 @@ class RawConfig(BaseModel): exogenous_features: list[str] event_filter: Literal['none', 'cusum_vol', 'cusum_fixed'] labeling: Literal['two_class', 'three_class_balanced', 'three_class_imbalanced'] + forecasting_horizon: int directional_models: list[str] meta_models: list[str] -class Config(BaseModel): +@dataclass +class Config: directional_models_meta: bool dimensionality_reduction: bool n_features_to_select: int @@ -55,14 +58,29 @@ class Config(BaseModel): exogenous_features: list[tuple[str, FeatureExtractor, list[int]]] event_filter: EventFilter labeling: EventLabeller + forecasting_horizon: int no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'] mode: Literal['training', 'inference'] - directional_models: list[Model] - meta_models: list[Model] + directional_model: Model + meta_model: Model - class Config: - arbitrary_types_allowed = True + @validator('directional_model', 'meta_model') + def check_model(cls, v): + assert isinstance(v, BaseEstimator) + return v + + @validator('event_filter') + def check_event_filter(cls, v): + assert isinstance(v, EventFilter) + return v + + @validator('labeling') + def check_labeling(cls, v): + assert isinstance(v, EventLabeller) + return v + # class Config: + # arbitrary_types_allowed = True diff --git a/data/5min_crypto/.gitkeep b/data/5min_crypto/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data_loader/collections.py b/data_loader/collections.py index 55d4094..9b583a5 100644 --- a/data_loader/collections.py +++ b/data_loader/collections.py @@ -10,7 +10,7 @@ __daily_crypto = ["ADA_USD", "BCH_USD", "BNB_USD", "BTC_USD", "DOT_USD", "ETC_US __daily_crypto_lightweight = ["ADA_USD", "BCH_USD"] -__minute_crypto = ["BTC_USD", "DASH_USD", "ETC_USD", "ETH_USD", "LTC_USD", "TRX_USD", "XLM_USD", "XMR_USD"] +__5min_crypto = ["BTC_USD", "DASH_USD", "ETC_USD", "ETH_USD", "LTC_USD", "TRX_USD", "XLM_USD", "XMR_USD"] __daily_glassnode = ['rhodl_ratio', # 'cvdd', @@ -45,7 +45,7 @@ data_collections = dict( daily_crypto = transform_to_data_collection("data/daily_crypto", __daily_crypto), daily_crypto_lightweight = transform_to_data_collection("data/daily_crypto", __daily_crypto_lightweight), daily_etf = transform_to_data_collection("data/daily_etf", __daily_etf), - minute_crypto = transform_to_data_collection("data/minute_crypto", __minute_crypto), + fivemin_crypto = transform_to_data_collection("data/5min_crypto", __5min_crypto), daily_glassnode =transform_to_data_collection("data/daily_glassnode", __daily_glassnode), ) diff --git a/data_loader/load.py b/data_loader/load.py index bcec4e7..93bf116 100644 --- a/data_loader/load.py +++ b/data_loader/load.py @@ -14,7 +14,7 @@ cache = Cache(".cachedir/data") -def load_data(**kwargs) -> tuple[XDataFrame, ReturnSeries, ForwardReturnSeries]: +def load_data(**kwargs) -> tuple[XDataFrame, ReturnSeries]: hashed = hash_data_config(kwargs) if hashed in cache: return cache.get(hashed) @@ -31,7 +31,7 @@ def __load_data(assets: DataCollection, own_features: list[tuple[str, FeatureExtractor, list[int]]], other_features: list[tuple[str, FeatureExtractor, list[int]]], exogenous_features: list[tuple[str, FeatureExtractor, list[int]]], - ) -> tuple[XDataFrame, ReturnSeries, ForwardReturnSeries]: + ) -> tuple[XDataFrame, ReturnSeries]: """ Loads asset data from the specified path. Returns: @@ -85,12 +85,8 @@ def __load_data(assets: DataCollection, ## Create target returns = df_target_asset_only_returns[target_asset[1] + '_returns'] returns.index = pd.DatetimeIndex(X.index) - forward_returns = __create_target_cum_forward_returns(returns, 1) - forward_returns.index = pd.DatetimeIndex(X.index) - # we need to null out the last forward returns row, because when doing forward-shifting, we automatically get the first row, which is definitely incorrect - forward_returns[forward_returns.index[-1]] = 0. - return X, returns, forward_returns + return X, returns @ray.remote @@ -130,11 +126,6 @@ def __apply_feature_extractors(df: pd.DataFrame, feature_extractors: list[tuple[ return df -def __create_target_cum_forward_returns(series: pd.Series, period: int) -> pd.Series: - assert period > 0 - return series.shift(-period).copy() - - def load_only_returns(assets: DataCollection, returns: Literal['price', 'returns']) -> pd.DataFrame: assets_future = [__load_df.remote( diff --git a/environment.yml b/environment.yml index e1683bb..1e1edf9 100644 --- a/environment.yml +++ b/environment.yml @@ -9,8 +9,7 @@ dependencies: - python=3.9 - scikit-learn-intelex=2021.4.0 - pip: - - torch - - pytorch-lightning + - skorch - fracdiff - ray - diskcache diff --git a/labeling/event_filters/cusum.py b/labeling/event_filters/cusum.py index 3ac35be..9349ddd 100644 --- a/labeling/event_filters/cusum.py +++ b/labeling/event_filters/cusum.py @@ -1,8 +1,11 @@ +from numpy import float32 from ..types import EventFilter from data_loader.types import ReturnSeries import pandas as pd +from numba import njit +from numba.typed import List class CUSUMVolatilityEventFilter(EventFilter): @@ -11,7 +14,7 @@ class CUSUMVolatilityEventFilter(EventFilter): def get_event_start_times(self, returns: ReturnSeries) -> pd.DatetimeIndex: - rolling_vol = returns.rolling(self.vol_period).std() * 0.15 + rolling_vol = returns.rolling(self.vol_period).std().mean() filtered_indices = [] pos_threshold = 0 @@ -39,22 +42,28 @@ class CUSUMFixedEventFilter(EventFilter): self.threshold = threshold def get_event_start_times(self, returns: ReturnSeries) -> pd.DatetimeIndex: - filtered_indices = [] - pos_threshold = 0 - neg_threshold = 0 - diff = returns.diff() - for index in diff.index[1:]: - pos_threshold, neg_threshold = ( - max(0, pos_threshold + diff.loc[index]), - min(0, neg_threshold + diff.loc[index]), - ) + diffed_returns = returns.diff() + int_indicies = _process(List(diffed_returns.to_list()), abs(returns.mean()) * self.threshold) - if neg_threshold < -self.threshold: - neg_threshold = 0 - filtered_indices.append(index) + return pd.DatetimeIndex([returns.index[i] for i in int_indicies]) - elif pos_threshold > self.threshold: - pos_threshold = 0 - filtered_indices.append(index) +@njit +def _process(diffed_returns: List, threshold: float32) -> List: + pos_threshold: float32 = 0.0 # type: ignore + neg_threshold: float32 = 0.0 # type: ignore + filtered_indicies = List() + for index in range(1, len(diffed_returns[1:])): + pos_threshold, neg_threshold = ( # type: ignore + max(0, pos_threshold + diffed_returns[index]), # type: ignore + min(0, neg_threshold + diffed_returns[index]), # type: ignore + ) - return pd.DatetimeIndex(filtered_indices) \ No newline at end of file + if neg_threshold < -threshold: + neg_threshold = 0.0 # type: ignore + filtered_indicies.append(index) + + elif pos_threshold > threshold: + pos_threshold = 0.0 # type: ignore + filtered_indicies.append(index) + + return filtered_indicies \ No newline at end of file diff --git a/labeling/eventfilters_map.py b/labeling/eventfilters_map.py index 5d87371..48c1569 100644 --- a/labeling/eventfilters_map.py +++ b/labeling/eventfilters_map.py @@ -4,5 +4,5 @@ from .event_filters.cusum import CUSUMVolatilityEventFilter, CUSUMFixedEventFilt eventfilters_map = dict( none = NoEventFilter(), cusum_vol = CUSUMVolatilityEventFilter(vol_period = 20), - cusum_fixed = CUSUMFixedEventFilter(threshold = 0.05) + cusum_fixed = CUSUMFixedEventFilter(threshold = 500) ) \ No newline at end of file diff --git a/labeling/labellers/fixed_time_three_class_balanced.py b/labeling/labellers/fixed_time_three_class_balanced.py index 68485fd..26db4d9 100644 --- a/labeling/labellers/fixed_time_three_class_balanced.py +++ b/labeling/labellers/fixed_time_three_class_balanced.py @@ -1,16 +1,20 @@ -from data_loader.types import ForwardReturnSeries +from data_loader.types import ReturnSeries, ForwardReturnSeries from ..types import EventLabeller, EventsDataFrame import pandas as pd +from .utils import create_forward_returns class FixedTimeHorionThreeClassBalancedEventLabeller(EventLabeller): time_horizon: int - def __init__(self, time_horizon: int = 1): + def __init__(self, time_horizon: int): self.time_horizon = time_horizon - def label_events(self, event_start_times: pd.DatetimeIndex, forward_returns: ForwardReturnSeries) -> EventsDataFrame: + def label_events(self, event_start_times: pd.DatetimeIndex, returns: ReturnSeries) -> tuple[EventsDataFrame, ForwardReturnSeries]: + forward_returns = create_forward_returns(returns, self.time_horizon) + cutoff_point = returns.index[-self.time_horizon] + event_start_times[event_start_times < cutoff_point] event_candidates = forward_returns[event_start_times] def get_bins_threeway(x): @@ -35,10 +39,10 @@ class FixedTimeHorionThreeClassBalancedEventLabeller(EventLabeller): return 1 labels = event_candidates.map(map_class_threeway) - return pd.DataFrame({ + return (pd.DataFrame({ 'start': event_start_times, 'end': event_start_times + pd.Timedelta(days=self.time_horizon), 'label': labels, 'returns': forward_returns[event_start_times] - }) + }), forward_returns[event_start_times]) diff --git a/labeling/labellers/fixed_time_three_class_imbalanced.py b/labeling/labellers/fixed_time_three_class_imbalanced.py index 4ef93f0..f56e373 100644 --- a/labeling/labellers/fixed_time_three_class_imbalanced.py +++ b/labeling/labellers/fixed_time_three_class_imbalanced.py @@ -1,15 +1,19 @@ -from ..types import EventLabeller, EventsDataFrame, ForwardReturnSeries +from ..types import EventLabeller, EventsDataFrame, ReturnSeries, ForwardReturnSeries import pandas as pd +from .utils import create_forward_returns class FixedTimeHorionThreeClassImbalancedEventLabeller(EventLabeller): time_horizon: int - def __init__(self, time_horizon: int = 1): + def __init__(self, time_horizon: int): self.time_horizon = time_horizon - def label_events(self, event_start_times: pd.DatetimeIndex, forward_returns: ForwardReturnSeries) -> EventsDataFrame: + def label_events(self, event_start_times: pd.DatetimeIndex, returns: ReturnSeries) -> tuple[EventsDataFrame, ForwardReturnSeries]: + forward_returns = create_forward_returns(returns, self.time_horizon) + cutoff_point = returns.index[-self.time_horizon] + event_start_times[event_start_times < cutoff_point] event_candidates = forward_returns[event_start_times] def get_bins_threeway(x): @@ -34,11 +38,11 @@ class FixedTimeHorionThreeClassImbalancedEventLabeller(EventLabeller): return 1 labels = event_candidates.map(map_class_threeway) - return pd.DataFrame({ + return (pd.DataFrame({ 'start': event_start_times, 'end': event_start_times + pd.Timedelta(days=self.time_horizon), 'label': labels, 'returns': forward_returns[event_start_times] - }) + }), forward_returns[event_start_times]) diff --git a/labeling/labellers/fixed_time_two_class.py b/labeling/labellers/fixed_time_two_class.py index 810b82b..2b24443 100644 --- a/labeling/labellers/fixed_time_two_class.py +++ b/labeling/labellers/fixed_time_two_class.py @@ -1,29 +1,31 @@ - - -from ..types import EventLabeller, EventsDataFrame, ForwardReturnSeries +from ..types import EventLabeller, EventsDataFrame, ReturnSeries, ForwardReturnSeries import pandas as pd +from .utils import create_forward_returns class FixedTimeHorionTwoClassEventLabeller(EventLabeller): time_horizon: int - def __init__(self, time_horizon: int = 1): + def __init__(self, time_horizon: int): self.time_horizon = time_horizon - def label_events(self, event_start_times: pd.DatetimeIndex, forward_returns: ForwardReturnSeries) -> EventsDataFrame: + def label_events(self, event_start_times: pd.DatetimeIndex, returns: ReturnSeries) -> tuple[EventsDataFrame, ForwardReturnSeries]: + forward_returns = create_forward_returns(returns, self.time_horizon) + cutoff_point = returns.index[-self.time_horizon] + event_start_times[event_start_times < cutoff_point] event_candidates = forward_returns[event_start_times] def get_class_binary(x: float) -> int: return -1 if x <= 0.0 else 1 labels = event_candidates.map(get_class_binary) - return pd.DataFrame({ + return (pd.DataFrame({ 'start': event_start_times, 'end': event_start_times + pd.Timedelta(days=self.time_horizon), 'label': labels, 'returns': forward_returns[event_start_times] - }) + }), forward_returns[event_start_times]) diff --git a/labeling/labellers/utils.py b/labeling/labellers/utils.py new file mode 100644 index 0000000..6e312f2 --- /dev/null +++ b/labeling/labellers/utils.py @@ -0,0 +1,9 @@ +import pandas as pd +from data_loader.types import ForwardReturnSeries + +def create_forward_returns(series: pd.Series, period: int) -> ForwardReturnSeries: + assert period > 0 + indexer = pd.api.indexers.FixedForwardWindowIndexer(window_size=period) + + forward_returns = series.rolling(window=indexer).sum() + return forward_returns \ No newline at end of file diff --git a/labeling/labellers_map.py b/labeling/labellers_map.py index 8e58390..b3176ad 100644 --- a/labeling/labellers_map.py +++ b/labeling/labellers_map.py @@ -3,7 +3,7 @@ from .labellers.fixed_time_three_class_imbalanced import FixedTimeHorionThreeCla from .labellers.fixed_time_two_class import FixedTimeHorionTwoClassEventLabeller labellers_map = dict( - two_class = FixedTimeHorionTwoClassEventLabeller(), - three_class_balanced = FixedTimeHorionThreeClassBalancedEventLabeller(), - three_class_imbalanced = FixedTimeHorionThreeClassImbalancedEventLabeller() + two_class = FixedTimeHorionTwoClassEventLabeller, + three_class_balanced = FixedTimeHorionThreeClassBalancedEventLabeller, + three_class_imbalanced = FixedTimeHorionThreeClassImbalancedEventLabeller ) \ No newline at end of file diff --git a/labeling/process.py b/labeling/process.py index 0520b44..9dd7a32 100644 --- a/labeling/process.py +++ b/labeling/process.py @@ -2,16 +2,18 @@ from .types import EventFilter, EventLabeller, EventsDataFrame from data_loader.types import ForwardReturnSeries, XDataFrame, ReturnSeries, ySeries def label_data( - event_filter: EventFilter, - event_labeller: EventLabeller, - X: XDataFrame, - returns: ReturnSeries, - forward_returns: ForwardReturnSeries) -> tuple[EventsDataFrame, XDataFrame, ySeries, ForwardReturnSeries]: - event_start_times = event_filter.get_event_start_times(returns) - events = event_labeller.label_events(event_start_times, forward_returns) - - X = X.filter(items = events.index, axis = 0) - y = events['label'] - forward_returns = events['returns'] + event_filter: EventFilter, + event_labeller: EventLabeller, + X: XDataFrame, + returns: ReturnSeries) -> tuple[EventsDataFrame, XDataFrame, ySeries, ForwardReturnSeries]: - return events, X, y, forward_returns \ No newline at end of file + event_start_times = event_filter.get_event_start_times(returns) + print("| Filtered out ", (1 - (len(event_start_times) / len(returns))) * 100, "% of timestamps" ) + + events, forward_returns = event_labeller.label_events(event_start_times, returns) + + X = X.filter(items = events.index, axis = 0) + y = events['label'] + forward_returns = events['returns'] + + return events, X, y, forward_returns \ No newline at end of file diff --git a/labeling/types.py b/labeling/types.py index 86abbeb..12eedeb 100644 --- a/labeling/types.py +++ b/labeling/types.py @@ -23,6 +23,6 @@ EventsDataFrame = DataFrame[EventSchema] class EventLabeller(ABC): @abstractmethod - def label_events(self, event_start_times: pd.DatetimeIndex, forward_returns: ForwardReturnSeries) -> EventsDataFrame: + def label_events(self, event_start_times: pd.DatetimeIndex, returns: ReturnSeries) -> tuple[EventsDataFrame, ForwardReturnSeries]: raise NotImplementedError diff --git a/models/base.py b/models/base.py index 5eea960..7ef71aa 100644 --- a/models/base.py +++ b/models/base.py @@ -6,10 +6,8 @@ import numpy as np class Model(ABC): name: str = "" - method: Literal["regression", "classification"] data_transformation: Literal["transformed", "original"] only_column: Optional[str] - model_type: Literal['ml', 'static'] predict_window_size: Literal['single_timestamp', 'window_size'] @abstractmethod @@ -17,17 +15,10 @@ class Model(ABC): raise NotImplementedError @abstractmethod - def predict(self, X: np.ndarray) -> tuple[float, np.ndarray]: + def predict(self, X: np.ndarray) -> np.ndarray: raise NotImplementedError @abstractmethod - def clone(self) -> Model: + def predict_proba(self, X: np.ndarray) -> np.ndarray: raise NotImplementedError - - @abstractmethod - def initialize_network(self, input_dim:int, output_dim:int): - pass - - - diff --git a/models/model_map.py b/models/model_map.py index 37bdc73..afa6708 100644 --- a/models/model_map.py +++ b/models/model_map.py @@ -3,7 +3,7 @@ from sklearnex.ensemble import RandomForestClassifier 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_classification = SKLearnModel(RandomForestClassifier(n_jobs=-1, max_depth=20, random_state=1)) def get_model(model_name: str) -> Model: @@ -11,85 +11,41 @@ def get_model(model_name: str) -> Model: model.name = model_name return model - if model_name == 'LinearRegression': - from sklearn.linear_model import LinearRegression - return set_name(SKLearnModel(LinearRegression(n_jobs=-1), 'regression')) - elif model_name == 'Lasso': - from sklearn.linear_model import Lasso - return set_name(SKLearnModel(Lasso(alpha=100, random_state=1), 'regression')) - elif model_name == 'Ridge': - from sklearn.linear_model import Ridge - return set_name(SKLearnModel(Ridge(alpha=0.1), 'regression')) - elif model_name == 'BayesianRidge': - from sklearn.linear_model import BayesianRidge - return set_name(SKLearnModel(BayesianRidge(), 'regression')) - elif model_name == 'KNN': - from sklearnex.neighbors import KNeighborsRegressor - return set_name(SKLearnModel(KNeighborsRegressor(n_neighbors=25), 'regression')) - elif model_name == 'AB': - from sklearn.ensemble import AdaBoostRegressor - return set_name(SKLearnModel(AdaBoostRegressor(random_state=1), 'regression')) - elif model_name == 'MLP': - from sklearn.neural_network import MLPRegressor - return set_name(SKLearnModel(MLPRegressor(hidden_layer_sizes=(100,20), max_iter=1000), 'regression')) - elif model_name == 'RFR': - 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 set_name(SKLearnModel(SVR(kernel='rbf', C=1e3, gamma=0.1), 'regression')) - elif model_name == 'StaticNaive': - from models.naive import 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 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': + if model_name == 'LogisticRegression_two_class': from sklearn.linear_model import LogisticRegression - return set_name(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))) elif model_name == 'LogisticRegression_three_class': from sklearnex.linear_model import LogisticRegression as LogisticRegression_EX - return set_name(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))) elif model_name == 'LDA': from sklearn.discriminant_analysis import LinearDiscriminantAnalysis - return set_name(SKLearnModel(LinearDiscriminantAnalysis(), 'classification')) + return set_name(SKLearnModel(LinearDiscriminantAnalysis())) elif model_name == 'KNN': from sklearn.neighbors import KNeighborsClassifier - return set_name(SKLearnModel(KNeighborsClassifier(), 'classification')) + return set_name(SKLearnModel(KNeighborsClassifier())) elif model_name == 'CART': from sklearn.tree import DecisionTreeClassifier - return set_name(SKLearnModel(DecisionTreeClassifier(max_depth=15, random_state=1), 'classification')) + return set_name(SKLearnModel(DecisionTreeClassifier(max_depth=15, random_state=1))) elif model_name == 'NB': from sklearn.naive_bayes import GaussianNB - return set_name(SKLearnModel(GaussianNB(), 'classification')) + return set_name(SKLearnModel(GaussianNB())) elif model_name == 'AB': from sklearn.ensemble import AdaBoostClassifier - return set_name(SKLearnModel(AdaBoostClassifier(n_estimators=15), 'classification')) + return set_name(SKLearnModel(AdaBoostClassifier(n_estimators=15))) elif model_name == 'RFC': - return set_name(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))) elif model_name == 'SVC': from sklearn.svm import SVC - 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 set_name(XGBoostModel(XGBClassifier(n_jobs=-1, max_depth = 20, random_state=1, objective='binary:logistic', use_label_encoder= False, eval_metric='mlogloss'))) + return set_name(SKLearnModel(SVC(kernel='rbf', C=1e3, probability=True, random_state=1))) + # elif model_name == 'XGB_two_class': + # from xgboost import XGBClassifier + # from models.xgboost import XGBoostModel + # 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 set_name(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))) elif model_name == 'StaticMom': from models.momentum import StaticMomentumModel return set_name(StaticMomentumModel(allow_short=True)) - elif model_name == 'Average': - from models.average import 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 c969fa7..5a77a65 100644 --- a/models/momentum.py +++ b/models/momentum.py @@ -1,16 +1,15 @@ from __future__ import annotations -from models.base import Model import numpy as np +from .base import Model +from sklearn.base import BaseEstimator, ClassifierMixin -class StaticMomentumModel(Model): +class StaticMomentumModel(BaseEstimator, ClassifierMixin, Model): ''' Model that uses only one feature: momentum. It's positive if momentum is greater than 0, otherwise it's negative. ''' - method = 'classification' data_transformation = 'original' only_column = 'mom' - model_type = 'static' predict_window_size = 'single_timestamp' def __init__(self, allow_short: bool) -> None: @@ -21,13 +20,10 @@ class StaticMomentumModel(Model): # This is a static model, it can' learn anything pass - def predict(self, X) -> tuple[float, np.ndarray]: + def predict(self, X) -> np.ndarray: negative_class = -1.0 if self.allow_short == True else 0.0 prediction = 1.0 if X[-1][0] > 0 else negative_class - return (prediction, np.array([])) - - def clone(self) -> StaticMomentumModel: - return self - - def initialize_network(self, input_dim:int, output_dim:int): - pass \ No newline at end of file + return np.array(prediction) + + def predict_proba(self, X) -> np.ndarray: + return np.array([]) diff --git a/models/naive.py b/models/naive.py deleted file mode 100644 index 81a9033..0000000 --- a/models/naive.py +++ /dev/null @@ -1,27 +0,0 @@ -from __future__ import annotations -from models.base import Model -import numpy as np - -class StaticNaiveModel(Model): - ''' - Model that carries the last observation (from returns) to the next one, naively. - ''' - - method = 'regression' - data_transformation = 'original' - only_column = None - 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]: - return (X[-1][0], np.array([])) - - def clone(self) -> StaticNaiveModel: - return self - - 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 deleted file mode 100644 index 114bbc5..0000000 --- a/models/neural.py +++ /dev/null @@ -1,38 +0,0 @@ -from __future__ import annotations -from models.base import Model -import numpy as np -from models.pytorch.pytorch_dataset import get_dataloader -import copy -import pytorch_lightning as pl - -class LightningNeuralNetModel(Model): - - method = 'regression' - data_transformation = 'transformed' - only_column = None - model_type = 'ml' - - ''' Standard lightning methods ''' - - def __init__(self, model, max_epochs=5): - self.model = model - self.trainer = pl.Trainer(max_epochs=max_epochs) - - def fit(self, X: np.ndarray, y: np.ndarray) -> None: - train_dataloader = self.__prepare_data(X.astype(float), y.astype(float)) - self.trainer.fit(self.model, train_dataloader) - - def predict(self, X: np.ndarray) -> tuple[float, np.ndarray]: - return self.model(X) - - def clone(self): - model_copy = copy.deepcopy(self.model) - return LightningNeuralNetModel(model_copy) - - ''' Non-standard lightning methods ''' - def __prepare_data(self, X:np.ndarray, y:np.ndarray): - dataloader = get_dataloader(X, y) - return dataloader - - def initialize_network(self, input_dim:int, output_dim:int): - self.model.initialize_network(input_dim, output_dim) diff --git a/models/pytorch/neural_nets.py b/models/pytorch/neural_nets.py deleted file mode 100644 index 3f61857..0000000 --- a/models/pytorch/neural_nets.py +++ /dev/null @@ -1,61 +0,0 @@ -import torch -from torch import nn -import torch.nn.functional as F -from torch.utils.data import DataLoader, random_split -import pytorch_lightning as pl -import math -import numpy as np - -class MultiLayerPerceptron(pl.LightningModule): - def __init__(self, hidden_layers_ratio: list[float] = [2.0, 2.0], probabilities: bool = False, loss_function=F.mse_loss): - super().__init__() - self.hidden_layers_ratio = hidden_layers_ratio - self.probabilities = probabilities - self.loss_function = loss_function - self.float() - - def initialize_network(self, input_dim: int, output_dim: int) -> None: - self.layers = nn.ModuleList() - current_dim = input_dim - - for hdim in self.hidden_layers_ratio: - hidden_layer_size = int(math.floor(current_dim * hdim)) - self.layers.append(nn.Linear(current_dim, hidden_layer_size)) - self.layers.append(nn.ReLU()) - current_dim = hidden_layer_size - - self.layers.append(nn.Linear(current_dim, output_dim)) - - def forward(self, x: torch.Tensor): - # in lightning, forward defines the prediction/inference actions - x = torch.from_numpy(x).float() - for layer in self.layers: - x = layer(x) - - if self.probabilities: - x = F.softmax(x, dim=1) - - return (x.item(), np.array([])) - - def training_step(self, batch: torch.Tensor, batch_idx): - # training_step defined the train loop. - # It is independent of forward - x, y = batch - x = x.view(x.size(0), -1) - - - loss = 0 - for layer in self.layers: - x = layer(x.float()) - - if self.probabilities: - p = F.softmax(x, dim=1) - loss = F.nll_loss(torch.log(p), y.float()) - - loss = self.loss_function(x, y.float()) - - return loss - - def configure_optimizers(self): - optimizer = torch.optim.Adam(self.parameters(), lr=1e-3) - return optimizer diff --git a/models/pytorch/pytorch_dataset.py b/models/pytorch/pytorch_dataset.py deleted file mode 100644 index 6ca16fd..0000000 --- a/models/pytorch/pytorch_dataset.py +++ /dev/null @@ -1,26 +0,0 @@ -import os -import pandas as pd - -import torch -from torch.utils.data import Dataset -from torch.utils.data import DataLoader - -import numpy as np - - -class TimeSeriesDataset(Dataset): - def __init__(self, X: np.ndarray, y: np.ndarray): - self.X = X - self.y = y - - def __len__(self): - return len(self.X) - - def __getitem__(self, idx): - return self.X[idx].astype(float), self.y[idx].astype(float) - -def get_dataloader(X: np.ndarray, y: np.ndarray, batch_size: int = 32, shuffle: bool = True): - training_data = TimeSeriesDataset(X, y) - train_dataloader = DataLoader(training_data, batch_size=batch_size, shuffle=shuffle) - - return train_dataloader \ No newline at end of file diff --git a/models/sklearn.py b/models/sklearn.py index 8b681ff..a43df5a 100644 --- a/models/sklearn.py +++ b/models/sklearn.py @@ -1,33 +1,23 @@ from __future__ import annotations from typing import Literal -from models.base import Model +from .base import Model import numpy as np -from sklearn.base import clone -class SKLearnModel(Model): - method: Literal["regression", "classification"] - data_transformation = 'transformed' - only_column = None - model_type = 'ml' - predict_window_size = 'single_timestamp' +def SKLearnModel(instance) -> Model: - def __init__(self, model, method: Literal['regression', 'classification']): - self.model = model - self.method = method + instance.data_transformation = 'transformed' + instance.only_column = None + instance.predict_window_size = 'single_timestamp' + instance.name = instance.__class__.__name__ + + return instance - def fit(self, X: np.ndarray, y: np.ndarray) -> None: - self.model.fit(X, y) - def predict(self, X) -> tuple[float, np.ndarray]: - pred = self.model.predict(X).item() - probability = self.model.predict_proba(X).squeeze() - return (pred, probability) - - def clone(self) -> SKLearnModel: - return SKLearnModel(clone(self.model), self.method) - - def initialize_network(self, input_dim:int, output_dim:int): - pass + # def predict(self, X) -> tuple[float, np.ndarray]: + # pred = self.model.predict(X).item() + # probability = self.model.predict_proba(X).squeeze() + # return (pred, probability) + \ No newline at end of file diff --git a/models/statsmodels.py b/models/statsmodels.py deleted file mode 100644 index 9e43c32..0000000 --- a/models/statsmodels.py +++ /dev/null @@ -1,31 +0,0 @@ -from __future__ import annotations -from statsmodels.tsa.base.tsa_model import TimeSeriesModel -from models.base import Model -import numpy as np -from copy import deepcopy - - -class StatsModel(Model): - - # This is work in progress - data_transformation = 'transformed' - only_column = None - model_type = 'ml' - predict_window_size = 'single_timestamp' - - def __init__(self, model: TimeSeriesModel): - self.model = model - - def fit(self, X: np.ndarray, y: np.ndarray) -> None: - self.model.fit(X, y) - - def predict(self, X) -> tuple[float, np.ndarray]: - pred = self.model.predict(X).item() - return (pred, np.array([0])) - - def clone(self) -> StatsModel: - return StatsModel(deepcopy(self.model)) - - def initialize_network(self, input_dim:int, output_dim:int): - pass - \ No newline at end of file diff --git a/models/xgboost.py b/models/xgboost.py index 3473615..eee592e 100644 --- a/models/xgboost.py +++ b/models/xgboost.py @@ -1,33 +1,22 @@ -from __future__ import annotations -from models.base import Model -import numpy as np -from xgboost import XGBClassifier -from sklearn.base import clone +# from __future__ import annotations +# from models.base import Model +# import numpy as np +# from xgboost import XGBClassifier -class XGBoostModel(Model): +# class XGBoostModel(XGBClassifier): - method = 'classification' - data_transformation = 'transformed' - only_column = None - model_type = 'ml' - predict_window_size = 'single_timestamp' - - def __init__(self, model: XGBClassifier): - self.model = model +# method = 'classification' +# data_transformation = 'transformed' +# only_column = None +# predict_window_size = 'single_timestamp' - def fit(self, X: np.ndarray, y: np.ndarray) -> None: - def map_to_xgb(y): return np.array([1 if i == 1 else 0 for i in y]) - self.model.fit(X, map_to_xgb(y)) +# def fit(self, X: np.ndarray, y: np.ndarray) -> None: +# def map_to_xgb(y): return np.array([1 if i == 1 else 0 for i in y]) +# self.fit(X, map_to_xgb(y)) + +# def predict(self, X) -> tuple[float, np.ndarray]: +# pred = self.predict(X).item() +# probability = self.predict_proba(X).squeeze() +# def map_from_xgb(y): return 1 if y == 1 else -1 +# return (map_from_xgb(pred), probability) - def predict(self, X) -> tuple[float, np.ndarray]: - pred = self.model.predict(X).item() - probability = self.model.predict_proba(X).squeeze() - def map_from_xgb(y): return 1 if y == 1 else -1 - return (map_from_xgb(pred), probability) - - def clone(self) -> XGBoostModel: - return XGBoostModel(clone(self.model)) - - def initialize_network(self, input_dim:int, output_dim:int): - pass - \ No newline at end of file diff --git a/mycodemod/replace_functions.py b/mycodemod/replace_functions.py index 377d454..aee7b30 100644 --- a/mycodemod/replace_functions.py +++ b/mycodemod/replace_functions.py @@ -22,7 +22,7 @@ class ReplaceFunctionCommand(VisitorBasedCodemodCommand): def leave_FunctionDef(self, original_node: cst.FunctionDef, updated_node: cst.FunctionDef) -> cst.FunctionDef: functions_docstring = updated_node.get_docstring() - docstring_should_be = '"""No docstring here yet."""' + docstring_should_be = '' if functions_docstring is not None: docstring_should_be = '"""\n{}\n\n"""'.format(functions_docstring) diff --git a/reporting/reporting.py b/reporting/reporting.py index 4989b4f..4ff31d9 100644 --- a/reporting/reporting.py +++ b/reporting/reporting.py @@ -4,7 +4,7 @@ from utils.helpers import weighted_average from config.types import Config from training.types import WeightsSeries, Stats -def report_results(directional_stats: list[Stats], output_stats: Stats, output_weights: WeightsSeries, config: Config, wandb, sweep: bool): +def report_results(directional_stats: Stats, output_stats: Stats, output_weights: WeightsSeries, config: Config, wandb, sweep: bool): # Only send the results of the final model to wandb send_report_to_wandb(output_stats, wandb) @@ -13,18 +13,16 @@ def report_results(directional_stats: list[Stats], output_stats: Stats, output_w output_weights.rename(config.target_asset[1]).to_csv('output/predictions.csv') print("\n--------\n") - 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: ", output_stats['benchmark_sharpe']) - 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)) + print("Level-1: Number of samples evaluated: ", directional_stats['no_of_samples']) + print("Mean Sharpe ratio for Level-1 models: ", round(directional_stats['sharpe'], 3)) + print("Mean Probabilistic Sharpe ratio for Level-1 models: ", round(directional_stats['prob_sharpe'], 3)) - 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']) + 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/run_fetch_data_minute.py b/run_fetch_data_5min.py similarity index 83% rename from run_fetch_data_minute.py rename to run_fetch_data_5min.py index d9c9e0f..81ce663 100644 --- a/run_fetch_data_minute.py +++ b/run_fetch_data_5min.py @@ -2,6 +2,7 @@ import pandas as pd import ssl from tqdm import tqdm from data_loader.utils import deduplicate_indexes +from utils.resample import resample_ohlc base_url = "https://www.cryptodatadownload.com/cdd/" @@ -27,6 +28,10 @@ for file in tqdm(files_to_download): data.drop(volume_column_to_delete + ['symbol'], axis=1, inplace=True) data.rename({'Volume USD': 'volume'}, axis=1, inplace=True) data.index.rename('time', inplace=True) - target_file = file.split('_')[1].replace('USD', '') + '_USD' + data.sort_index(inplace=True) data = deduplicate_indexes(data) - data.to_csv(f"data/minute_crypto/{target_file}.csv") + data.index = pd.to_datetime(data.index) + data = data.resample('1Min').ffill() + data = resample_ohlc(data, '5Min') + target_file = file.split('_')[1].replace('USD', '') + '_USD' + data.to_csv(f"data/5min_crypto/{target_file}.csv") diff --git a/run_inference.py b/run_inference.py index f934ef2..b1dedd6 100644 --- a/run_inference.py +++ b/run_inference.py @@ -8,9 +8,8 @@ from config.presets import get_dev_config, get_default_ensemble_config, get_ligh from labeling.process import label_data import pandas as pd -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.directional_training import train_directional_model +from training.bet_sizing import bet_sizing_with_meta_model from training.types import PipelineOutcome def run_inference(preload_models:bool, fallback_raw_config: RawConfig): @@ -26,7 +25,7 @@ def run_inference(preload_models:bool, fallback_raw_config: RawConfig): def __inference(config: Config, pipeline_outcome: PipelineOutcome): # 1. Load data, check for validity and process data - X, returns, forward_returns = load_data( + X, returns = load_data( assets = config.assets, other_assets = config.other_assets, exogenous_data = config.exogenous_data, @@ -38,26 +37,18 @@ def __inference(config: Config, pipeline_outcome: PipelineOutcome): ) assert check_data(X, config) == True, "Data is not valid. Cancelling Inference." - events, X, y, forward_returns = label_data(config.event_filter, config.labeling, X, returns, forward_returns) + # 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) inference_from: pd.Timestamp = X.index[len(X.index) - 1] - # 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 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) + directional_training_outcome = train_directional_model(X, y, forward_returns, config, config.directional_model, 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)] + bet_sizing_outcome = bet_sizing_with_meta_model(X, directional_training_outcome.training.predictions, y, forward_returns, config.meta_model, config, 'meta', from_index = inference_from, transformations_over_time = pipeline_outcome.bet_sizing.meta_transformations, preloaded_models = pipeline_outcome.bet_sizing.meta_training.model_over_time) - # 4. Ensemble weights - ensemble_outcome = ensemble_weights([o.weights for o in bet_sizing_outcomes], forward_returns, y, config.no_of_classes, config.mode == 'training') - - # 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) + return PipelineOutcome(directional_training_outcome, bet_sizing_outcome) if __name__ == '__main__': diff --git a/run_pipeline.py b/run_pipeline.py index 4d7f247..fe1059d 100644 --- a/run_pipeline.py +++ b/run_pipeline.py @@ -1,7 +1,7 @@ from typing import Optional from config.types import Config, RawConfig -from config.preprocess import preprocess_config, validate_config +from config.preprocess import preprocess_config from config.presets import get_default_ensemble_config, get_lightweight_ensemble_config from data_loader.load import load_data @@ -13,9 +13,8 @@ from reporting.wandb import launch_wandb, override_config_with_wandb_values from reporting.reporting import report_results from reporting.saving import save_models -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.directional_training import train_directional_model +from training.bet_sizing import bet_sizing_with_meta_model from training.types import PipelineOutcome import ray @@ -25,7 +24,7 @@ ray.init() 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) 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) + report_results(pipeline_outcome.directional_training.training.stats, pipeline_outcome.get_output_stats(), pipeline_outcome.get_output_weights(), config, wandb, sweep) save_models(pipeline_outcome, config) return pipeline_outcome, config @@ -42,11 +41,9 @@ def __setup_config(project_name:str, with_wandb: bool, sweep: bool, raw_config: def __run_training(config: Config) -> PipelineOutcome: - - validate_config(config) - # 1. Load data, check for validity - X, returns, forward_returns = load_data( + print("---> Load data, check for validity") + X, returns = load_data( assets = config.assets, other_assets = config.other_assets, exogenous_data = config.exogenous_data, @@ -59,22 +56,16 @@ def __run_training(config: Config) -> PipelineOutcome: assert check_data(X, config) == True, "Data is not valid." - # 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) + print("---> 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) - # 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) + print("---> Train directional models") + directional_training_outcome = train_directional_model(X, y, forward_returns, config, config.directional_model, 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] + print("---> Run bet sizing on directional model's output") + bet_sizing_outcomes = bet_sizing_with_meta_model(X, directional_training_outcome.training.predictions, y, forward_returns, config.meta_model, config, 'meta', None, None, None) - # 4. Ensemble weights - ensemble_outcome = ensemble_weights([o.weights for o in bet_sizing_outcomes], forward_returns, y, config.no_of_classes, config.mode == 'training') - - # 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) + return PipelineOutcome(directional_training_outcome, bet_sizing_outcomes) 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 + run_pipeline(project_name='price-prediction', with_wandb = False, sweep = False, raw_config=get_lightweight_ensemble_config()) \ No newline at end of file diff --git a/tests/test_evaluation.py b/tests/test_evaluation.py index 9c48439..dcc3317 100644 --- a/tests/test_evaluation.py +++ b/tests/test_evaluation.py @@ -4,6 +4,7 @@ import pandas as pd from training.walk_forward import walk_forward_train, walk_forward_inference from models.base import Model from utils.evaluate import evaluate_predictions +from sklearn.base import BaseEstimator, ClassifierMixin no_of_rows = 100 @@ -29,7 +30,8 @@ def __generate_even_odd_test_data(no_of_rows) -> tuple[pd.DataFrame, pd.Series]: return X, y -class EvenOddStubModel(Model): + +class EvenOddStubModel(BaseEstimator, ClassifierMixin, Model): ''' A deteministic model that can predict the future with 100% accuracy It verifies that the X[n][any_column] == 1 if n is even, @@ -49,14 +51,10 @@ class EvenOddStubModel(Model): assert y[i] == -1 if X[i][0] == 1 else 1 def predict(self, X): - return (-1 if X[0][0] == 1 else 1, np.array([])) + return np.array([-1 if row[0] == 1 else 1 for row in X]) - def clone(self): - return self - - - def initialize_network(self, input_dim: int, output_dim: int): - pass + def predict_proba(self, X): + return np.array([[row[0] + 1, 0] for row in X]) def test_evaluation(): diff --git a/tests/test_walk_forward.py b/tests/test_walk_forward.py index 0347efc..c0d381d 100644 --- a/tests/test_walk_forward.py +++ b/tests/test_walk_forward.py @@ -2,6 +2,7 @@ import numpy as np import pandas as pd from training.walk_forward import walk_forward_train, walk_forward_inference from models.base import Model +from sklearn.base import BaseEstimator, ClassifierMixin no_of_rows = 100 @@ -27,7 +28,7 @@ def __generate_incremental_test_data(no_of_rows) -> tuple[pd.DataFrame, pd.Serie -class IncrementingStubModel(Model): +class IncrementingStubModel(Model, BaseEstimator, ClassifierMixin): ''' A deteministic model that can predict the future with 100% accuracy It verifies that the X[n][any_column]+1 == y[n] @@ -48,13 +49,10 @@ class IncrementingStubModel(Model): assert X[i][0] + 1 == y[i] def predict(self, X): - return (X[0][0] + 1, np.array([])) + return np.array([row[0] + 1 for row in X]) - def clone(self): - return self - - def initialize_network(self, input_dim: int, output_dim: int): - pass + def predict_proba(self, X): + return np.array([[row[0] + 1, 0] for row in X]) def test_walk_forward_train_test(): X, y = __generate_incremental_test_data(no_of_rows) diff --git a/training/bet_sizing.py b/training/bet_sizing.py index ea389a3..bc1591b 100644 --- a/training/bet_sizing.py +++ b/training/bet_sizing.py @@ -1,7 +1,7 @@ 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 +from .train_model import train_model import pandas as pd from models.base import Model from models.model_map import default_feature_selector_classification @@ -13,17 +13,17 @@ from transformations.scaler import get_scaler from transformations.rfe import RFETransformation from transformations.pca import PCATransformation -def bet_sizing_with_meta_models( +def bet_sizing_with_meta_model( X: XDataFrame, input_predictions: pd.Series, y: ySeries, forward_returns: ForwardReturnSeries, - models: list[Model], + model: Model, config: Config, model_suffix: str, from_index: Optional[pd.Timestamp], transformations_over_time: Optional[TransformationsOverTime] = None, - preloaded_models: Optional[list[ModelOverTime]] = None + preloaded_models: Optional[ModelOverTime] = None ) -> BetSizingWithMetaOutcome: input_predictions.name = "model_predictions" @@ -50,12 +50,12 @@ def bet_sizing_with_meta_models( ], ) - meta_outcomes = train_models( + meta_outcome = train_model( ticker_to_predict = "prediction_correct", X = meta_X, y = meta_y, forward_returns = forward_returns, - models = models, + model = model, expanding_window = config.expanding_window_meta, sliding_window_size = config.sliding_window_size_meta, retrain_every = config.retrain_every, @@ -64,16 +64,11 @@ def bet_sizing_with_meta_models( level = 'meta', output_stats = config.mode == 'training', transformations_over_time = transformations_over_time, - models_over_time = preloaded_models, + model_over_time = preloaded_models, ) - # Ensemble predictions if necessary - if len(models) > 1: - meta_predictions = pd.concat([outcome.predictions for outcome in meta_outcomes], axis = 1).mean(axis = 1).apply(discretize_threeway_threshold(0.5)) - bet_size = pd.concat([outcome.probabilities[outcome.probabilities.columns[1::2]] for outcome in meta_outcomes], axis = 1).mean(axis = 1) - else: - meta_predictions = meta_outcomes[0].predictions - bet_size = meta_outcomes[0].probabilities.iloc[:,1] + meta_predictions = meta_outcome.predictions + bet_size = meta_outcome.probabilities.iloc[:,1] avg_predictions_with_sizing = input_predictions * meta_predictions * bet_size if config.mode == 'training': @@ -89,4 +84,4 @@ def bet_sizing_with_meta_models( stats = None model_id = "model_" + config.target_asset[1] + "_" + model_suffix - return BetSizingWithMetaOutcome(model_id, meta_outcomes, transformations_over_time, avg_predictions_with_sizing, stats) + return BetSizingWithMetaOutcome(model_id, meta_outcome, transformations_over_time, avg_predictions_with_sizing, stats) diff --git a/training/directional_training.py b/training/directional_training.py index bfb44b8..a4b4dcb 100644 --- a/training/directional_training.py +++ b/training/directional_training.py @@ -13,12 +13,12 @@ from transformations.scaler import get_scaler from transformations.rfe import RFETransformation from transformations.pca import PCATransformation -def train_directional_models( +def train_directional_model( X: pd.DataFrame, y: pd.Series, forward_returns: pd.Series, config: Config, - models: list[Model], + model: Model, from_index: Optional[pd.Timestamp], preloaded_training_step: Optional[DirectionalTrainingOutcome] = None, ) -> DirectionalTrainingOutcome: @@ -42,12 +42,7 @@ def train_directional_models( else: transformations_over_time = preloaded_training_step.transformations - def print_stats(outcome: TrainingOutcome) -> TrainingOutcome: - if config.mode == 'training': - print(outcome.stats) - return outcome - - training_outcomes = [print_stats(train_model( + training_outcome = train_model( ticker_to_predict = config.target_asset[1], X = X, y = y, @@ -55,13 +50,16 @@ def train_directional_models( model = model, expanding_window = config.expanding_window_base, sliding_window_size = config.sliding_window_size_base, - retrain_every = config.retrain_every, + retrain_every = config.retrain_every, from_index = from_index, no_of_classes = config.no_of_classes, level = 'primary', output_stats= config.mode == 'training', 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) + model_over_time = preloaded_training_step.training.model_over_time if preloaded_training_step else None + ) + if config.mode == 'training': + print(training_outcome.stats) + + return DirectionalTrainingOutcome(training_outcome, transformations_over_time) diff --git a/training/ensemble.py b/training/ensemble.py deleted file mode 100644 index 02d1c67..0000000 --- a/training/ensemble.py +++ /dev/null @@ -1,26 +0,0 @@ -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'], - output_stats: bool - ) -> EnsembleOutcome: - weights = pd.concat(input_weights, axis=1).mean(axis=1) - if output_stats: - stats = evaluate_predictions( - forward_returns = forward_returns, - y_pred = weights, - y_true = y, - no_of_classes = no_of_classes, - discretize = True, - ) - print(stats) - else: - stats = None - return EnsembleOutcome(weights, stats) diff --git a/training/train_model.py b/training/train_model.py index c900cbd..59d4a44 100644 --- a/training/train_model.py +++ b/training/train_model.py @@ -1,29 +1,10 @@ import pandas as pd from typing import Literal, Optional -from training.walk_forward import walk_forward_train, walk_forward_inference +from training.walk_forward import walk_forward_train, walk_forward_inference, walk_forward_inference_batched 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, - output_stats: 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, output_stats, 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, @@ -61,7 +42,8 @@ def train_model( else: model_id = model_over_time.name - predictions, probabilities = walk_forward_inference( + inference_function = walk_forward_inference if from_index is not None else walk_forward_inference_batched + predictions, probabilities = inference_function( model_name = model_id, model_over_time= model_over_time, transformations_over_time = transformations_over_time, diff --git a/training/types.py b/training/types.py index 2e44970..17a74f5 100644 --- a/training/types.py +++ b/training/types.py @@ -19,33 +19,26 @@ class TrainingOutcome: stats: Optional[Stats] model_over_time: ModelOverTime -@dataclass -class EnsembleOutcome: - weights: WeightsSeries - stats: Optional[Stats] - @dataclass class BetSizingWithMetaOutcome: model_id: str - meta_training: list[TrainingOutcome] + meta_training: TrainingOutcome meta_transformations: TransformationsOverTime weights: WeightsSeries stats: Optional[Stats] @dataclass class DirectionalTrainingOutcome: - training: list[TrainingOutcome] + training: TrainingOutcome transformations: TransformationsOverTime @dataclass class PipelineOutcome: directional_training: DirectionalTrainingOutcome - bet_sizing: list[BetSizingWithMetaOutcome] - ensemble: EnsembleOutcome - secondary_bet_sizing: Optional[BetSizingWithMetaOutcome] + bet_sizing: BetSizingWithMetaOutcome def get_output_weights(self) -> WeightsSeries: - return self.secondary_bet_sizing.weights if self.secondary_bet_sizing else self.ensemble.weights + return self.bet_sizing.weights - def get_output_stats(self) -> Optional[Stats]: - return self.secondary_bet_sizing.stats if self.secondary_bet_sizing else self.ensemble.stats \ No newline at end of file + def get_output_stats(self) -> Stats: + return self.bet_sizing.stats \ No newline at end of file diff --git a/training/walk_forward/__init__.py b/training/walk_forward/__init__.py index f3865ed..70759af 100644 --- a/training/walk_forward/__init__.py +++ b/training/walk_forward/__init__.py @@ -1,3 +1,4 @@ +from .inference_batched import walk_forward_inference_batched 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 +from .process_transformations 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 index 05465b5..2fedf74 100644 --- a/training/walk_forward/inference.py +++ b/training/walk_forward/inference.py @@ -16,7 +16,7 @@ def walk_forward_inference( retrain_every: int, from_index: Optional[pd.Timestamp], ) -> tuple[PredictionsSeries, ProbabilitiesDataFrame]: - predictions = pd.Series(index=X.index).rename(model_name) + predictions = pd.Series(index=X.index, dtype='object').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) @@ -49,7 +49,9 @@ def walk_forward_inference( next_timestep = next_timestep.to_numpy() - prediction, probs = current_model.predict(next_timestep) + prediction = current_model.predict(next_timestep) + probs = current_model.predict_proba(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))]) diff --git a/training/walk_forward/inference_batched.py b/training/walk_forward/inference_batched.py new file mode 100644 index 0000000..4013d8a --- /dev/null +++ b/training/walk_forward/inference_batched.py @@ -0,0 +1,56 @@ +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 tqdm import tqdm + +def walk_forward_inference_batched( + 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, dtype='object').rename(model_name) + probabilities = pd.DataFrame(index=X.index, columns=['0', '1']) + + 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 = [] + + batch_indices = range(inference_from, inference_till, retrain_every) if inference_till - inference_from > retrain_every else [inference_from] + batched_results = [__inference_from_window(index, index + retrain_every, X, model_over_time, transformations_over_time, expanding_window, window_size) for index in tqdm(batch_indices)] + for batch in batched_results: + for index, prediction, probs in batch: + predictions[X.index[index]] = prediction + probabilities.loc[X.index[index]] = probs + + return predictions, probabilities + +def __inference_from_window(index_start: int, index_end: int, X: XDataFrame, model_over_time: ModelOverTime, transformations_over_time: TransformationsOverTime, expanding_window: bool, window_size: int) -> list[tuple[int, float, pd.Series]]: + current_model = model_over_time[X.index[index_start]] + current_transformations = [transformation_over_time[X.index[index_start]] for transformation_over_time in transformations_over_time] + + input_data = X.iloc[index_start:index_end] + + for transformation in current_transformations: + input_data = transformation.transform(input_data) + + input_data = input_data.to_numpy() + + predictions = current_model.predict(input_data) + probs = current_model.predict_proba(input_data) + results = [(index_start + index, predictions[index], probs[index]) for index in range(len(predictions))] + + return results \ No newline at end of file diff --git a/training/walk_forward/inference_parallel.py b/training/walk_forward/inference_parallel.py index 326f660..775f669 100644 --- a/training/walk_forward/inference_parallel.py +++ b/training/walk_forward/inference_parallel.py @@ -18,7 +18,7 @@ def walk_forward_inference( retrain_every: int, from_index: Optional[pd.Timestamp], ) -> tuple[PredictionsSeries, ProbabilitiesDataFrame]: - predictions = pd.Series(index=X.index).rename(model_name) + predictions = pd.Series(index=X.index, dtype='object').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) @@ -31,32 +31,38 @@ def walk_forward_inference( 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 + batch_size = int((inference_till - inference_from) / 10) + batched_results = ray.get([__inference_from_window.remote(index, index + batch_size, inference_from, retrain_every, X, model_over_time, transformations_over_time, expanding_window, window_size) for index in range(inference_from, inference_till)]) + for batch in batched_results: + for index, prediction, probs in batch: + 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]] +def __inference_from_window(index_start: int, index_end: int, inference_from: int, retrain_every: int, X: XDataFrame, model_over_time: ModelOverTime, transformations_over_time: TransformationsOverTime, expanding_window: bool, window_size: int) -> list[tuple[int, float, pd.Series]]: - for transformation in current_transformations: - next_timestep = transformation.transform(next_timestep) + results = [] + for index in range(index_start, index_end): + 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] - next_timestep = next_timestep.to_numpy() + 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] - prediction, probs = current_model.predict(next_timestep) - return index, prediction, probs \ No newline at end of file + 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) + results.append((index, prediction, probs)) + + return results \ No newline at end of file diff --git a/training/walk_forward/process_transformations.py b/training/walk_forward/process_transformations.py index c0da95a..d0e815c 100644 --- a/training/walk_forward/process_transformations.py +++ b/training/walk_forward/process_transformations.py @@ -17,7 +17,7 @@ def walk_forward_process_transformations( 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] + transformations_over_time = [pd.Series(index=y.index, dtype='object').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) diff --git a/training/walk_forward/process_transformations_parallel.py b/training/walk_forward/process_transformations_parallel.py index 269b2bc..b8f1f07 100644 --- a/training/walk_forward/process_transformations_parallel.py +++ b/training/walk_forward/process_transformations_parallel.py @@ -6,6 +6,7 @@ from transformations.base import Transformation from typing import Optional from data_loader.types import ForwardReturnSeries, XDataFrame, ySeries import ray +from utils.parallel import parallel_compute_with_bar def walk_forward_process_transformations( X: XDataFrame, @@ -23,7 +24,7 @@ def walk_forward_process_transformations( 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)]) + processed_transformations = parallel_compute_with_bar([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): diff --git a/training/walk_forward/train.py b/training/walk_forward/train.py index f2a772f..3d85297 100644 --- a/training/walk_forward/train.py +++ b/training/walk_forward/train.py @@ -5,7 +5,7 @@ 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 - +from copy import deepcopy def walk_forward_train( model: Model, @@ -18,7 +18,7 @@ def walk_forward_train( from_index: Optional[pd.Timestamp], transformations_over_time: TransformationsOverTime, ) -> ModelOverTime: - models_over_time = pd.Series(index=y.index).rename(model.name) + models_over_time = pd.Series(index=y.index, dtype='object').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) @@ -43,13 +43,9 @@ def walk_forward_train( 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 = deepcopy(model) 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/training/walk_forward/train_parallel.py b/training/walk_forward/train_parallel.py new file mode 100644 index 0000000..f6ffee7 --- /dev/null +++ b/training/walk_forward/train_parallel.py @@ -0,0 +1,58 @@ +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 +import ray +from utils.parallel import parallel_compute_with_bar +from copy import deepcopy + +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 = [] + + models = parallel_compute_with_bar([train_on_window.remote(index, first_nonzero_return, window_size, X, y, model, expanding_window, transformations_over_time) for index in tqdm(range(train_from, train_till, retrain_every))]) + for index, current_model in models: + models_over_time[X.index[index]] = current_model + + return models_over_time + +@ray.remote +def train_on_window(index: int, first_nonzero_return: int, window_size: int, X: XDataFrame, y: ySeries, model: Model, expanding_window: bool, transformations_over_time: TransformationsOverTime) -> tuple[int, Model]: + 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 = deepcopy(model) + + current_model.fit(X_slice, y_slice) + return index, model diff --git a/transformations/rfe.py b/transformations/rfe.py index 2c2b05a..cc5cbb3 100644 --- a/transformations/rfe.py +++ b/transformations/rfe.py @@ -4,21 +4,17 @@ from typing import Optional from copy import deepcopy from sklearn.feature_selection import RFE import pandas as pd -from models.base import Model -from models.model_map import default_feature_selector_classification +from models.sklearn import SKLearnModel class RFETransformation(Transformation): rfe: RFE n_feature_to_select: int - def __init__(self, n_feature_to_select: int, model: Model, step = 0.1): + def __init__(self, n_feature_to_select: int, model: SKLearnModel, step = 0.1): self.n_feature_to_keep = n_feature_to_select self.model = model - if hasattr(self.model, 'model') == False: return - if hasattr(self.model.model, 'feature_importances_') == False and hasattr(self.model.model, 'coef_') == False: - model = default_feature_selector_classification - self.rfe = RFE(model.model, n_features_to_select= n_feature_to_select, step=step) + self.rfe = RFE(model, n_features_to_select= n_feature_to_select, step=step) def fit(self, X: pd.DataFrame, y: Optional[pd.Series] = None) -> None: if self.rfe is None: return diff --git a/utils/evaluate.py b/utils/evaluate.py index 665fd99..ddee5a2 100644 --- a/utils/evaluate.py +++ b/utils/evaluate.py @@ -49,7 +49,7 @@ def evaluate_predictions( return len(series[series != 0]) no_of_samples = count_non_zero(df.y_pred) scorecard['no_of_samples'] = no_of_samples - sharpe = sharpe_ratio(df.result) + sharpe = sharpe_ratio(df.result + 1e-20) scorecard['sharpe'] = sharpe benchmark_sharpe = sharpe_ratio(df.forward_returns) scorecard['benchmark_sharpe'] = benchmark_sharpe diff --git a/utils/parallel.py b/utils/parallel.py new file mode 100644 index 0000000..9890e87 --- /dev/null +++ b/utils/parallel.py @@ -0,0 +1,15 @@ +from tqdm import tqdm +import ray + +def parallel_compute_with_bar(computations) -> list: + + def to_iterator(obj_ids): + while obj_ids: + done, obj_ids = ray.wait(obj_ids) + yield ray.get(done[0]) + + ret = [] + for x in tqdm(to_iterator(computations), total=len(computations)): + ret.append(x) + + return ret \ No newline at end of file diff --git a/utils/resample.py b/utils/resample.py new file mode 100644 index 0000000..ba98692 --- /dev/null +++ b/utils/resample.py @@ -0,0 +1,9 @@ +import pandas as pd + +def resample_ohlc(df, period): + output = pd.DataFrame() + output['open'] = df.open.resample(period).first() + output['high'] = df.high.resample(period).max() + output['low'] = df.low.resample(period).min() + output['close'] = df.close.resample(period).last() + return output \ No newline at end of file