Skip to content

trading_agent.terminal.screens.checkpoints_screen

The Checkpoints page: snapshots of a desk's own files, with restore.

It copies the web page's CheckpointsPage.tsx. Restoring puts the desk's Memory.md, memory/ and notes/ back as they were; the gateway first takes a checkpoint of the present state, so a restore can be undone. The newest checkpoint is the present state and cannot be restored.

Typical usage example:

app.push_screen(CheckpointsScreen('general'))

CheckpointsScreen 

CheckpointsScreen(desk_name)

Bases: PageScreen

The Checkpoints 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/checkpoints_screen.py 
33
34
35
36
37
38
39
40
41
def __init__(self, desk_name: str):
    """Creates the page.

    Args:
        desk_name (str): The desk the chat shows.
    """
    super().__init__(desk_name)
    self._checkpoints: list[dict[str, Any]] = []
    self._formatter = formatter.Formatter()

compose_page 

compose_page()

Builds the checkpoints table.

Yields:

Type Description
Widget

The note and the table.
Source code in src/trading_agent/terminal/screens/checkpoints_screen.py 
43
44
45
46
47
48
49
50
51
52
53
def compose_page(self):
    """Builds the checkpoints table.

    Yields:
        Widget: The note and the table.
    """
    yield widgets.Static(
        "[dim]Restoring puts the desk's Memory.md, memory/ and notes/ back as they were. A checkpoint of the present is taken first, so it can be undone.[/dim]",
        classes='page-summary',
    )
    yield widgets.DataTable(id='checkpoints', cursor_type='row')

on_mount 

on_mount()

Sets up the table, then reads the checkpoints.

Source code in src/trading_agent/terminal/screens/checkpoints_screen.py 
55
56
57
58
def on_mount(self) -> None:
    """Sets up the table, then reads the checkpoints."""
    self.table('checkpoints').add_columns('Taken', 'Why', 'Id')
    super().on_mount()

refresh_page async 

refresh_page()

Reads the desk's checkpoints.

Source code in src/trading_agent/terminal/screens/checkpoints_screen.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@work(exclusive=True, group='page')
async def refresh_page(self) -> None:
    """Reads the desk's checkpoints."""
    try:
        answer = await self.app.gateway.checkpoints(self.desk_name)
    except gateway_client.GatewayError as error:
        self.problem(error.message)
        return
    self._checkpoints = answer.get('checkpoints') or []
    table = self.table('checkpoints')
    table.clear()
    for index, checkpoint in enumerate(self._checkpoints):
        why = str(checkpoint.get('reason', ''))
        if index == 0:
            why += '  (the present state)'
        table.add_row(
            self._formatter.day_time(str(checkpoint.get('created_at', ''))),
            why,
            str(checkpoint.get('checkpoint_id', '')),
            key=str(checkpoint.get('checkpoint_id', '')),
        )

action_restore 

action_restore()

Asks for a reason, then restores the highlighted checkpoint.

Source code in src/trading_agent/terminal/screens/checkpoints_screen.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def action_restore(self) -> None:
    """Asks for a reason, then restores the highlighted checkpoint."""
    checkpoint_id = self.selected_key('checkpoints')
    if checkpoint_id is None:
        return
    if self._checkpoints and checkpoint_id == self._checkpoints[0].get(
        'checkpoint_id'
    ):
        self.notify(
            'The newest checkpoint is the present state; choose an older one.'
        )
        return

    def restore(reason: str | None) -> None:
        """Restores once a reason is given.

        Args:
            reason (str | None): The reason, or None when cancelled.
        """
        if reason is not None:
            self._restore(checkpoint_id)

    self.app.push_screen(
        confirm_dialog.ConfirmDialog(
            f'Restore checkpoint {checkpoint_id}',
            f"{self.desk_name}'s Memory.md, memory/ and notes/ go back to how they were at this checkpoint.",
            'Restore',
        ),
        restore,
    )