feat(Data): resample exogenous/other datasource when their frequency is different (#234)

* feat(Data): resample exogenous/other datasource when their frequency is different

* fix(Linter): ran

* fix(Data): resampling done properly

config/hashing.py

@@ -3,7 +3,7 @@ from data_loader.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])
+        return "".join([a.path + a.file_name for a in data_collection])
 
     def hash_feature_extractors(feature_extractos) -> str:
         return "".join([f[0] for f in feature_extractos])
@@ -17,7 +17,8 @@ def hash_data_config(data_config: dict) -> 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["target_asset"].path
+                + data_config["target_asset"].file_name,
                 data_config["load_non_target_asset"],
                 hash_feature_extractors(data_config["own_features"]),
                 hash_feature_extractors(data_config["other_features"]),

config/preprocess.py

@@ -103,7 +103,7 @@ def __preprocess_data_collections_config(data_dict: dict) -> dict:
             [
                 asset
                 for asset in data_dict["assets"]
-                if asset[1] == data_dict["target_asset"]
+                if asset.file_name == data_dict["target_asset"]
             ]
         ),
         None,

config/presets.py

@@ -22,11 +22,11 @@ def get_default_config() -> RawConfig:
         assets=["fivemin_crypto"],
         target_asset="BTCUSDT",
         other_assets=[],
-        exogenous_data=[],
+        exogenous_data=["daily_glassnode"],
         load_non_target_asset=True,
         own_features=["level_2"],
         other_features=["z_score"],
-        exogenous_features=[],
+        exogenous_features=["z_score"],
         directional_models=classification_models,
         meta_models=meta_models,
         event_filter="cusum_vol",

data_loader/collections.py

@@ -1,9 +1,11 @@
-from .types import Path, FileName, DataSource, DataCollection
-from utils.helpers import flatten
+from typing import Literal
+from .types import DataSource, DataCollection
 
 
-def transform_to_data_collection(path: str, file_names: list[str]) -> DataCollection:
-    return list(zip([path] * len(file_names), file_names))
+def transform_to_data_collection(
+    path: str, file_names: list[str], freq: Literal["5m", "1h", "1d"]
+) -> DataCollection:
+    return [DataSource(path, file_name, freq) for file_name in file_names]
 
 
 __daily_etf = ["GLD", "IEF", "QQQ", "SPY", "TLT"]
