trading_agent.risk.risk_engine

The risk engine: plain code that every order must pass before it reaches a broker.

No model can change it, talk to it or skip it. Before each order it reads the limits from TradingRules.md and the account from UBI, checks the loss limits and fires the kill switch when one is reached, and then checks the order against every rule, collecting every reason it fails so the model can fix the order in one step.

Two principles shape it. When the limits or the account cannot be read, the answer is no. And an order that only reduces or closes a desk's own position is never blocked by a limit or a halt, because getting out must always be possible.

Typical usage example:

decision = await engine.evaluate(order_request) if decision.approved: ...

RiskDecision dataclass

RiskDecision(approved, reasons, estimated_margin)

The engine's answer for one order.

Attributes:

Name	Type	Description
approved	bool	Whether the order may be sent.
reasons	tuple[str, ...]	Why it was refused; empty when approved.
estimated_margin	float	The margin the order is estimated to block, in rupees.

RiskServices dataclass

RiskServices(limits_reader, snapshot_reader, ledger, kill_switch, margin_estimator, exchange_hours)

The parts the engine reads from and writes to.

Attributes:

Name	Type	Description
limits_reader	TradingLimitsReader	Reads the limits from TradingRules.md.
snapshot_reader	AccountSnapshotReader	Reads the account from UBI.
ledger	RiskLedger	The engine's memory.
kill_switch	KillSwitch	Halts and flattens the account.
margin_estimator	MarginEstimator	Estimates an order's margin.
exchange_hours	ExchangeHours	Says whether an exchange is open.

RiskEngine

RiskEngine(services, event_bus, ownership_grace, clock=None)

Checks every order against the limits and the account.

Creates the engine.

Parameters:

Name	Type	Description	Default
services	RiskServices	What the engine reads and writes.	required
event_bus	EventBus	Where every decision is published.	required
ownership_grace	timedelta	How long a desk keeps a claim on an instrument it does not hold yet.	required
clock	Callable[[], datetime] | None	The source of the current time, replaceable in tests; None for the real clock.	None

Source code in src/trading_agent/risk/risk_engine.py

def __init__(
    self,
    services: RiskServices,
    event_bus: event_bus_module.EventBus,
    ownership_grace: datetime.timedelta,
    clock: Callable[[], datetime.datetime] | None = None,
):
    """Creates the engine.

    Args:
        services (RiskServices): What the engine reads and writes.
        event_bus (event_bus_module.EventBus): Where every decision is published.
        ownership_grace (datetime.timedelta): How long a desk keeps a claim on an instrument it does not hold yet.
        clock (Callable[[], datetime.datetime] | None): The source of the current time, replaceable in tests; None for the real clock.
    """
    self._services = services
    self._event_bus = event_bus
    self._ownership_grace = ownership_grace
    if clock is None:
        clock = self._real_now
    self._clock = clock

evaluate async

evaluate(order)

Decides whether one order may be sent, and records an approval.

Parameters:

Name	Type	Description	Default
order	OrderRequest	The order.	required

Returns:

Type	Description
RiskDecision	The decision, with every reason when refused.

Source code in src/trading_agent/risk/risk_engine.py

