trading_agent.gateway.routes.account_routes

Routes for the account: balances, positions, profit and loss, orders, ownership, halts and the kill switch.

GET /api/account the account now, each desk's profit and loss, ownership and halts GET /api/orders?kind=orders UBI's order book: orders, trades or open_parents GET /api/desk-orders each desk's open orders in the desk order book POST /api/account/kill halt every desk and flatten the account; needs {"confirm": "FLATTEN"}

Typical usage example:

application.include_router(AccountRoutes(guard, shared, desk_names).router())

KillRequest

Bases: BaseModel

The body of a kill-switch request.

Attributes:

Name	Type	Description
confirm	str	Must be exactly FLATTEN, so the switch cannot fire by accident.
reason	str	Why, for the ledger and the journal.

AccountRoutes

AccountRoutes(guard, shared, desk_names, clock=None)

Bases: RouteGroup

The account routes.

Creates the routes.

Parameters:

Name	Type	Description	Default
guard	AccessGuard	Checks the access token.	required
shared	SharedParts	The shared parts.	required
desk_names	Sequence[str]	The desks, for their order lists.	required
clock	Callable[[], datetime] | None	Returns the current time; None uses the system clock.	None

Source code in src/trading_agent/gateway/routes/account_routes.py

def __init__(
    self,
    guard: access_guard.AccessGuard,
    shared: gateway_parts.SharedParts,
    desk_names: Sequence[str],
    clock: Callable[[], datetime.datetime] | None = None,
):
    """Creates the routes.

    Args:
        guard (access_guard.AccessGuard): Checks the access token.
        shared (gateway_parts.SharedParts): The shared parts.
        desk_names (Sequence[str]): The desks, for their order lists.
        clock (Callable[[], datetime.datetime] | None): Returns the current time; None uses the system clock.
    """
    super().__init__(guard)
    self._shared = shared
    self._desk_names = list(desk_names)
    self._clock = clock

account async

account()

The account now.

Returns:

Type	Description
dict[str, Any]	Balance, margin, today's and this week's profit and loss, each desk's profit and loss, positions, ownership and halts.

Raises:

Type	Description
HTTPException	UBI cannot be read (status 503).

Source code in src/trading_agent/gateway/routes/account_routes.py

async def account(self) -> dict[str, Any]:
    """The account now.

    Returns:
        dict[str, Any]: Balance, margin, today's and this week's profit and loss, each desk's profit and loss, positions, ownership and halts.

    Raises:
        fastapi.HTTPException: UBI cannot be read (status 503).
    """
    try:
        snapshot = await self._shared.snapshot_reader.read()
    except ubi_exceptions.UnifiedBrokerInterfaceError as error:
        raise fastapi.HTTPException(
            status_code=503, detail=f'UBI cannot be read: {error}'
        ) from error
    today = self._now().date()
    positions = []
    symbols = {}
    for row in snapshot.positions:
        entry = dataclasses.asdict(row)
        entry['day_pnl'] = snapshot.day_pnl_by_instrument.get(
            row.instrument_id
        )
        positions.append(entry)
        symbols[row.instrument_id] = row.symbol
    return {
        'total_balance': snapshot.total_balance,
        'margin_utilized': snapshot.margin_utilized,
        'day_pnl': snapshot.day_pnl,
        'week_pnl': self._shared.ledger.week_pnl(today),
        'desk_pnl': self._shared.risk_engine.desk_day_pnl(snapshot),
        'positions': positions,
        'holdings': snapshot.held_quantities,
        'owners': self._owners(symbols),
        'halts': self._shared.ledger.halts_in_force(today),
    }

orders async

orders(kind='orders')

UBI's order book, as UBI serves it.

Parameters:

Name	Type	Description	Default
kind	Literal['orders', 'trades', 'open_parents']	Which book.	'orders'

Returns:

Type	Description
Any	UBI's document.

Raises:

Type	Description
HTTPException	UBI cannot be read (status 503).

Source code in src/trading_agent/gateway/routes/account_routes.py

async def orders(
    self,
    kind: Literal['orders', 'trades', 'open_parents'] = 'orders',
) -> Any:
    """UBI's order book, as UBI serves it.

    Args:
        kind (Literal['orders', 'trades', 'open_parents']): Which book.

    Returns:
        Any: UBI's document.

    Raises:
        fastapi.HTTPException: UBI cannot be read (status 503).
    """
    reader = self._shared.reader
    try:
        return await reader.call(
            functools.partial(reader.get, _ORDER_BOOK_PATHS[kind])
        )
    except ubi_exceptions.UnifiedBrokerInterfaceError as error:
        raise fastapi.HTTPException(
            status_code=503, detail=f'UBI cannot be read: {error}'
        ) from error

desk_orders

desk_orders()

Each desk's open orders in the desk order book, dry runs included.

Returns:

Type	Description
dict[str, Any]	The orders, keyed by desk name.

Source code in src/trading_agent/gateway/routes/account_routes.py

def desk_orders(self) -> dict[str, Any]:
    """Each desk's open orders in the desk order book, dry runs included.

    Returns:
        dict[str, Any]: The orders, keyed by desk name.
    """
    orders = {}
    for desk_name in self._desk_names:
        records = []
        for record in self._shared.order_book.open_orders(desk_name):
            records.append(dataclasses.asdict(record))
        orders[desk_name] = records
    return orders

kill async

kill(request)

Halts every desk until tomorrow and flattens the account.

In dry_run mode the flatten is a dry run, exactly as when the loss limit fires it.

Parameters:

Name	Type	Description	Default
request	KillRequest	The confirmation and the reason.	required

Returns:

Type	Description
dict[str, Any]	UBI's answer to the flatten, or a note that the switch had already fired.

Raises:

Type	Description
HTTPException	The confirmation is not FLATTEN (status 400).

Source code in src/trading_agent/gateway/routes/account_routes.py

async def kill(self, request: KillRequest) -> dict[str, Any]:
    """Halts every desk until tomorrow and flattens the account.

    In dry_run mode the flatten is a dry run, exactly as when the loss limit fires it.

    Args:
        request (KillRequest): The confirmation and the reason.

    Returns:
        dict[str, Any]: UBI's answer to the flatten, or a note that the switch had already fired.

    Raises:
        fastapi.HTTPException: The confirmation is not FLATTEN (status 400).
    """
    if request.confirm != _CONFIRMATION:
        raise fastapi.HTTPException(
            status_code=400,
            detail='Send {"confirm": "FLATTEN"} to fire the kill switch.',
        )
    now = self._now()
    return await self._shared.kill_switch.trigger(
        f'The user fired the kill switch: {request.reason}',
        now.date() + datetime.timedelta(days=1),
        now,
    )