trading_agent.scheduling.cron_job

One scheduled job of a desk, and when it next runs.

A job either repeats on a five-field cron schedule in India time ("45 15 * * 1-5" is 15:45 on weekdays) or runs once at a given moment. Two options narrow a repeating schedule: trading_days_only skips weekends and exchange holidays, and market_hours_only keeps only the times when NSE equities are open.

There are three kinds of job:

• an agent job gives the desk a message, as if the user had sent it; with a subagent target, a subagent does the task instead and its report is published;
• a command job runs a shell command in the desk's sandbox, with no model;
• the built-in jobs, the heartbeat and the after-close rollover, come from trading_agent.toml rather than from the jobs file.

Typical usage example:

job = CronJob.repeating('Morning check', '0 9 * * 1-5', 'Review overnight news', created_by='user') moment = job.next_run_after(now, calendar, hours)

AGENT module-attribute

AGENT = 'agent'

COMMAND module-attribute

COMMAND = 'command'

DESK_TARGET module-attribute

DESK_TARGET = 'desk'

SUBAGENT_TARGET_PREFIX module-attribute

SUBAGENT_TARGET_PREFIX = 'subagent:'

CronJobError

Bases: ValueError

A job's schedule, time or fields are not valid.

CronJob dataclass

CronJob(job_id, name, kind, schedule, run_at, text, target=DESK_TARGET, trading_days_only=True, market_hours_only=False, deliver_url=None, created_by='user', created_at='', catch_up=True)

One scheduled job.

Attributes:

Name	Type	Description
job_id	str	A short identifier, used to delete the job.
name	str	A name for people.
kind	str	"agent" or "command".
schedule	str | None	The cron expression, or None for a one-off job.
run_at	str | None	The moment of a one-off job, or None for a repeating one.
text	str	For an agent job, the message for the desk or the subagent's task; for a command job, the command.
target	str	"desk", or "subagent:" for an agent job a subagent does.
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.
created_by	str	"user", "desk" or "built-in".
created_at	str	When the job was created, as an ISO 8601 string.
catch_up	bool	Whether a run missed while the gateway was down is made up once after a restart.

__post_init__

__post_init__()

Checks the fields.

Raises:

Type	Description
CronJobError	The job has neither or both of a schedule and a time, an invalid schedule or time, an unknown kind, or an invalid target.

Source code in src/trading_agent/scheduling/cron_job.py

def __post_init__(self) -> None:
    """Checks the fields.

    Raises:
        CronJobError: The job has neither or both of a schedule and a time, an invalid schedule or time, an unknown kind, or an invalid target.
    """
    if (self.schedule is None) == (self.run_at is None):
        raise CronJobError(
            'A job needs either a schedule or a run_at time, not both'
        )
    if self.schedule is not None and not croniter.croniter.is_valid(
        self.schedule
    ):
        raise CronJobError(
            f'{self.schedule!r} is not a valid cron schedule; use five fields: minute hour day month weekday, such as "45 15 * * 1-5"'
        )
    if self.run_at is not None:
        self.run_moment()
    if self.kind not in (AGENT, COMMAND):
        raise CronJobError(
            f'kind must be "agent" or "command", not {self.kind!r}'
        )
    if not self.text.strip():
        raise CronJobError('A job needs a message, task or command')
    is_desk = self.target == DESK_TARGET
    is_subagent = self.target.startswith(SUBAGENT_TARGET_PREFIX)
    if not is_desk and not is_subagent:
        raise CronJobError(
            f'target must be "desk" or "subagent:<name>", not {self.target!r}'
        )

new classmethod

new(**fields)

Creates a job with a fresh identifier and creation time.

Parameters:

Name	Type	Description	Default
**fields	Any	The job's other fields.	{}

Returns:

Type	Description
CronJob	The job.

Raises:

Type	Description
CronJobError	A field is not valid.

Source code in src/trading_agent/scheduling/cron_job.py

@classmethod
def new(cls, **fields: Any) -> Self:
    """Creates a job with a fresh identifier and creation time.

    Args:
        **fields (Any): The job's other fields.

    Returns:
        CronJob: The job.

    Raises:
        CronJobError: A field is not valid.
    """
    return cls(
        job_id=uuid.uuid4().hex[:8],
        created_at=datetime.datetime.now(_INDIA).isoformat(
            timespec='seconds'
        ),
        **fields,
    )

from_json_dictionary classmethod

from_json_dictionary(data)

Reads a job saved by to_json_dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The saved fields.	required

Returns:

Type	Description
CronJob	The job.

Raises:

Type	Description
CronJobError	A field is not valid.
TypeError	A field is missing or unknown.

Source code in src/trading_agent/scheduling/cron_job.py

@classmethod
def from_json_dictionary(cls, data: dict[str, Any]) -> Self:
    """Reads a job saved by to_json_dictionary.

    Args:
        data (dict[str, Any]): The saved fields.

    Returns:
        CronJob: The job.

    Raises:
        CronJobError: A field is not valid.
        TypeError: A field is missing or unknown.
    """
    return cls(**data)

to_json_dictionary

