Source code for tensortrade.data.cdd

"""Contains methods and classes to collect data from
https://www.cryptodatadownload.com.
"""

import ssl

import pandas as pd


ssl._create_default_https_context = ssl._create_unverified_context


[docs]class CryptoDataDownload: """Provides methods for retrieving data on different cryptocurrencies from https://www.cryptodatadownload.com/cdd/. Attributes ---------- url : str The url for collecting data from CryptoDataDownload. Methods ------- fetch(exchange_name,base_symbol,quote_symbol,timeframe,include_all_volumes=False) Fetches data for different exchanges and cryptocurrency pairs. """ def __init__(self) -> None: self.url = "https://www.cryptodatadownload.com/cdd/"
[docs] def fetch_default(self, exchange_name: str, base_symbol: str, quote_symbol: str, timeframe: str, include_all_volumes: bool = False) -> pd.DataFrame: """Fetches data from all exchanges that match the evaluation structure. Parameters ---------- exchange_name : str The name of the exchange. base_symbol : str The base symbol fo the cryptocurrency pair. quote_symbol : str The quote symbol fo the cryptocurrency pair. timeframe : {"d", "h", "m"} The timeframe to collect data from. include_all_volumes : bool, optional Whether or not to include both base and quote volume. Returns ------- `pd.DataFrame` A open, high, low, close and volume for the specified exchange and cryptocurrency pair. """ filename = "{}_{}{}_{}.csv".format(exchange_name, quote_symbol, base_symbol, timeframe) base_vc = "Volume {}".format(base_symbol) new_base_vc = "volume_base" quote_vc = "Volume {}".format(quote_symbol) new_quote_vc = "volume_quote" df = pd.read_csv(self.url + filename, skiprows=1) df = df[::-1] df = df.drop(["symbol"], axis=1) df = df.rename({base_vc: new_base_vc, quote_vc: new_quote_vc, "Date": "date"}, axis=1) df["unix"] = df["unix"].astype(int) df["unix"] = df["unix"].apply( lambda x: int(x / 1000) if len(str(x)) == 13 else x ) df["date"] = pd.to_datetime(df["unix"], unit="s") df = df.set_index("date") df.columns = [name.lower() for name in df.columns] df = df.reset_index() if not include_all_volumes: df = df.drop([new_quote_vc], axis=1) df = df.rename({new_base_vc: "volume"}, axis=1) return df return df
[docs] def fetch_gemini(self, base_symbol: str, quote_symbol: str, timeframe: str) -> pd.DataFrame: """ Fetches data from the gemini exchange. Parameters ---------- base_symbol : str The base symbol fo the cryptocurrency pair. quote_symbol : str The quote symbol fo the cryptocurrency pair. timeframe : {"d", "h", "m"} The timeframe to collect data from. Returns ------- `pd.DataFrame` A open, high, low, close and volume for the specified cryptocurrency pair. """ if timeframe.endswith("h"): timeframe = timeframe[:-1] + "hr" filename = "{}_{}{}_{}.csv".format("gemini", quote_symbol, base_symbol, timeframe) df = pd.read_csv(self.url + filename, skiprows=1) df = df[::-1] df = df.drop(["Symbol", "Unix Timestamp"], axis=1) df.columns = [name.lower() for name in df.columns] df = df.set_index("date") df = df.reset_index() return df
[docs] def fetch(self, exchange_name: str, base_symbol: str, quote_symbol: str, timeframe: str, include_all_volumes: bool = False) -> pd.DataFrame: """Fetches data for different exchanges and cryptocurrency pairs. Parameters ---------- exchange_name : str The name of the exchange. base_symbol : str The base symbol fo the cryptocurrency pair. quote_symbol : str The quote symbol fo the cryptocurrency pair. timeframe : {"d", "h", "m"} The timeframe to collect data from. include_all_volumes : bool, optional Whether or not to include both base and quote volume. Returns ------- `pd.DataFrame` A open, high, low, close and volume for the specified exchange and cryptocurrency pair. """ if exchange_name.lower() == "gemini": return self.fetch_gemini(base_symbol, quote_symbol, timeframe) return self.fetch_default(exchange_name, base_symbol, quote_symbol, timeframe, include_all_volumes=include_all_volumes)