trading_agent.terminal.chat_transcript

Groups one desk's events into chat turns, one per trigger, as the web page does.

It follows the rules of the web page's chatTurnBuilder.ts and eventSocket.ts, but it is fed one event at a time, and each call says which item changed, so a screen only redraws that item.

Typical usage example:

transcript = ChatTranscript() for event in journal_events: transcript.add(event) changed = transcript.add(live_event)

ChatTranscript

ChatTranscript()

The chat of one desk, built from its events.

Attributes:

Name	Type	Description
items	list[ChatItem]	The turns and loose events, oldest first.

Creates an empty transcript.

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

def __init__(self):
    """Creates an empty transcript."""
    self.items: list[ChatItem] = []
    self._turns: dict[str, chat_items.ChatTurn] = {}
    self._shown: set[str] = set()
    self._open_turn: chat_items.ChatTurn | None = None
    self._request_steps: dict[str, chat_items.RequestStep] = {}
    self._request_turns: dict[str, chat_items.ChatTurn] = {}
    self._seen_keys: set[str] = set()
    self._describer = event_describer.EventDescriber()

add

add(event)

Adds one event.

An event already added, such as one read from the journal and received live too, is ignored.

Parameters:

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

Returns:

Type	Description
ChatItem | None	The turn or loose event that changed or was added, or None when nothing shown changed.

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

def add(self, event: dict[str, Any]) -> ChatItem | None:
    """Adds one event.

    An event already added, such as one read from the journal and received live too, is ignored.

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

    Returns:
        ChatItem | None: The turn or loose event that changed or was added, or None when nothing shown changed.
    """
    kind = event.get('kind', '')
    if kind != 'text_delta':
        key = self.key(event)
        if key in self._seen_keys:
            return None
        self._seen_keys.add(key)
    payload = event.get('payload') or {}
    if kind == 'user_response':
        return self._add_response(payload)
    trigger_id = payload.get('trigger_id')
    if not isinstance(trigger_id, str):
        trigger_id = None
    if kind == 'trigger_received' and trigger_id is not None:
        return self._add_trigger(event, trigger_id)
    if kind == 'text_delta':
        turn = self._turns.get(trigger_id or '')
        if turn is None:
            return None
        turn.streaming_text += str(payload.get('text', ''))
        return self._shown_item(turn)
    newly_shown = None
    if (
        kind == 'turn_started'
        and trigger_id is not None
        and trigger_id not in self._shown
    ):
        queued_message = self._turns.get(trigger_id)
        if queued_message is not None:
            self.items.append(queued_message)
            self._shown.add(trigger_id)
            newly_shown = queued_message
    turn = self._open_turn
    if trigger_id is not None:
        turn = self._turns.get(trigger_id)
    if kind == 'user_request':
        step = chat_items.RequestStep(request=dict(payload))
        request_id = str(payload.get('request_id', ''))
        self._request_steps[request_id] = step
        if turn is None:
            return None
        turn.steps.append(step)
        self._request_turns[request_id] = turn
        return turn
    if turn is None:
        if kind not in _TURN_LESS_KINDS_SHOWN:
            return None
        if self._describer.describe(event).tone == 'quiet':
            return None
        loose = chat_items.LooseEvent(event=event)
        self.items.append(loose)
        return loose
    if kind == 'turn_started':
        turn.started = True
        turn.started_at = event.get('created_at')
        self._open_turn = turn
        return turn
    if kind == 'turn_finished':
        turn.finished = True
        turn.finished_at = event.get('created_at')
        for step in turn.steps:
            if (
                isinstance(step, chat_items.ToolStep)
                and step.result_text is None
            ):
                step.ended_early = True
        if self._open_turn is turn:
            self._open_turn = None
        return turn
    if self._add_to_turn(turn, event):
        return self._shown_item(turn)
    return newly_shown

turn

turn(trigger_id)

Finds a turn by its trigger.

Parameters:

Name	Type	Description	Default
trigger_id	str	The trigger's id.	required

Returns:

Type	Description
ChatTurn | None	The turn, or None when no such trigger was seen.

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

def turn(self, trigger_id: str) -> chat_items.ChatTurn | None:
    """Finds a turn by its trigger.

    Args:
        trigger_id (str): The trigger's id.

    Returns:
        chat_items.ChatTurn | None: The turn, or None when no such trigger was seen.
    """
    return self._turns.get(trigger_id)

running_turn

running_turn()

The turn that has started and not finished, if any.

Returns:

Type	Description
ChatTurn | None	The open turn, or None.

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

def running_turn(self) -> chat_items.ChatTurn | None:
    """The turn that has started and not finished, if any.

    Returns:
        chat_items.ChatTurn | None: The open turn, or None.
    """
    return self._open_turn

key

key(event)

A key that is the same for the journal's copy and the live copy of one event; the web page's key plus the trigger id.

Parameters:

Name	Type	Description	Default
event	dict[str, Any]	The event.	required

Returns:

Type	Description
str	The key.

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

def key(self, event: dict[str, Any]) -> str:
    """A key that is the same for the journal's copy and the live copy of one event; the web page's key plus the trigger id.

    Args:
        event (dict[str, Any]): The event.

    Returns:
        str: The key.
    """
    payload = event.get('payload') or {}
    call_id = payload.get('call_id')
    if not isinstance(call_id, str):
        call_id = ''
    trigger_id = payload.get('trigger_id')
    if not isinstance(trigger_id, str):
        trigger_id = ''
    desk = event.get('desk') or ''
    return f'{event.get("created_at", "")}|{event.get("kind", "")}|{desk}|{trigger_id}|{call_id}'