trading_agent.gateway.routes.schedule_routes

Routes for each desk's scheduled jobs and price alerts.

GET /api/desks/{desk}/jobs the jobs, built-in ones included, with their next runs, and the recent runs POST /api/desks/{desk}/jobs create a job, as cron_create does DELETE /api/desks/{desk}/jobs/{job_id} delete a job; the built-in ones cannot be deleted here POST /api/desks/{desk}/jobs/{job_id}/run start a job now GET /api/desks/{desk}/alerts the desk's price alerts DELETE /api/desks/{desk}/alerts/{alert_id} remove an alert

Typical usage example:

application.include_router(ScheduleRoutes(guard, parts, scheduler).router())

JobRequest

Bases: BaseModel

The body of a new job.

Attributes:

Name	Type	Description
name	str	A short name.
schedule	str | None	Five cron fields in India time, or None for a one-off job.
run_at	str | None	The ISO 8601 time of a one-off job, or None.
instructions	str | None	For an agent job, the message or the subagent's task.
command	str | None	For a command job instead, the shell command.
target	str	"desk" or "subagent:".
trading_days_only	bool	Whether to skip weekends and exchange holidays.
market_hours_only	bool	Whether to run only while NSE equities trade.
deliver_url	str | None	An address to POST the result to, or None.

ScheduleRoutes

ScheduleRoutes(guard, parts, scheduler)

Bases: RouteGroup

The job and alert routes.

Creates the routes.

Parameters:

Name	Type	Description	Default
guard	AccessGuard	Checks the access token.	required
parts	Mapping[str, DeskParts]	Each desk's stores.	required
scheduler	CronScheduler	Works out next runs and starts jobs now.	required

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

def __init__(
    self,
    guard: access_guard.AccessGuard,
    parts: Mapping[str, gateway_parts.DeskParts],
    scheduler: cron_scheduler_module.CronScheduler,
):
    """Creates the routes.

    Args:
        guard (access_guard.AccessGuard): Checks the access token.
        parts (Mapping[str, gateway_parts.DeskParts]): Each desk's stores.
        scheduler (cron_scheduler_module.CronScheduler): Works out next runs and starts jobs now.
    """
    super().__init__(guard)
    self._parts = parts
    self._scheduler = scheduler

list_jobs

list_jobs(desk_name)

The desk's jobs and recent runs.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	Each job with its description, next three runs and last run, and the latest runs.

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

def list_jobs(self, desk_name: str) -> dict[str, Any]:
    """The desk's jobs and recent runs.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: Each job with its description, next three runs and last run, and the latest runs.
    """
    parts = self._find(self._parts, desk_name, 'desk')
    jobs = []
    for job in [*parts.built_in_jobs, *parts.jobs.jobs()]:
        entry = dataclasses.asdict(job)
        entry['description'] = job.describe()
        entry['next_runs'] = [
            moment.isoformat()
            for moment in self._scheduler.upcoming(job, 3)
        ]
        last_run = parts.jobs.last_run(job.job_id)
        entry['last_run'] = last_run.isoformat() if last_run else None
        jobs.append(entry)
    return {
        'jobs': jobs,
        'recent_runs': parts.jobs.recent_runs(_RECENT_RUNS),
    }

create_job

create_job(desk_name, request)

Creates a job.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
request	JobRequest	The job.	required

Returns:

Type	Description
dict[str, Any]	The job, with its description and next runs.

Raises:

Type	Description
HTTPException	The job is not valid (status 400).

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

def create_job(self, desk_name: str, request: JobRequest) -> dict[str, Any]:
    """Creates a job.

    Args:
        desk_name (str): The desk.
        request (JobRequest): The job.

    Returns:
        dict[str, Any]: The job, with its description and next runs.

    Raises:
        fastapi.HTTPException: The job is not valid (status 400).
    """
    parts = self._find(self._parts, desk_name, 'desk')
    if (request.instructions is None) == (request.command is None):
        raise fastapi.HTTPException(
            status_code=400, detail='Give instructions or command, not both'
        )
    kind = cron_job.AGENT if request.command is None else cron_job.COMMAND
    try:
        job = cron_job.CronJob.new(
            name=request.name,
            kind=kind,
            schedule=request.schedule,
            run_at=request.run_at,
            text=request.instructions or request.command,
            target=request.target,
            trading_days_only=request.trading_days_only,
            market_hours_only=request.market_hours_only,
            deliver_url=request.deliver_url,
            created_by='user',
        )
    except cron_job.CronJobError as error:
        raise fastapi.HTTPException(
            status_code=400, detail=str(error)
        ) from error
    parts.jobs.add(job)
    entry = dataclasses.asdict(job)
    entry['description'] = job.describe()
    entry['next_runs'] = [
        moment.isoformat() for moment in self._scheduler.upcoming(job, 3)
    ]
    return entry

