trading_agent.tools.cron_create_tool

The cron_create tool: schedules a repeating job for the desk.

Typical usage example:

tool = CronCreateTool(store, calendar, hours)

CronCreateArguments

Bases: BaseModel

The arguments of cron_create.

Attributes:

Name	Type	Description
name	str	A short name for the job.
schedule	str	Five cron fields in India time: minute hour day month weekday, such as "30 9 * * 1-5".
instructions	str | None	For an agent job: the message you will receive at each run, or the subagent's task.
command	str | None	For a command job instead: the exact shell command to run, such as "python3 scan.py".
target	str	"desk" for yourself, or "subagent:" to have a subagent do it.
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.

CronCreateTool

CronCreateTool(store, calendar, hours)

Bases: Tool

Adds a repeating job.

Creates the tool.

Parameters:

Name	Type	Description	Default
store	CronJobStore	The desk's jobs.	required
calendar	MarketCalendar	Decides trading days.	required
hours	ExchangeHours	Decides market hours.	required

Source code in src/trading_agent/tools/cron_create_tool.py

def __init__(
    self,
    store: cron_job_store_module.CronJobStore,
    calendar: market_calendar_module.MarketCalendar,
    hours: exchange_hours_module.ExchangeHours,
):
    """Creates the tool.

    Args:
        store (cron_job_store_module.CronJobStore): The desk's jobs.
        calendar (market_calendar_module.MarketCalendar): Decides trading days.
        hours (exchange_hours_module.ExchangeHours): Decides market hours.
    """
    self._store = store
    self._calendar = calendar
    self._hours = hours

run async

run(arguments)

Checks and stores the job, and shows its next runs.

Parameters:

Name	Type	Description	Default
arguments	CronCreateArguments	The job.	required

Returns:

Type	Description
ToolResult	The job's id and next three runs, or why it was refused.

Source code in src/trading_agent/tools/cron_create_tool.py

async def run(self, arguments: CronCreateArguments) -> tool.ToolResult:
    """Checks and stores the job, and shows its next runs.

    Args:
        arguments (CronCreateArguments): The job.

    Returns:
        tool.ToolResult: The job's id and next three runs, or why it was refused.
    """
    kind = cron_job_module.AGENT
    text = arguments.instructions
    if arguments.command is not None:
        kind = cron_job_module.COMMAND
        text = arguments.command
    try:
        job = cron_job_module.CronJob.new(
            name=arguments.name,
            kind=kind,
            schedule=arguments.schedule.strip(),
            run_at=None,
            text=text,
            target=arguments.target,
            trading_days_only=arguments.trading_days_only,
            market_hours_only=arguments.market_hours_only,
            deliver_url=arguments.deliver_url,
            created_by='desk',
        )
    except cron_job_module.CronJobError as error:
        return tool.ToolResult.error(str(error))
    runs = []
    moment = datetime.datetime.now(_INDIA)
    for _ in range(3):
        moment = job.next_run_after(moment, self._calendar, self._hours)
        if moment is None:
            break
        runs.append(f'{moment:%a %d %b %H:%M}')
    if not runs:
        return tool.ToolResult.error(
            'That schedule never runs with these options; check the fields and the trading-day and market-hours settings.'
        )
    self._store.add(job)
    return tool.ToolResult.success(
        f'Scheduled {job.describe()} as job {job.job_id}. Next runs: {", ".join(runs)}.'
    )