trading_agent.channels.pairing_store

Which Telegram chats may command the desks, and the pairing codes waiting for the user's approval.

Anyone who finds the bot can message it, so the bot obeys only chats the user has approved. A chat that is not paired gets an eight-character code. The user approves the code in the web page or with trading-agent pairing approve, which proves the chat is theirs: only someone who can see the bot's reply and reach this machine can do both. A code expires after an hour.

The store is one JSON file in workspace/shared, which the desks can read but never write, so no desk can pair a chat by itself. The file is read afresh for every question, so an approval made by the command line reaches the running gateway at once.

Typical usage example:

store = PairingStore(layout.telegram_pairing_file) code = store.request(chat_id, 'Pramod') store.approve(code) store.is_paired(chat_id)

PairingError

Bases: ValueError

A pairing code that does not exist or has expired, or a chat that is not paired.

PendingPairing dataclass

PendingPairing(code, chat_id, name, requested_at)

A chat waiting for the user to approve its code.

Attributes:

Name	Type	Description
code	str	The eight-character code the bot sent to the chat.
chat_id	int	Telegram's id for the chat.
name	str	The sender's name, as Telegram gives it.
requested_at	str	When the code was made, as an ISO 8601 string in UTC.

PairedChat dataclass

PairedChat(chat_id, name, paired_at, desk)

A chat the desks obey.

Attributes:

Name	Type	Description
chat_id	int	Telegram's id for the chat.
name	str	The sender's name when it was paired.
paired_at	str	When it was approved, as an ISO 8601 string in UTC.
desk	str	The desk the chat talks to.

PairingStore

PairingStore(path, clock=None)

The paired chats and pending codes, kept in one JSON file.

Creates the store without reading the file.

Parameters:

Name	Type	Description	Default
path	Path	The JSON file.	required
clock	Callable[[], datetime] | None	Returns the current time in UTC; None uses the system clock.	None

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

def __init__(
    self, path: Path, clock: Callable[[], datetime.datetime] | None = None
):
    """Creates the store without reading the file.

    Args:
        path (Path): The JSON file.
        clock (Callable[[], datetime.datetime] | None): Returns the current time in UTC; None uses the system clock.
    """
    self._path = path
    self._clock = clock
    self._lock = threading.Lock()

request

request(chat_id, name)

Gives an unpaired chat a pairing code, reusing its code while it is still valid.

Parameters:

Name	Type	Description	Default
chat_id	int	The chat.	required
name	str	The sender's name.	required

Returns:

Type	Description
str | None	The code, or None when too many codes are already waiting, so strangers cannot fill the list.

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

def request(self, chat_id: int, name: str) -> str | None:
    """Gives an unpaired chat a pairing code, reusing its code while it is still valid.

    Args:
        chat_id (int): The chat.
        name (str): The sender's name.

    Returns:
        str | None: The code, or None when too many codes are already waiting, so strangers cannot fill the list.
    """
    with self._lock:
        data = self._read_live()
        for pending in data['pending']:
            if pending['chat_id'] == chat_id:
                return pending['code']
        if len(data['pending']) >= _MOST_PENDING:
            return None
        code = self._new_code()
        data['pending'].append(
            dataclasses.asdict(
                PendingPairing(
                    code=code,
                    chat_id=chat_id,
                    name=name,
                    requested_at=self._now().isoformat(timespec='seconds'),
                )
            )
        )
        self._write(data)
        return code

approve

approve(code, desk)

Pairs the chat that was given a code.

Parameters:

Name	Type	Description	Default
code	str	The code, with or without a dash, in any case.	required
desk	str	The desk the chat starts talking to.	required

Returns:

Type	Description
PairedChat	The newly paired chat.

Raises:

Type	Description
PairingError	No valid code matches.

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

