trading_agent.channels.telegram_commands

The Telegram commands answered by code, with no model.

These work even when every model is unavailable or a desk is paused, which is exactly when the user most needs them: what is happening, what is open, how much is made or lost, stop a desk, or stop everything. Any other message, including the desks' own slash commands such as /usage or /stop, goes to the chat's desk.

Typical usage example:

commands = TelegramCommands(desks, shared, pairing, 'dry_run') if commands.handles('/pnl'): reply = await commands.answer(chat, '/pnl')

TelegramCommands

TelegramCommands(desks, shared, pairing, mode, clock=None, request_store=None, responder=None)

Answers the code-only commands of one gateway.

Creates the commands.

Parameters:

Name	Type	Description	Default
desks	Mapping[str, DeskRuntime]	The running desks.	required
shared	SharedParts	The snapshot reader, risk engine, ledger and kill switch.	required
pairing	PairingStore	Where each chat's desk is kept.	required
mode	str	"dry_run" or "live".	required
clock	Callable[[], datetime] | None	Returns the current India time; None uses the system clock.	None
request_store	UserRequestStore | None	The desks' requests, searched by /approve and /deny, or None.	None
responder	UserRequestResponder | None	Decides permission requests for /approve and /deny, or None.	None

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

def __init__(
    self,
    desks: Mapping[str, desk_runtime.DeskRuntime],
    shared: gateway_parts.SharedParts,
    pairing: pairing_store_module.PairingStore,
    mode: str,
    clock: Callable[[], datetime.datetime] | None = None,
    request_store: user_request_store.UserRequestStore | None = None,
    responder: user_request_responder.UserRequestResponder | None = None,
):
    """Creates the commands.

    Args:
        desks (Mapping[str, desk_runtime.DeskRuntime]): The running desks.
        shared (gateway_parts.SharedParts): The snapshot reader, risk engine, ledger and kill switch.
        pairing (pairing_store_module.PairingStore): Where each chat's desk is kept.
        mode (str): "dry_run" or "live".
        clock (Callable[[], datetime.datetime] | None): Returns the current India time; None uses the system clock.
        request_store (user_request_store.UserRequestStore | None): The desks' requests, searched by /approve and /deny, or None.
        responder (user_request_responder.UserRequestResponder | None): Decides permission requests for /approve and /deny, or None.
    """
    self._desks = desks
    self._shared = shared
    self._pairing = pairing
    self._mode = mode
    self._clock = clock
    self._request_store = request_store
    self._responder = responder

handles

handles(text)

Says whether a message is one of these commands.

Parameters:

Name	Type	Description	Default
text	str	The message.	required

Returns:

Type	Description
bool	True for /start, /help, /desk, /status, /positions, /pnl, /pause, /resume, /kill, /approve and /deny.

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

def handles(self, text: str) -> bool:
    """Says whether a message is one of these commands.

    Args:
        text (str): The message.

    Returns:
        bool: True for /start, /help, /desk, /status, /positions, /pnl, /pause, /resume, /kill, /approve and /deny.
    """
    return self._command_name(text) in _COMMANDS

menu

menu()

The command menu Telegram shows when the user types a slash.

Returns:

Type	Description
tuple[tuple[str, str], ...]	Each command as (name without the slash, description).

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

def menu(self) -> tuple[tuple[str, str], ...]:
    """The command menu Telegram shows when the user types a slash.

    Returns:
        tuple[tuple[str, str], ...]: Each command as (name without the slash, description).
    """
    return _MENU

answer async

answer(chat, text)

Answers a command.

Parameters:

Name	Type	Description	Default
chat	PairedChat	The paired chat that sent it.	required
text	str	The command and its argument.	required

Returns:

Type	Description
str	The reply.

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

async def answer(
    self, chat: pairing_store_module.PairedChat, text: str
) -> str:
    """Answers a command.

    Args:
        chat (pairing_store_module.PairedChat): The paired chat that sent it.
        text (str): The command and its argument.

    Returns:
        str: The reply.
    """
    name = self._command_name(text)
    argument = text.strip()[len(text.strip().split()[0]) :].strip()
    if name == '/desk':
        return self._desk(chat, argument)
    if name == '/status':
        return await self._status(chat)
    if name == '/positions':
        return await self._positions()
    if name == '/pnl':
        return await self._pnl()
    if name == '/pause':
        return self._set_paused(chat, argument, True)
    if name == '/resume':
        return self._set_paused(chat, argument, False)
    if name == '/kill':
        return await self._kill(argument)
    if name == '/approve':
        return await self._decide(argument, True)
    if name == '/deny':
        return await self._decide(argument, False)
    return self._help(chat)