to_json_dictionary()

Converts the job to a dictionary that json.dumps can write.

Returns:

Type	Description
dict[str, Any]	The fields.

Source code in src/trading_agent/scheduling/cron_job.py

def to_json_dictionary(self) -> dict[str, Any]:
    """Converts the job to a dictionary that json.dumps can write.

    Returns:
        dict[str, Any]: The fields.
    """
    return dataclasses.asdict(self)

run_moment

run_moment()

The moment of a one-off job, in India time.

Returns:

Type	Description
datetime	The moment.

Raises:

Type	Description
CronJobError	The time is not ISO 8601.

Source code in src/trading_agent/scheduling/cron_job.py

def run_moment(self) -> datetime.datetime:
    """The moment of a one-off job, in India time.

    Returns:
        datetime.datetime: The moment.

    Raises:
        CronJobError: The time is not ISO 8601.
    """
    try:
        moment = datetime.datetime.fromisoformat(str(self.run_at))
    except ValueError as error:
        raise CronJobError(
            f'run_at {self.run_at!r} is not an ISO 8601 time'
        ) from error
    if moment.tzinfo is None:
        moment = moment.replace(tzinfo=_INDIA)
    return moment.astimezone(_INDIA)

next_run_after

next_run_after(moment, calendar, hours)

The first moment after the given one when the job runs.

Parameters:

Name	Type	Description	Default
moment	datetime	The moment to count from, not included.	required
calendar	MarketCalendar	Decides trading days.	required
hours	ExchangeHours	Decides market hours.	required

Returns:

Type	Description
datetime | None	The next run in India time, or None when a one-off job's time has passed or no allowed time is found.

Source code in src/trading_agent/scheduling/cron_job.py

def next_run_after(
    self,
    moment: datetime.datetime,
    calendar: market_calendar_module.MarketCalendar,
    hours: exchange_hours_module.ExchangeHours,
) -> datetime.datetime | None:
    """The first moment after the given one when the job runs.

    Args:
        moment (datetime.datetime): The moment to count from, not included.
        calendar (market_calendar_module.MarketCalendar): Decides trading days.
        hours (exchange_hours_module.ExchangeHours): Decides market hours.

    Returns:
        datetime.datetime | None: The next run in India time, or None when a one-off job's time has passed or no allowed time is found.
    """
    if self.run_at is not None:
        run_moment = self.run_moment()
        if run_moment > moment:
            return run_moment
        return None
    candidates = croniter.croniter(self.schedule, moment.astimezone(_INDIA))
    for _ in range(_MOST_CANDIDATES):
        candidate = candidates.get_next(datetime.datetime)
        if self._allowed(candidate, calendar, hours):
            return candidate
    return None

previous_run_before

previous_run_before(moment, calendar, hours)

The last moment before the given one when the job should have run.

Parameters:

Name	Type	Description	Default
moment	datetime	The moment to count back from.	required
calendar	MarketCalendar	Decides trading days.	required
hours	ExchangeHours	Decides market hours.	required

Returns:

Type	Description
datetime | None	The previous run in India time, or None for a one-off job still to come or when none is found.

Source code in src/trading_agent/scheduling/cron_job.py

def previous_run_before(
    self,
    moment: datetime.datetime,
    calendar: market_calendar_module.MarketCalendar,
    hours: exchange_hours_module.ExchangeHours,
) -> datetime.datetime | None:
    """The last moment before the given one when the job should have run.

    Args:
        moment (datetime.datetime): The moment to count back from.
        calendar (market_calendar_module.MarketCalendar): Decides trading days.
        hours (exchange_hours_module.ExchangeHours): Decides market hours.

    Returns:
        datetime.datetime | None: The previous run in India time, or None for a one-off job still to come or when none is found.
    """
    if self.run_at is not None:
        run_moment = self.run_moment()
        if run_moment <= moment:
            return run_moment
        return None
    candidates = croniter.croniter(self.schedule, moment.astimezone(_INDIA))
    for _ in range(_MOST_CANDIDATES):
        candidate = candidates.get_prev(datetime.datetime)
        if self._allowed(candidate, calendar, hours):
            return candidate
    return None

describe

describe()

The job in one line, for lists.

Returns:

Type	Description
str	Such as "Morning check (agent, 0 9 * * 1-5, trading days only)".

Source code in src/trading_agent/scheduling/cron_job.py

def describe(self) -> str:
    """The job in one line, for lists.

    Returns:
        str: Such as "Morning check (agent, 0 9 * * 1-5, trading days only)".
    """
    when = (
        self.schedule
        if self.schedule is not None
        else f'once at {self.run_at}'
    )
    options = []
    if self.trading_days_only and self.schedule is not None:
        options.append('trading days only')
    if self.market_hours_only:
        options.append('market hours only')
    if self.target != DESK_TARGET:
        options.append(self.target)
    if self.deliver_url:
        options.append(f'delivered to {self.deliver_url}')
    suffix = ''
    if options:
        suffix = ', ' + ', '.join(options)
    return f'{self.name} ({self.kind}, {when}{suffix})'