def approve(self, code: str, desk: str) -> PairedChat:
    """Pairs the chat that was given a code.

    Args:
        code (str): The code, with or without a dash, in any case.
        desk (str): The desk the chat starts talking to.

    Returns:
        PairedChat: The newly paired chat.

    Raises:
        PairingError: No valid code matches.
    """
    wanted = code.replace('-', '').replace(' ', '').upper()
    with self._lock:
        data = self._read_live()
        for pending in data['pending']:
            if pending['code'] == wanted:
                data['pending'].remove(pending)
                chat = PairedChat(
                    chat_id=pending['chat_id'],
                    name=pending['name'],
                    paired_at=self._now().isoformat(timespec='seconds'),
                    desk=desk,
                )
                data['paired'] = [
                    entry
                    for entry in data['paired']
                    if entry['chat_id'] != chat.chat_id
                ]
                data['paired'].append(dataclasses.asdict(chat))
                self._write(data)
                return chat
    raise PairingError(
        f'No pairing code {wanted!r} is waiting; codes expire after an hour'
    )

revoke

revoke(chat_id)

Unpairs a chat, so the desks stop obeying it.

Parameters:

Name	Type	Description	Default
chat_id	int	The chat.	required

Returns:

Type	Description
PairedChat	The chat that was removed.

Raises:

Type	Description
PairingError	The chat is not paired.

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

def revoke(self, chat_id: int) -> PairedChat:
    """Unpairs a chat, so the desks stop obeying it.

    Args:
        chat_id (int): The chat.

    Returns:
        PairedChat: The chat that was removed.

    Raises:
        PairingError: The chat is not paired.
    """
    with self._lock:
        data = self._read_live()
        for entry in data['paired']:
            if entry['chat_id'] == chat_id:
                data['paired'].remove(entry)
                self._write(data)
                return PairedChat(**entry)
    raise PairingError(f'Chat {chat_id} is not paired')

paired_chat

paired_chat(chat_id)

Finds a paired chat.

Parameters:

Name	Type	Description	Default
chat_id	int	The chat.	required

Returns:

Type	Description
PairedChat | None	The chat, or None when it is not paired.

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

def paired_chat(self, chat_id: int) -> PairedChat | None:
    """Finds a paired chat.

    Args:
        chat_id (int): The chat.

    Returns:
        PairedChat | None: The chat, or None when it is not paired.
    """
    for chat in self.paired_chats():
        if chat.chat_id == chat_id:
            return chat
    return None

paired_chats

paired_chats()

Every paired chat.

Returns:

Type	Description
list[PairedChat]	The chats, in the order they were paired.

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

def paired_chats(self) -> list[PairedChat]:
    """Every paired chat.

    Returns:
        list[PairedChat]: The chats, in the order they were paired.
    """
    with self._lock:
        data = self._read_live()
    chats = []
    for entry in data['paired']:
        chats.append(PairedChat(**entry))
    return chats

pending

pending()

Every code still waiting for approval.

Returns:

Type	Description
list[PendingPairing]	The codes, oldest first.

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

def pending(self) -> list[PendingPairing]:
    """Every code still waiting for approval.

    Returns:
        list[PendingPairing]: The codes, oldest first.
    """
    with self._lock:
        data = self._read_live()
    pending = []
    for entry in data['pending']:
        pending.append(PendingPairing(**entry))
    return pending

choose_desk

choose_desk(chat_id, desk)

Changes the desk a paired chat talks to.

Parameters:

Name	Type	Description	Default
chat_id	int	The chat.	required
desk	str	The desk.	required

Raises:

Type	Description
PairingError	The chat is not paired.

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

def choose_desk(self, chat_id: int, desk: str) -> None:
    """Changes the desk a paired chat talks to.

    Args:
        chat_id (int): The chat.
        desk (str): The desk.

    Raises:
        PairingError: The chat is not paired.
    """
    with self._lock:
        data = self._read_live()
        for entry in data['paired']:
            if entry['chat_id'] == chat_id:
                entry['desk'] = desk
                self._write(data)
                return
    raise PairingError(f'Chat {chat_id} is not paired')