Custom Timetables

Concepts covered: Schedules and timetables

A close-of-day run for a trading desk happens at 18:00 New York on each day the market is open, and its window runs from the previous close to this one. Weekdays are a cron matter. The 9 or so exchange holidays a year are a list no cron string holds, and the window after a Thursday holiday has to span 48 hours. When neither built-in cron timetable can say when the next window ends, you write the rule as a Timetable subclass, and the scheduler calls it exactly the way it calls the built-in ones.

The 2 questions a timetable answers

restriction carries the DAG's bounds: earliest is the start_date, latest is the end_date or None, and catchup is the DAG's catchup flag. With catchup on, the first window starts at the first close on or after start_date; with it off, the first window starts at the most recent close before now, the same jump to the latest window the built-in timetables make.

from pendulum import DateTime, timezone
from airflow.timetables.base import DagRunInfo, DataInterval, Timetable

NY = timezone("America/New_York")
HOLIDAYS = {"2026-11-26", "2026-12-25"}


def is_trading_day(day: DateTime) -> bool:
    return day.weekday() < 5 and day.to_date_string() not in HOLIDAYS


class TradingCloseTimetable(Timetable):
    description = "Close of each trading day, 18:00 New York"

    def _close_after(self, moment: DateTime) -> DateTime:
        day = moment.in_timezone(NY).start_of("day")
        while True:
            close = day.at(18, 0)
            if close > moment and is_trading_day(day):
                return close
            day = day.add(days=1)

    def _close_before(self, moment: DateTime) -> DateTime:
        day = moment.in_timezone(NY).start_of("day")
        while True:
            close = day.at(18, 0)
            if close <= moment and is_trading_day(day):
                return close
            day = day.subtract(days=1)

    def infer_manual_data_interval(self, *, run_after):
        end = self._close_before(run_after)
        return DataInterval(start=self._close_before(end.subtract(seconds=1)), end=end)

    def next_dagrun_info(self, *, last_automated_data_interval, restriction):
        if last_automated_data_interval is not None:
            start = last_automated_data_interval.end
        elif restriction.earliest is None:
            return None
        elif restriction.catchup:
            start = self._close_after(restriction.earliest.subtract(seconds=1))
        else:
            start = self._close_before(DateTime.utcnow())
        end = self._close_after(start)
        if restriction.latest is not None and end > restriction.latest:
            return None
        return DagRunInfo.interval(start=start, end=end)

    def serialize(self):
        return {}

    @classmethod
    def deserialize(cls, data):
        return cls()

The 2 private helpers walk the calendar a day at a time, which is slow in theory and instant in practice: a loop that skips a 3-day weekend and a holiday runs 5 iterations. serialize returns an empty dict because this timetable has no settings; a version that took the holiday list as a constructor argument would return it there and rebuild from it in deserialize.

Registering and using the timetable

from datetime import datetime
from airflow.plugins_manager import AirflowPlugin
from airflow.sdk import dag

from trading_calendar import TradingCloseTimetable


class TradingCalendarPlugin(AirflowPlugin):
    name = "trading_calendar"
    timetables = [TradingCloseTimetable]


@dag(
    schedule=TradingCloseTimetable(),
    start_date=datetime(2026, 11, 23),
    catchup=True,
)
def close_of_day():
    ...


close_of_day()

The plugin must be present wherever the DAG is read: the DAG processor parses the file and needs the class to build the DAG, the scheduler needs it to call next_dagrun_info, and the API server needs it to show the description in the UI. The serialized DAG stores the class's import path and the dict from serialize, and a timetable class that is not registered through a plugin fails at parse time, with the DAG shown as broken in the UI.

The windows around Thanksgiving 2026

Thursday Nov 26 is in HOLIDAYS, so _close_after(Wed 18:00) steps past Thursday and returns Friday's close, and run 3 covers 48 hours of trades. The weekend gives run 4 a 72-hour window. The logical date is the window's start, as it is for every data interval timetable, so run 3's logical date is Wednesday evening even though it is created on Friday. A person who triggers the DAG by hand on Saturday gets infer_manual_data_interval(Saturday): the window from Friday's close back to the close before it, the same window run 3 read.

About this lesson

This section is part of the Scheduling: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.