PostgreSQL Module

This module contains the PostgreSQL handler for database operations.

Functions:

setup(symbol, tick_interval, ...[, ...])

Setups the connection to the PostgreSQL database.

get_data_and_parse(cursor, table, symbol, ...)

Gets data from a table in the database and parses it to a dataframe.

create_connection(user, password, host, ...)

Crea una conexión a la base de datos PostgreSQL.

is_cursor_alive(cursor)

Verifica si un cursor de PostgreSQL está activo y en buen estado.

close_connection(connection, cursor)

It closes the connection to the PostgreSQL database.

create_database(cursor, db_name)

Crea una nueva base de datos en PostgreSQL.

list_tables_with_suffix(cursor[, suffix])

Retorna una lista con los nombres de las tablas que terminan con el sufijo especificado.

count_rows_in_tables(cursor, table_names[, ...])

is_hypertable(cursor, table_name)

Si la tabla es una hypertable, retorna True.

get_column_names(cursor, table_name, ...)

Returns a list with the names of the columns in a table.

fetch_hypertable(cursor, table[, startime, ...])

Returns a list with integrity data, that is, trades or klines.

fetch_hypertable_selective(cursor, table[, ...])

Returns a list with the specified data from a table.

delete_record(cursor, table_name, ...[, ...])

Delete records from a hypertable in TimescaleDB where a specific field matches a specific value.

delete_table(cursor, table_name[, schema])

Delete a table from the database to which the cursor is connected.

delete_bulk_tables(cursor, tables[, batch])

It deletes a list of tables from the database to which the cursor is connected.

sanitize_table_name(table_name)

Sanitizes a table name.

get_valid_table_list(cursor)

Gets the list of tables in the database.

infer_sql_type(value, key[, time_col])

Infers the SQL type of a value. Because hypertables will be used, NOT NULL or UNIQUE are not allowed. Uniqueness must be

check_unique_index_with_time(cursor, ...)

Checks if a unique index with the time column exists for the given table.

create_table_and_hypertable(cursor, ...[, ...])

Creates a regular table and converts it to a hypertable.

insert_data(cursor, table_name, records, ...)

Inserts data into a table.

update_table_columns(cursor, table_name, ...)

Verificar igualdad de columnas e insertar columnas faltantes en su caso.

flexible_tables_and_data_insert(cursor, ...)

Checks if the tables exist and creates them if they do not exist.

data_type_table(table_name)

Gets the data type of the table name.

delete_dupes(cursor, table, dupes, column, ...)

Delete a dupe by continuity column.

data_type_from_table(table)

Gets the data type of the table name.

check_standard_table_exists(cursor, table_name)

Checks if a standard table exists.

get_indexed_columns(cursor, table_name)

No tengo muy claro la respuesta tan extendida de campos que da esta consulta.

get_hypertable_indexes(cursor, hypertable_name)

Return example:

add_index_to_hypertable(cursor, ...[, ...])

Añade un índice a una hypertable en TimescaleDB.

check_index_exists(cursor, table_name, ...)

create_simple_table(cursor, table_name, columns)

Create a simple table in PostgreSQL.

create_missed_table(cursor, ...)

Create a table to store the missed ids.

get_missed_ids(cursor, table, ...[, ...])

Get the missed timestamps or trade_id from the database for each table.

insert_missed_from_api_ids(cursor, table, ...)

Insert missed ids from the API into the database.
setup(symbol: str, tick_interval: str, postgresql_host: str, postgresql_user: str, postgresql_database: str, postgres_klines: bool, postgres_atomic_trades: bool, postgres_agg_trades: bool, postgresql_port: int = 5432) → tuple[source]

Setups the connection to the PostgreSQL database.

Parameters:

  • symbol – A symbol like “BTCUSDT”

  • tick_interval – A tick interval like “1m”

  • postgresql_host – A host name or ip address

  • postgresql_user – A user name

  • postgresql_database – A database name

  • postgres_klines – A boolean to indicate if klines are requested

  • postgres_atomic_trades – A boolean to indicate if atomic trades are requested

  • postgres_agg_trades – A boolean to indicate if aggregate trades are requested

  • postgresql_port – A port number

