feat(DataLoader): caching MVP, added ability to use standard scaling for exogenous data, scaling is now also done before feature selection (#105)

* fix(FeatureExtractor): apply log to transform some series to normality

* feat(DataLoader): add ability of not returning returns when they're not needed (exogenous data), applied log to certain features

* feat(FeatureExtractors): added standard scaling for exogenous data

* feat(FeatureSelection): scale data with the passed in scaler before doing feature-selection

* fix(Config): sweep config

* feat(Models): output probability, store it

* feat(Core): added caching to select_features() and load_data()

* fix(Dependencies): added diskcache

* fix(Training): error when creating results DF

* feat(Models): added xgboost, fixed tests

* refactor(Cache): moved hashing to a separate function, created wrapper functions to separate business logic and caching

* fix(Tests): new syntax

* fix(Model): XGboost can't handle -1 class, so we'll use the deprecated label_encoder fornow

* fix(Model): XGBoost config

* feat(Cache): add run_clear_cache script

* fix(Pipeline) accidentally re-instatiating all_predictions for each asset

.gitignore

@@ -132,3 +132,5 @@ lightning/lightning_logs/
 results.csv
 predictions.csv
 wandb/
+
+.cachedir/**

config/config.py

@@ -15,7 +15,7 @@ def get_default_level_1_daily_config() -> tuple[dict, dict, dict]:
     )
 
     data_config = dict(
-        assets = ['hourly_crypto'],
+        assets = ['daily_crypto'],
         other_assets = [],
         exogenous_data = [],
         load_non_target_asset= True,
@@ -23,10 +23,11 @@ def get_default_level_1_daily_config() -> tuple[dict, dict, dict]:
         forecasting_horizon = 1,
         own_features = ['level_2', 'date_days'],
         other_features = ['single_mom'],
-        exogenous_features = ['fracdiff'],
+        exogenous_features = ['standard_scaling'],
         index_column= 'int',
         method= 'classification',
-        no_of_classes= 'three-balanced'
+        no_of_classes= 'three-balanced',
+        narrow_format = False,
     )
 
     regression_models = ["Lasso"]
@@ -64,10 +65,11 @@ def get_default_level_2_hourly_config() -> tuple[dict, dict, dict]:
         forecasting_horizon = 1,
         own_features = ['level_2', 'date_days', 'lags_up_to_5'],
         other_features = ['level_2'],
-        exogenous_features = ['fracdiff'],
+        exogenous_features = ['standard_scaling'],
         index_column= 'int',
         method= 'classification',
-        no_of_classes= 'three-balanced'
+        no_of_classes= 'three-balanced',
+        narrow_format = False,
     )
 
     regression_models = ["Lasso", "KNN", "RF"]
@@ -105,17 +107,18 @@ def get_default_level_2_daily_config() -> tuple[dict, dict, dict]:
         load_non_target_asset= True,
         log_returns= True,
         forecasting_horizon = 1,
-        own_features = ['level_2', 'date_days', 'fracdiff'],
-        other_features = ['level_2', 'fracdiff'],
-        exogenous_features = ['fracdiff'],
+        own_features = ['level_2', 'date_days', 'lags_up_to_5'],
+        other_features = ['level_2', 'lags_up_to_5'],
+        exogenous_features = ['standard_scaling'],
         index_column= 'int',
         method= 'classification',
-        no_of_classes= 'three-balanced'
+        no_of_classes= 'three-balanced',
+        narrow_format = False,
     )
 
     regression_models = ["Lasso", "KNN", "RF"]
     regression_ensemble_model = 'KNN'
-    classification_models = ["LDA", "KNN", "CART", "RF", "StaticMom"]
+    classification_models = ['LR', 'LDA', 'KNN', 'CART', 'NB', 'AB', 'RF', 'XGB', 'StaticMom']
     classification_ensemble_model = 'Ensemble_Average'
 
     model_config = dict(

config/hashing.py

@@ -0,0 +1,24 @@
+
+from utils.types import DataCollection
+
+def hash_data_config(data_config: dict) -> str:
+    
+    def hash_data_collection(data_collection: DataCollection) -> str: return ''.join([a[0] + a[1] for a in data_collection])
+    def hash_feature_extractors(feature_extractos) -> str: return ''.join([f[0] for f in feature_extractos])
+    def to_str(x): return ''.join([str(i) for i in x])
+    return '_'.join(to_str([
+        hash_data_collection(data_config['assets']),
+        hash_data_collection(data_config['other_assets']),
+        hash_data_collection(data_config['exogenous_data']),
+        data_config['target_asset'][0] + data_config['target_asset'][1],
+        data_config['load_non_target_asset'],
+        data_config['log_returns'],
+        data_config['forecasting_horizon'],
+        hash_feature_extractors(data_config['own_features']),
+        hash_feature_extractors(data_config['other_features']),
+        hash_feature_extractors(data_config['exogenous_features']),
+        data_config['index_column'],
+        data_config['method'],
+        data_config['no_of_classes'],
+        data_config['narrow_format']
+    ]))

data_loader/load_data.py

@@ -6,8 +6,20 @@ from data_loader.collections import DataCollection
 from typing import Literal
 import ray
 import os
+from config.hashing import hash_data_config
+from diskcache import Cache
+cache = Cache(".cachedir/data")
 
-def load_data(assets: DataCollection,
+def load_data(**kwargs):
+    hashed = hash_data_config(kwargs)
+    if hashed in cache:
+        return cache.get(hashed)
+    else:
+        return_value = __load_data(**kwargs)
+        cache[hashed] = return_value
+        return return_value
+
+def __load_data(assets: DataCollection,
             other_assets: DataCollection,
             exogenous_data: DataCollection,
             target_asset: DataSource,
@@ -33,13 +45,22 @@ def load_data(assets: DataCollection,
     target_file = [f for f in assets if f[1].startswith(target_asset[1])]
     assert len(target_file) == 1, "There should be exactly one target file"
     other_files = [f for f in assets if load_non_target_asset == True and f[1].startswith(target_asset[1]) == False]
-    files = target_file + other_files + other_assets
-    def is_target_asset(target_asset: str, file: str): return file.split('.')[0].startswith(target_asset)
+    files = other_files + other_assets
+    
+    target_asset_future = [__load_df.remote(
+        data_source=data_source,
+        prefix=data_source[1],
+        returns='log_returns' if log_returns else 'returns',
+        feature_extractors=own_features,
+        narrow_format=narrow_format,
+    ) for data_source in target_file]
+    target_asset_df = ray.get(target_asset_future)
+
     asset_futures = [__load_df.remote(
         data_source=data_source,
         prefix=data_source[1],
         returns='log_returns' if log_returns else 'returns',
-        feature_extractors=own_features if is_target_asset(target_asset[1], data_source[1]) else other_features,
+        feature_extractors=other_features,
         narrow_format=narrow_format,
     ) for data_source in files]
     asset_dfs = ray.get(asset_futures)
@@ -47,19 +68,19 @@ def load_data(assets: DataCollection,
     exogenous_futures = [__load_df.remote(
         data_source=data_source,
         prefix=data_source[1],
-        returns='returns',
+        returns='none',
         feature_extractors=exogenous_features,
         narrow_format=narrow_format,
     ) for data_source in exogenous_data]
     exogenous_dfs = ray.get(exogenous_futures)
 
-    dfs = asset_dfs + exogenous_dfs
+    dfs = target_asset_df + asset_dfs + exogenous_dfs
     dfs = [deduplicate_indexes(df) for df in dfs]
-    longest_df = max(dfs, key=lambda df: df.shape[0])
+    target_df = dfs[0]
     if narrow_format:
-        dfs = pd.concat([df.sort_index().reindex(longest_df.index) for df in dfs], axis=0).fillna(0.)
+        dfs = pd.concat([df.sort_index().reindex(target_df.index) for df in dfs], axis=0).fillna(0.)
     else:
-        dfs = pd.concat([df.sort_index().reindex(longest_df.index) for df in dfs], axis=1).fillna(0.)
+        dfs = pd.concat([df.sort_index().reindex(target_df.index) for df in dfs], axis=1).fillna(0.)
 
     dfs.index = pd.DatetimeIndex(dfs.index)
 
@@ -90,7 +111,7 @@ def load_data(assets: DataCollection,
 @ray.remote
 def __load_df(data_source: DataSource,
             prefix: str,
-            returns: Literal['price', 'returns', 'log_returns'],
+            returns: Literal['none', 'price', 'returns', 'log_returns'],
             feature_extractors: list[tuple[str, FeatureExtractor, list[int]]],
             narrow_format: bool = False) -> pd.DataFrame:
     df = pd.read_csv(os.path.join(data_source[0], data_source[1] + '.csv'), header=0, index_col=0).fillna(0)
@@ -99,7 +120,7 @@ def __load_df(data_source: DataSource,
         df['returns'] = np.log(df['close']).diff(1)
     elif returns == 'price':
         df['returns'] = df['close']
-    else:
+    elif returns == 'returns':
         df['returns'] = df['close'].pct_change()
 
     df = __apply_feature_extractors(df, log_returns=True if returns == 'log_returns' else False, feature_extractors = feature_extractors)
@@ -124,7 +145,7 @@ def __apply_feature_extractors(df: pd.DataFrame,
             if type(features) == pd.DataFrame:
                 df = pd.concat([df, features], axis=1)
             elif type(features) == pd.Series:
-                df[name + '_' + str(period)] = extractor(df, period, log_returns)
+                df[name + '_' + str(period)] = features
             else:
                 assert False, "Feature extractor must return a pd.DataFrame or pd.Series"
     return df

environment.yml

@@ -24,7 +24,9 @@ dependencies:
   - tqdm
   - pip
   - pandas-ta
+  - xgboost
   - pip:
     - fracdiff
     - ray
+    - diskcache
 prefix: /usr/local/anaconda3/envs/quant

feature_extractors/feature_extractor_presets.py

@@ -1,7 +1,7 @@
-from feature_extractors.feature_extractors import feature_lag, feature_mom, feature_ROC, feature_RSI, feature_STOD, feature_STOK, feature_vol, feature_day_of_month, feature_day_of_week, feature_month, feature_debug_future_lookahead
+from feature_extractors.feature_extractors import feature_lag, feature_mom, feature_ROC, feature_RSI, feature_STOD, feature_STOK, feature_standard_scaling, feature_vol, feature_day_of_month, feature_day_of_week, feature_month, feature_debug_future_lookahead
 from utils.types import FeatureExtractorConfig
 from utils.helpers import flatten
-from feature_extractors.fractional_differentiation import feature_fractional_differentiation
+from feature_extractors.fractional_differentiation import feature_fractional_differentiation, feature_fractional_differentiation_log
 
 __presets = dict(
     debug_future_lookahead = [('debug_future', feature_debug_future_lookahead, [1])],
@@ -24,6 +24,8 @@ __presets = dict(
     stod = [('stod', feature_STOD, [10, 30, 200])],
     stok = [('stok', feature_STOK, [10, 30, 200])],
     fracdiff = [('fracdiff', feature_fractional_differentiation, [10, 30])],
+    fracdiff_log = [('fracdiff_log', feature_fractional_differentiation_log, [10, 30])],
+    standard_scaling = [('standard_scaling', feature_standard_scaling, [0])],
 )
 
 presets = __presets | dict(

feature_extractors/feature_extractors.py

@@ -1,6 +1,8 @@
 import pandas as pd
 import numpy as np
 from feature_extractors.utils import get_close_low_high
+from feature_extractors.utils import apply_log_if_necessary_series
+from sklearn.preprocessing import StandardScaler
 
 def feature_debug_future_lookahead(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
     return df['returns'].shift(-period)
@@ -9,6 +11,10 @@ def feature_lag(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series
     assert period > 0
     return df['returns'].shift(period)
 
+def feature_standard_scaling(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
+    scaler = StandardScaler()
+    return pd.Series(scaler.fit_transform(df['close'].to_numpy().reshape(-1, 1)).squeeze(), index = df.index)
+
 def feature_day_of_week(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.DataFrame:
     return pd.get_dummies(pd.DatetimeIndex(df.index).dayofweek, drop_first=True, prefix="date_day_week").set_index(df.index)
 
@@ -31,7 +37,7 @@ def feature_STOK(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Serie
     close, low, high = get_close_low_high(df)
 
     STOK = ((close - low.rolling(period).min()) / (high.rolling(period).max() - low.rolling(period).min())) * 100
-    return STOK
+    return apply_log_if_necessary_series(STOK, "stok")
 
 def feature_STOD(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
     stok = feature_STOK(df, period, is_log_return)
@@ -48,10 +54,11 @@ def feature_RSI(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series
     d[d.index[period-1]] = np.mean( d[:period] ) #first value is sum of avg losses d = d.drop(d.index[:(period-1)])
     rs = u.ewm(com=period-1, adjust=False).mean() / \
     d.ewm(com=period-1, adjust=False).mean()
-    return 100-100/(1+rs)
+    return apply_log_if_necessary_series(100-100/(1+rs), "rsi")
 
 def feature_ROC(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
     returns = df['returns']
     M = returns.diff(period - 1)
     N = returns.shift(period - 1)
-    return pd.Series(((M / N) * 100), name = 'ROC_' + str(period))
+    roc = pd.Series(((M / N) * 100), name = 'ROC_' + str(period))
+    return apply_log_if_necessary_series(roc, "roc")

feature_extractors/fractional_differentiation.py

@@ -1,9 +1,15 @@
 from fracdiff.sklearn import FracdiffStat
 import pandas as pd
 import numpy as np
+from feature_extractors.utils import apply_log_if_necessary_series
 
 def feature_fractional_differentiation(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
     frac_diff = FracdiffStat(window = period)
     input_series = df["close"].to_numpy().reshape(-1, 1)
     result = frac_diff.fit_transform(input_series)
-    return pd.Series(result.squeeze(), index = df.index)
+    return pd.Series(result.squeeze(), index = df.index)
+
+def feature_fractional_differentiation_log(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
+    series = feature_fractional_differentiation(df, period, is_log_return)
+    return apply_log_if_necessary_series(series, "fracdiff")
+    

feature_extractors/utils.py

@@ -1,7 +1,25 @@
 import pandas as pd
+import numpy as np
+from scipy.stats import shapiro
 
 def get_close_low_high(df: pd.DataFrame) -> tuple[pd.Series, pd.Series, pd.Series]:
     close = df['close']
     low = df['low']
     high = df['high']
-    return close, low, high
+    return close, low, high
+
+
+def apply_log_if_necessary_series(series: pd.Series, name: str) -> pd.Series:
+    values = series.to_numpy()
+    no_of_unique_values = np.unique(values)
+    if len(no_of_unique_values) < 4:
+        return series
+    is_normal = shapiro(values).pvalue > 0.05
+    if not is_normal:
+        # print("Applying log to column: " + column)
+        min_value = np.min(series)
+        series = (series + min_value).apply(lambda x: np.log(x))
+        is_normal_after_log = shapiro(series).pvalue > 0.05
+        if not is_normal_after_log:
+            print("Failed to normalize column: ", name)
+    return series

feature_selection/feature_selection.py

@@ -2,14 +2,31 @@ from sklearn.feature_selection import RFE
 from sklearn.model_selection import TimeSeriesSplit
 import pandas as pd
 from models.base import Model, SKLearnModel
-from sklearn.decomposition import PCA
+from utils.scaler import get_scaler
+from utils.types import ScalerTypes
+from utils.hashing import hash_df, hash_series
+from diskcache import Cache
+cache = Cache(".cachedir/feature_selection")
 
-def select_features(X: pd.DataFrame, y: pd.Series, model: Model, n_features_to_select: int, backup_model: SKLearnModel) -> pd.DataFrame:
+def select_features(**kwargs):
+    hashed = kwargs['data_config_hash'] + kwargs['model'].get_name() + str(kwargs['n_features_to_select']) + kwargs['backup_model'].get_name() + kwargs['scaling']
+    if hashed in cache:
+        return cache.get(hashed)
+    else:
+        return_value = __select_features(**kwargs)
+        cache[hashed] = return_value
+        return return_value
+
+def __select_features(X: pd.DataFrame, y: pd.Series, model: Model, n_features_to_select: int, backup_model: SKLearnModel, scaling: ScalerTypes, data_config_hash: str) -> pd.DataFrame:
     ''' Select features using RFECV, returns a pd.DataFrame (X) with only the selected features.'''
     if model.model_type != 'ml': return X
 
     # 2. Recursive feature selection
     cv = TimeSeriesSplit(n_splits=5)
+    scaler = get_scaler(scaling)
+    X_scaled = X.copy()
+    if scaler is not None:
+        X_scaled = scaler.fit_transform(X_scaled)
 
     feat_selector_model = model.model
     if hasattr(feat_selector_model, 'feature_importances_') == False and hasattr(feat_selector_model, 'coef_') == False:
@@ -17,7 +34,7 @@ def select_features(X: pd.DataFrame, y: pd.Series, model: Model, n_features_to_s
 
 #     selector = RFECV(feat_selector_model, cv = cv, step=5, min_features_to_select=min_features_to_select)
     selector = RFE(feat_selector_model, n_features_to_select= n_features_to_select)
-    selector = selector.fit(X, y)
-    print("Kept %d features out of %d" % (selector.n_features_, X.shape[1]))
+    selector = selector.fit(X_scaled, y)
+    print("Kept %d features out of %d" % (selector.n_features_, X_scaled.shape[1]))
 
     return pd.DataFrame(X[X.columns[selector.support_]], index= X.index)

models/average.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 from models.base import Model
 import numpy as np
 
@@ -10,16 +11,20 @@ class StaticAverageModel(Model):
     only_column = 'model_'
     feature_selection = 'off'
     model_type = 'static'
+    predict_window_size = 'single_timestamp'
 
-    def fit(self, X, y, prev_model):
+    def fit(self, X: np.ndarray, y: np.ndarray) -> None:
         # This is a static model, it can' learn anything
         pass
 
-    def predict(self, X):
+    def predict(self, X) -> tuple[float, np.ndarray]:
         # Make sure there's data to average
         assert X.shape[1] > 0
         prediction = np.average(X[-1])
-        return np.array([prediction])
+        return (prediction, np.array([]))
 
-    def clone(self):
-        return self
+    def clone(self) -> StaticAverageModel:
+        return self
+
+    def get_name(self) -> str:
+        return 'static_average'

models/base.py

@@ -1,7 +1,8 @@
-
+from __future__ import annotations
 from typing import Literal, Optional
 from sklearn.base import clone
-from abc import ABC, abstractmethod, abstractproperty
+from abc import ABC, abstractmethod
+import numpy as np
 
 class Model(ABC):
 
@@ -10,18 +11,22 @@ class Model(ABC):
     # data_format: Literal["wide", "narrow"]
     only_column: Optional[str]
     model_type: Literal['ml', 'static']
+    predict_window_size: Literal['single_timestamp', 'window_size']
 
     @abstractmethod
-    def fit(self, X, y, prev_model):
-        pass
+    def fit(self, X: np.ndarray, y: np.ndarray) -> None:
+        raise NotImplementedError
 
     @abstractmethod
-    def predict(self, X):
-        pass
+    def predict(self, X) -> tuple[float, np.ndarray]:
+        raise NotImplementedError
 
     @abstractmethod
-    def clone(self):
-        pass
+    def clone(self) -> Model:
+        raise NotImplementedError
+
+    def get_name(self) -> str:
+        raise NotImplementedError
 
 
 class SKLearnModel(Model):
@@ -30,15 +35,21 @@ class SKLearnModel(Model):
     only_column = None
     feature_selection = 'on'
     model_type = 'ml'
+    predict_window_size = 'single_timestamp'
 
     def __init__(self, model):
         self.model = model
 
-    def fit(self, X, y, prev_model):
+    def fit(self, X: np.ndarray, y: np.ndarray) -> None:
         self.model.fit(X, y)
 
-    def predict(self, X):
-        return self.model.predict(X)
+    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):
-        return SKLearnModel(clone(self.model))
+    def clone(self) -> SKLearnModel:
+        return SKLearnModel(clone(self.model))
+
+    def get_name(self) -> str:
+        return self.model.__class__.__name__

models/model_map.py

@@ -11,6 +11,7 @@ from models.base import SKLearnModel
 from models.momentum import StaticMomentumModel
 from models.average import StaticAverageModel
 from models.naive import StaticNaiveModel
+from xgboost import XGBClassifier
 
 
 model_map = {
@@ -34,6 +35,7 @@ model_map = {
         NB= SKLearnModel(GaussianNB()),
         AB= SKLearnModel(AdaBoostClassifier(n_estimators=15)),
         RF= SKLearnModel(RandomForestClassifier(n_jobs=-1, max_depth=20, random_state=1)),
+        XGB= SKLearnModel(XGBClassifier(n_jobs=-1, max_depth = 20, random_state=1, use_label_encoder=True, objective='multi:softprob', eval_metric='mlogloss')),
         StaticMom= StaticMomentumModel(allow_short=True),
         Ensemble_Average = StaticAverageModel(),
     ),

models/momentum.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 from models.base import Model
 import numpy as np
 
@@ -10,19 +11,23 @@ class StaticMomentumModel(Model):
     only_column = 'mom'
     feature_selection = 'off'
     model_type = 'static'
+    predict_window_size = 'single_timestamp'
 
     def __init__(self, allow_short: bool) -> None:
         super().__init__()
         self.allow_short = allow_short
 
-    def fit(self, X, y, prev_model):
+    def fit(self, X: np.ndarray, y: np.ndarray) -> None:
         # This is a static model, it can' learn anything
         pass
 
-    def predict(self, X):
+    def predict(self, X) -> tuple[float, 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 np.array([prediction])
+        return (prediction, np.array([]))
 
-    def clone(self):
-        return self
+    def clone(self) -> StaticMomentumModel:
+        return self
+    
+    def get_name(self) -> str:
+        return 'static_mom'

models/naive.py

@@ -1,3 +1,4 @@
+from __future__ import annotations
 from models.base import Model
 import numpy as np
 
@@ -10,13 +11,17 @@ class StaticNaiveModel(Model):
     only_column = None
     feature_selection = 'off'
     model_type = 'static'
+    predict_window_size = 'single_timestamp'
 
-    def fit(self, X, y, prev_model):
+    def fit(self, X: np.ndarray, y: np.ndarray) -> None:
         # This is a static model, it can' learn anything
         pass
 
-    def predict(self, X):
-        return np.array([X[-1][0]])
+    def predict(self, X) -> tuple[float, np.ndarray]:
+        return (X[-1][0], np.array([]))
 
-    def clone(self):
-        return self
+    def clone(self) -> StaticNaiveModel:
+        return self
+
+    def get_name(self) -> str:
+        return 'static_naive'

run_clean_cache.py

@@ -0,0 +1,5 @@
+from diskcache import Cache
+cache_1 = Cache(".cachedir/feature_selection")
+cache_2 = Cache(".cachedir/data")
+cache_1.clear()
+cache_2.clear()

run_pipeline.py

@@ -1,3 +1,4 @@
+from config.hashing import hash_data_config
 from data_loader.load_data import load_data
 import pandas as pd
 from training.training import run_single_asset_trainig
@@ -26,11 +27,11 @@ def setup_pipeline(project_name:str, with_wandb: bool, sweep: bool):
 def pipeline(project_name:str, wandb, sweep:bool, model_config:dict, training_config:dict, data_config:dict):
     results = pd.DataFrame()
     all_predictions = pd.DataFrame()
+    all_probabilities = pd.DataFrame()
     validate_config(model_config, training_config, data_config)
 
     for asset in data_config['assets']:
         print('--------\nPredicting: ', asset[1])
-        all_predictions = pd.DataFrame()
 
         # 1. Load data
         data_params = data_config.copy()
@@ -53,11 +54,11 @@ def pipeline(project_name:str, wandb, sweep:bool, model_config:dict, training_co
             print("Feature Selection started")
             # TODO: this needs to be done per model!
             backup_model = default_feature_selector_regression if data_config['method'] == 'regression' else default_feature_selector_classification
-            X = select_features(X, y, model_config['level_1_models'][0][1], n_features_to_select = training_config['n_features_to_select'], backup_model = backup_model)
+            X = select_features(X = X, y = y, model = model_config['level_1_models'][0][1], n_features_to_select = training_config['n_features_to_select'], backup_model = backup_model, scaling = training_config['scaler'], data_config_hash = hash_data_config(data_params))
             print("Feature Selection ended")
 
         # 3. Train Level-1 models
-        current_result, current_predictions = run_single_asset_trainig(
+        current_result, current_predictions, current_probabilities = run_single_asset_trainig(
             ticker_to_predict = asset[1],
             original_X = original_X,
             X = X,
@@ -75,14 +76,15 @@ def pipeline(project_name:str, wandb, sweep:bool, model_config:dict, training_co
         results = pd.concat([results, current_result], axis=1)
         # With static models, because of the lag in the indicator, the first prediction is NA, so we fill it with zero.
         all_predictions = pd.concat([all_predictions, current_predictions], axis=1).fillna(0.)
+        all_probabilities = pd.concat([all_probabilities, current_probabilities], axis=1).fillna(0.)
 
         # 3. Train Level-2 (Ensemble) model (Optional)
         if model_config['level_2_model'] is not None: 
-            ensemble_X = all_predictions
+            ensemble_X = pd.concat([all_predictions, all_probabilities], axis = 1)
             if training_config['include_original_data_in_ensemble']:
                 ensemble_X = pd.concat([ensemble_X, X], axis=1)
 
-            ensemble_result, ensemble_preds = run_single_asset_trainig(
+            ensemble_result, ensemble_preds, ensemble_probabilities = run_single_asset_trainig(
                 ticker_to_predict = asset[1],
                 original_X = ensemble_X,
                 X = ensemble_X,
@@ -100,6 +102,7 @@ def pipeline(project_name:str, wandb, sweep:bool, model_config:dict, training_co
 
             results = pd.concat([results, ensemble_result], axis=1)
             all_predictions = pd.concat([all_predictions, ensemble_preds], axis=1)
+            all_probabilities = pd.concat([all_probabilities, ensemble_probabilities], axis=1).fillna(0.)
         
 
     results.to_csv('results.csv')

sweep_exo_data.yaml

@@ -1,7 +1,7 @@
 program: run_sweep.py
 method: grid
 project: price-forecasting
-name: Fractional differentiation / number of features
+name: Exogenous data / data transformation
 metric:
   goal: maximize
   name: sharpe
@@ -24,8 +24,7 @@ parameters:
     feature_selection:
       value: True
     n_features_to_select:
-      values: [30, 40]
-      distribution: categorical
+      value: 30
     dimensionality_reduction:
       value: True
     retrain_every: 
@@ -57,5 +56,5 @@ parameters:
       values: [['level_2', 'lags_up_to_5'], ['level_2', 'fracdiff']]
       distribution: categorical
     exogenous_features:
-      values: [[], ['fracdiff']]
+      values: [[], ['fracdiff'], ['standard_scaling']]
       distribution: categorical

tests/test_evaluation.py

@@ -42,13 +42,13 @@ class EvenOddStubModel(Model):
         super().__init__()
         self.window_length = window_length
 
-    def fit(self, X, y, prev_model):
+    def fit(self, X, y):
         assert len(X) == self.window_length
         for i in range(len(X)):
             assert y[i] == -1 if X[i][0] == 1 else 1
 
     def predict(self, X):
-        return np.array([-1 if X[0][0] == 1 else 1])
+        return (-1 if X[0][0] == 1 else 1, np.array([]))
 
     def clone(self):
         return self
@@ -62,7 +62,7 @@ def test_evaluation():
     model = EvenOddStubModel(window_length = window_length)
     scaler = None
     
-    models, predictions = walk_forward_train_test(
+    models, predictions, probs = walk_forward_train_test(
         model_name='test',
         model=model,
         X=X,

tests/test_walk_forward.py

@@ -40,13 +40,13 @@ class IncrementingStubModel(Model):
         super().__init__()
         self.window_length = window_length
 
-    def fit(self, X, y, prev_model):
+    def fit(self, X, y):
         assert len(X) == self.window_length
         for i in range(len(X)):
             assert X[i][0] + 1 == y[i]
 
     def predict(self, X):
-        return np.array([X[0][0] + 1])
+        return (X[0][0] + 1, np.array([]))
 
     def clone(self):
         return self
@@ -59,7 +59,7 @@ def test_walk_forward_train_test():
     model = IncrementingStubModel(window_length = window_length)
     scaler = None
 
-    models, predictions = walk_forward_train_test(
+    models, predictions, probs = walk_forward_train_test(
         model_name='test',
         model=model,
         X=X,

training/training.py

@@ -1,19 +1,10 @@
 import pandas as pd
 from typing import Literal
 from training.walk_forward import walk_forward_train_test
-from sklearn.preprocessing import MinMaxScaler, Normalizer, StandardScaler
 from utils.evaluate import evaluate_predictions
 from models.base import Model
-
-def __get_scaler(type: Literal['normalize', 'minmax', 'standardize', 'none']):
-    if type == 'normalize':
-        return Normalizer()
-    elif type == 'minmax':
-        return MinMaxScaler(feature_range= (-1, 1))
-    elif type == 'standardize':
-        return StandardScaler()
-    else:
-        return None
+from utils.scaler import get_scaler
+from utils.types import ScalerTypes
 
 def run_single_asset_trainig(
                     ticker_to_predict: str,
@@ -26,19 +17,20 @@ def run_single_asset_trainig(
                     expanding_window: bool,
                     sliding_window_size: int,
                     retrain_every: int,
-                    scaler: Literal['normalize', 'minmax', 'standardize', 'none'],
+                    scaler: ScalerTypes,
                     no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'],
                     level: int
-    ) -> tuple[pd.DataFrame, pd.DataFrame]:
+    ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
 
 
-    scaler = __get_scaler(scaler)
+    scaler = get_scaler(scaler)
 
     results = pd.DataFrame()
-    predictions = pd.DataFrame()
+    predictions = pd.DataFrame(index=y.index)
+    probabilities = pd.DataFrame(index=y.index)
     
     for model_name, model in models:
-        model_over_time, preds = walk_forward_train_test(
+        model_over_time, preds, probs = walk_forward_train_test(
             model_name=model_name,
             model = model,
             X = X if model.feature_selection == 'on' else original_X,
@@ -58,10 +50,13 @@ def run_single_asset_trainig(
             method = method,
             no_of_classes=no_of_classes
         )
-        column_name = ticker_to_predict + "_" + model_name + "_lvl" + str(level)
+        column_name = "model_" + ticker_to_predict + "_" + model_name + "_lvl" + str(level)
         results[column_name] = result
         # column names for model outputs should be different, so we can differentiate between original data and model predictions later, where necessary
-        predictions["model_" + column_name] = preds
+        predictions[column_name] = preds
+        probs_column_name = "probs_" + ticker_to_predict + "_" + model_name + "_lvl" + str(level)
+        probs.columns = [probs_column_name + "_" + c for c in probs.columns]
+        probabilities = pd.concat([probabilities, probs], axis=1)
         
 
-    return results, predictions
+    return results, predictions, probabilities

training/walk_forward.py

@@ -16,9 +16,10 @@ def walk_forward_train_test(
                             window_size: int,
                             retrain_every: int,
                             scaler,
-                        ) -> tuple[pd.Series, pd.Series]:
+                        ) -> tuple[pd.Series, pd.Series, pd.DataFrame]:
     assert len(X) == len(y)
     predictions = pd.Series(index=y.index).rename(model_name)
+    probabilities = pd.DataFrame(index=y.index)
     models = pd.Series(index=y.index).rename(model_name)
 
     first_nonzero_return = max(get_first_valid_return_index(target_returns), get_first_valid_return_index(X.iloc[:,0]))
@@ -45,8 +46,8 @@ def walk_forward_train_test(
             train_window_end = index - 1
             
             if is_scaling_on:
-                # First we need to fit on the expanding window data slice
-                # This is our only way to avoid lookahead bia
+                # We need to fit on the expanding window data slice
+                # This is our only way to avoid lookahead bias
                 X_expanding_window = X[first_nonzero_return:train_window_end]
                 scaler.fit(X_expanding_window.values)
 
@@ -59,7 +60,7 @@ def walk_forward_train_test(
                 X_slice = X_slice.to_numpy()
 
             current_model = model.clone()
-            current_model.fit(X_slice, y_slice.to_numpy(), models[index-1])
+            current_model.fit(X_slice, y_slice.to_numpy())
             iterations_before_retrain = retrain_every
         else:
             current_model = models[index-1]
@@ -70,8 +71,12 @@ def walk_forward_train_test(
         if is_scaling_on:
             next_timestep = scaler.transform(next_timestep)
 
-        prediction = current_model.predict(next_timestep).item()
+        prediction, probs = current_model.predict(next_timestep)
         predictions[index] = prediction
+        if len(probabilities.columns) != len(probs):
+            probabilities = probabilities.reindex(columns = ["prob_" + str(num) for num in range(0, len(probs.T))])
+        probabilities.iloc[index] = probs
+
         iterations_before_retrain -= 1
 
-    return models, predictions
+    return models, predictions, probabilities

utils/hashing.py

@@ -0,0 +1,10 @@
+from hashlib import sha256
+import pandas as pd
+
+def hash_df(df: pd.DataFrame) -> str:
+    s = str(df.columns) + str(df.index) + str(df.values)
+    return sha256(s.encode()).hexdigest()
+
+def hash_series(df: pd.Series) -> str:
+    s = str(df.name) + str(df.index) + str(df.values)
+    return sha256(s.encode()).hexdigest()

utils/normalize.py

@@ -1,6 +0,0 @@
-import pandas as pd
-
-def normalize(data: pd.DataFrame) -> pd.DataFrame:
-    data_mean = data.mean(axis=0)
-    data_std = data.std(axis=0)
-    return ((data - data_mean) / data_std).fillna(0.)

utils/scaler.py

@@ -0,0 +1,13 @@
+from sklearn.preprocessing import MinMaxScaler, Normalizer, StandardScaler
+from typing import Optional, Union
+from utils.types import ScalerTypes
+
+def get_scaler(type: ScalerTypes) -> Optional[Union[MinMaxScaler, Normalizer, StandardScaler]]:
+    if type == 'normalize':
+        return Normalizer()
+    elif type == 'minmax':
+        return MinMaxScaler(feature_range= (-1, 1))
+    elif type == 'standardize':
+        return StandardScaler()
+    else:
+        return None

utils/types.py

@@ -1,4 +1,4 @@
-from typing import Callable, Union
+from typing import Callable, Union, Literal
 import pandas as pd
 
 Period = int
@@ -9,4 +9,6 @@ FeatureExtractorConfig = tuple[Name, FeatureExtractor, list[Period]]
 Path = str
 FileName = str
 DataSource = list[tuple[Path, FileName]]
-DataCollection = list[DataSource]
+DataCollection = list[DataSource]
+
+ScalerTypes = Literal['normalize', 'minmax', 'standardize', 'none']