Skip to content

trading_agent.trading.broker_reader

The one place trading-agent reads from UBI, through tradingmachine.

tradingmachine is synchronous, so every call runs in a worker thread and the desks keep working while UBI answers. A lock lets one UBI call run at a time, as the tradingmachine_mcp server does, because UBI holds a single access token and broker rate limits are shared.

The client is built from tradingmachine's own .env, named in trading_agent.toml. Left to itself, tradingmachine would look for a .env in the working folder and find trading-agent's, which has no UBI settings.

Typical usage example:

reader = BrokerReader(Path('/home/pramod/Projects/tradingmachine/.env')) quote = await reader.call(lambda: reader.instrument(arguments).quote)

BrokerReader 

BrokerReader(environment_file)

Reads instruments, prices, the portfolio and the order book from UBI, and hands out the instruments the order tools trade.

Remembers where tradingmachine's settings are, without connecting.

Parameters:

Name Type Description Default
environment_file Path

tradingmachine's .env file.

 required
Source code in src/trading_agent/trading/broker_reader.py 
35
36
37
38
39
40
41
42
43
def __init__(self, environment_file: Path):
    """Remembers where tradingmachine's settings are, without connecting.

    Args:
        environment_file (Path): tradingmachine's `.env` file.
    """
    self._environment_file = environment_file
    self._client = None
    self._lock = threading.Lock()

call async 

call(work)

Runs one piece of UBI work in a worker thread, one at a time.

Parameters:

Name Type Description Default
work Callable[[], _Result]

The work, such as reading a quote.

 required

Returns:

Type Description
_Result

What the work returned.

Raises:

Type Description
UnifiedBrokerInterfaceError

UBI refused or failed the request.
InstrumentError

The instrument does not exist.
Source code in src/trading_agent/trading/broker_reader.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
async def call(self, work: Callable[[], _Result]) -> _Result:
    """Runs one piece of UBI work in a worker thread, one at a time.

    Args:
        work (Callable[[], _Result]): The work, such as reading a quote.

    Returns:
        _Result: What the work returned.

    Raises:
        tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI refused or failed the request.
        tradingmachine.assets.exceptions.InstrumentError: The instrument does not exist.
    """
    return await asyncio.to_thread(self._locked, work)

get 

get(path, parameters=None)

Sends a GET request to one of UBI's read routes. Call it inside call().

Parameters:

Name Type Description Default
path str

The route, such as /api/portfolio/funds.

 required
parameters dict[str, Any] | None

The query parameters, or None.

 None

Returns:

Type Description
Any

UBI's decoded JSON answer.
Source code in src/trading_agent/trading/broker_reader.py 
60
61
62
63
64
65
66
67
68
69
70
def get(self, path: str, parameters: dict[str, Any] | None = None) -> Any:
    """Sends a GET request to one of UBI's read routes. Call it inside call().

    Args:
        path (str): The route, such as /api/portfolio/funds.
        parameters (dict[str, Any] | None): The query parameters, or None.

    Returns:
        Any: UBI's decoded JSON answer.
    """
    return self._shared_client().get(path, params=parameters)

client 

client()

The shared UBI client, for tradingmachine objects that need one, such as Account. Call it inside call().

Returns:

Type Description
UnifiedBrokerInterface

The client.
Source code in src/trading_agent/trading/broker_reader.py 
72
73
74
75
76
77
78
def client(self) -> ubi_client.UnifiedBrokerInterface:
    """The shared UBI client, for tradingmachine objects that need one, such as Account. Call it inside call().

    Returns:
        ubi_client.UnifiedBrokerInterface: The client.
    """
    return self._shared_client()

instrument 

instrument(arguments)

Looks an instrument up as a plain Instrument, which has no order methods. Call it inside call().

Parameters:

Name Type Description Default
arguments InstrumentArguments

How the model named the instrument.

 required

Returns:

Type Description
Instrument

The instrument.
Source code in src/trading_agent/trading/broker_reader.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def instrument(
    self,
    arguments: instrument_arguments_module.InstrumentArguments,
) -> instruments.Instrument:
    """Looks an instrument up as a plain Instrument, which has no order methods. Call it inside call().

    Args:
        arguments (instrument_arguments_module.InstrumentArguments): How the model named the instrument.

    Returns:
        instruments.Instrument: The instrument.
    """
    shared_client = self._shared_client()
    if arguments.instrument_id is not None:
        return instruments.Instrument(
            instrument_id=arguments.instrument_id,
            unified_broker_interface=shared_client,
        )
    return instruments.Instrument(
        exchange=arguments.exchange,
        segment=arguments.segment,
        symbol=arguments.symbol,
        underlying_symbol=arguments.underlying_symbol,
        expiry_date=arguments.expiry_date,
        strike_price=arguments.strike_price,
        option_type=arguments.option_type,
        unified_broker_interface=shared_client,
    )

tradeable_instrument 

tradeable_instrument(arguments)

Looks an instrument up as a TradeableInstrument, which has the order methods. Call it inside call().

Parameters:

Name Type Description Default
arguments InstrumentArguments

How the model named the instrument.

 required

Returns:

Type Description
TradeableInstrument

The instrument.

Raises:

Type Description
TradeableInstrumentError

The instrument is an index, which cannot be traded.
Source code in src/trading_agent/trading/broker_reader.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def tradeable_instrument(
    self,
    arguments: instrument_arguments_module.InstrumentArguments,
) -> instruments.TradeableInstrument:
    """Looks an instrument up as a TradeableInstrument, which has the order methods. Call it inside call().

    Args:
        arguments (instrument_arguments_module.InstrumentArguments): How the model named the instrument.

    Returns:
        instruments.TradeableInstrument: The instrument.

    Raises:
        tradingmachine.assets.exceptions.TradeableInstrumentError: The instrument is an index, which cannot be traded.
    """
    shared_client = self._shared_client()
    if arguments.instrument_id is not None:
        return instruments.TradeableInstrument(
            instrument_id=arguments.instrument_id,
            unified_broker_interface=shared_client,
        )
    return instruments.TradeableInstrument(
        exchange=arguments.exchange,
        segment=arguments.segment,
        symbol=arguments.symbol,
        underlying_symbol=arguments.underlying_symbol,
        expiry_date=arguments.expiry_date,
        strike_price=arguments.strike_price,
        option_type=arguments.option_type,
        unified_broker_interface=shared_client,
    )