feat(Data): added script to download data from binance (#224)

* feat(Data): added script to download data from binance

* feat(Data): saving unified parquet file/loading

* fix(Config): tweak the cusum filter's threshold

* fix(Dependencies): added binance_historical_data

config/presets.py

@@ -20,7 +20,7 @@ def get_default_config() -> RawConfig:
         retrain_every=1000,
         scaler="minmax",  # 'normalize' 'minmax' 'standardize' 'robust'
         assets=["fivemin_crypto"],
-        target_asset="BTC_USD",
+        target_asset="BTCUSDT",
         other_assets=[],
         exogenous_data=[],
         load_non_target_asset=True,

data_loader/collections.py

@@ -9,13 +9,17 @@ def transform_to_data_collection(path: str, file_names: list[str]) -> DataCollec
 __daily_etf = ["GLD", "IEF", "QQQ", "SPY", "TLT"]
 
 __5min_crypto = [
-    "BTC_USD",
+    "TRXUSDT",
-    "ETC_USD",
+    "XRPUSDT",
-    "ETH_USD",
+    "ADAUSDT",
-    "LTC_USD",
+    "SOLUSDT",
-    "TRX_USD",
+    "AVAXUSDT",
-    "XLM_USD",
+    "DOTUSDT",
-    "XMR_USD",
+    "ETHUSDT",
+    "LTCUSDT",
+    "BNBUSDT",
+    "BTCUSDT",
+    "ETCUSDT",
 ]
 
 __daily_glassnode = [

data_loader/load.py

@@ -1,3 +1,5 @@
+from re import S
+from shutil import ExecError
 import pandas as pd
 import numpy as np
 from .types import DataSource
@@ -111,9 +113,14 @@ def __load_df(
     returns: Literal["none", "price", "returns", "log_returns"],
     feature_extractors: list[tuple[str, FeatureExtractor, list[int]]],
 ) -> pd.DataFrame:
-    df = pd.read_csv(
+    csv_file = os.path.join(data_source[0], data_source[1] + ".csv")
-        os.path.join(data_source[0], data_source[1] + ".csv"), header=0, index_col=0
+    parquet_file = os.path.join(data_source[0], data_source[1] + ".parquet")
-    ).fillna(0)
+    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])
 
     if returns == "log_returns":
         df["returns"] = np.log(df["close"]).diff(1)

environment.yml

@@ -35,4 +35,5 @@ dependencies:
     - vectorbt
     - pydantic
     - pandera[mypy]
+    - binance_historical_data
 prefix: /usr/local/anaconda3/envs/quant

feature_extractors/feature_extractor_presets.py

@@ -15,7 +15,6 @@ from .feature_extractors import (
 )
 from .fractional_differentiation import (
     feature_fractional_differentiation,
-    feature_fractional_differentiation_log,
 )
 
 __presets = dict(
@@ -40,7 +39,6 @@ __presets = dict(
     stod=[("stod", feature_STOD, [100, 300, 2000])],
     stok=[("stok", feature_STOK, [100, 300, 2000])],
     fracdiff=[("fracdiff", feature_fractional_differentiation, [100, 300])],
-    fracdiff_log=[("fracdiff_log", feature_fractional_differentiation_log, [100, 300])],
     z_score=[("z_score", feature_expanding_zscore, [100])],
 )
 

feature_extractors/feature_extractors.py

@@ -1,7 +1,6 @@
 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
 
 
 def feature_debug_future_lookahead(df: pd.DataFrame, period: int) -> pd.Series:
@@ -51,7 +50,7 @@ def feature_STOK(df: pd.DataFrame, period: int) -> pd.Series:
         (close - low.rolling(period).min())
         / (high.rolling(period).max() - low.rolling(period).min())
     ) * 100
-    return apply_log_if_necessary_series(STOK, "stok")
+    return STOK
 
 
 def feature_STOD(df: pd.DataFrame, period: int) -> pd.Series:
@@ -76,7 +75,7 @@ def feature_RSI(df: pd.DataFrame, period: int) -> pd.Series:
         u.ewm(com=period - 1, adjust=False).mean()
         / d.ewm(com=period - 1, adjust=False).mean()
     )
-    return apply_log_if_necessary_series(100 - 100 / (1 + rs), "rsi")
+    return 100 - 100 / (1 + rs)
 
 
 def feature_ROC(df: pd.DataFrame, period: int) -> pd.Series:
@@ -84,4 +83,4 @@ def feature_ROC(df: pd.DataFrame, period: int) -> pd.Series:
     M = returns.diff(period - 1)
     N = returns.shift(period - 1)
     roc = pd.Series(((M / N) * 100), name="ROC_" + str(period))
-    return apply_log_if_necessary_series(roc, "roc")
+    return roc

feature_extractors/fractional_differentiation.py

@@ -1,7 +1,6 @@
 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) -> pd.Series:
@@ -9,8 +8,3 @@ def feature_fractional_differentiation(df: pd.DataFrame, period: int) -> pd.Seri
     input_series = df["close"].to_numpy().reshape(-1, 1)
     result = frac_diff.fit_transform(input_series)
     return pd.Series(result.squeeze(), index=df.index)
-
-
-def feature_fractional_differentiation_log(df: pd.DataFrame, period: int) -> pd.Series:
-    series = feature_fractional_differentiation(df, period, is_log_return)
-    return apply_log_if_necessary_series(series, "fracdiff")

feature_extractors/utils.py

