BotRepository API Reference

tradingbot.utils.bot_repository.BotRepository

Handles database operations for Bot entities.

create_or_get_bot(name: str, session: Optional[Session] = None) -> BotModel staticmethod

Create or retrieve bot from database.

Parameters:

Name	Type	Description	Default
name	str	Bot name	required
session	Optional[Session]	Optional existing database session	None

Returns:

Type	Description
Bot	BotModel instance

Source code in tradingbot/utils/bot_repository.py

@staticmethod
def create_or_get_bot(name: str, session: Optional[Session] = None) -> BotModel:
    """
    Create or retrieve bot from database.

    Args:
        name: Bot name
        session: Optional existing database session

    Returns:
        BotModel instance
    """
    def _get_or_create(sess: Session):
        bot = sess.query(BotModel).filter_by(name=name).first()
        if not bot:
            bot = BotModel(name=name)
            sess.add(bot)
            sess.flush()
            sess.refresh(bot)
        _ = bot.portfolio
        return bot

    if session:
        return _get_or_create(session)

    with get_db_session() as session:
        bot = _get_or_create(session)
        session.expunge(bot)
        return bot

get_bot_locked(session: Session, name: str) -> BotModel staticmethod

Get a bot by name with a row-level lock (FOR UPDATE). MUST be called within an active transaction.

Parameters:

Name	Type	Description	Default
session	Session	Active database session	required
name	str	Bot name	required

Returns:

Type	Description
Bot	Bot model instance

Source code in tradingbot/utils/bot_repository.py

@staticmethod
def get_bot_locked(session: Session, name: str) -> BotModel:
    """
    Get a bot by name with a row-level lock (FOR UPDATE).
    MUST be called within an active transaction.

    Args:
        session: Active database session
        name: Bot name

    Returns:
        Bot model instance
    """
    return session.query(BotModel).filter_by(name=name).with_for_update().one()

log_trade(bot_name: str, symbol: str, quantity: float, price: float, is_buy: bool, profit: Optional[float] = None, session: Optional[Session] = None) -> Trade staticmethod

Log a trade to the database.

Parameters:

Name	Type	Description	Default
bot_name	str	Name of the bot executing the trade	required
symbol	str	Trading symbol	required
quantity	float	Number of shares/units	required
price	float	Price per unit	required
is_buy	bool	True for buy, False for sell	required
profit	Optional[float]	Profit from the trade (for sells)	None
session	Optional[Session]	Optional existing database session	None

Returns:

Type	Description
Trade	Created Trade object

Source code in tradingbot/utils/bot_repository.py

@staticmethod
def log_trade(
    bot_name: str,
    symbol: str,
    quantity: float,
    price: float,
    is_buy: bool,
    profit: Optional[float] = None,
    session: Optional[Session] = None,
) -> Trade:
    """
    Log a trade to the database.

    Args:
        bot_name: Name of the bot executing the trade
        symbol: Trading symbol
        quantity: Number of shares/units
        price: Price per unit
        is_buy: True for buy, False for sell
        profit: Profit from the trade (for sells)
        session: Optional existing database session

    Returns:
        Created Trade object
    """
    def _create_trade(sess: Session):
        trade = Trade(
            bot_name=bot_name,
            symbol=symbol,
            isBuy=is_buy,
            quantity=float(quantity),
            price=float(price),
            timestamp=datetime.utcnow(),
            profit=float(profit) if profit is not None else None,
        )
        sess.add(trade)
        sess.flush()
        sess.refresh(trade)
        return trade

    if session:
        return _create_trade(session)

    with get_db_session() as session:
        return _create_trade(session)

update_bot(bot: BotModel, session: Optional[Session] = None) -> BotModel staticmethod

Update bot state in database.

Parameters:

Name	Type	Description	Default
bot	Bot	BotModel instance to update	required
session	Optional[Session]	Optional existing database session	None

Returns:

Type	Description
Bot	Updated BotModel instance

Source code in tradingbot/utils/bot_repository.py

@staticmethod
def update_bot(bot: BotModel, session: Optional[Session] = None) -> BotModel:
    """
    Update bot state in database.

    Args:
        bot: BotModel instance to update
        session: Optional existing database session

    Returns:
        Updated BotModel instance
    """
    if session:
        session.add(bot)
        session.flush()
        return bot

    with get_db_session() as session:
        session.merge(bot)
        return bot