Skip to content

trading_agent.risk.account_snapshot

The state of the account that the risk engine needs, read fresh from UBI before each order.

Typical usage example:

snapshot = await AccountSnapshotReader(broker_reader).read() snapshot.net_quantity('f310e1b0-...')

PositionRow dataclass 

PositionRow(instrument_id, symbol, segment, product, quantity, last_price)

One open position, in one instrument and product, merged across brokers.

Attributes:

Name Type Description
instrument_id str

UBI's id for the instrument.
symbol str

The instrument's symbol.
segment str

UBI's segment.
product str

UBI's position product, such as intraday or delivery.
quantity int

Positive when long, negative when short.
last_price float

The latest price.

AccountSnapshot dataclass 

AccountSnapshot(total_balance, margin_utilized, day_pnl, positions, held_quantities=dict(), held_symbols=frozenset(), day_pnl_by_instrument=dict())

Balances, today's profit and loss, and open positions.

Attributes:

Name Type Description
total_balance float

The total balance across every broker: available balance plus margin in use.
margin_utilized float

Margin already blocked across every broker.
day_pnl float

Today's realised plus unrealised profit or loss, negative for a loss.
positions tuple[PositionRow, ...]

The open positions.
held_quantities dict[str, int]

The quantity held as delivery holdings, by instrument id.
held_symbols frozenset[str]

The upper-case symbols of every delivery holding, including holdings UBI gives no instrument id.
day_pnl_by_instrument dict[str, float]

Today's profit or loss of each position in UBI's day view, closed ones included, by instrument id.

net_quantity 

net_quantity(instrument_id)

The account's total position in one instrument, across products.

Parameters:

Name Type Description Default
instrument_id str

UBI's id for the instrument.

 required

Returns:

Type Description
int

Positive when long, negative when short, zero when flat.
Source code in src/trading_agent/risk/account_snapshot.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def net_quantity(self, instrument_id: str) -> int:
    """The account's total position in one instrument, across products.

    Args:
        instrument_id (str): UBI's id for the instrument.

    Returns:
        int: Positive when long, negative when short, zero when flat.
    """
    total = 0
    for row in self.positions:
        if row.instrument_id == instrument_id:
            total += row.quantity
    return total

held_instrument_ids 

held_instrument_ids()

The instruments with a non-zero position or a holding.

Returns:

Type Description
set[str]

Their ids.
Source code in src/trading_agent/risk/account_snapshot.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def held_instrument_ids(self) -> set[str]:
    """The instruments with a non-zero position or a holding.

    Returns:
        set[str]: Their ids.
    """
    held = set()
    for row in self.positions:
        if row.quantity != 0:
            held.add(row.instrument_id)
    for instrument_id, quantity in self.held_quantities.items():
        if quantity != 0:
            held.add(instrument_id)
    return held

AccountSnapshotReader 

AccountSnapshotReader(reader)

Reads an AccountSnapshot from UBI.

Creates the reader.

Parameters:

Name Type Description Default
reader BrokerReader

The shared reader of UBI.

 required
Source code in src/trading_agent/risk/account_snapshot.py 
 98
 99
100
101
102
103
104
def __init__(self, reader: broker_reader_module.BrokerReader):
    """Creates the reader.

    Args:
        reader (broker_reader_module.BrokerReader): The shared reader of UBI.
    """
    self._reader = reader

read async 

read()

Reads funds and positions.

Returns:

Type Description
AccountSnapshot

The account now.

Raises:

Type Description
UnifiedBrokerInterfaceError

UBI could not answer.
Source code in src/trading_agent/risk/account_snapshot.py 
106
107
108
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
140
141
142
143
144
145
146
async def read(self) -> AccountSnapshot:
    """Reads funds and positions.

    Returns:
        AccountSnapshot: The account now.

    Raises:
        tradingmachine.unified_broker_interface.exceptions.UnifiedBrokerInterfaceError: UBI could not answer.
    """
    funds = await self._reader.call(self._read_funds)
    positions = await self._reader.call(self._read_positions)
    holdings = await self._reader.call(self._read_holdings)
    summary = funds['summary']
    rows = []
    for row in positions.get('net', []):
        rows.append(self._position_row(row))
    held_quantities = {}
    held_symbols = set()
    for row in holdings.get('holdings', []):
        if row.get('symbol'):
            held_symbols.add(str(row['symbol']).upper())
        instrument_id = row.get('instrument_id')
        if instrument_id:
            held_quantities[str(instrument_id)] = int(
                row.get('quantity') or 0
            )
    day_pnl_by_instrument = {}
    for row in positions.get('day', []):
        instrument_id = str(row.get('instrument_id') or row.get('symbol'))
        day_pnl_by_instrument[instrument_id] = day_pnl_by_instrument.get(
            instrument_id, 0.0
        ) + self._row_pnl(row)
    return AccountSnapshot(
        total_balance=float(summary['total_balance']),
        margin_utilized=float(summary.get('margin_utilized') or 0),
        day_pnl=float(positions['summary']['day']['total_pnl']),
        positions=tuple(rows),
        held_quantities=held_quantities,
        held_symbols=frozenset(held_symbols),
        day_pnl_by_instrument=day_pnl_by_instrument,
    )