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

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)

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)
 

training/ensemble.py

@@ -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)

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,

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
+    def get_output_stats(self) -> Stats:
+        return self.bet_sizing.stats

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
+from .process_transformations import walk_forward_process_transformations

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))])

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

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
+        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

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)

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):

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

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