client

Initializes the client for connecting to the Binance API.

Source code in make_us_rich/client/binance_client.py

def __init__(self):
    """
    Initializes the client for connecting to the Binance API.
    """
    try:
        self._config = load_env("binance")
    except:
        self._config = {"API_KEY": getenv("API_KEY"), "SECRET_KEY": getenv("SECRET_KEY")}
    self.client = Client(self._config["API_KEY"], self._config["SECRET_KEY"])
    self.columns = ["timestamp", "open", "high", "low", "close", "volume", "close_time", 
        "quote_av", "trades", "tb_base_av", "tb_quote_av", "ignore"]

get_data(symbol, interval, start_time, end_time=None)

Gets the data for the given symbol, interval, and time range.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol to get the data for.	required
interval	str	Interval to get the data for.	required
start_time	str	Start time of the data.	required
end_time	Optional[str]	End time of the data.	None

Returns:

Type	Description
pd.DataFrame	Dataframe for the given symbol, interval, and time range.

Source code in make_us_rich/client/binance_client.py

def get_data(self, symbol: str, interval: str, start_time: str, end_time: Optional[str] = None) -> pd.DataFrame:
    """
    Gets the data for the given symbol, interval, and time range.

    Parameters
    ----------
    symbol: str
        Symbol to get the data for.
    interval: str
        Interval to get the data for.
    start_time: str
        Start time of the data.
    end_time: Optional[str]
        End time of the data. 

    Returns
    -------
    pd.DataFrame
        Dataframe for the given symbol, interval, and time range.
    """
    klines = self.client.get_historical_klines(symbol, interval, start_time, end_time)
    data = pd.DataFrame(klines, columns=self.columns)
    data["timestamp"] = pd.to_datetime(data["timestamp"], unit="ms")
    return data

get_five_days_data(symbol)

Gets the data for the last five days.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol to get the data for.	required

Returns:

Type	Description
pd.DataFrame	Dataframe for the last five days.

Source code in make_us_rich/client/binance_client.py

def get_five_days_data(self, symbol: str) -> pd.DataFrame:
    """
    Gets the data for the last five days.

    Parameters
    ----------
    symbol: str
        Symbol to get the data for.

    Returns
    -------
    pd.DataFrame
        Dataframe for the last five days.
    """
    symbol = symbol.upper()
    klines = self.client.get_historical_klines(symbol, "1h", "5 day ago UTC")
    data = pd.DataFrame(klines, columns=self.columns)
    data["timestamp"] = pd.to_datetime(data["timestamp"], unit="ms")
    return data

get_one_year_data(symbol)

Gets the data for the last year.

Parameters:

Name	Type	Description	Default
symbol	str	Symbol to get the data for.	required

Returns:

Type	Description
pd.DataFrame	Dataframe for the last year.

Source code in make_us_rich/client/binance_client.py

def get_one_year_data(self, symbol: str) -> pd.DataFrame:
    """
    Gets the data for the last year.

    Parameters
    ----------
    symbol: str
        Symbol to get the data for.

    Returns
    -------
    pd.DataFrame
        Dataframe for the last year.
    """
    klines = self.client.get_historical_klines(symbol, "1h", "1 year ago UTC")
    data = pd.DataFrame(klines, columns=self.columns)
    data["timestamp"] = pd.to_datetime(data["timestamp"], unit="ms")
    return data

Initializes the Minio client.

Returns:

Type	Description
None

Source code in make_us_rich/client/minio_client.py

def __init__(self) -> None:
    """
    Initializes the Minio client.

    Returns
    -------
    None
    """
    try:
        self._config = load_env("minio")
    except:
        self._config = {
            "ACCESS_KEY": getenv("ACCESS_KEY"), "SECRET_KEY": getenv("SECRET_KEY"), 
            "ENDPOINT": getenv("ENDPOINT"), "BUCKET": getenv("BUCKET")
        }
    print(self._config)
    self.client = Minio(
        endpoint=str(self._config["ENDPOINT"]),
        access_key=self._config["ACCESS_KEY"],
        secret_key=self._config["SECRET_KEY"],
        secure=False
    )
    self.bucket = self._config["BUCKET"]
    if not self.client.bucket_exists(self.bucket):
        self.client.make_bucket(self.bucket)

download(bucket, object_name, file_path)

Downloads a file from Minio.

Parameters:

Name	Type	Description	Default
bucket	str	Bucket name.	required
object_name	str	Object name.	required
file_path	str	File path.	required

Returns:

Type	Description
None

Source code in make_us_rich/client/minio_client.py

def download(self, bucket: str, object_name: str, file_path: str) -> None:
    """
    Downloads a file from Minio.

    Parameters
    ----------
    bucket: str
        Bucket name.
    object_name: str
        Object name.
    file_path: str
        File path.

    Returns
    -------
    None
    """
    self.client.fget_object(
        bucket_name=bucket, object_name=object_name, file_path=file_path
    )

upload(bucket, object_name, file_path)

Uploads a file to Minio.

Parameters:

Name	Type	Description	Default
bucket	str	Bucket name.	required
object_name	str	Object name.	required
file_path	str	File path.	required

Returns:

Type	Description
None

Source code in make_us_rich/client/minio_client.py

def upload(self, bucket: str, object_name: str, file_path: str) -> None:
    """
    Uploads a file to Minio.

    Parameters
    ----------
    bucket: str
        Bucket name.
    object_name: str
        Object name.
    file_path: str
        File path.

    Returns
    -------
    None
    """
    self.client.fput_object(
        bucket_name=bucket, object_name=object_name, file_path=file_path
    )

---

Last update: 2022-05-04