trading_agent.channels.telegram_requests

The desks' questions and plans on Telegram, answered with buttons.

Each question becomes one message with a button per option. A single-choice question is settled by one press; a multiple-choice one ticks options and is settled with Done. "Other…" and a plan's "Keep planning" ask the user to type a reply, and the chat's next message is taken as that reply instead of going to the desk. Once every question of a request is settled, the answers go to the desk through the same responder the web page uses.

When a request is answered anywhere, on the web page or in another chat, its Telegram messages lose their buttons and say how it was settled.

Typical usage example:

requests = TelegramRequests(api, pairing, store, responder) await requests.show(request) await requests.press(chat, button)

TelegramRequests

TelegramRequests(api, pairing, store, responder)

Shows requests in the paired chats, and turns button presses and typed replies into answers.

Each request's progress is kept in memory while it is open: the answers chosen so far, which questions are settled, and which messages hold its buttons. After a restart, the request is shown again when it is next published.

Creates the handler.

Parameters:

Name	Type	Description	Default
api	TelegramApi	The connection to Telegram.	required
pairing	PairingStore	The paired chats.	required
store	UserRequestStore	Where requests are kept.	required
responder	UserRequestResponder	Records the answers and tells the desks.	required

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

def __init__(
    self,
    api: telegram_api_module.TelegramApi,
    pairing: pairing_store_module.PairingStore,
    store: user_request_store_module.UserRequestStore,
    responder: user_request_responder_module.UserRequestResponder,
):
    """Creates the handler.

    Args:
        api (telegram_api_module.TelegramApi): The connection to Telegram.
        pairing (pairing_store_module.PairingStore): The paired chats.
        store (user_request_store_module.UserRequestStore): Where requests are kept.
        responder (user_request_responder_module.UserRequestResponder): Records the answers and tells the desks.
    """
    self._api = api
    self._pairing = pairing
    self._store = store
    self._responder = responder
    self._drafts: dict[str, list[dict[str, Any]]] = {}
    self._settled: dict[str, set[int]] = {}
    self._messages: dict[str, list[tuple[int, int, int]]] = {}
    self._awaiting_text: dict[int, tuple[str, str, int]] = {}

show async

show(request)

Sends a request to every paired chat, with its buttons.

Parameters:

Name	Type	Description	Default
request	UserRequest	The request.	required

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

async def show(self, request: user_request_module.UserRequest) -> None:
    """Sends a request to every paired chat, with its buttons.

    Args:
        request (user_request_module.UserRequest): The request.
    """
    if request.kind == user_request_module.QUESTION:
        questions = request.content['questions']
        self._drafts[request.request_id] = []
        for _ in questions:
            self._drafts[request.request_id].append(
                {
                    'selected': [],
                    'other': '',
                }
            )
        self._settled[request.request_id] = set()
        for chat in self._pairing.paired_chats():
            for index in range(len(questions)):
                await self._send(
                    request,
                    chat.chat_id,
                    index,
                    self._question_text(request, index),
                    self._question_buttons(request, index),
                )
        return
    if request.kind == user_request_module.PERMISSION:
        for chat in self._pairing.paired_chats():
            await self._send(
                request,
                chat.chat_id,
                -1,
                self._permission_text(request),
                self._permission_buttons(request),
            )
        return
    if request.kind == user_request_module.PLAN:
        text = f'📋 {request.desk} presents a plan for your approval: {request.content["title"]}\n\n{request.content["plan"]}\n\nApprove it, or ask for changes?'
        for chat in self._pairing.paired_chats():
            await self._send(
                request,
                chat.chat_id,
                -1,
                text,
                self._plan_buttons(request),
            )

wants_text

wants_text(chat_id)

Says whether the chat's next message is a typed reply to a request.

Parameters:

Name	Type	Description	Default
chat_id	int	The chat.	required

Returns:

Type	Description
bool	True after "Other…" or "Keep planning" was pressed in that chat.

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

def wants_text(self, chat_id: int) -> bool:
    """Says whether the chat's next message is a typed reply to a request.

    Args:
        chat_id (int): The chat.

    Returns:
        bool: True after "Other…" or "Keep planning" was pressed in that chat.
    """
    return chat_id in self._awaiting_text

receive_text async

receive_text(chat_id, text)

Takes a typed reply as the "Other" answer or as plan feedback.

Parameters:

Name	Type	Description	Default
chat_id	int	The chat.	required
text	str	The reply.	required

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