async def evaluate(
    self,
    order: order_request_module.OrderRequest,
) -> RiskDecision:
    """Decides whether one order may be sent, and records an approval.

    Args:
        order (order_request_module.OrderRequest): The order.

    Returns:
        RiskDecision: The decision, with every reason when refused.
    """
    now = self._clock().astimezone(_INDIA)
    reasons = []
    limits_error = None
    try:
        limits = self._services.limits_reader.read()
    except trading_limits_module.LimitsError as error:
        limits = None
        limits_error = f'The trading rules cannot be read, so nothing new can be opened: {error}.'
    try:
        snapshot = await self._services.snapshot_reader.read()
    except ubi_exceptions.UnifiedBrokerInterfaceError as error:
        return self._finish(
            order,
            [
                f'The account cannot be read from UBI right now ({type(error).__name__}), so no order can be checked.',
            ],
            0.0,
        )
    ledger = self._services.ledger
    ledger.record_day_pnl(now.date(), snapshot.day_pnl)
    ledger.release_unheld(
        snapshot.held_instrument_ids(), now, self._ownership_grace
    )
    if limits is not None:
        await self._check_loss_limits(limits.account, snapshot, now)
        self._check_desk_loss_limits(limits, snapshot, now)
    margin = self._services.margin_estimator.margin(order)
    reducing = self._reduces_own_position(order, snapshot)
    reasons.extend(self._market_reasons(order, limits, now))
    reasons.extend(self._ownership_reasons(order, snapshot))
    if not reducing and limits_error is not None:
        reasons.append(limits_error)
    if not reducing and limits is not None:
        reasons.extend(
            self._opening_reasons(order, limits, snapshot, margin, now)
        )
    if not reasons:
        ledger.claim(order.instrument_id, order.desk, now)
        ledger.record_approval(order.desk, now)
    return self._finish(order, reasons, margin)

check_account async

check_account()

Reads the account and fires the kill switch at a loss limit, without an order.

The market watcher calls this every cycle, because a losing position can reach the daily or weekly loss limit while no desk is placing orders.

Returns:

Type	Description
AccountSnapshot | None	The account now, or None when UBI cannot be read.

Source code in src/trading_agent/risk/risk_engine.py

async def check_account(
    self,
) -> account_snapshot_module.AccountSnapshot | None:
    """Reads the account and fires the kill switch at a loss limit, without an order.

    The market watcher calls this every cycle, because a losing position can reach the daily or weekly loss limit while no desk is placing orders.

    Returns:
        account_snapshot_module.AccountSnapshot | None: The account now, or None when UBI cannot be read.
    """
    now = self._clock().astimezone(_INDIA)
    try:
        snapshot = await self._services.snapshot_reader.read()
    except ubi_exceptions.UnifiedBrokerInterfaceError:
        _LOGGER.warning(
            'The account could not be read for the loss-limit check'
        )
        return None
    self._services.ledger.record_day_pnl(now.date(), snapshot.day_pnl)
    try:
        limits = self._services.limits_reader.read()
    except trading_limits_module.LimitsError:
        return snapshot
    await self._check_loss_limits(limits.account, snapshot, now)
    self._check_desk_loss_limits(limits, snapshot, now)
    return snapshot

desk_day_pnl

desk_day_pnl(snapshot)

Today's profit or loss of each desk, from the instruments it owned today.

An instrument counts for the desk that last claimed it today, even after the position was closed and the claim released, so a loss taken on a closed trade still counts against the desk that took it. Positions no desk claimed, such as the user's own, count for no desk.

Parameters:

Name	Type	Description	Default
snapshot	AccountSnapshot	The account now.	required

Returns:

Type	Description
dict[str, float]	Each desk's profit or loss, keyed by desk name; desks with nothing today are absent.

Source code in src/trading_agent/risk/risk_engine.py

def desk_day_pnl(
    self,
    snapshot: account_snapshot_module.AccountSnapshot,
) -> dict[str, float]:
    """Today's profit or loss of each desk, from the instruments it owned today.

    An instrument counts for the desk that last claimed it today, even after the position was closed and the claim released, so a loss taken on a closed trade still counts against the desk that took it. Positions no desk claimed, such as the user's own, count for no desk.

    Args:
        snapshot (account_snapshot_module.AccountSnapshot): The account now.

    Returns:
        dict[str, float]: Each desk's profit or loss, keyed by desk name; desks with nothing today are absent.
    """
    today = self._clock().astimezone(_INDIA).date()
    owners = self._services.ledger.day_owners(today)
    pnl = {}
    for instrument_id, amount in snapshot.day_pnl_by_instrument.items():
        desk = owners.get(instrument_id)
        if desk is None:
            continue
        pnl[desk] = pnl.get(desk, 0.0) + amount
    return pnl