trading_agent.memory.session_search

Keyword search over everything a desk has said and done before.

It looks through four kinds of record in the desk folder, newest days first:

• exchanges: each trigger and the desk's answer, paired from the journal's trigger_received and text events;
• journal: the desk's actions, such as orders, risk decisions and tool calls, from journal/*.jsonl;
• summaries: each paragraph of the daily summaries, from sessions/summaries/*.md;
• notes: each note, from notes/*.md.

The journal is the complete record, going back before sessions were kept, so exchanges come from it rather than from the session logs, which exist to resume a session. The desk's own search_sessions calls are left out, so a search never finds itself.

A record matches when it contains every keyword, ignoring case. When nothing matches every keyword, records that match some of them are returned instead, best first.

Typical usage example:

search = SessionSearch(Path('workspace/desks/main')) hits = search.search('NIFTY gap down', limit=10)

SearchRecord dataclass

SearchRecord(day, kind, text)

One searchable record.

Attributes:

Name	Type	Description
day	str	The India date it belongs to, as YYYY-MM-DD.
kind	str	"exchange", "summary", "note" or "journal".
text	str	The record's text.

SearchHit dataclass

SearchHit(record, matched_words, snippet)

One record that matched.

Attributes:

Name	Type	Description
record	SearchRecord	The record.
matched_words	int	How many of the keywords it contains.
snippet	str	The part of its text around the first keyword.

SearchResult dataclass

SearchResult(hits, matched_every_word, words)

The outcome of one search.

Attributes:

Name	Type	Description
hits	list[SearchHit]	The matching records, best and newest first.
matched_every_word	bool	Whether the hits contain every keyword, or only some.
words	list[str]	The keywords searched for.

SessionSearch

SessionSearch(desk_directory)

Searches one desk's records.

Creates the search.

Parameters:

Name	Type	Description	Default
desk_directory	Path	The desk folder.	required

Source code in src/trading_agent/memory/session_search.py

def __init__(self, desk_directory: Path):
    """Creates the search.

    Args:
        desk_directory (Path): The desk folder.
    """
    self._desk_directory = desk_directory

search

search(query, limit)

Finds the records that best match the keywords.

Parameters:

Name	Type	Description	Default
query	str	Keywords separated by spaces.	required
limit	int	The most hits to return.	required

Returns:

Type	Description
SearchResult	The hits.

Source code in src/trading_agent/memory/session_search.py

def search(self, query: str, limit: int) -> SearchResult:
    """Finds the records that best match the keywords.

    Args:
        query (str): Keywords separated by spaces.
        limit (int): The most hits to return.

    Returns:
        SearchResult: The hits.
    """
    words = self._words(query)
    if not words:
        return SearchResult(hits=[], matched_every_word=False, words=[])
    full_hits = []
    partial_hits = []
    for record in self._records(words):
        lowered = record.text.lower()
        matched = 0
        for word in words:
            if word in lowered:
                matched += 1
        if matched == 0:
            continue
        hit = SearchHit(
            record=record,
            matched_words=matched,
            snippet=self._snippet(record.text, lowered, words),
        )
        if matched == len(words):
            full_hits.append(hit)
        else:
            partial_hits.append(hit)
    if full_hits:
        return SearchResult(
            hits=self._best(full_hits, limit),
            matched_every_word=True,
            words=words,
        )
    return SearchResult(
        hits=self._best(partial_hits, limit),
        matched_every_word=False,
        words=words,
    )