trading_agent.doctor.checks.journal_search

Searches the desks' journals for proof that something has already happened, for the going-live checks.

Typical usage example:

event = JournalSearch(layout, ['main']).latest('order', 'dry_run', True)

JournalSearch

JournalSearch(layout, desk_names)

Reads every desk's journal files, newest day first.

Creates the search.

Parameters:

Name	Type	Description	Default
layout	WorkspaceLayout	Where the desks' folders are.	required
desk_names	Sequence[str]	The desks to search.	required

Source code in src/trading_agent/doctor/checks/journal_search.py

def __init__(
    self,
    layout: workspace_layout.WorkspaceLayout,
    desk_names: Sequence[str],
):
    """Creates the search.

    Args:
        layout (workspace_layout.WorkspaceLayout): Where the desks' folders are.
        desk_names (Sequence[str]): The desks to search.
    """
    self._layout = layout
    self._desk_names = list(desk_names)

events

events(kind)

Every journaled event of one kind, from every desk, newest day first.

Parameters:

Name	Type	Description	Default
kind	str	The event kind, such as "order".	required

Returns:

Type	Description
list[dict[str, Any]]	The events as the journal stores them.

Source code in src/trading_agent/doctor/checks/journal_search.py

def events(self, kind: str) -> list[dict[str, Any]]:
    """Every journaled event of one kind, from every desk, newest day first.

    Args:
        kind (str): The event kind, such as "order".

    Returns:
        list[dict[str, Any]]: The events as the journal stores them.
    """
    found = []
    for desk_name in self._desk_names:
        folder = self._layout.desk_directory(desk_name) / 'journal'
        if not folder.is_dir():
            continue
        for path in sorted(folder.glob('*.jsonl'), reverse=True):
            for line in path.read_text(encoding='utf-8').splitlines():
                try:
                    event = json.loads(line)
                except ValueError:
                    continue
                if event.get('kind') == kind:
                    found.append(event)
    found.sort(key=self._time_of, reverse=True)
    return found