feat(Data): added feature extractors, and feature extractor presets, removed a bunch of custom arguments from load_data (#42)

feature_extractors/feature_extractor_presets.py

@@ -0,0 +1,22 @@
+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
+
+lags = [('lag', feature_lag, [1,2,3,4,5,6,7,8,9])]
+
+only_mom = [('mom', feature_mom, [30])]
+
+date = [
+    ('day_of_week', feature_day_of_week, [0]),
+    ('day_of_month', feature_day_of_month, [0]),
+    ('month', feature_month, [0])]
+
+level1 = [
+    ('mom', feature_mom, [10, 20, 30, 60, 90]),
+    ('vol', feature_vol, [10, 20, 30, 60]),
+]
+
+level2 = level1 + [
+    ('roc', feature_ROC, [10, 30]),
+    ('rsi', feature_RSI, [10, 30, 100]),
+    ('stod', feature_STOD, [10, 30, 200]),
+    ('stok', feature_STOK, [10, 30, 200]),
+]

feature_extractors/feature_extractors.py

@@ -0,0 +1,62 @@
+import pandas as pd
+import numpy as np
+
+## Utility functions
+
+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
+
+## Feature extractors
+
+def feature_lag(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
+    return df['returns'].shift(period)
+
+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)
+
+def feature_day_of_month(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.DataFrame:
+    return pd.get_dummies(pd.DatetimeIndex(df.index).day, drop_first=True, prefix="date_day_month").set_index(df.index)
+
+def feature_month(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.DataFrame:
+    return pd.get_dummies(pd.DatetimeIndex(df.index).month, drop_first=True, prefix="date_month").set_index(df.index)
+
+def feature_vol(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
+    return df['returns'].rolling(period).std() * (252**0.5)
+
+def feature_mom(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
+    if is_log_return:
+        return np.log(df['close']).diff(period)
+    else:
+        return df['close'].pct_change(period)
+
+def feature_STOK(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
+    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
+
+def feature_STOD(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
+    stok = feature_STOK(df, period, is_log_return)
+    return stok.rolling(3).mean()
+
+def feature_RSI(df: pd.DataFrame, period: int, is_log_return: bool) -> pd.Series:
+    returns = df['returns']
+    delta = returns.diff().dropna()
+    u=delta*0
+    d = u.copy()
+    u[delta > 0] = delta[delta > 0]
+    d[delta < 0] = -delta[delta < 0]
+    u[u.index[period-1]] = np.mean( u[:period] ) #first value is sum of avg gains u = u.drop(u.index[:(period-1)])
+    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)
+
+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))

run_pipeline.py

@@ -14,6 +14,7 @@ from sklearn.naive_bayes import GaussianNB
 from sklearn.neural_network import MLPRegressor, MLPClassifier
 from sklearn.ensemble import AdaBoostRegressor, RandomForestRegressor, ExtraTreesRegressor, AdaBoostClassifier, GradientBoostingClassifier, RandomForestClassifier, ExtraTreesClassifier
 
+import feature_extractors.feature_extractor_presets as feature_extractor_presets
 from training.pipeline import run_single_asset_trainig_pipeline
 
 
@@ -51,15 +52,13 @@ retrain_every = 20
 scaler = 'minmax' # 'normalize' 'minmax' 'standardize' 'none'
 include_original_data_in_ensemble = True
 method = 'classification'
+forecasting_horizon = 1
 data_parameters = dict(path=path,
-    target_asset_lags= [1,2,3,4,5,6,8,10,15],
     load_other_assets= False,
-    other_asset_lags= [],
     log_returns= True,
-    add_date_features= False,
-    own_technical_features= 'level2',
-    other_technical_features= 'none',
-    exogenous_features= 'none',
+    forecasting_horizon = forecasting_horizon,
+    own_features= feature_extractor_presets.date + feature_extractor_presets.level1,
+    other_features= [],
     index_column= 'int',
     method= method,
 )

utils/load_data.py

@@ -4,7 +4,8 @@ import os
 import numpy as np
 from pandas.core.frame import DataFrame
 from utils.technical_indicators import ROC, RSI, STOK, STOD
-from typing import Literal
+from utils.typing import FeatureExtractor
+from typing import Callable, Literal
 from sklearn.preprocessing import OneHotEncoder
 
 #%%
@@ -18,14 +19,11 @@ def get_etf_assets(path: str) -> list[str]:
 
 def load_data(path: str,
             target_asset: str,
-            target_asset_lags: list[int],
             load_other_assets: bool,
-            other_asset_lags: list[int],
             log_returns: bool,
-            add_date_features: bool,
-            own_technical_features: Literal['none', 'level1', 'level2'],
-            other_technical_features: Literal['none', 'level1', 'level2'],
-            exogenous_features: Literal['none', 'level1'],
+            forecasting_horizon: int,
+            own_features: list[tuple[str, FeatureExtractor, list[int]]],
+            other_features: list[tuple[str, FeatureExtractor, list[int]]],
             index_column: Literal['date', 'int'],
             method: Literal['regression', 'classification'],
             narrow_format: bool = False,
@@ -45,8 +43,7 @@ def load_data(path: str,
         path=os.path.join(path,f),
         prefix=f.split('.')[0],
         returns='log_returns' if log_returns else 'returns',
-        technical_features=own_technical_features if is_target_asset(target_asset, f) else other_technical_features,
-        lags= target_asset_lags if is_target_asset(target_asset, f) else other_asset_lags,
+        feature_extractors=own_features if is_target_asset(target_asset, f) else other_features,
         narrow_format=narrow_format,
     ) for f in files]
     if narrow_format:
@@ -56,11 +53,6 @@ def load_data(path: str,
 
     dfs.index = pd.DatetimeIndex(dfs.index)
 
-    if add_date_features:
-        dfs = pd.concat([dfs, pd.get_dummies(dfs.index.day, drop_first=True, prefix="day_month").set_index(dfs.index)], axis=1)
-        dfs = pd.concat([dfs, pd.get_dummies(dfs.index.dayofweek, drop_first=True, prefix="day_week").set_index(dfs.index)] , axis=1)
-        dfs = pd.concat([dfs, pd.get_dummies(dfs.index.month, drop_first=True, prefix="month").set_index(dfs.index)], axis = 1)
-
     if index_column == 'int':
         dfs.reset_index(drop=True, inplace=True)
 
@@ -70,15 +62,14 @@ def load_data(path: str,
     ## Create target 
     target_col = 'target'
     returns_col = target_asset + '_returns'
-    prediction_horizon = 1
-    forward_returns = __create_target_cum_forward_returns(dfs, returns_col, prediction_horizon)
+    forward_returns = __create_target_cum_forward_returns(dfs, returns_col, forecasting_horizon)
     if method == 'regression':
         dfs[target_col] = forward_returns
     elif method == 'classification':
-        dfs[target_col] = __create_target_classes(dfs, returns_col, prediction_horizon, 'two')
+        dfs[target_col] = __create_target_classes(dfs, returns_col, forecasting_horizon, 'two')
     # we need to drop the last row, because we forward-shift the target (see what happens if you call .shift[-1] on a pd.Series) 
-    dfs = dfs.iloc[:-prediction_horizon]
-    forward_returns = forward_returns.iloc[:-prediction_horizon]
+    dfs = dfs.iloc[:-forecasting_horizon]
+    forward_returns = forward_returns.iloc[:-forecasting_horizon]
     
     X = dfs.drop(columns=[target_col])
     y = dfs[target_col]
@@ -93,8 +84,7 @@ def load_crypto_only_returns(path: str, index_column: Literal['date', 'int'], re
         path=os.path.join(path,f),
         prefix=f.split('.')[0],
         returns=returns,
-        technical_features='none',
-        lags=[],
+        feature_extractors=[],
         narrow_format=False,
     ) for f in files]
     dfs = pd.concat(dfs, axis=1)
@@ -110,7 +100,11 @@ def load_crypto_assets_availability(path: str, index_column: Literal['date', 'in
     return load_crypto_only_returns(path, index_column, 'returns').applymap(lambda x: 0 if x == 0.0 or x == 0 or np.isnan(x) else 1)
 
 
-def __load_df(path: str, prefix: str, returns: Literal['price', 'returns', 'log_returns'], technical_features: Literal['none', 'level1', 'level2'], lags: list[int], narrow_format: bool = False) -> pd.DataFrame:
+def __load_df(path: str,
+            prefix: str,
+            returns: Literal['price', 'returns', 'log_returns'],
+            feature_extractors: list[tuple[str, FeatureExtractor, list[int]]],
+            narrow_format: bool = False) -> pd.DataFrame:
     df = pd.read_csv(path, header=0, index_col=0).fillna(0)
 
     if returns == 'log_returns':
@@ -119,11 +113,8 @@ def __load_df(path: str, prefix: str, returns: Literal['price', 'returns', 'log_
         df['returns'] = df['close']
     else:
         df['returns'] = df['close'].pct_change()
-    
-    for lag in lags:
-        df[f'lag_{lag}'] = df['returns'].shift(lag)
 
-    df = __augment_derived_features(df, log_returns=True if returns == 'log_returns' else False, technical_features=technical_features)
+    df = __apply_feature_extractors(df, log_returns=True if returns == 'log_returns' else False, feature_extractors = feature_extractors)
 
     df = df.replace([np.inf, -np.inf], 0.)
     df = df.drop(columns=['open', 'high', 'low', 'close'])
@@ -134,49 +125,26 @@ def __load_df(path: str, prefix: str, returns: Literal['price', 'returns', 'log_
     if narrow_format:
         df["ticker"] = np.repeat(prefix, df.shape[0])
     else: 
-        df.columns = [prefix + "_" + c for c in df.columns]
+        df.columns = [prefix + "_" + c if 'date' not in c else c for c in df.columns]
     return df
 
-def __augment_derived_features(df: pd.DataFrame, log_returns: bool, technical_features: Literal['none', 'level1', 'level2']) -> pd.DataFrame:
-    if technical_features == 'level1' or technical_features == 'level2':
-        # volatility (10, 20, 30, 60 days)
-        df['vol_10'] = df['returns'].rolling(10).std()*(252**0.5)
-        df['vol_20'] = df['returns'].rolling(20).std()*(252**0.5)
-        df['vol_30'] = df['returns'].rolling(30).std()*(252**0.5)
-        df['vol_60'] = df['returns'].rolling(60).std()*(252**0.5)
 
-        # momentum (10, 20, 30, 60, 90 days)
-        if log_returns:
-            df['mom_10'] = np.log(df['close']).diff(10)
-            df['mom_20'] = np.log(df['close']).diff(20)
-            df['mom_30'] = np.log(df['close']).diff(30)
-            df['mom_60'] = np.log(df['close']).diff(60)
-            df['mom_90'] = np.log(df['close']).diff(90)
-        else:
-            df['mom_10'] = df['close'].pct_change(10)
-            df['mom_20'] = df['close'].pct_change(20)
-            df['mom_30'] = df['close'].pct_change(30)
-            df['mom_60'] = df['close'].pct_change(60)
-            df['mom_90'] = df['close'].pct_change(90)
+def __apply_feature_extractors(df: pd.DataFrame,
+                            log_returns: bool,
+                            feature_extractors: list[tuple[str, FeatureExtractor, list[int]]]) -> pd.DataFrame:
 
-    if technical_features == 'level2':
-        df['roc_10'] = ROC(df['close'], 10)
-        df['roc_30'] = ROC(df['close'], 30)
-
-        df['rsi_10'] = RSI(df['close'], 10)
-        df['rsi_30'] = RSI(df['close'], 30)
-        df['rsi_100'] = RSI(df['close'], 30)
-
-        df['stok_10'] = STOK(df['close'], df['low'], df['high'], 10)
-        df['stod_10'] = STOD(df['close'], df['low'], df['high'], 10)
-        df['stok_30'] = STOK(df['close'], df['low'], df['high'], 30)
-        df['stod_30'] = STOD(df['close'], df['low'], df['high'], 30)
-        df['stok_200'] = STOK(df['close'], df['low'], df['high'], 200)
-        df['stod_200'] = STOD(df['close'], df['low'], df['high'], 200)
+    for name, extractor, periods in feature_extractors:
+        for period in periods:
+            features = extractor(df, period, log_returns)
+            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)
+            else:
+                assert False, "Feature extractor must return a pd.DataFrame or pd.Series"
     return df
 
 
-
 # %%
 def __create_target_cum_forward_returns(df: pd.DataFrame, source_column: str, period: int) -> pd.Series:
     assert period > 0

utils/typing.py

@@ -1,7 +1,13 @@
-from typing import Protocol
+from typing import Protocol, Callable, Union
+import pandas as pd
 
 class SKLearnModel(Protocol):
     def fit(self, X, y, sample_weight=None): ...
     def predict(self, X): ...
     def score(self, X, y, sample_weight=None): ...
-    def set_params(self, **params): ...
+    def set_params(self, **params): ...
+
+
+Period = int
+IsLogReturn = bool
+FeatureExtractor = Callable[[pd.DataFrame, Period, IsLogReturn], Union[pd.DataFrame, pd.Series]]