Data Container Access¶

The following provide information on how the high level operations on the data container are performed

class crypto_history.data_container.data_container_access.TimeStampIndexedDataContainer¶

Responsible for transforming the Primitive DataArray to a DataArray indexed by the timestamp of choice with possibility to approximate for easier handling

calculate_tolerance_value_from_ratio(ratio: float = 0.001) → float¶

Calculates the tolerance value based on the tolerance ratio :param ratio: ratio of the tolerance :type ratio: float

Returns:	float, value of the tolerance value

classmethod create_time_stamp_indexed_data_container(exchange_factory: crypto_history.stock_market.stock_market_factory.StockMarketFactory, base_assets: List[T], reference_assets: List[T], reference_ticker: tuple, aggregate_coordinate_by: str, ohlcv_fields: List[T], weight: str, start_time: Union[str, datetime.datetime, int], end_time: Union[str, datetime.datetime, int])¶

The factory for creating the time stamp indexed data container :param exchange_factory: The exchange factory :type exchange_factory: StockMarketFactory :param base_assets: List of base assets :type base_assets: List :param reference_assets: List of reference assets :type reference_assets: List :param reference_ticker: (‘xxx’, ‘yyy’) where xxx is the base_asset and yyy is the reference asset for indexing the timestamp :type reference_ticker: tuple :param aggregate_coordinate_by: The direction in which the coordinates should be aggregated by :type aggregate_coordinate_by: str :param ohlcv_fields: list of ohlcv-fields necessary to capture :type ohlcv_fields: List :param weight: weight/interval of the kline/candle :type weight: str :param start_time: start time/date of the candles :type start_time: str/datetime.datetime/int :param end_time: end time/date of the candles :type end_time: str/datetime.datetime/int

Yields:	TimeStampIndexedDataContainer constructed with above details

static generate_empty_dataarray_indexed_by_timestamp(coords: Dict[KT, VT]) → xarray.core.dataarray.DataArray¶

Generates the new dataarray from the coordinates provided :param coords: the coordinates for the new dataarray :type coords: Dict

Returns:	xr.DataArray the new empty dataarray

generate_empty_df_with_new_timestamps(do_approximation: bool, old_dataarray: xarray.core.dataarray.DataArray) → xarray.core.dataarray.DataArray¶

Generates an empty dataframe with new timestamps in the coordinates :param do_approximation: Check if the timestamps can be approximated :type do_approximation: bool :param old_dataarray: The old dataarray used for reference :type old_dataarray: xr.DataArray :param for constructing new dataarray:

Returns:	xr.DataArray The new empty dataarray which has the coordinates set

static get_all_unique_values(dataarray: xarray.core.dataarray.DataArray, coord_name: str, field_in_coord: str) → List[T]¶

Gets all the unique values in the xr.DataArray in the particular item of the coordinate :param dataarray: whose data is to be selected :type dataarray: xr.DataArray :param coord_name: name of the coordinate which is the primary selector :type coord_name: str :param field_in_coord: name of the item in the coordinate which is the second level of select :type field_in_coord: str

Returns:	list of all the items in the selected coordinate, field sorted increasingly

get_coords_for_timestamp_indexed_datarray(old_dataarray: xarray.core.dataarray.DataArray, reference_ts: List[int]) → Dict[KT, VT]¶

Gets the coordinates for the timestamp index dataarray :param old_dataarray: The old dataarray taken as the building block for the new dataarray :type old_dataarray: xr.DataArray :param reference_ts: The new timestamps which serve as the references instead of index_number :type reference_ts: list

Returns:	dictionary of coordinates for the new dataarray

static get_df_to_insert(sub_dataarray: xarray.core.dataarray.DataArray, index_of_integrating_ts: int, reference_ts: List[T], tolerance: float) → pandas.core.frame.DataFrame¶

Calculates the dataframe from the dataarray with one reduced shape size :param sub_dataarray: the sub dataarray whose dataframe is to be extracted :type sub_dataarray: xr.DataArray :param index_of_integrating_ts: index of the integrating field in the coordinate :type index_of_integrating_ts: int :param reference_ts: list of the indexes according to what the dataframe needs to be adjusted :type reference_ts: List :param tolerance: the value of the tolerance which can be considered while matching indices :type tolerance: float

Returns:	pd.DataFrame the reindexed dataframe which contains the data in the required indices
Raises:	EmptyDataFrameException if the sub_dataarray contains only na values

static get_index_of_field(data_array: xarray.core.dataarray.DataArray, coord_name: str, item_in_coord: str) → int¶

Gets the index of the particular field in the coordinate :param data_array: in which the index is to be identified :type data_array: xr.DataArray :param coord_name: name of the coordinate :type coord_name: str :param item_in_coord: name of the field in the coordinate :type item_in_coord: str

Returns:	int, the index number which can be used to identify the item

get_primitive_full_xr_dataarray() → xarray.core.dataarray.DataArray¶: Gets the full data container :returns: the complete data container :rtype: xr.DataArray

get_primitive_reference_xr_dataarray() → xarray.core.dataarray.DataArray¶: Gets the reference data container :returns: the reference data container (eg. ETHBTC history) :rtype: xr.DataArray

