DataBase

To avoid spamming APIs and to save time, this library save locally the price data retrieved.

Tables

class CryptoPrice.storage.tables.KlineCacheTable(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME)[source]

__init__(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME)[source]

class CryptoPrice.storage.tables.KlineTable(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME)[source]

__init__(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME)[source]

class CryptoPrice.storage.tables.Table(name: str, columns_names: List[str], columns_sql_types: List[str], primary_key: Optional[str] = None, primary_key_sql_type: Optional[str] = None)[source]

This class represent a table of a database @DynamicAttrs

__init__(name: str, columns_names: List[str], columns_sql_types: List[str], primary_key: Optional[str] = None, primary_key_sql_type: Optional[str] = None)[source]

DataBase

This class is the base of all database, it contains the main logic to communicate with a sqlite db.

class CryptoPrice.storage.DataBase.DataBase(name: str)[source]

This class will be used to interact with sqlite3 databases without having to generates sqlite commands

__init__(name: str)[source]

Instantiate a database object, the name will be used for the saving file

Parameters: name (str) – name of the database

add_row(table: CryptoPrice.storage.tables.Table, row: Tuple, auto_commit: bool = True, update_if_exists: bool = False)[source]

add_rows(table: CryptoPrice.storage.tables.Table, rows: List[Tuple], auto_commit: bool = True, update_if_exists: bool = False)[source]

commit()[source]: submit and save the database state :return:

create_table(table: CryptoPrice.storage.tables.Table)[source]: create a table in the database :param table: Table instance with the config of the table to create :return:

drop_all_tables()[source]

drop all the tables existing in the database

Returns: None
Return type: None

drop_table(table: Union[CryptoPrice.storage.tables.Table, str])[source]

delete a table from the database

Parameters: table (str or Table instance) – table or table name to drop
Returns: None
Return type: None

drop_tables(tables: List[Union[CryptoPrice.storage.tables.Table, str]])[source]

Delete a list of tables if they exist

Parameters: tables (List[Union[Table, str]]) – list of tables to delete
Returns: None
Return type: None

get_all_rows(table: CryptoPrice.storage.tables.Table) → List[source]

get_conditions_rows(table: CryptoPrice.storage.tables.Table, selection: Optional[Union[str, List[str]]] = None, conditions_list: Optional[List[Tuple[str, CryptoPrice.storage.DataBase.SQLConditionEnum, Any]]] = None, order_list: Optional[List[str]] = None) → List[source]

static get_create_cmd(table: CryptoPrice.storage.tables.Table)[source]: return the command in string format to create a table in the database :param table: Table instance with the config if the table to create :return: execution command for the table creation

get_row_by_key(table: CryptoPrice.storage.tables.Table, key_value) → Optional[Tuple][source]: get the row identified by a primary key value from a table :param table: table to fetch the row from :param key_value: key value of the row :return: None or the row of value

get_tables_descriptions() → List[Tuple][source]

return the descriptions of all the tables existing in the database

Returns: tables descriptions
Return type: List[Tuple]

update_row(table: CryptoPrice.storage.tables.Table, row: Tuple, auto_commit=True)[source]

class CryptoPrice.storage.DataBase.SQLConditionEnum(value)[source]

https://www.techonthenet.com/sqlite/comparison_operators.php

diff = '!='

equal = '='

greater = '>'

greater_equal = '>='

lower = '<'

lower_equal = '<='

KlineDataBase

This is a child class of DataBase, specifi for candle data (ohlc).

class CryptoPrice.storage.KlineDataBase.KlineDataBase(name: str)[source]

__init__(name: str)[source]

Instantiate a kline database object, the name will be used for the saving file

Parameters: name (str) – name of the database

add_cache_closest(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME, timestamp: int, closest_timestamp: int, window: int)[source]

Save the result of a previous closest price request

Parameters

asset (str) – asset of the trading pair
ref_asset (str) – reference asset of the trading pair
timeframe (TIMEFRAME) – timeframe for the kline
timestamp (int) – request timestamp
closest_timestamp (int) – the timestamp that got selected as the closest
window (int) – time window in seconds that got used to fetch the klines

Returns

None

Return type

None

add_klines(klines: List[CryptoPrice.common.prices.Kline], ignore_if_exists: bool = False)[source]

add several klines to the database

Parameters

klines (List[Kline]) – list of klines to add to the database
ignore_if_exists (bool) – if integrity errors should be ignored, default False

Returns

None

Return type

None

drop_cache_tables()[source]

Delete all the cache tables stored in the database

Returns: None
Return type: None

drop_pair_table(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME)[source]

drop the table associated with a trading pair and a time frame

Parameters

asset (str) – asset of the trading pair
ref_asset (str) – reference asset of the trading pair
timeframe (TIMEFRAME) – timeframe for the kline

Returns

None

Return type

None

get_cache_closest(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME, timestamp: int) → Tuple[Optional[int], int][source]

Look if a request for the timestamp has been saved in the cache Return the cached timestamp of the previously selected kline along with the window used at that time a timestamp of -1 means that no result were found

Parameters

asset (str) – asset of the trading pair
ref_asset (str) – reference asset of the trading pair
timeframe (TIMEFRAME) – timeframe for the kline
timestamp (int) – request timestamp

Returns

cached_timestamp, window

Return type

Optional[int], int

get_closest_kline(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME, timestamp: int, window: int = 120) → Optional[CryptoPrice.common.prices.Kline][source]

Return the closest Kline in a time window for a trading pair and a timeframe. If there is no Kline, None is returned

Parameters

asset (str) – asset of the trading pair
ref_asset (str) – reference asset of the trading pair
timeframe (TIMEFRAME) – timeframe for the kline
timestamp (int) – time of interest in seconds
window (int) – time window in seconds for the kline to look

Returns

the Kline with an open time the closest to the provided timestamp

Return type

Optional[Kline]

get_klines(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME, start_time: Optional[int] = None, end_time: Optional[int] = None) → List[CryptoPrice.common.prices.Kline][source]

return the klines corresponding to a trading pair and a timeframe a time window can also be provided.

Parameters

asset (str) – asset of the trading pair
ref_asset (str) – reference asset of the trading pair
timeframe (TIMEFRAME) – timeframe for the kline
start_time (Optional[int]) – fetch only klines with an open time greater or equal than start_time
end_time (Optional[int]) – fetch only klines with an open time lower than end_time

Returns

list of klines

Return type

List[Klines]

row_to_kline(asset: str, ref_asset: str, timeframe: CryptoPrice.utils.time.TIMEFRAME, row: Tuple)[source]

Take a row from the KlineDatabase and transform it into a Kline object

Parameters

asset (str) – asset of the trading pair
ref_asset (str) – reference asset of the trading pair
timeframe (TIMEFRAME) – timeframe for the kline
row (Tuple) – raw data from the database

Returns

the Kline corresponding to the args

Return type

Kline