Funding_Rate/modules/extended_db.py

import logging
from typing import AsyncContextManager
from sqlalchemy import text


### Orders Table ####
async def create_fr_extended_user_order(
        CON: AsyncContextManager,
        engine: str = 'mysql', # mysql | duckdb
    ) -> None:
    if CON is None:
        logging.info("NO DB CONNECTION, SKIPPING Create Statements")
    else:
        if engine == 'mysql':
            logging.info('Creating Table if Does Not Exist: fr_extended_user_order')
            await CON.execute(text("""
            CREATE TABLE IF NOT EXISTS fr_extended_user_order (
                sequence_id INT,
                timestamp_arrival BIGINT,
                timestamp_msg BIGINT,
                order_id VARCHAR(100),
                account_id VARCHAR(100),
                external_id VARCHAR(100),
                market VARCHAR(20),
                type VARCHAR(20),
                side VARCHAR(20),
                status VARCHAR(20),
                status_reason VARCHAR(100),
                price DOUBLE,
                averagePrice DOUBLE,
                qty DOUBLE,
                filled_qty DOUBLE,
                payed_fee DOUBLE,
                tp_sl_type VARCHAR(20),
                reduce_only BOOL,
                post_only BOOL,
                created_time_ts BIGINT,
                updated_time_ts BIGINT,
                expire_time_ts BIGINT
            );
            """))
            await CON.commit()
        else:
            raise ValueError('Only MySQL engine is implemented')

### Orders Table ####
async def create_fr_extended_user_trade(
        CON: AsyncContextManager,
        engine: str = 'mysql', # mysql | duckdb
    ) -> None:
    if CON is None:
        logging.info("NO DB CONNECTION, SKIPPING Create Statements")
    else:
        if engine == 'mysql':
            logging.info('Creating Table if Does Not Exist: fr_extended_user_trade')
            await CON.execute(text("""
            CREATE TABLE IF NOT EXISTS fr_extended_user_trade (
                sequence_id INT,
                timestamp_arrival BIGINT,
                timestamp_msg BIGINT,
                trade_id VARCHAR(100),
                account_id VARCHAR(100),
                market VARCHAR(20),
                order_id VARCHAR(100),
                external_order_id VARCHAR(100),
                side VARCHAR(20),
                price DOUBLE,
                qty DOUBLE,
                value DOUBLE,
                fee DOUBLE,
                trade_type VARCHAR(20),
                created_time_ts BIGINT,
                is_taker BOOL
            );
            """))
            await CON.commit()
        else:
            raise ValueError('Only MySQL engine is implemented')

### Balance Table ####
async def create_fr_extended_user_balance(
        CON: AsyncContextManager,
        engine: str = 'mysql', # mysql | duckdb
    ) -> None:
    if CON is None:
        logging.info("NO DB CONNECTION, SKIPPING Create Statements")
    else:
        if engine == 'mysql':
            logging.info('Creating Table if Does Not Exist: fr_extended_user_balance')
            await CON.execute(text("""
            CREATE TABLE IF NOT EXISTS fr_extended_user_balance (
                sequence_id INT,
                timestamp_arrival BIGINT,
                timestamp_msg BIGINT,
                collateral_name VARCHAR(20),
                balance DOUBLE,
                equity DOUBLE,
                available_for_trade DOUBLE,
                available_for_withdrawal DOUBLE,
                unrealised_pnl DOUBLE,
                initial_margin DOUBLE,
                margin_ratio DOUBLE,
                updated_time_ts BIGINT,
                exposure DOUBLE,
                leverage DOUBLE
            );
            """))
            await CON.commit()
        else:
            raise ValueError('Only MySQL engine is implemented')

### Balance Table ####
async def create_fr_extended_user_position(
        CON: AsyncContextManager,
        engine: str = 'mysql', # mysql | duckdb
    ) -> None:
    if CON is None:
        logging.info("NO DB CONNECTION, SKIPPING Create Statements")
    else:
        if engine == 'mysql':
            logging.info('Creating Table if Does Not Exist: fr_extended_user_position')
            await CON.execute(text("""
            CREATE TABLE IF NOT EXISTS fr_extended_user_position (
                sequence_id INT,
                timestamp_arrival BIGINT,
                timestamp_msg BIGINT,
                position_id VARCHAR(100),
                account_id VARCHAR(100),
                market VARCHAR(20),
                side VARCHAR(20),
                leverage DOUBLE,
                size DOUBLE,
                value DOUBLE,
                open_price DOUBLE,
                mark_price DOUBLE,
                liquidation_price DOUBLE,
                margin DOUBLE,
                unrealised_pnl DOUBLE,
                realised_pnl DOUBLE,
                tp_trigger_price DOUBLE,
                tp_limit_price DOUBLE,
                sl_trigger_price DOUBLE,
                sl_limit_price DOUBLE,
                adl_percentile INT,
                created_at_ts BIGINT,
                updated_at_ts BIGINT
            );
            """))
            await CON.commit()
        else:
            raise ValueError('Only MySQL engine is implemented')