@@ -8,19 +8,3 @@ def get_close_low_high(df: pd.DataFrame) -> tuple[pd.Series, pd.Series, pd.Serie
     low = df["low"]
     high = df["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

labeling/eventfilters_map.py

@@ -4,5 +4,5 @@ from .event_filters.cusum import CUSUMVolatilityEventFilter, CUSUMFixedEventFilt
 eventfilters_map = dict(
     none=NoEventFilter(),
     cusum_vol=CUSUMVolatilityEventFilter(vol_period=20),
-    cusum_fixed=CUSUMFixedEventFilter(threshold=500),
+    cusum_fixed=CUSUMFixedEventFilter(threshold=120),
 )

run_fetch_data_5min.py

@@ -1,41 +1,62 @@
-import pandas as pd
+from binance_historical_data import CandleDataDumper
-import ssl
+import datetime
-from tqdm import tqdm
-from data_loader.utils import deduplicate_indexes
-from utils.resample import resample_ohlc
 
-base_url = "https://www.cryptodatadownload.com/cdd/"
+data_dumper = CandleDataDumper(
+    path_dir_where_to_dump="./data/5min_crypto/",
+    str_data_frequency="5m",
+)
 
-exchange_name = "Bitfinex"
+assets = [
-period = "minute"  # 1h
+    "TRXUSDT",
-files_to_download = [
+    "XRPUSDT",
-    exchange_name + "_TRXUSD_" + period + ".csv",
+    "ADAUSDT",
-    exchange_name + "_ETHUSD_" + period + ".csv",
+    "SOLUSDT",
-    exchange_name + "_XLMUSD_" + period + ".csv",
+    "AVAXUSDT",
-    exchange_name + "_XMRUSD_" + period + ".csv",
+    "DOTUSDT",
-    exchange_name + "_LTCUSD_" + period + ".csv",
+    "ETHUSDT",
-    exchange_name + "_DASHUSD_" + period + ".csv",
+    "LTCUSDT",
-    exchange_name + "_BTCUSD_" + period + ".csv",
+    "BNBUSDT",
-    exchange_name + "_ETCUSD_" + period + ".csv",
+    "BTCUSDT",
+    "ETCUSDT",
 ]
 
+data_dumper.dump_data(
+    list_tickers=assets,
+    date_start=datetime.date(2018, 1, 1),
+    date_end=datetime.date(2022, 1, 1),
+    is_to_update_existing=False,
+)
 
-for file in tqdm(files_to_download):
+import os
-    data_location = base_url + file
+from tqdm import tqdm
-    ssl._create_default_https_context = ssl._create_unverified_context
+import pandas as pd
-    data = pd.read_csv(data_location, skiprows=1, index_col=1, parse_dates=True).drop(
+
-        columns=["unix"]
+for asset in tqdm(assets):
-    )
+    path = f"./data/5min_crypto/{asset}/5m/monthly/"
-    volume_column_to_delete = [
+    files = os.listdir(path)
-        c for c in data.columns if c.startswith("Volume") and "USD" not in c
+
-    ]
+    def load_df(path):
-    data.drop(volume_column_to_delete + ["symbol"], axis=1, inplace=True)
+        df = pd.read_csv(
-    data.rename({"Volume USD": "volume"}, axis=1, inplace=True)
+            path,
-    data.index.rename("time", inplace=True)
+            names=[
-    data.sort_index(inplace=True)
+                "timestamp",
-    data = deduplicate_indexes(data)
+                "open",
-    data.index = pd.to_datetime(data.index)
+                "low",
-    data = data.resample("1Min").ffill()
+                "high",
-    data = resample_ohlc(data, "5Min")
+                "close",
-    target_file = file.split("_")[1].replace("USD", "") + "_USD"
+                "volume",
-    data.to_csv(f"data/5min_crypto/{target_file}.csv")
+                "Closetime",
+                "Quote asset volume",
+                "Number of trades",
+                "Taker buy base asset volume",
+                "Taker buy quote asset volume",
+                "Ignore",
+            ],
+        )
+        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
+        df = df[["timestamp", "open", "high", "low", "close", "volume"]]
+        df.set_index("timestamp", inplace=True)
+        return df
+
+    dfs = pd.concat([load_df(path + file) for file in files], axis=0)
+    dfs.to_parquet(f"./data/5min_crypto/{asset}.parquet")

run_fetch_data_daily.py

@@ -3,26 +3,8 @@ import pandas as pd
 from data_loader.get_prices import get_crypto_price_crypto_compare, get_stock_price_av
 
 #%%
-crypto_tickers = [
+
-    "BTC",
-    "ETH",
-    "BNB",
-    "ADA",
-    "SOL",
-    "XRP",
-    "DOT",
-    "LTC",
-    "UNI",
-    "TRX",
-    "XLM",
-    "BCH",
-    "FIL",
-    "ETC",
-    "THETA",
-    "XTZ",
-]
 etf_tickers = ["GLD", "IEF", "TLT", "SPY", "QQQ"]
-crypto_path = "data/daily_crypto"
 etf_path = "data/daily_etf"
 
 #%%
@@ -31,9 +13,3 @@ for ticker in etf_tickers:
     df = get_stock_price_av(ticker, "2017-11-10")
 
     df.to_csv(f"{etf_path}/{ticker}.csv", index=True)
-
-for src_ticker in crypto_tickers:
-    print("Fetching ", src_ticker, "USD")
-    df = get_crypto_price_crypto_compare(src_ticker, "USD", 1500)
-
-    df.to_csv(f"{crypto_path}/{src_ticker}_USD.csv", index=True)