Skip to content

trading_agent.tools.cron_list_tool

The cron_list tool: lists the desk's scheduled jobs and their next runs.

Typical usage example:

tool = CronListTool(store, built_in_jobs, calendar, hours)

CronListTool 

CronListTool(store, built_in_jobs, calendar, hours)

Bases: Tool

Lists the jobs.

Creates the tool.

Parameters:

Name Type Description Default
store CronJobStore

The desk's jobs.

 required
built_in_jobs Sequence[CronJob]

The heartbeat and rollover.

 required
calendar MarketCalendar

Decides trading days.

 required
hours ExchangeHours

Decides market hours.

 required
Source code in src/trading_agent/tools/cron_list_tool.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def __init__(
    self,
    store: cron_job_store_module.CronJobStore,
    built_in_jobs: Sequence[cron_job_module.CronJob],
    calendar: market_calendar_module.MarketCalendar,
    hours: exchange_hours_module.ExchangeHours,
):
    """Creates the tool.

    Args:
        store (cron_job_store_module.CronJobStore): The desk's jobs.
        built_in_jobs (Sequence[cron_job_module.CronJob]): The heartbeat and rollover.
        calendar (market_calendar_module.MarketCalendar): Decides trading days.
        hours (exchange_hours_module.ExchangeHours): Decides market hours.
    """
    self._store = store
    self._built_in_jobs = list(built_in_jobs)
    self._calendar = calendar
    self._hours = hours

run async 

run(arguments)

Lists every job with its next run.

Parameters:

Name Type Description Default
arguments NoArguments

No arguments.

 required

Returns:

Type Description
ToolResult

One line per job.
Source code in src/trading_agent/tools/cron_list_tool.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def run(self, arguments: tool.NoArguments) -> tool.ToolResult:
    """Lists every job with its next run.

    Args:
        arguments (tool.NoArguments): No arguments.

    Returns:
        tool.ToolResult: One line per job.
    """
    del arguments
    now = datetime.datetime.now(_INDIA)
    lines = []
    for job in [*self._built_in_jobs, *self._store.jobs()]:
        next_run = job.next_run_after(now, self._calendar, self._hours)
        when = (
            f'next {next_run:%a %d %b %H:%M}'
            if next_run
            else 'no further runs'
        )
        lines.append(
            f'{job.job_id}: {job.describe()}; {when}; by {job.created_by}'
        )
    return tool.ToolResult.success('\n'.join(lines))