Skip to content

trading_agent.terminal.screens.activity_screen

The Activity page: every event of today, from every desk, newest first, with live events added as they arrive.

It copies the web page's ActivityPage.tsx: a filter by desk, a text search, and routine events hidden unless asked for. Up to 600 rows are shown.

Typical usage example:

app.push_screen(ActivityScreen('general'))

ActivityScreen 

ActivityScreen(desk_name)

Bases: PageScreen

The Activity page.

Creates the page.

Parameters:

Name Type Description Default
desk_name str

The desk the chat shows.

 required
Source code in src/trading_agent/terminal/screens/activity_screen.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(self, desk_name: str):
    """Creates the page.

    Args:
        desk_name (str): The desk the chat shows.
    """
    super().__init__(desk_name)
    self._events: list[dict[str, Any]] = []
    self._keys: set[str] = set()
    self._filter_desk = _ALL_DESKS
    self._show_routine = False
    self._describer = event_describer.EventDescriber()
    self._formatter = formatter.Formatter()

compose_page 

compose_page()

Builds the filter line, the search box and the feed.

Yields:

Type Description
Widget

The filters, the search box and the table.
Source code in src/trading_agent/terminal/screens/activity_screen.py 
52
53
54
55
56
57
58
59
60
61
62
def compose_page(self):
    """Builds the filter line, the search box and the feed.

    Yields:
        Widget: The filters, the search box and the table.
    """
    yield widgets.Static('', id='activity-filters', classes='page-summary')
    yield widgets.Input(
        placeholder='Search (/ to focus)', id='activity-search'
    )
    yield widgets.DataTable(id='activity', cursor_type='row')

on_mount 

on_mount()

Sets up the table, then reads the journals.

Source code in src/trading_agent/terminal/screens/activity_screen.py 
64
65
66
67
68
69
70
71
72
def on_mount(self) -> None:
    """Sets up the table, then reads the journals."""
    table = self.table('activity')
    table.add_column('Time', width=8)
    table.add_column('Desk', width=10)
    table.add_column('What', width=46)
    table.add_column('Detail')
    self._show_filters()
    super().on_mount()

refresh_page async 

refresh_page()

Reads today's journal of every desk and the events of no desk.

Source code in src/trading_agent/terminal/screens/activity_screen.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@work(exclusive=True, group='page')
async def refresh_page(self) -> None:
    """Reads today's journal of every desk and the events of no desk."""
    gateway = self.app.gateway
    events = []
    try:
        for desk in self.app.desks_document.get('desks', []):
            journal = await gateway.desk_journal(
                desk.get('name'), _JOURNAL_LIMIT
            )
            events.extend(journal.get('events', []))
        journal = await gateway.journal(_JOURNAL_LIMIT)
        events.extend(journal.get('events', []))
    except gateway_client.GatewayError as error:
        self.problem(error.message)
        return
    self._events = []
    self._keys = set()
    for event in events:
        self._remember(event)
    self._show_rows()

take_event 

take_event(event)

Adds a live event to the feed.

Parameters:

Name Type Description Default
event dict[str, Any]

The event.

 required
Source code in src/trading_agent/terminal/screens/activity_screen.py 
 96
 97
 98
 99
100
101
102
103
104
105
def take_event(self, event: dict[str, Any]) -> None:
    """Adds a live event to the feed.

    Args:
        event (dict[str, Any]): The event.
    """
    if event.get('kind') == 'text_delta':
        return
    if self._remember(event):
        self._show_rows()

on_input_changed 

on_input_changed(event)

Filters the feed by the search text.

Parameters:

Name Type Description Default
event Changed

The new search text.

 required
Source code in src/trading_agent/terminal/screens/activity_screen.py 
107
108
109
110
111
112
113
114
def on_input_changed(self, event: widgets.Input.Changed) -> None:
    """Filters the feed by the search text.

    Args:
        event (widgets.Input.Changed): The new search text.
    """
    event.stop()
    self._show_rows()

action_next_filter_desk 

action_next_filter_desk()

Cycles the desk filter through all desks and each desk.

Source code in src/trading_agent/terminal/screens/activity_screen.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
def action_next_filter_desk(self) -> None:
    """Cycles the desk filter through all desks and each desk."""
    choices = [
        _ALL_DESKS,
    ]
    for desk in self.app.desks_document.get('desks', []):
        choices.append(desk.get('name'))
    index = 0
    if self._filter_desk in choices:
        index = (choices.index(self._filter_desk) + 1) % len(choices)
    self._filter_desk = choices[index]
    self._show_filters()
    self._show_rows()

action_toggle_routine 

action_toggle_routine()

Shows or hides routine events, such as turn starts and model usage.

Source code in src/trading_agent/terminal/screens/activity_screen.py 
130
131
132
133
134
def action_toggle_routine(self) -> None:
    """Shows or hides routine events, such as turn starts and model usage."""
    self._show_routine = not self._show_routine
    self._show_filters()
    self._show_rows()

action_focus_search()

Moves to the search box.

Source code in src/trading_agent/terminal/screens/activity_screen.py 
136
137
138
def action_focus_search(self) -> None:
    """Moves to the search box."""
    self.query_one('#activity-search', widgets.Input).focus()