trading_agent.scheduling.cron_scheduler

Runs every desk's scheduled jobs: the user's, the desk's own, the heartbeat and the after-close rollover.

The scheduler is a loop with no model. When it starts, it makes up once for each run missed while the gateway was down, if the missed run was within the last 24 hours and the job allows it. Then, every 30 seconds or sooner, it starts each job whose next run has come:

• an agent job for the desk joins the desk's queue as a trigger, so it never runs beside one of the desk's turns; the heartbeat is skipped when the desk is busy or has triggers waiting, so heartbeats cannot pile up;
• an agent job for a subagent and a command job run on their own, and their result is published as a cron_result event;
• a job with a delivery address has its result POSTed there, with secrets masked;
• a one-off job is deleted after it starts.

Each run is recorded in the desk's runs.jsonl before it starts, so a crash cannot make a job run twice.

Typical usage example:

scheduler = CronScheduler(desk_schedules, calendar, hours, event_bus, redactor) task = asyncio.create_task(scheduler.run())

HEARTBEAT_JOB_ID module-attribute

HEARTBEAT_JOB_ID = 'heartbeat'

ROLLOVER_JOB_ID module-attribute

ROLLOVER_JOB_ID = 'rollover'

DeskSchedule dataclass

DeskSchedule(desk, store, built_in_jobs, command_runner, subagent_runner)

What the scheduler needs for one desk.

Attributes:

Name	Type	Description
desk	DeskRuntime	The running desk.
store	CronJobStore	Its jobs and run record.
built_in_jobs	list[CronJob]	The heartbeat and rollover jobs from trading_agent.toml.
command_runner	CommandJobRunner	Runs its command jobs.
subagent_runner	SubagentRunner	Runs its subagent jobs.

all_jobs

all_jobs()

The built-in jobs, then the stored ones.

Returns:

Type	Description
list[CronJob]	Every job of the desk.

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

def all_jobs(self) -> list[cron_job_module.CronJob]:
    """The built-in jobs, then the stored ones.

    Returns:
        list[cron_job_module.CronJob]: Every job of the desk.
    """
    return [*self.built_in_jobs, *self.store.jobs()]

CronScheduler

CronScheduler(schedules, calendar, hours, event_bus, redactor=None, clock=None)

Starts scheduled jobs on time.

Subscribes to the bus at once, so no result to deliver is missed.

Parameters:

Name	Type	Description	Default
schedules	Mapping[str, DeskSchedule]	Each desk's schedule, keyed by desk name.	required
calendar	MarketCalendar	Decides trading days.	required
hours	ExchangeHours	Decides market hours.	required
event_bus	EventBus	Where runs and results are published, and desk answers are read for delivery.	required
redactor	SecretRedactor | None	Masks secrets in delivered results, or None.	None
clock	Callable[[], datetime] | None	Returns the current time; None uses the system clock. Tests replace it.	None

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

def __init__(
    self,
    schedules: Mapping[str, DeskSchedule],
    calendar: market_calendar_module.MarketCalendar,
    hours: exchange_hours_module.ExchangeHours,
    event_bus: event_bus_module.EventBus,
    redactor: secret_redactor_module.SecretRedactor | None = None,
    clock: Callable[[], datetime.datetime] | None = None,
):
    """Subscribes to the bus at once, so no result to deliver is missed.

    Args:
        schedules (Mapping[str, DeskSchedule]): Each desk's schedule, keyed by desk name.
        calendar (market_calendar_module.MarketCalendar): Decides trading days.
        hours (exchange_hours_module.ExchangeHours): Decides market hours.
        event_bus (event_bus_module.EventBus): Where runs and results are published, and desk answers are read for delivery.
        redactor (secret_redactor_module.SecretRedactor | None): Masks secrets in delivered results, or None.
        clock (Callable[[], datetime.datetime] | None): Returns the current time; None uses the system clock. Tests replace it.
    """
    self._schedules = schedules
    self._calendar = calendar
    self._hours = hours
    self._event_bus = event_bus
    self._redactor = redactor
    self._clock = clock
    self._started_at = self.now()
    self._pending_deliveries: dict[
        str, tuple[cron_job_module.CronJob, str]
    ] = {}
    self._running: set[asyncio.Task] = set()
    self._subscription = event_bus.subscribe()

