Database API Reference

Models

tradingbot.utils.db.Bot

Bases: Base

Bot model representing a trading bot instance.

Attributes:

Name	Type	Description
name		Unique bot name (primary key)
description		Optional description of the bot
portfolio		JSON dictionary representing portfolio holdings (default: {"USD": 10000}) Format: {"USD": cash_amount, "SYMBOL": quantity, ...}
created_at		Timestamp when bot was created
updated_at		Timestamp when bot was last updated

tradingbot.utils.db.Trade

Bases: Base

Trade model representing a single trade execution.

Attributes:

Name	Type	Description
id		Auto-incrementing trade ID (primary key)
bot_name		Name of the bot that executed the trade (foreign key to Bot.name)
symbol		Trading symbol (e.g., "QQQ", "EURUSD=X")
isBuy		True for buy orders, False for sell orders
quantity		Number of shares/units traded
price		Price per unit at time of trade
timestamp		Timestamp when trade was executed
profit		Profit from the trade (for sells, nullable)

tradingbot.utils.db.HistoricData

Bases: Base

Historic market data model for storing OHLCV data.

Attributes:

Name	Type	Description
symbol		Trading symbol (primary key, part of composite key)
timestamp		Timestamp of the data point (primary key, part of composite key)
open		Opening price
high		Highest price
low		Lowest price
close		Closing price
volume		Trading volume

tradingbot.utils.db.RunLog

Bases: Base

Run log model for tracking bot execution history.

Attributes:

Name	Type	Description
id		Auto-incrementing log ID (primary key)
bot_name		Name of the bot (foreign key to Bot.name)
start_time		Timestamp when the run started
success		Whether the run completed successfully
result		Result message (nullable, contains decision/error info)

tradingbot.utils.db.PortfolioWorth

Bases: Base

Portfolio worth model for tracking portfolio value over time.

Attributes:

Name	Type	Description
bot_name		Name of the bot (primary key, part of composite key, foreign key to Bot.name)
date		Date of the portfolio valuation (primary key, part of composite key)
portfolio_worth		Total portfolio value in USD
holdings		JSON dictionary of holdings at this date
created_at		Timestamp when this record was created

Session Management

tradingbot.utils.db.get_db_session(max_retries: int = 3, retry_delay: float = 1.0) -> Generator[Session, None, None]

Context manager for database sessions with retry logic.

Ensures proper session cleanup and rollback on exceptions. Automatically retries on transient connection errors.

Parameters:

Name	Type	Description	Default
max_retries	int	Maximum number of retry attempts (default: 3)	3
retry_delay	float	Initial delay between retries in seconds (default: 1.0)	1.0

Usage

with get_db_session() as session: # Use session here session.query(Bot).all()

Source code in tradingbot/utils/db.py

@contextmanager
def get_db_session(max_retries: int = 3, retry_delay: float = 1.0) -> Generator[Session, None, None]:
    """
    Context manager for database sessions with retry logic.

    Ensures proper session cleanup and rollback on exceptions.
    Automatically retries on transient connection errors.

    Args:
        max_retries: Maximum number of retry attempts (default: 3)
        retry_delay: Initial delay between retries in seconds (default: 1.0)

    Usage:
        with get_db_session() as session:
            # Use session here
            session.query(Bot).all()
    """
    session = None
    last_exception = None

    for attempt in range(max_retries):
        try:
            session = SessionLocal()
            yield session
            session.commit()
            return  # Success, exit retry loop
        except (OperationalError, SQLAlchemyError) as e:
            # These are potentially transient errors that might benefit from retry
            if session:
                try:
                    session.rollback()
                except Exception:
                    pass  # Ignore rollback errors during retry
                finally:
                    try:
                        session.close()
                    except Exception:
                        pass  # Ignore close errors during retry

            last_exception = e
            is_last_attempt = attempt == max_retries - 1

            if is_last_attempt:
                logger.error(
                    f"Database error after {max_retries} attempts: "
                    f"{type(e).__name__}: {e}"
                )
                raise
            else:
                # Exponential backoff: delay increases with each retry
                delay = retry_delay * (2 ** attempt)
                logger.warning(
                    f"Database error (attempt {attempt + 1}/{max_retries}): "
                    f"{type(e).__name__}: {e}. Retrying in {delay:.2f}s..."
                )
                time.sleep(delay)
        except Exception as e:
            # Non-retryable errors - close session and raise immediately
            if session:
                try:
                    session.rollback()
                except Exception:
                    pass
                finally:
                    try:
                        session.close()
                    except Exception:
                        pass
            logger.error(f"Unexpected error in database session: {type(e).__name__}: {e}")
            raise

    # Should never reach here, but just in case
    if last_exception:
        raise last_exception