@@ -52,9 +54,11 @@ __daily_glassnode = [
 
 
 data_collections = dict(
-    daily_etf=transform_to_data_collection("data/daily_etf", __daily_etf),
-    fivemin_crypto=transform_to_data_collection("data/5min_crypto", __5min_crypto),
+    daily_etf=transform_to_data_collection("data/daily_etf", __daily_etf, "1d"),
+    fivemin_crypto=transform_to_data_collection(
+        "data/5min_crypto", __5min_crypto, "5m"
+    ),
     daily_glassnode=transform_to_data_collection(
-        "data/daily_glassnode", __daily_glassnode
+        "data/daily_glassnode", __daily_glassnode, "1d"
     ),
 )

data_loader/load.py

@@ -1,16 +1,14 @@
-from re import S
-from shutil import ExecError
+import os
 import pandas as pd
 import numpy as np
 from .types import DataSource
 from feature_extractors.types import FeatureExtractor
 from utils.helpers import drop_columns_if_exist
 from data_loader.collections import DataCollection
-from typing import Literal
-
-import os
+from typing import Literal, Optional
+from utils.resample import resample_ohlc
 from config.hashing import hash_data_config
-from .types import XDataFrame, ReturnSeries, ForwardReturnSeries
+from .types import XDataFrame, ReturnSeries
 from diskcache import Cache
 
 cache = Cache(".cachedir/data")
@@ -44,38 +42,43 @@ def __load_data(
         - Series `forward_returns` with the target asset returns shifted by 1 day
     """
 
-    target_file = [f for f in assets if f[1].startswith(target_asset[1])]
+    target_file = [f for f in assets if f.file_name.startswith(target_asset.file_name)]
     assert len(target_file) == 1, "There should be exactly one target file"
+    target_freq = target_file[0].freq
     other_files = [
         f
         for f in assets
-        if load_non_target_asset == True and f[1].startswith(target_asset[1]) == False
+        if load_non_target_asset == True
+        and f.file_name.startswith(target_asset.file_name) == False
     ]
     files = other_files + other_assets
 
     target_asset_df = [
         __load_df(
             data_source=data_source,
-            prefix=data_source[1],
+            prefix=data_source.file_name,
             returns="log_returns",
             feature_extractors=own_features,
+            resample_to_freq=None,
         )
         for data_source in target_file
     ]
 
     df_target_asset_only_returns = __load_df(
         data_source=target_file[0],
-        prefix=target_file[0][1],
+        prefix=target_file[0].file_name,
         returns="returns",
         feature_extractors=[],
+        resample_to_freq=None,
     )
 
     asset_dfs = [
         __load_df(
             data_source=data_source,
-            prefix=data_source[1],
+            prefix=data_source.file_name,
             returns="log_returns",
             feature_extractors=other_features,
+            resample_to_freq=target_freq,
         )
         for data_source in files
     ]
@@ -83,9 +86,10 @@ def __load_data(
     exogenous_dfs = [
         __load_df(
             data_source=data_source,
-            prefix=data_source[1],
+            prefix=data_source.file_name,
             returns="none",
             feature_extractors=exogenous_features,
+            resample_to_freq=target_freq,
         )
         for data_source in exogenous_data
     ]
@@ -97,7 +101,7 @@ def __load_data(
     X.sort_index(inplace=True)
 
     ## Create target
-    returns = df_target_asset_only_returns[target_asset[1] + "_returns"]
+    returns = df_target_asset_only_returns[target_asset.file_name + "_returns"]
     returns.index = pd.DatetimeIndex(X.index)
 
     return X, returns
@@ -108,15 +112,16 @@ def __load_df(
     prefix: str,
     returns: Literal["none", "price", "returns", "log_returns"],
     feature_extractors: list[tuple[str, FeatureExtractor, list[int]]],
+    resample_to_freq: Optional[Literal["5m", "1h", "1d"]] = None,
 ) -> pd.DataFrame:
-    csv_file = os.path.join(data_source[0], data_source[1] + ".csv")
-    parquet_file = os.path.join(data_source[0], data_source[1] + ".parquet")
+    csv_file = os.path.join(data_source.path, data_source.file_name + ".csv")
+    parquet_file = os.path.join(data_source.path, data_source.file_name + ".parquet")
     if os.path.isfile(csv_file):
         df = pd.read_csv(csv_file, header=0, index_col=0).fillna(0)
     elif os.path.isfile(parquet_file):
         df = pd.read_parquet(parquet_file).fillna(0)
     else:
-        raise Exception("File not found: " + data_source[0] + data_source[1])
+        raise Exception("File not found: " + data_source.path + data_source.file_name)
 
     if returns == "log_returns":
         df["returns"] = np.log(df["close"]).diff(1)
@@ -128,6 +133,10 @@ def __load_df(
     df = __apply_feature_extractors(df, feature_extractors=feature_extractors)
 
     df = df.replace([np.inf, -np.inf], 0.0)
+    df.index = pd.DatetimeIndex(df.index)
+    if resample_to_freq is not None and resample_to_freq != data_source.freq:
+        df = resample_ohlc(df, resample_to_freq)
+
     df = drop_columns_if_exist(df, ["open", "high", "low", "close", "volume"])
 
     df.columns = [prefix + "_" + c if "date" not in c else c for c in df.columns]
@@ -159,7 +168,7 @@ def load_only_returns(
     assets_future = [
         __load_df(
             data_source=data_source,
-            prefix=data_source[1],
+            prefix=data_source.file_name,
             returns=returns,
             feature_extractors=[],
         )

data_loader/types.py

@@ -1,8 +1,18 @@
+from dataclasses import dataclass
+from typing import Literal
 import pandas as pd
 
 Path = str
 FileName = str
-DataSource = tuple[Path, FileName]
+
+
+@dataclass
+class DataSource:
+    path: Path
+    file_name: FileName
+    freq: Literal["5m", "1h", "1d"]
+
+
 DataCollection = list[DataSource]
 
 ReturnSeries = pd.Series

reporting/reporting.py

@@ -18,7 +18,9 @@ def report_results(
     send_report_to_wandb(output_stats, wandb)
     pd.Series(output_stats).to_csv("output/results.csv")
 
-    output_weights.rename(config.target_asset[1]).to_csv("output/predictions.csv")
+    output_weights.rename(config.target_asset.file_name).to_csv(
+        "output/predictions.csv"
+    )
 
     print("\n--------\n")
 

training/bet_sizing.py

@@ -96,7 +96,7 @@ def bet_sizing_with_meta_model(
         pp.pprint(stats)
     else:
         stats = None
-    model_id = "model_" + config.target_asset[1] + "_meta"
+    model_id = "model_" + config.target_asset.file_name + "_meta"
 
     outcome_dict = vars(meta_outcome)
     outcome_dict["model_id"] = model_id

training/directional_training.py

@@ -39,7 +39,7 @@ def train_directional_model(
         transformations_over_time = preloaded_training_step.transformations
 
     training_outcome = train_model(
-        ticker_to_predict=config.target_asset[1],
+        ticker_to_predict=config.target_asset.file_name,
         X=X,
         y=y,
         forward_returns=forward_returns,

utils/resample.py

@@ -3,8 +3,14 @@ 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()
+    period = period.replace("m", "T")
+
+    if "open" in df.columns:
+        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()
+    else:
+        output = df.resample(period).ffill()
+
     return output