trading_agent.channels.notification_formatter

Turns the events worth a phone notification into short messages.

The user chooses which kinds are pushed in trading_agent.toml's [telegram] notifications list. Every message starts with the desk's name, because one chat hears from every desk.

Typical usage example:

formatter = NotificationFormatter(['fills', 'halts']) text = formatter.format(event, trigger_source)

NotificationFormatter

NotificationFormatter(notifications)

Decides which events are pushed and how they read.

Creates the formatter.

Parameters:

Name	Type	Description	Default
notifications	Sequence[str]	The kinds to push, from [telegram] notifications.	required

Source code in src/trading_agent/channels/notification_formatter.py

def __init__(self, notifications: Sequence[str]):
    """Creates the formatter.

    Args:
        notifications (Sequence[str]): The kinds to push, from [telegram] notifications.
    """
    self._notifications = set(notifications)

format

format(published_event, trigger_source, trigger_text)

Builds the notification for one event.

Parameters:

Name	Type	Description	Default
published_event	Event	The event.	required
trigger_source	str | None	For a text event, what woke the desk for that turn, or None when unknown.	required
trigger_text	str | None	For a text event, the trigger's text, or None when unknown.	required

Returns:

Type	Description
str | None	The message, or None when this event is not pushed.

Source code in src/trading_agent/channels/notification_formatter.py

def format(
    self,
    published_event: event_module.Event,
    trigger_source: str | None,
    trigger_text: str | None,
) -> str | None:
    """Builds the notification for one event.

    Args:
        published_event (event_module.Event): The event.
        trigger_source (str | None): For a text event, what woke the desk for that turn, or None when unknown.
        trigger_text (str | None): For a text event, the trigger's text, or None when unknown.

    Returns:
        str | None: The message, or None when this event is not pushed.
    """
    kind = published_event.kind
    payload = published_event.payload
    desk = published_event.desk or 'account'
    if kind == event_module.EventKind.FILL and self._wants('fills'):
        return self._fill(desk, payload)
    if (
        kind == event_module.EventKind.RISK_DECISION
        and not payload.get('approved')
        and self._wants('rejections')
    ):
        reasons = ' '.join(
            str(reason) for reason in payload.get('reasons', [])
        )
        return f'{desk}: the risk engine refused {payload.get("order", "an order")}. {reasons}'
    if (
        kind == event_module.EventKind.BACKEND_SWITCH
        and payload.get('from')
        and self._wants('model_switches')
    ):
        return f'{desk}: switched model from {payload.get("from")} to {payload.get("to")}.'
    if kind == event_module.EventKind.CRON_RESULT and self._wants(
        'job_results'
    ):
        return f'{desk}: job {payload.get("name")} finished ({payload.get("outcome")}).\n{payload.get("text", "")}'.strip()
    if kind == event_module.EventKind.TEXT and self._wants('desk_reports'):
        return self._report(desk, payload, trigger_source, trigger_text)
    if kind == event_module.EventKind.MESSAGE and self._wants('messages'):
        prefix = 'Message from'
        if payload.get('attachments'):
            prefix = 'Picture from'
        if payload.get('urgent'):
            prefix = 'URGENT from'
        return f'{prefix} {desk}:\n{payload.get("text", "")}'.strip()
    if kind == event_module.EventKind.PRICE_ALERT and self._wants('alerts'):
        note = f' Note: {payload["note"]}' if payload.get('note') else ''
        return f'{desk}: {payload.get("symbol")} traded at {payload.get("last_price")}, {payload.get("condition")} the alert at {payload.get("price")}.{note}'
    if kind == event_module.EventKind.DESK_HALT and self._wants('halts'):
        return f'{desk} is halted for the day: {payload.get("reason")}'
    if (
        kind == event_module.EventKind.KILL_SWITCH
        and payload.get('stage') == 'fired'
        and self._wants('halts')
    ):
        mode = ' (dry run)' if payload.get('dry_run') else ''
        return f'KILL SWITCH fired{mode}: {payload.get("reason")}. Every desk is halted until {payload.get("until_date")}.'
    if kind == event_module.EventKind.ERROR and self._wants('errors'):
        return f'{desk}: a turn failed: {payload.get("error")}'
    return None

wants_daily_summaries

wants_daily_summaries()

Says whether the day's summaries are pushed at each rollover.

Returns:

Type	Description
bool	True when daily_summaries is in the notifications list.

Source code in src/trading_agent/channels/notification_formatter.py

def wants_daily_summaries(self) -> bool:
    """Says whether the day's summaries are pushed at each rollover.

    Returns:
        bool: True when daily_summaries is in the notifications list.
    """
    return self._wants('daily_summaries')