delete_job

delete_job(desk_name, job_id)

Deletes a job.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
job_id	str	The job.	required

Returns:

Type	Description
dict[str, Any]	The deleted job's description.

Raises:

Type	Description
HTTPException	The job is built in (status 400) or unknown (status 404).

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

def delete_job(self, desk_name: str, job_id: str) -> dict[str, Any]:
    """Deletes a job.

    Args:
        desk_name (str): The desk.
        job_id (str): The job.

    Returns:
        dict[str, Any]: The deleted job's description.

    Raises:
        fastapi.HTTPException: The job is built in (status 400) or unknown (status 404).
    """
    parts = self._find(self._parts, desk_name, 'desk')
    for job in parts.built_in_jobs:
        if job.job_id == job_id:
            raise fastapi.HTTPException(
                status_code=400,
                detail='Built-in jobs are set in trading_agent.toml',
            )
    removed = parts.jobs.delete(job_id)
    if removed is None:
        raise fastapi.HTTPException(
            status_code=404, detail=f'No job {job_id!r}'
        )
    parts.jobs.forget_run(job_id)
    return {
        'deleted': removed.describe(),
    }

run_job async

run_job(desk_name, job_id)

Starts a job now.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
job_id	str	The job.	required

Returns:

Type	Description
dict[str, Any]	A confirmation; results arrive as events.

Raises:

Type	Description
HTTPException	The desk or job is unknown (status 404).

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

async def run_job(self, desk_name: str, job_id: str) -> dict[str, Any]:
    """Starts a job now.

    Args:
        desk_name (str): The desk.
        job_id (str): The job.

    Returns:
        dict[str, Any]: A confirmation; results arrive as events.

    Raises:
        fastapi.HTTPException: The desk or job is unknown (status 404).
    """
    self._find(self._parts, desk_name, 'desk')
    if not await self._scheduler.run_now(desk_name, job_id):
        raise fastapi.HTTPException(
            status_code=404, detail=f'No job {job_id!r}'
        )
    return {
        'started': job_id,
    }

list_alerts

list_alerts(desk_name)

The desk's price alerts.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required

Returns:

Type	Description
dict[str, Any]	The alerts, oldest first.

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

def list_alerts(self, desk_name: str) -> dict[str, Any]:
    """The desk's price alerts.

    Args:
        desk_name (str): The desk.

    Returns:
        dict[str, Any]: The alerts, oldest first.
    """
    parts = self._find(self._parts, desk_name, 'desk')
    alerts = []
    for alert in parts.alerts.alerts():
        entry = dataclasses.asdict(alert)
        entry['description'] = alert.describe()
        alerts.append(entry)
    return {
        'alerts': alerts,
    }

delete_alert

delete_alert(desk_name, alert_id)

Removes a price alert.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk.	required
alert_id	str	The alert.	required

Returns:

Type	Description
dict[str, Any]	The removed alert's description.

Raises:

Type	Description
HTTPException	The alert is unknown (status 404).

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

def delete_alert(self, desk_name: str, alert_id: str) -> dict[str, Any]:
    """Removes a price alert.

    Args:
        desk_name (str): The desk.
        alert_id (str): The alert.

    Returns:
        dict[str, Any]: The removed alert's description.

    Raises:
        fastapi.HTTPException: The alert is unknown (status 404).
    """
    parts = self._find(self._parts, desk_name, 'desk')
    removed = parts.alerts.remove(alert_id)
    if removed is None:
        raise fastapi.HTTPException(
            status_code=404, detail=f'No alert {alert_id!r}'
        )
    return {
        'deleted': removed.describe(),
    }