async def receive_text(self, chat_id: int, text: str) -> None:
    """Takes a typed reply as the "Other" answer or as plan feedback.

    Args:
        chat_id (int): The chat.
        text (str): The reply.
    """
    request_id, purpose, index = self._awaiting_text.pop(chat_id)
    if not text.strip():
        return
    if purpose == _FEEDBACK:
        await self._respond(
            chat_id,
            lambda: self._responder.decide_plan(request_id, 'revise', text, _VIA),
        )
        return
    drafts = self._drafts.get(request_id)
    if drafts is None:
        await self._note(chat_id, 'That question is no longer open.')
        return
    request = self._store.get(request_id)
    question = request.content['questions'][index]
    if not question.get('multi_select'):
        drafts[index]['selected'] = []
    drafts[index]['other'] = text.strip()
    await self._settle(request, index, chat_id)

press async

press(button)

Handles a press on one of a request's buttons.

Parameters:

Name	Type	Description	Default
button	IncomingButton	The press.	required

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

async def press(
    self,
    button: telegram_api_module.IncomingButton,
) -> None:
    """Handles a press on one of a request's buttons.

    Args:
        button (telegram_api_module.IncomingButton): The press.
    """
    parts = button.data.split(_SEPARATOR)
    if len(parts) != 4:
        await self._acknowledge(button, '')
        return
    action, request_id, question_text, option_text = parts
    index = int(question_text)
    option = int(option_text)
    try:
        request = self._store.pending_request(request_id)
    except user_request_store_module.UserRequestError as error:
        await self._acknowledge(button, str(error))
        return
    if action in _PERMISSION_DECISIONS:
        await self._acknowledge(button, '')
        try:
            await self._responder.decide_permission(
                request_id,
                _PERMISSION_DECISIONS[action],
                '',
                _VIA,
            )
        except ValueError as error:
            await self._note(button.chat_id, f'Not done: {error}')
    elif action == _DISMISS:
        await self._acknowledge(button, 'Dismissed')
        await self._respond(
            button.chat_id,
            lambda: self._responder.dismiss(request_id, _VIA),
        )
    elif action == _APPROVE:
        await self._acknowledge(button, 'Approved')
        await self._respond(
            button.chat_id,
            lambda: self._responder.decide_plan(request_id, 'approve', '', _VIA),
        )
    elif action == _REVISE:
        self._awaiting_text[button.chat_id] = (request_id, _FEEDBACK, -1)
        await self._acknowledge(button, '')
        await self._note(button.chat_id, 'Type what should change in the plan and send it.')
    elif action == _OTHER:
        self._awaiting_text[button.chat_id] = (request_id, _OTHER, index)
        await self._acknowledge(button, '')
        question = request.content['questions'][index]['question']
        await self._note(button.chat_id, f'Type your own answer to "{question}" and send it.')
    elif action == _PICK:
        await self._pick(request, index, option, button)
    elif action == _DONE:
        drafts = self._drafts.get(request_id, [])
        if index >= len(drafts) or not (drafts[index]['selected'] or drafts[index]['other']):
            await self._acknowledge(button, 'Choose at least one first')
            return
        await self._acknowledge(button, '')
        await self._settle(request, index, button.chat_id)

close async

close(payload)

Removes a request's buttons once it is answered anywhere, saying how it was settled.

Parameters:

Name	Type	Description	Default
payload	dict[str, Any]	The user_response event's payload.	required

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

async def close(self, payload: dict[str, Any]) -> None:
    """Removes a request's buttons once it is answered anywhere, saying how it was settled.

    Args:
        payload (dict[str, Any]): The user_response event's payload.
    """
    request_id = str(payload.get('request_id', ''))
    sent = self._messages.pop(request_id, [])
    self._drafts.pop(request_id, None)
    self._settled.pop(request_id, None)
    for chat_id in list(self._awaiting_text):
        if self._awaiting_text[chat_id][0] == request_id:
            self._awaiting_text.pop(chat_id)
    if not sent:
        return
    request = self._store.get(request_id)
    outcome = self._outcome(request)
    for chat_id, message_id, index in sent:
        if index >= 0:
            text = f'{self._question_text(request, index)}\n\n{outcome}'
        elif request.kind == user_request_module.PERMISSION:
            text = f'{self._permission_text(request)}\n\n{outcome}'
        else:
            text = f'📋 {request.content.get("title", "Plan")}\n\n{outcome}'
        try:
            await self._api.edit(chat_id, message_id, text, None)
        except telegram_api_module.TelegramApiError as error:
            _LOGGER.warning('Could not update a Telegram request message: %s', error)