trading_agent.tools.order_tool

The shallow base class of the tools that place, change, cancel and close orders.

Every order tool asks the risk engine before it sends anything that adds risk, tags each order with the desk's name, records the order in the desk order book, and publishes what it did on the event bus, where the journal keeps it with the model's stated reason. In dry_run mode every request goes to UBI as a dry run, which UBI checks and answers without sending anything to a broker.

Typical usage example:

class PlaceOrderTool(OrderTool): async def _act(self, arguments): ...

OrderServices dataclass

OrderServices(reader, risk_engine, order_book, snapshot_reader, ledger, event_bus, formatter)

What the order tools use, shared by every desk.

Attributes:

Name	Type	Description
reader	BrokerReader	The shared reader of UBI, whose lock every UBI call goes through.
risk_engine	RiskEngine	The engine every order that adds risk must pass.
order_book	DeskOrderBook	The record of which desk placed which order.
snapshot_reader	AccountSnapshotReader	Reads the account's positions, for closing a desk's positions.
ledger	RiskLedger	The risk ledger, for the instruments a desk owns.
event_bus	EventBus	Where every order action is published.
formatter	ResultFormatter	Turns UBI's answers into text.

OrderTool

OrderTool(services, desk, mode)

Bases: Tool

A tool that sends orders, or changes or cancels them, for one desk.

Creates the tool.

Parameters:

Name	Type	Description	Default
services	OrderServices	What the tool uses.	required
desk	str	The desk the tool works for.	required
mode	str	"dry_run" to send everything to UBI as a dry run, "live" to trade.	required

Source code in src/trading_agent/tools/order_tool.py

def __init__(self, services: OrderServices, desk: str, mode: str):
    """Creates the tool.

    Args:
        services (OrderServices): What the tool uses.
        desk (str): The desk the tool works for.
        mode (str): "dry_run" to send everything to UBI as a dry run, "live" to trade.
    """
    self._services = services
    self._desk = desk
    self._dry_run = mode != 'live'
    tag_characters = []
    for character in desk:
        if character.isalnum():
            tag_characters.append(character)
    self._tag = ''.join(tag_characters)[:_TAG_LENGTH]

run async

run(arguments)

Does the tool's work, and explains any failure in words the model can act on.

Parameters:

Name	Type	Description	Default
arguments	BaseModel	The checked arguments.	required

Returns:

Type	Description
ToolResult	What happened, or why it could not happen.

Source code in src/trading_agent/tools/order_tool.py

async def run(self, arguments: pydantic.BaseModel) -> tool.ToolResult:
    """Does the tool's work, and explains any failure in words the model can act on.

    Args:
        arguments (pydantic.BaseModel): The checked arguments.

    Returns:
        tool.ToolResult: What happened, or why it could not happen.
    """
    try:
        return await self._act(arguments)
    except ubi_exceptions.OrderOutcomeUnknownError as error:
        return tool.ToolResult.error(
            f'UBI did not answer in time, so the request may still have gone through: {error}. Check get_orders before trying again, or the order may be sent twice.'
        )
    except instrument_exceptions.InstrumentError as error:
        return tool.ToolResult.error(
            f'Instrument problem: {error}. Check the fields, or find the instrument with search_instruments.'
        )
    except ubi_exceptions.UnifiedBrokerInterfaceError as error:
        return tool.ToolResult.error(
            f'UBI refused or failed the request: {type(error).__name__}: {error}'
        )