Returns:

Connection and cursor to the database

get_data_and_parse(cursor, table: str, symbol: str, tick_interval: str, time_zone: str, start_time: int, end_time: int, data_type: str) → DataFrame[source]

Gets data from a table in the database and parses it to a dataframe.

Parameters:

  • cursor – Database cursor

  • table – Table name

  • symbol – Symbol like “BTCUSDT”

  • tick_interval – Tick interval like “1m”

  • time_zone – Time zone like “Europe/Madrid”

  • start_time – Timestamp in milliseconds

  • end_time – Timestamp in milliseconds

  • data_type – Data type like “kline”, “trade”, “aggTrade”. Types are Binance websocket stream name types.

Returns:

A dataframe with the data and pertinent columns.

create_connection(user: str, password: str, host: str, port: int, database: str, timeout: int = 10) → tuple[connection, cursor][source]

Crea una conexión a la base de datos PostgreSQL.

Parameters:

  • user – Nombre de usuario

  • password – Contraseña en texto plano (gestionada por panzer).

  • host – Nombre del host o dirección IP

  • port – Número de puerto

  • database – Nombre de la base de datos

  • timeout – Tiempo de espera en segundos

Returns:

Devuelve una conexión y un cursor a la base de datos

is_cursor_alive(cursor) → bool[source]

Verifica si un cursor de PostgreSQL está activo y en buen estado.

Parámetros: - cursor: Cursor de la conexión a la base de datos PostgreSQL.

Retorna: - True si el cursor está en buen estado, False en caso contrario.

close_connection(connection, cursor) → None[source]

It closes the connection to the PostgreSQL database.

Parameters:

  • connection – A psycopg2 connection to the database.

  • cursor – A psycopg2 cursor to the database.

Returns:

It returns nothing.

create_database(cursor, db_name: str) → None[source]

Crea una nueva base de datos en PostgreSQL.

Parameters:

  • cursor – Cursor de psycopg2 a la base de datos.

  • db_name – Nombre de la base de datos a crear.

list_tables_with_suffix(cursor, suffix='') → list[source]

Retorna una lista con los nombres de las tablas que terminan con el sufijo especificado.

Parameters:

  • cursor – Un cursor de psycopg2 conectado a la base de datos.

  • suffix – Opcional. Sufijo de las tablas a buscar. Ejemplo: “_trade”.

Returns:

Una lista con los nombres de las tablas que terminan con el sufijo especificado.

count_rows_in_tables(cursor, table_names: list[str], approximated: bool = True) → dict[str, int][source]
is_hypertable(cursor, table_name) → bool[source]

Si la tabla es una hypertable, retorna True. Si no, retorna False. :param cursor: :param table_name: :return:

get_column_names(cursor, table_name: str, own_transaction: bool) → list[source]

Returns a list with the names of the columns in a table.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table_name – Name of the table.

  • own_transaction – If True, the function will create its own transaction. If False, the function will use the transaction of the cursor.

Returns:

fetch_hypertable(cursor, table: str, startime: int = None, endtime: int = None) → list[tuple][source]

Returns a list with integrity data, that is, trades or klines. Obtained from the table.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table – Name of the table.

  • startime – A timestamp in milliseconds. If not specified, the first record will be used.

  • endtime – A timestamp in milliseconds. If not specified, it will fetch up to the last record.

Returns:

It returns a list of tuples with the data retrieved.

fetch_hypertable_selective(cursor, table: str, columns: list[str] | None = None, startime: int = None, endtime: int = None) → list[tuple][source]

Returns a list with the specified data from a table.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table – Name of the table.

  • columns – List of columns to select.

  • startime – A timestamp in milliseconds. If not specified, the first record will be used.

  • endtime – A timestamp in milliseconds. If not specified, it will fetch up to the last record.

Returns:

List of tuples with the recovered data.

delete_record(cursor, table_name: str, field_name: str, value: any, is_timestamp: bool = False, own_transaction=True) → None[source]

Delete records from a hypertable in TimescaleDB where a specific field matches a specific value.

