trading_agent.gateway.routes.file_routes

Routes for the user's files and the desks' proposals to change them.

GET /api/files every editable file: the shared files, each desk's Persona.md and Memory.md GET /api/files/content?name=&desk= one file's text PUT /api/files/content replace one file's text GET /api/proposals every proposal, with the change it would make POST /api/proposals/{id}/accept apply a proposal, keeping a copy of the old file POST /api/proposals/{id}/reject reject a proposal

The user may edit their own files and a desk's Memory.md. A desk's Memory.md is snapshotted first, as when the desk writes it. A new TradingRules.md is checked before it is saved: limits the risk engine could not read would stop every desk from opening positions, so such a version is refused.

Typical usage example:

application.include_router(FileRoutes(guard, layout, parts, proposals).router())

FileWrite

Bases: BaseModel

The body of a file write.

Attributes:

Name	Type	Description
name	str	The file's name, such as TradingRules.md or Persona.md.
desk	str | None	The desk, for Persona.md and Memory.md.
text	str	The complete new text.

FileRoutes

FileRoutes(guard, layout, parts, proposals)

Bases: RouteGroup

The file and proposal routes.

Creates the routes.

Parameters:

Name	Type	Description	Default
guard	AccessGuard	Checks the access token.	required
layout	WorkspaceLayout	Where the files are.	required
parts	Mapping[str, DeskParts]	Each desk's parts, for its checkpoints.	required
proposals	ProposalStore	The proposals.	required

Source code in src/trading_agent/gateway/routes/file_routes.py

def __init__(
    self,
    guard: access_guard.AccessGuard,
    layout: workspace_layout.WorkspaceLayout,
    parts: Mapping[str, gateway_parts.DeskParts],
    proposals: proposal_store_module.ProposalStore,
):
    """Creates the routes.

    Args:
        guard (access_guard.AccessGuard): Checks the access token.
        layout (workspace_layout.WorkspaceLayout): Where the files are.
        parts (Mapping[str, gateway_parts.DeskParts]): Each desk's parts, for its checkpoints.
        proposals (proposal_store_module.ProposalStore): The proposals.
    """
    super().__init__(guard)
    self._layout = layout
    self._parts = parts
    self._proposals = proposals

list_files

list_files()

Every editable file.

Returns:

Type	Description
dict[str, Any]	Each file's name, desk (or None for shared files) and size in characters.

Source code in src/trading_agent/gateway/routes/file_routes.py

def list_files(self) -> dict[str, Any]:
    """Every editable file.

    Returns:
        dict[str, Any]: Each file's name, desk (or None for shared files) and size in characters.
    """
    files = []
    for name in workspace_layout.SHARED_FILE_NAMES:
        files.append(self._file_entry(name, None))
    for desk_name in self._parts:
        files.append(self._file_entry('Persona.md', desk_name))
        files.append(self._file_entry(_MEMORY_FILE_NAME, desk_name))
    return {
        'files': files,
    }

read_file

read_file(name, desk=None)

One file's text.

Parameters:

Name	Type	Description	Default
name	str	The file's name.	required
desk	str | None	The desk, for Persona.md and Memory.md.	None

Returns:

Type	Description
dict[str, Any]	The name, desk and text; an empty text for a file not created yet.

Source code in src/trading_agent/gateway/routes/file_routes.py

def read_file(self, name: str, desk: str | None = None) -> dict[str, Any]:
    """One file's text.

    Args:
        name (str): The file's name.
        desk (str | None): The desk, for Persona.md and Memory.md.

    Returns:
        dict[str, Any]: The name, desk and text; an empty text for a file not created yet.
    """
    path = self._path(name, desk)
    text = path.read_text(encoding='utf-8') if path.is_file() else ''
    return {
        'name': name,
        'desk': desk,
        'text': text,
    }

write_file

write_file(request)

Replaces one file's text.

Parameters:

Name	Type	Description	Default
request	FileWrite	The file and its new text.	required

Returns:

Type	Description
dict[str, Any]	The name, desk and new size.

Raises:

Type	Description
HTTPException	The new TradingRules.md cannot be read by the risk engine (status 400).

Source code in src/trading_agent/gateway/routes/file_routes.py

def write_file(self, request: FileWrite) -> dict[str, Any]:
    """Replaces one file's text.

    Args:
        request (FileWrite): The file and its new text.

    Returns:
        dict[str, Any]: The name, desk and new size.

    Raises:
        fastapi.HTTPException: The new TradingRules.md cannot be read by the risk engine (status 400).
    """
    path = self._path(request.name, request.desk)
    if request.name == 'TradingRules.md':
        self._check_limits(request.text)
    if request.name == _MEMORY_FILE_NAME:
        parts = self._find(self._parts, request.desk or '', 'desk')
        try:
            parts.checkpoints.snapshot('before the user edited Memory.md')
        except checkpoints_module.CheckpointError:
            _LOGGER.exception(
                'Could not snapshot before the user edited Memory.md'
            )
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(request.text, encoding='utf-8')
    return {
        'name': request.name,
        'desk': request.desk,
        'characters': len(request.text),
    }

list_proposals

list_proposals()

Every proposal, newest first, with the change each would make.

Returns:

Type	Description
dict[str, Any]	The proposals.

Source code in src/trading_agent/gateway/routes/file_routes.py

def list_proposals(self) -> dict[str, Any]:
    """Every proposal, newest first, with the change each would make.

    Returns:
        dict[str, Any]: The proposals.
    """
    proposals = []
    for proposal in self._proposals.all():
        entry = proposal.to_json_dictionary()
        entry['difference'] = self._proposals.difference(
            proposal.proposal_id
        )
        proposals.append(entry)
    return {
        'proposals': proposals,
    }

accept

accept(proposal_id)

Applies a proposal.

Parameters:

Name	Type	Description	Default
proposal_id	str	The proposal.	required

Returns:

Type	Description
dict[str, Any]	The file that now holds the proposed text.

Raises:

Type	Description
HTTPException	The proposal does not exist or is not pending (status 409).

Source code in src/trading_agent/gateway/routes/file_routes.py

def accept(self, proposal_id: str) -> dict[str, Any]:
    """Applies a proposal.

    Args:
        proposal_id (str): The proposal.

    Returns:
        dict[str, Any]: The file that now holds the proposed text.

    Raises:
        fastapi.HTTPException: The proposal does not exist or is not pending (status 409).
    """
    try:
        target = self._proposals.accept(proposal_id)
    except proposal_store_module.ProposalError as error:
        raise fastapi.HTTPException(
            status_code=409, detail=str(error)
        ) from error
    return {
        'accepted': proposal_id,
        'file': str(target),
    }

reject

reject(proposal_id)

Rejects a proposal.

Parameters:

Name	Type	Description	Default
proposal_id	str	The proposal.	required

Returns:

Type	Description
dict[str, Any]	A confirmation.

Raises:

Type	Description
HTTPException	The proposal does not exist or is not pending (status 409).

Source code in src/trading_agent/gateway/routes/file_routes.py

def reject(self, proposal_id: str) -> dict[str, Any]:
    """Rejects a proposal.

    Args:
        proposal_id (str): The proposal.

    Returns:
        dict[str, Any]: A confirmation.

    Raises:
        fastapi.HTTPException: The proposal does not exist or is not pending (status 409).
    """
    try:
        self._proposals.reject(proposal_id)
    except proposal_store_module.ProposalError as error:
        raise fastapi.HTTPException(
            status_code=409, detail=str(error)
        ) from error
    return {
        'rejected': proposal_id,
    }