Data Handling Module

class DataHandler(*args, **kwargs)

Bases: object

Responsible for data handling. Including data collection and preparation.

async prepare_data(trading_pair: str, start_date: str, end_date: str, granularity: Granularity, list_of_indicators: Optional[list[source.indicators.indicator_handler_base.IndicatorHandlerBase]] = None) → DataFrame

Collects data from coinbase API and extends it with list of indicators.

Parameters:

  • trading_pair (str) – String representing unique trading pair symbol.

  • start_date (str) – String representing date that collected data should start from.

  • end_date (str) – String representing date that collected data should finish at.

  • granularity (Granularity) – Enum specifying resolution of collected data - e.g. each 15 minutes or 1 hour or 6 hours is treated separately

  • list_of_indicators (list[IndicatorHandlerBase]) – List of indicators that should be calculated and added to the data. Defaults to None, which means no indicators will be added.

Raises:

RuntimeError – If given trading pair symbol is not recognized.

Returns:

Collected data extended with given indicators.

Return type:

(pd.DataFrame)

read_extended_data_from_csv_formatted_string_buffer(file_content_string_buffer: StringIO) → tuple[pandas.core.frame.DataFrame, Optional[dict[str, Any]]]

Reads extended data from a CSV formatted string buffer.

Parameters:

file_content_string_buffer (io.StringIO) – StringIO buffer containing the CSV formatted data.

Returns:

Tuple containing the data frame with extended data and optional metadata.

Return type:

(tuple[pd.DataFrame, Optional[dict[str, Any]]])

register_api_data_collectors(api_data_collectors: list[source.data_handling.api_data_collector_base.ApiDataCollectorBase]) → None

Registers API data collectors for data collection.

Parameters:

api_data_collectors (list[ApiDataCollectorBase]) – A list of instances of ApiDataCollectorBase or its subclasses.

save_extended_data_into_csv_formatted_string_buffer(data: DataFrame, meta_data: Optional[dict[str, Any]] = None) → StringIO

Saves extended data into a CSV formatted string buffer.

Parameters:

  • data (pd.DataFrame) – Data frame to be saved.

  • meta_data (Optional[dict[str, Any]]) – Optional metadata to include in the CSV.

Returns:

StringIO buffer containing the CSV formatted data.

Return type:

(io.StringIO)