Parameters:

  • cursor – Cursor from psycopg2 to the database.

  • table_name – Name of the hypertable.

  • field_name – Name of the field to compare.

  • value – Value to look for to delete.

  • is_timestamp – If True, converts the millisecond value to TIMESTAMPZ.

  • own_transaction – If True, the function will create its own transaction. If False, the function will use the

delete_table(cursor, table_name, schema='public') → bool[source]

Delete a table from the database to which the cursor is connected. Also removes the hypertable if it exists.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table_name – Name of the table to delete.

  • schema – Table schema.

Returns:

Returns True if the table was deleted successfully, False otherwise.

delete_bulk_tables(cursor, tables: list, batch=20) → None[source]

It deletes a list of tables from the database to which the cursor is connected.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • tables – A list of tables to delete.

  • batch – A batch size. Default is 20. Each batch will be committed.

Returns:

It returns nothing.

sanitize_table_name(table_name: str) → str[source]

Sanitizes a table name. It replaces @ with _ and adds a prefix if the name starts with a number.

Parameters:

table_name – A table name.

Returns:

A sanitized table name.

get_valid_table_list(cursor) → list[source]

Gets the list of tables in the database.

Parameters:

cursor – A psycopg2 cursor connected to the database.

Returns:

infer_sql_type(value: any, key: str, time_col: str = 'time') → str[source]
Infers the SQL type of a value. Because hypertables will be used, NOT NULL or UNIQUE are not allowed. Uniqueness must be

enforced by the index.

Parameters:

  • value – A value.

  • key – A key.

  • time_col – The name of the time column. Default is “time”.

Returns:

It returns a string with the SQL type. Like “BIGINT” or “TEXT”.

check_unique_index_with_time(cursor, table_name: str, time_col: str) → bool[source]

Checks if a unique index with the time column exists for the given table.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table_name – A table name.

  • time_col – A time column name.

Returns:

Returns True if the unique index exists, False otherwise.

create_table_and_hypertable(cursor, table_name: str, column_definitions: dict, time_col: str, additional_index_column=None) → None[source]

Creates a regular table and converts it to a hypertable. If the table already exists, it does nothing.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table_name – Name of the table.

  • column_definitions – Definitions of the columns. A dictionary where the key is the name of the column and the value is

  • time_col – The name of the time column.

  • additional_index_column – Any additional column to be indexed. If None, just time_col will be indexed.

Returns:

It returns nothing.

insert_data(cursor, table_name: str, records: list[dict], time_column: str, unique_column: str = None) → None[source]

Inserts data into a table.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table_name – Name of the table.

  • records – A list of dictionaries with the data to insert.

  • time_column – The name of the time column.

  • unique_column – The name of the unique column. If None, no extra uniqueness will be enforced.

Returns:

It returns nothing.

update_table_columns(cursor, table_name: str, record: dict, checked_columns: list) → list[source]

Verificar igualdad de columnas e insertar columnas faltantes en su caso.

Parameters:

  • cursor – Cursor de la conexión a la base de datos.

  • table_name – Nombre de la tabla.

  • record – Un diccionario con los datos a typo a insertar para deducir el tipo de datos.

  • checked_columns – Lista de columnas ya verificadas.

Returns:

Retorna una lista con las tablas verificadas actualizadas.

flexible_tables_and_data_insert(cursor, parsed_dict: dict[str, list[dict]], verified_tables: list = None, checked_columns: list = None, time_column='time', batch=10) → tuple[list, list][source]

Checks if the tables exist and creates them if they do not exist. Then insert the data into the tables.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • parsed_dict – A dictionary with the data to be inserted. The key is the name of the table and the value is a list of dictionaries with the data to be inserted.

  • verified_tables – A set of verified tables.

  • checked_columns – A set of checked columns.

  • time_column – Name of the time column.

  • batch – Size of the batch to insert data.

Returns:

It returns a tuple with the verified tables and the checked columns.

data_type_table(table_name: str) → str[source]

Gets the data type of the table name.

Parameters:

table_name – The name of the table.

Returns:

Websockets channel type.

delete_dupes(cursor, table: str, dupes: list[int], column: str, is_timestamp: bool, ignore_errors: bool = False) → None[source]

