Refactor(Training): new outcome types, representative pipeline steps, bet-sizing (#187)

* refactor(Training): added InferenceResult & TrainedModel types

* refactor(Pipeline): introduced TrainingOutcome, BetSizingWithMetaOutcome, etc.

* fix(Pipeline): getting it to compile

* refactor(WalkForward): separate preprocessing step

* feat(Pipeline): separate out transformations processing step

* refactor(Pipeline): use the Directional model terminology, put bet_sizing into pipeline instead of hiding it in a step

* refactor(WalkForward): moved functions to separate folder

* fix(WalkForward): use sparse array to store models, process transformations in parallel (lot faster)

* fix(Tests): and evaluation

* fix(Tests): for realz

* fix(Inference): preloading everything now, renamed primary models to directional models

* fix(BetSizing): was running transformations on the wrong data, oops

* fix(BetSizing): concatenated on the wrong axis accidentally

* fix(Reporting): able to use the new Stats type

* fix(BetSizing): renamed int column names

* fix(Portfolio): name the column properly

* fix(Reporting): rename the correct Series, lol

* fix(Inference): walk_forwad_inference() can deal with models not being aligned with the starting index

* fix(WalkForward): accidentally using the wrong index

* fix(WalkForward): use the correct indicies to fetch last model/transformations

* fix(CI): changed the name of the results

training/walk_forward/__init__.py

@@ -0,0 +1,3 @@
+from .inference import walk_forward_inference
+from .train import walk_forward_train
+from .process_transformations_parallel import walk_forward_process_transformations

training/walk_forward/inference.py

@@ -0,0 +1,58 @@
+import pandas as pd
+from training.types import ModelOverTime, TransformationsOverTime, PredictionsSeries, ProbabilitiesDataFrame
+from utils.helpers import get_first_valid_return_index
+from tqdm import tqdm
+from typing import Optional
+from data_loader.types import XDataFrame
+from utils.helpers import get_last_non_na_index
+
+def walk_forward_inference(
+                    model_name: str,
+                    model_over_time: ModelOverTime,
+                    transformations_over_time: TransformationsOverTime,
+                    X: XDataFrame,
+                    expanding_window: bool,
+                    window_size: int,
+                    retrain_every: int,
+                    from_index: Optional[pd.Timestamp],
+                ) -> tuple[PredictionsSeries, ProbabilitiesDataFrame]:
+    predictions = pd.Series(index=X.index).rename(model_name)
+    probabilities = pd.DataFrame(index=X.index)
+
+    inference_from = max(get_first_valid_return_index(model_over_time), get_first_valid_return_index(X.iloc[:,0])) if from_index is None else X.index.to_list().index(from_index)
+    inference_till = X.shape[0]
+    model_index_offset = get_last_non_na_index(model_over_time, inference_from) if pd.isna(model_over_time[inference_from]) else 0 
+    first_model = model_over_time[inference_from - model_index_offset] if pd.isna(model_over_time[inference_from]) else model_over_time[inference_from]
+
+    if first_model.only_column is not None:
+        X = X[[column for column in X.columns if first_model.only_column in column]]
+    
+    if first_model.data_transformation == 'original':
+        transformations_over_time = []
+
+    for index in tqdm(range(inference_from, inference_till)):
+        
+        last_model_index = index - ((index - inference_from) % retrain_every) - model_index_offset
+        train_window_start = X.index[inference_from] if expanding_window else X.index[index - window_size - 1]
+
+        current_model = model_over_time[X.index[last_model_index]]
+        current_transformations = [transformation_over_time[X.index[last_model_index]] for transformation_over_time in transformations_over_time]
+
+        if current_model.predict_window_size == 'window_size': 
+            next_timestep = X.loc[train_window_start:X.index[index]]
+        else: 
+            # we need to get a Dataframe out of it, since the transformation step always expects a 2D array, but it's equivalent to X.iloc[index]
+            next_timestep = X.loc[X.index[index]:X.index[index]]
+        
+        for transformation in current_transformations:
+            next_timestep = transformation.transform(next_timestep)
+
+        next_timestep = next_timestep.to_numpy()
+
+        prediction, probs = current_model.predict(next_timestep)
+        predictions[X.index[index]] = prediction
+        if inference_from == index and len(probabilities.columns) != len(probs):
+            probabilities = probabilities.reindex(columns = ["prob_" + str(num) for num in range(0, len(probs.T))])
+        probabilities.loc[X.index[index]] = probs
+
+    return predictions, probabilities

training/walk_forward/inference_parallel.py

@@ -0,0 +1,62 @@
+import pandas as pd
+from models.base import Model
+from training.types import ModelOverTime, TransformationsOverTime, PredictionsSeries, ProbabilitiesDataFrame
+from transformations.base import Transformation
+from utils.helpers import get_first_valid_return_index
+from tqdm import tqdm
+from typing import Optional
+from data_loader.types import XDataFrame
+import ray
+
+def walk_forward_inference(
+                    model_name: str,
+                    model_over_time: ModelOverTime,
+                    transformations_over_time: TransformationsOverTime,
+                    X: XDataFrame,
+                    expanding_window: bool,
+                    window_size: int,
+                    retrain_every: int,
+                    from_index: Optional[pd.Timestamp],
+                ) -> tuple[PredictionsSeries, ProbabilitiesDataFrame]:
+    predictions = pd.Series(index=X.index).rename(model_name)
+    probabilities = pd.DataFrame(index=X.index)
+
+    inference_from = max(get_first_valid_return_index(model_over_time), get_first_valid_return_index(X.iloc[:,0])) if from_index is None else X.index.to_list().index(from_index)
+    inference_till = X.shape[0]
+    first_model = model_over_time[inference_from]
+
+    if first_model.only_column is not None:
+        X = X[[column for column in X.columns if first_model.only_column in column]]
+    
+    if first_model.data_transformation == 'original':
+        transformations_over_time = []
+
+    results = ray.get([__inference_from_window.remote(index, inference_from, retrain_every, X, model_over_time, transformations_over_time, expanding_window, window_size) for index in range(inference_from, inference_till)])
+    for index, prediction, probs in results:
+        predictions[X.index[index]] = prediction
+        probabilities.loc[X.index[index]] = probs
+
+    return predictions, probabilities
+
+@ray.remote
+def __inference_from_window(index: int, inference_from: int, retrain_every: int, X: XDataFrame, model_over_time: ModelOverTime, transformations_over_time: TransformationsOverTime, expanding_window: bool, window_size: int) -> tuple[int, float, pd.Series]:
+            
+    last_model_index = index - ((index - inference_from) % retrain_every)
+    train_window_start = X.index[inference_from] if expanding_window else X.index[index - window_size - 1]
+
+    current_model = model_over_time[X.index[last_model_index]]
+    current_transformations = [transformation_over_time[X.index[last_model_index]] for transformation_over_time in transformations_over_time]
+
+    if current_model.predict_window_size == 'window_size': 
+        next_timestep = X.loc[train_window_start:X.index[index]]
+    else: 
+        # we need to get a Dataframe out of it, since the transformation step always expects a 2D array, but it's equivalent to X.iloc[index]
+        next_timestep = X.loc[X.index[index]:X.index[index]]
+    
+    for transformation in current_transformations:
+        next_timestep = transformation.transform(next_timestep)
+
+    next_timestep = next_timestep.to_numpy()
+
+    prediction, probs = current_model.predict(next_timestep)
+    return index, prediction, probs

training/walk_forward/process_transformations.py

@@ -0,0 +1,48 @@
+import pandas as pd
+from training.types import TransformationsOverTime
+from utils.helpers import get_first_valid_return_index
+from tqdm import tqdm
+from transformations.base import Transformation
+from typing import Optional
+from data_loader.types import ForwardReturnSeries, XDataFrame, ySeries
+
+
+def walk_forward_process_transformations(
+                        X: XDataFrame,
+                        y: ySeries,
+                        forward_returns: ForwardReturnSeries,
+                        expanding_window: bool,
+                        window_size: int,
+                        retrain_every: int,
+                        from_index: Optional[pd.Timestamp],
+                        transformations: list[Transformation],
+                    ) -> TransformationsOverTime:
+    transformations_over_time = [pd.Series(index=y.index).rename(t.get_name()) for t in transformations]
+
+    first_nonzero_return = max(get_first_valid_return_index(forward_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y))
+    train_from = first_nonzero_return + window_size + 1 if from_index is None else X.index.to_list().index(from_index)
+    train_till = len(y)
+    iterations_before_retrain = 0
+    
+    for index in tqdm(range(train_from, train_till)):
+        train_window_start = X.index[first_nonzero_return] if expanding_window else X.index[index - window_size - 1]
+
+        if iterations_before_retrain <= 0 or pd.isna(transformations_over_time[0][index-1]):
+
+            train_window_end = X.index[index - 1]
+            
+            X_expanding_window = X[train_window_start:train_window_end]
+            y_expanding_window = y[train_window_start:train_window_end]
+
+            current_transformations = [t.clone() for t in transformations]
+            for transformation_index, transformation in enumerate(current_transformations):
+                X_expanding_window = transformation.fit_transform(X_expanding_window, y_expanding_window)
+
+            iterations_before_retrain = retrain_every
+
+        for transformation_index, transformation in enumerate(current_transformations):
+            transformations_over_time[transformation_index][X.index[index]] = transformation
+
+        iterations_before_retrain -= 1
+
+    return transformations_over_time

training/walk_forward/process_transformations_parallel.py

@@ -0,0 +1,46 @@
+import pandas as pd
+from training.types import TransformationsOverTime
+from utils.helpers import get_first_valid_return_index
+from tqdm import tqdm
+from transformations.base import Transformation
+from typing import Optional
+from data_loader.types import ForwardReturnSeries, XDataFrame, ySeries
+import ray
+
+def walk_forward_process_transformations(
+                        X: XDataFrame,
+                        y: ySeries,
+                        forward_returns: ForwardReturnSeries,
+                        expanding_window: bool,
+                        window_size: int,
+                        retrain_every: int,
+                        from_index: Optional[pd.Timestamp],
+                        transformations: list[Transformation],
+                    ) -> TransformationsOverTime:
+    transformations_over_time = [pd.Series(index=y.index).rename(t.get_name()) for t in transformations]
+
+    first_nonzero_return = max(get_first_valid_return_index(forward_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y))
+    train_from = first_nonzero_return + window_size + 1 if from_index is None else X.index.to_list().index(from_index)
+    train_till = len(y)
+    
+    processed_transformations = ray.get([preprocess_transformations_window.remote(X, y, expanding_window, window_size, transformations, first_nonzero_return, index) for index in range(train_from, train_till, retrain_every)])
+        
+    for transformation, index_time in processed_transformations:
+        for transformation_index, transformation in enumerate(transformation):
+            transformations_over_time[transformation_index][X.index[index_time]] = transformation
+
+    return transformations_over_time
+
+@ray.remote
+def preprocess_transformations_window(X: XDataFrame, y: ySeries, expanding_window: bool, window_size: int, transformations: list[Transformation], first_nonzero_return: int, index: int) -> tuple[list[Transformation], int]:
+    train_window_start = X.index[first_nonzero_return] if expanding_window else X.index[index - window_size - 1]
+    train_window_end = X.index[index - 1]
+
+    X_expanding_window = X[train_window_start:train_window_end]
+    y_expanding_window = y[train_window_start:train_window_end]
+
+    current_transformations = [t.clone() for t in transformations]
+    for transformation in current_transformations:
+        X_expanding_window = transformation.fit_transform(X_expanding_window, y_expanding_window)
+
+    return (current_transformations, index)

training/walk_forward/train.py

@@ -0,0 +1,55 @@
+import pandas as pd
+from models.base import Model
+from training.types import ModelOverTime, TransformationsOverTime
+from utils.helpers import get_first_valid_return_index
+from tqdm import tqdm
+from typing import Optional
+from data_loader.types import ForwardReturnSeries, XDataFrame, ySeries
+
+
+def walk_forward_train(
+                    model: Model,
+                    X: XDataFrame,
+                    y: ySeries,
+                    forward_returns: ForwardReturnSeries,
+                    expanding_window: bool,
+                    window_size: int,
+                    retrain_every: int,
+                    from_index: Optional[pd.Timestamp],
+                    transformations_over_time: TransformationsOverTime,
+                ) -> ModelOverTime:
+    models_over_time = pd.Series(index=y.index).rename(model.name)
+
+    first_nonzero_return = max(get_first_valid_return_index(forward_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y))
+    train_from = first_nonzero_return + window_size + 1 if from_index is None else X.index.to_list().index(from_index)
+    train_till = len(y)
+
+    if model.only_column is not None:
+        X = X[[column for column in X.columns if model.only_column in column]]
+    
+    if model.data_transformation == 'original':
+        transformations_over_time = []
+    
+    for index in tqdm(range(train_from, train_till, retrain_every)):
+        train_window_start = X.index[first_nonzero_return] if expanding_window else X.index[index - window_size - 1]
+
+        train_window_end = X.index[index - 1]
+        current_transformations = [transformation_over_time[index] for transformation_over_time in transformations_over_time]
+        X_slice = X[train_window_start:train_window_end]
+
+        for transformation in current_transformations:
+            X_slice = transformation.transform(X_slice)
+        
+        X_slice = X_slice.to_numpy()
+        y_slice = y[train_window_start:train_window_end].to_numpy()
+
+        current_model = model.clone()
+
+        current_model.initialize_network(input_dim = len(X_slice[0]), output_dim=1) 
+        current_model.fit(X_slice, y_slice)
+
+        models_over_time[X.index[index]] = current_model
+        for transformation_index, transformation in enumerate(current_transformations):
+            transformations_over_time[transformation_index][X.index[index]] = transformation
+
+    return models_over_time