trading_agent.terminal.event_describer

Turns gateway events into short sentences for the activity feed and the chat.

It is a port of the web page's eventDescriber.ts, so the terminal and the browser describe events in the same words.

Typical usage example:

description = EventDescriber().describe(event) description.title # 'Risk engine refused: BUY 10 RELIANCE'

EventDescription dataclass

EventDescription(title, detail, tone)

One event in words.

Attributes:

Name	Type	Description
title	str	A short title.
detail	str	A longer detail, or an empty string.
tone	str	How much attention it deserves: "normal", "quiet", "good", "warning" or "bad".

EventDescriber

EventDescriber()

Describes events by their kind and payload.

Creates the describer.

Source code in src/trading_agent/terminal/event_describer.py

def __init__(self):
    """Creates the describer."""
    self._formatter = formatter_module.Formatter()

describe

describe(event)

Describes one event.

Parameters:

Name	Type	Description	Default
event	dict[str, Any]	The event, with kind, desk, payload and created_at.	required

Returns:

Type	Description
EventDescription	Its title, detail and tone.

Source code in src/trading_agent/terminal/event_describer.py

def describe(self, event: dict[str, Any]) -> EventDescription:
    """Describes one event.

    Args:
        event (dict[str, Any]): The event, with kind, desk, payload and created_at.

    Returns:
        EventDescription: Its title, detail and tone.
    """
    kind = event.get('kind', '')
    payload = event.get('payload') or {}
    if kind == 'trigger_received':
        return self._make(
            self.source_words(payload.get('source')),
            self._text(payload.get('text')),
            'normal',
        )
    if kind == 'turn_started':
        return self._make('Turn started', '', 'quiet')
    if kind == 'turn_finished':
        return self._make('Turn finished', '', 'quiet')
    if kind == 'text':
        if payload.get('silent') is True:
            return self._make(
                'Nothing to do', self._text(payload.get('text')), 'quiet'
            )
        return self._make(
            'Answered', self._text(payload.get('text')), 'normal'
        )
    if kind == 'tool_call':
        return self._make(
            f'Calls {self._text(payload.get("name"))}',
            self._json(payload.get('arguments')),
            'normal',
        )
    if kind == 'tool_result':
        name = self._text(payload.get('name'))
        if payload.get('is_error') is True:
            return self._make(
                f'{name} failed', self._text(payload.get('text')), 'warning'
            )
        return self._make(
            f'{name} answered', self._text(payload.get('text')), 'quiet'
        )
    if kind == 'model_usage':
        prompt = self._formatter.number(payload.get('prompt_tokens'))
        answer = self._formatter.number(payload.get('completion_tokens'))
        return self._make(
            f'{self._text(payload.get("backend"))} used {prompt} prompt and {answer} answer tokens',
            '',
            'quiet',
        )
    if kind == 'rate_limit':
        return self._make(
            f'{self._text(payload.get("backend"))} {self._text(payload.get("limit"))} limit: {self._text(payload.get("status"))}',
            '',
            'quiet',
        )
    if kind == 'backend_unavailable':
        waiting = self._formatter.number(payload.get('waiting_seconds'))
        return self._make(
            f'{self._text(payload.get("backend"))} is unavailable; retrying in {waiting} s',
            self._text(payload.get('reason')),
            'warning',
        )
    if kind == 'backend_switch':
        if payload.get('from') is None:
            return self._make(
                f'Model chosen: {self._text(payload.get("to"))}',
                '',
                'quiet',
            )
        return self._make(
            f'Model switched: {self._text(payload.get("from"))} → {self._text(payload.get("to"))}',
            '',
            'warning',
        )
    if kind == 'risk_decision':
        return self._risk_decision(payload)
    if kind == 'kill_switch':
        if payload.get('stage') == 'fired':
            return self._make(
                'Kill switch fired',
                self._text(payload.get('reason')),
                'bad',
            )
        return self._make(
            'Kill switch: flatten finished',
            self._json(payload.get('outcome')),
            'bad',
        )
    if kind == 'order':
        return self._order(payload)
    if kind == 'fill':
        return self._fill(payload)
    if kind == 'price_alert':
        return self._price_alert(payload)
    if kind == 'desk_halt':
        return self._make(
            'Desk halted at its loss limit',
            self._text(payload.get('reason')),
            'bad',
        )
    if kind == 'account_status':
        return self._make(
            f'Account {self._formatter.money(payload.get("total_balance"))}, today {self._formatter.signed_money(payload.get("day_pnl"))}',
            '',
            'quiet',
        )
    if kind == 'session_started':
        session = self._text(payload.get('session_id'))
        if payload.get('reason') == 'resumed':
            return self._make(
                f'Session {session} resumed after a restart', '', 'quiet'
            )
        return self._make(
            f'Session {session} started',
            self._text(payload.get('reason')).replace('_', ' '),
            'normal',
        )
    if kind == 'context_compressed':
        before = self._formatter.number(payload.get('tokens_before'))
        summarised = self._formatter.number(
            payload.get('messages_summarised')
        )
        return self._make(
            f'Conversation compressed from {before} tokens',
            f'{summarised} messages summarised',
            'normal',
        )
    if kind == 'subagent_started':
        return self._make(
            f'Subagent {self._text(payload.get("subagent"))} started',
            self._text(payload.get('task')),
            'normal',
        )
    if kind == 'subagent_finished':
        seconds = self._formatter.number(payload.get('seconds'))
        return self._make(
            f'Subagent {self._text(payload.get("subagent"))} {self._text(payload.get("outcome"))} in {seconds} s',
            '',
            'normal',
        )
    if kind == 'cron_fired':
        detail = ''
        if payload.get('catching_up') is True:
            detail = 'catching up a missed run'
        return self._make(
            f'Scheduled job: {self._text(payload.get("name"))}',
            detail,
            'normal',
        )
    if kind == 'cron_result':
        return self._make(
            f'Job {self._text(payload.get("name"))}: {self._text(payload.get("outcome"))}',
            self._text(payload.get('text')),
            'normal',
        )
    if kind == 'message':
        if payload.get('urgent') is True:
            return self._make(
                'Urgent message', self._text(payload.get('text')), 'warning'
            )
        return self._make(
            'Message', self._text(payload.get('text')), 'normal'
        )
    if kind == 'error':
        return self._make(
            'The turn failed', self._text(payload.get('error')), 'bad'
        )
    if kind == 'steering':
        if payload.get('delivered') is True:
            return self._make(
                'Your message was handed to the running turn', '', 'good'
            )
        return self._make(
            'The turn ended before your message reached it, so it is the next turn',
            '',
            'normal',
        )
    if kind == 'channel':
        return self._channel(payload)
    if kind == 'user_request':
        return self._user_request(payload)
    if kind == 'user_response':
        title = f'You answered a {self._text(payload.get("kind"))}'
        if payload.get('status') == 'dismissed':
            title = f'You dismissed a {self._text(payload.get("kind"))}'
        return self._make(title, self._text(payload.get('title')), 'good')
    if kind == 'stopped':
        return self._make('Turn stopped by you', '', 'warning')
    if kind == 'desk_started':
        return self._make('Desk started', '', 'quiet')
    if kind == 'desk_stopped':
        return self._make('Desk stopped', '', 'warning')
    tone = 'normal'
    if kind in _QUIET_KINDS:
        tone = 'quiet'
    return self._make(kind.replace('_', ' '), self._json(payload), tone)

source_words

source_words(source)

Names what woke a desk.

Parameters:

Name	Type	Description	Default
source	object	The trigger's source, such as "user_message".	required

Returns:

Type	Description
str	Plain words, such as "Message from you".

Source code in src/trading_agent/terminal/event_describer.py

def source_words(self, source: object) -> str:
    """Names what woke a desk.

    Args:
        source (object): The trigger's source, such as "user_message".

    Returns:
        str: Plain words, such as "Message from you".
    """
    key = self._text(source)
    if key in _SOURCE_WORDS:
        return _SOURCE_WORDS[key]
    return f'Woken by {key.replace("_", " ")}'