Delete a dupe by continuity column.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table – The name of the table.

  • dupes – A list with the values of the dupe.

  • column – The name of the continuity column.

  • is_timestamp – If True, convert the values to timestamp.

  • ignore_errors – If True, ignore errors.

Returns:

None

data_type_from_table(table: str) → str | None[source]

Gets the data type of the table name.

Parameters:

table – The name of the table.

Returns:

Websockets channel type.

check_standard_table_exists(cursor, table_name: str) → bool[source]

Checks if a standard table exists.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table_name – Name of the table.

Returns:

Returns True if the table exists, False otherwise.

get_indexed_columns(cursor, table_name) → list[source]

No tengo muy claro la respuesta tan extendida de campos que da esta consulta.

Parameters:

  • cursor – Un cursor de psycopg2 conectado a la base de datos.

  • table_name – Nombre de la tabla.

Returns:

Una lista con los nombres de las columnas indexadas.

get_hypertable_indexes(cursor, hypertable_name, schema_name='public') → list[tuple][source]

Return example:

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • hypertable_name – Name of the hypertable.

  • schema_name – Name of the schema. Default is ‘public’.

Returns:

A list of tuples with the index name and the index definition.

add_index_to_hypertable(cursor, hypertable_name: str, column_name: str, index_name: str = None) → bool[source]

Añade un índice a una hypertable en TimescaleDB.

Ejemplo:

add_index_to_hypertable(cursor, "simple_table", column_name="time", index_name="pepe")

[('simple_table_time_idx',
  'CREATE INDEX simple_table_time_idx ON public.simple_table USING btree ("time" DESC)'),
 ('pepe', 'CREATE INDEX pepe ON public.simple_table USING btree ("time")')]
Parameters:

  • cursor – Un cursor de psycopg2 conectado a la base de datos.

  • hypertable_name – Nombre de la hypertable.

  • column_name – Nombre de la columna a la que se añadirá el índice.

  • index_name – Nombre opcional para el nuevo índice. Si no se proporciona, se generará automáticamente.

Returns:

Retorna True si la operación fue exitosa, False en caso contrario.

check_index_exists(cursor, table_name, index_name) → bool[source]
create_simple_table(cursor, table_name: str, columns: list) → None[source]

Create a simple table in PostgreSQL.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table_name – Name of the new table.

  • columns – List of tuples describing the columns. Each tuple must have the column name and the data type (e.g. [(“id”, “SERIAL PRIMARY KEY”), (“name”, “VARCHAR(50)”)]).

create_missed_table(cursor, continuity_field: str, miss_table: str) → None[source]

Create a table to store the missed ids.

Parameters:

  • continuity_field – A continuity field. Like “time” or “trade_id”.

  • cursor – A psycopg2 cursor connected to the database.

  • miss_table – A table name. Like “ltcusdt_trade_missed” or “ltcusdt_kline_1m_missed”.

Returns:

None

get_missed_ids(cursor, table: str, continuity_field: str, previously_used_missed_tables: list, own_transaction: bool = True) → tuple[list, list][source]

Get the missed timestamps or trade_id from the database for each table. Each table has a parallel table with api misses.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table – Missed ids table name. Default is “table_name_missed”.

  • continuity_field – The name of the continuity field.

  • previously_used_missed_tables – A list of existing tables. If None, it will be retrieved from the database.

  • own_transaction – If True, the function will create its own transaction. If False, the function will use the

Returns:

A tuple with a list with the missed timestamps and a list with the previously used missed tables.

insert_missed_from_api_ids(cursor, table: str, misses: list[int], continuity_field: str, previously_used_missed_tables: list = None, own_transaction: bool = True) → list[source]

Insert missed ids from the API into the database.

Parameters:

  • cursor – A psycopg2 cursor connected to the database.

  • table – A table name. Like “ltcusdt_trade”.

  • misses – A list of missed ids. Each id is a timestamp or a trade_id. Example: [1620000000000, 1620000000001, …]

  • continuity_field – A continuity field. Like “time” or “trade_id”.

  • previously_used_missed_tables – A list of existing tables. If None, it will be retrieved from the database.

  • own_transaction – A boolean. If True, the function will create its own transaction. If False, the function will use the

Returns:

It returns nothing.