static get_primitive_xr_dataarray(data_container_object: crypto_history.data_container.data_container_pre.PrimitiveDataArrayOperations) → xarray.core.dataarray.DataArray¶: Gets the primitive xr.DataArray from the data container object :returns: the actual xr.DataArray :rtype: xr.DataArray

get_timestamps_for_new_dataarray(do_approximation: bool, dataarray: xarray.core.dataarray.DataArray) → List[T]¶

Obtains the time stamp for the new dataarray. It may either select it from the reference dataarray or get it from the actual dataarray if approximation is not desired :param do_approximation: if True, timestamps are approximated and taken from the reference dataarray. if False, timestamps are not approximated. All timestamps from all various tickers will be the coordinate for the timestamp :type do_approximation: bool :param dataarray: the original dataarray which contains all the timestamps in it :type dataarray: xr.DataArray

Returns:	list of the timestamps for the new dataarray

get_timestamps_of_reference_dataarray(field_in_coord: str) → List[T]¶

Obtains the time stamp list from the reference dataarray :param field_in_coord: the item in the ohlcv_fields which is of interest :type field_in_coord: str

Returns:	list of the time stamps from the reference dataarray

get_value_of_tolerance(do_approximation: bool, tolerance_ratio: float = 0.001)¶

Gets the value of tolerance used for reindexing :param do_approximation: Check if approximation can be made :type do_approximation: bool :param tolerance_ratio: The ratio of maximum tolerance of time difference for generating coordinates of timestamp :type tolerance_ratio: float

Returns:	float, value of the tolerance

get_xr_dataarray_indexed_by_timestamps(do_approximation: bool = True, tolerance_ratio: float = 0.001) → xarray.core.dataarray.DataArray¶

Gets the xr.DataSet of the coin histories of the particular chunk. Obtains the data and then transforms it to the xr.DataSet :param do_approximation: check if the timestamps can be approximated to the reference datarray :type do_approximation: bool :param tolerance_ratio: the ratio of the maximum tolerance for :type tolerance_ratio: float :param approximations of timestamps:

Returns:	xr.DataSet data of the coin history

class crypto_history.data_container.data_container_access.TimeAggregatedDataContainer¶

Aggregates various time-ranges into the data container

static concatenate_dataarray_in_coord(*dataarray, coordinate='timestamp') → xarray.core.dataarray.DataArray¶

Concatenates two histories in the desired coordinate/dimension :param dataarray List[xr.DataArray]: List/Tuple of xr.DataArrays that need to be concatednated :param coordinate: coordinate direction to concat by

Returns:	Concatenated xr.DataArray consisting of all the data concatenated

classmethod create_instance(exchange_factory: crypto_history.stock_market.stock_market_factory.StockMarketFactory, base_assets: List[str], reference_assets: List[str], ohlcv_fields: List[str], time_range_dict: Dict[Tuple[Union[datetime.timedelta, str], Union[datetime.timedelta, str]], str], reference_ticker: Tuple = ('ETH', 'BTC'), aggregate_coordinate_by: str = 'open_ts') → TypeVarPlaceHolder¶

Creates the time-aggregator from time-deltas which go: back from the current time

Parameters:

exchange_factory (StockMarketFactory) – The exchange factory
base_assets (List) – base-asset coins
reference_assets (List) – reference-asset coins
ohlcv_fields (List) – ohlcv-fields to calculate
time_range_dict (Dict) – dictionary of time-ranges where the keys are tuples of timedeltas
reference_ticker (Tuple) – reference-ticker for reference timestamps
aggregate_coordinate_by (str) – Timestamp indexed by this value

Returns:

TimeAggregatedDataContainer

get_chunk_history(kline_width: str, start_time: Union[str, datetime.datetime, int], end_time: Union[str, datetime.datetime, int]) → xarray.core.dataarray.DataArray¶

Gets the history of the particular chunk whose start and end times are well defined :param kline_width: The exchanges interval :type kline_width: str :param start_time: the start-time of the chunk of history :type start_time: datetime/str/int :param end_time: (datetime/str/int): the end-time of the chunk of history

Returns:	xr.DataArray the history of the chunk

static get_sorted_dataarray(dataarray: xarray.core.dataarray.DataArray, coordinate: str) → xarray.core.dataarray.DataArray¶

Sorts the given dataarray in the given coordinate direction :param dataarray: The dataarray which has to be sorted :type dataarray: xr.DataArray :param coordinate: sort the dataarray in the particular direction of the coordinate

Returns:	xr.DataArray The sorted dataarray

get_time_aggregated_data_container(sort: bool = True) → xarray.core.dataarray.DataArray¶

Gets time aggregated data container by splitting: the containers as required

Parameters:	sort (bool) – to check if the data is to be sorted in an increasing order
Returns:	xr.DataArray The complete history of the interval

get_time_interval_chunks(time_range: Dict[KT, VT]) → List[tuple]¶

Gets the time interval chunks from the raw dict provided :param time_range: Dict with key=tuple of start and end time value=type of klines/intervals

Returns:	List of tuples with the individual requests for histories ((start time, end time), interval)