now

now()

The current time in India.

Returns:

Type	Description
datetime	The time.

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

def now(self) -> datetime.datetime:
    """The current time in India.

    Returns:
        datetime.datetime: The time.
    """
    if self._clock is not None:
        return self._clock()
    return datetime.datetime.now(_INDIA)

run async

run()

Catches up, then starts due jobs until cancelled.

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

async def run(self) -> None:
    """Catches up, then starts due jobs until cancelled."""
    delivery = asyncio.create_task(self._deliver_desk_answers())
    try:
        await self.catch_up()
        while True:
            soonest = await self.tick()
            wait = (soonest - self.now()).total_seconds()
            await asyncio.sleep(max(1.0, wait))
    finally:
        delivery.cancel()
        for task in list(self._running):
            task.cancel()
        self._subscription.close()

catch_up async

catch_up()

Starts once each job that missed a run while the gateway was down.

Returns:

Type	Description
list[str]	The names of the jobs started.

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

async def catch_up(self) -> list[str]:
    """Starts once each job that missed a run while the gateway was down.

    Returns:
        list[str]: The names of the jobs started.
    """
    started = []
    for schedule in self._schedules.values():
        for job in schedule.all_jobs():
            missed = self._missed_run(schedule, job)
            if missed is not None:
                await self._start(schedule, job, missed, catching_up=True)
                started.append(job.name)
    return started

tick async

tick()

Starts every job whose next run has come.

Returns:

Type	Description
datetime	When to look again: the earliest next run, but no later than 30 seconds from now.

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

async def tick(self) -> datetime.datetime:
    """Starts every job whose next run has come.

    Returns:
        datetime.datetime: When to look again: the earliest next run, but no later than 30 seconds from now.
    """
    now = self.now()
    soonest = now + _LONGEST_SLEEP
    for schedule in self._schedules.values():
        for job in schedule.all_jobs():
            due = self._next_run(schedule, job)
            if due is None:
                continue
            if due <= now:
                await self._start(schedule, job, due, catching_up=False)
                due = self._next_run(schedule, job)
            if due is not None and due < soonest:
                soonest = due
    return soonest

run_now async

run_now(desk_name, job_id)

Starts one job at once, as if its time had come.

Parameters:

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

Returns:

Type	Description
bool	True when the job was found and started, False when the desk or job is unknown.

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

async def run_now(self, desk_name: str, job_id: str) -> bool:
    """Starts one job at once, as if its time had come.

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

    Returns:
        bool: True when the job was found and started, False when the desk or job is unknown.
    """
    schedule = self._schedules.get(desk_name)
    if schedule is None:
        return False
    for job in schedule.all_jobs():
        if job.job_id == job_id:
            await self._start(schedule, job, self.now(), catching_up=False)
            return True
    return False

upcoming

upcoming(job, count)

The next few runs of a job, for lists.

Parameters:

Name	Type	Description	Default
job	CronJob	The job.	required
count	int	How many runs.	required

Returns:

Type	Description
list[datetime]	The runs, soonest first.

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

def upcoming(
    self,
    job: cron_job_module.CronJob,
    count: int,
) -> list[datetime.datetime]:
    """The next few runs of a job, for lists.

    Args:
        job (cron_job_module.CronJob): The job.
        count (int): How many runs.

    Returns:
        list[datetime.datetime]: The runs, soonest first.
    """
    runs = []
    moment = self.now()
    for _ in range(count):
        moment = job.next_run_after(moment, self._calendar, self._hours)
        if moment is None:
            break
        runs.append(moment)
    return runs