Sources API Reference

Auto-generated documentation from source code docstrings. Note these are low level source details; for user-facing guides see the User Guide.

RTDSM (Real-Time Data Set for Macroeconomists)

rtdsm

Philadelphia Fed Real-Time Data Set for Macroeconomists (RTDSM) source.

RTDSM has no API. The Philadelphia Fed publishes, for each macroeconomic series, a single spreadsheet containing the complete history of every vintage: rows are observation dates and columns are vintages (the data as they were known at successive points in time). This source downloads those spreadsheets and stores every vintage in the macrotrace data model, with no requirement to keep the Excel files on disk (an optional excel_dir lets callers archive them if they wish).

File-naming convention (confirmed across all 115 standard series), where the leading token is the series dataset_id (the Philadelphia Fed's mnemonic, e.g. ROUTPUT):

{DATASET_ID}{V}v{D}d.xlsx

where V is the vintage frequency (M or Q) and D is the data/observation frequency (M or Q); the two are independent. The series_key selects the vintage frequency, e.g. {"frequency": "Q"} for quarterly vintages or {"frequency": "M"} for monthly vintages. The data frequency is fixed per series and is looked up from the bundled catalog, so we never probe the server to discover a filename.

Vintage labels map to calendar dates per the Philadelphia Fed documentation: a quarterly vintage YYYY:Qq is dated to the middle (15th) of the middle month of the quarter (February, May, August, November); a monthly vintage YYYY:Mm is dated to the middle (15th) of month m.

To respect the provider (the files refresh only at the start of the month), the request cache for a file is set to expire at the start of the next calendar month, so repeated loads within the same month are served from the local cache without contacting philadelphiafed.org.

RTDSMSeries dataclass

One series' static metadata in the RTDSM catalog.

Attributes:

Name	Type	Description
title	str	Human-readable series title.
data_freq	str	Observation frequency, "Q" or "M" (fixed per series).
vintage_freqs	Tuple[str, ...]	The vintage frequencies the series is published at, a subset of ("Q", "M").

Source code in macrotrace/sources/rtdsm.py

@dataclass
class RTDSMSeries:
    """
    One series' static metadata in the RTDSM catalog.

    Attributes:
        title: Human-readable series title.
        data_freq: Observation frequency, "Q" or "M" (fixed per series).
        vintage_freqs: The vintage frequencies the series is published at, a
            subset of ("Q", "M").
    """

    title: str
    data_freq: str
    vintage_freqs: Tuple[str, ...]

ParsedVintageFile dataclass

Structured contents of one parsed vintage spreadsheet.

Attributes:

Name	Type	Description
vintages	List[Tuple[str, datetime]]	(vintage_label, release_date) pairs in column order.
cells	Dict[str, List[Tuple[datetime, float]]]	Maps each vintage label to its non-missing (observation_timestamp, value) pairs (#N/A cells are omitted).

Source code in macrotrace/sources/rtdsm.py

@dataclass
class ParsedVintageFile:
    """
    Structured contents of one parsed vintage spreadsheet.

    Attributes:
        vintages: (vintage_label, release_date) pairs in column order.
        cells: Maps each vintage label to its non-missing
            (observation_timestamp, value) pairs (#N/A cells are omitted).
    """

    vintages: List[Tuple[str, datetime]]
    cells: Dict[str, List[Tuple[datetime, float]]]

RTDSMAPIClient

Bases: APIClient

Downloads and parses a single RTDSM spreadsheet.

One client is created per (series, vintage frequency). The parsed workbook is memoized so the managers that make up one update share a single download and parse.

Source code in macrotrace/sources/rtdsm.py

class RTDSMAPIClient(APIClient):
    """
    Downloads and parses a single RTDSM spreadsheet.

    One client is created per (series, vintage frequency). The parsed workbook
    is memoized so the managers that make up one update share a single download
    and parse.
    """

    def __init__(
        self,
        dataset_id: str,
        filename: str,
        data_freq: str,
        excel_dir: Optional[str] = None,
        cache_settings: Optional[Dict[str, Any]] = None,
        cache_path: Optional[str] = None,
    ):
        self.dataset_id = dataset_id
        self.filename = filename
        self.data_freq = data_freq
        self.excel_dir = excel_dir
        self._parsed: Optional[ParsedVintageFile] = None
        super().__init__(
            base_url=RTDSM_BASE_URL,
            cache_settings=cache_settings,
            cache_path=cache_path,
        )

    def _get_request_headers(self) -> Dict[str, Any]:
        """RTDSM downloads need no special headers beyond the user agent."""
        return {}

    def _get_default_params(self) -> Dict[str, str]:
        """RTDSM downloads need no query parameters."""
        return {}

    @retry(
        stop=stop_after_attempt(4),
        wait=wait_exponential(max=30),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        retry=retry_if_exception_type(requests.RequestException),
        reraise=True,
    )
    def _download(self) -> bytes:
        """
        Download the spreadsheet bytes, honoring the month-aligned cache.

        Returns:
            bytes: The raw .xlsx content.

        Raises:
            ValueError: If the response is not a valid .xlsx file (the media
                server returns an HTML error page with HTTP 200 for missing
                files, so content is validated by its zip magic bytes).
        """
        url = self.base_url + self.filename
        headers = {"User-Agent": self.user_agent}
        logger.info(f"Downloading RTDSM file: {self.filename}")

        if isinstance(self.session, requests_cache.CachedSession):
            # Expire the cached copy at the start of next month so repeated
            # loads within a month never re-request the provider's server.
            expire_at = _first_of_next_month(datetime.now(UTC))
            response = self.session.get(url, headers=headers, expire_after=expire_at)
        else:
            response = self.session.get(url, headers=headers)

        is_cached = getattr(response, "from_cache", False)
        response.raise_for_status()
        content = response.content
        logger.debug(
            f"RTDSM download {self.filename}: status={response.status_code}, "
            f"cached={is_cached}, size={len(content)} bytes"
        )

        if content[:4] != b"PK\x03\x04":
            raise ValueError(
                f"RTDSM download for {self.filename} did not return a valid "
                f".xlsx file ({len(content)} bytes). The series may not exist "
                f"at the requested vintage frequency, or the provider returned "
                f"an error page."
            )

        if self.excel_dir:
            self._save_excel(content)

        return content

    def _save_excel(self, content: bytes) -> None:
        """
        Write the downloaded spreadsheet to ``excel_dir`` if requested.

        Args:
            content: The raw .xlsx content.
        """
        os.makedirs(self.excel_dir, exist_ok=True)
        path = os.path.join(self.excel_dir, self.filename)
        with open(path, "wb") as handle:
            handle.write(content)
        logger.info(f"Saved RTDSM spreadsheet to {path}")

    def get_parsed_file(self) -> ParsedVintageFile:
        """
        Return the parsed workbook, downloading and parsing once.

        Returns:
            ParsedVintageFile: The parsed vintages and observations.

        Raises:
            ValueError: If the download does not return a valid .xlsx file.
        """
        if self._parsed is None:
            content = self._download()
            self._parsed = _parse_workbook(content, self.dataset_id, self.data_freq)
            logger.info(
                f"Parsed RTDSM file {self.filename}: "
                f"{len(self._parsed.vintages)} vintage(s)"
            )
        return self._parsed

get_parsed_file()

Return the parsed workbook, downloading and parsing once.

Returns:

Name	Type	Description
ParsedVintageFile	ParsedVintageFile	The parsed vintages and observations.

Raises:

Type	Description
ValueError	If the download does not return a valid .xlsx file.

Source code in macrotrace/sources/rtdsm.py

def get_parsed_file(self) -> ParsedVintageFile:
    """
    Return the parsed workbook, downloading and parsing once.

    Returns:
        ParsedVintageFile: The parsed vintages and observations.

    Raises:
        ValueError: If the download does not return a valid .xlsx file.
    """
    if self._parsed is None:
        content = self._download()
        self._parsed = _parse_workbook(content, self.dataset_id, self.data_freq)
        logger.info(
            f"Parsed RTDSM file {self.filename}: "
            f"{len(self._parsed.vintages)} vintage(s)"
        )
    return self._parsed

RTDSMDatasetManager

Bases: DatasetManager

Source code in macrotrace/sources/rtdsm.py

class RTDSMDatasetManager(DatasetManager):
    def __init__(self, api_client: RTDSMAPIClient):
        super().__init__(api_client)

    def fetch_new_dataset_dimensions(
        self, state: UpdateState
    ) -> List[DatasetDimension]:
        """
        Create the single numeric dimension that defines the series.

        Unlike FRED (which versions its one dimension by realtime period) an
        RTDSM series has a single, static definition: one numeric dimension
        spanning every vintage. We create it once, on first load, with a
        ``valid_from`` early enough to cover every vintage in the file so all
        releases associate with it.

        Args:
            state: The update state.

        Returns:
            List[DatasetDimension]: The new dimension, or an empty list if it
                already exists or there are no releases to anchor it.
        """
        existing = self._get_all_local_dataset_dimensions(state.dataset.id)
        if existing:
            logger.debug(
                f"RTDSM dimension already exists for {self.api_client.dataset_id}; "
                f"no new dimensions."
            )
            return []

        parsed = self.api_client.get_parsed_file()
        if not parsed.vintages:
            logger.debug("No vintages found in RTDSM file; no dimension created.")
            return []

        earliest_release = min(release_date for _, release_date in parsed.vintages)
        info = RTDSM_CATALOG[self.api_client.dataset_id]
        dimension = DatasetDimension(
            dataset=state.dataset,
            dataset_dimension_id=self.api_client.dataset_id,
            title=info.title,
            type="numeric",
            frequency=_PANDAS_FREQ[self.api_client.data_freq],
            description=None,
            units=None,
            seasonal_adjustment=None,
            valid_from=earliest_release,
            valid_to=None,
        )
        logger.info(
            f"Created RTDSM dataset dimension for {self.api_client.dataset_id} "
            f"(valid_from={earliest_release})"
        )
        return [dimension]

fetch_new_dataset_dimensions(state)

Create the single numeric dimension that defines the series.

Unlike FRED (which versions its one dimension by realtime period) an RTDSM series has a single, static definition: one numeric dimension spanning every vintage. We create it once, on first load, with a valid_from early enough to cover every vintage in the file so all releases associate with it.

Parameters:

Name	Type	Description	Default
state	UpdateState	The update state.	required

Returns:

Type	Description
List[DatasetDimension]	List[DatasetDimension]: The new dimension, or an empty list if it already exists or there are no releases to anchor it.

Source code in macrotrace/sources/rtdsm.py

def fetch_new_dataset_dimensions(
    self, state: UpdateState
) -> List[DatasetDimension]:
    """
    Create the single numeric dimension that defines the series.

    Unlike FRED (which versions its one dimension by realtime period) an
    RTDSM series has a single, static definition: one numeric dimension
    spanning every vintage. We create it once, on first load, with a
    ``valid_from`` early enough to cover every vintage in the file so all
    releases associate with it.

    Args:
        state: The update state.

    Returns:
        List[DatasetDimension]: The new dimension, or an empty list if it
            already exists or there are no releases to anchor it.
    """
    existing = self._get_all_local_dataset_dimensions(state.dataset.id)
    if existing:
        logger.debug(
            f"RTDSM dimension already exists for {self.api_client.dataset_id}; "
            f"no new dimensions."
        )
        return []

    parsed = self.api_client.get_parsed_file()
    if not parsed.vintages:
        logger.debug("No vintages found in RTDSM file; no dimension created.")
        return []

    earliest_release = min(release_date for _, release_date in parsed.vintages)
    info = RTDSM_CATALOG[self.api_client.dataset_id]
    dimension = DatasetDimension(
        dataset=state.dataset,
        dataset_dimension_id=self.api_client.dataset_id,
        title=info.title,
        type="numeric",
        frequency=_PANDAS_FREQ[self.api_client.data_freq],
        description=None,
        units=None,
        seasonal_adjustment=None,
        valid_from=earliest_release,
        valid_to=None,
    )
    logger.info(
        f"Created RTDSM dataset dimension for {self.api_client.dataset_id} "
        f"(valid_from={earliest_release})"
    )
    return [dimension]

RTDSMReleaseManager

Bases: ReleaseManager

Source code in macrotrace/sources/rtdsm.py

class RTDSMReleaseManager(ReleaseManager):
    def __init__(self, api_client: RTDSMAPIClient):
        super().__init__(api_client)

    def fetch_new_releases(self, state: UpdateState) -> List[Release]:
        """
        Create a Release for each vintage column not already stored.

        The whole vintage history is downloaded, so releases are filtered
        client-side against any requested vintage window and against what is
        already in the database.

        Args:
            state: The update state.

        Returns:
            List[Release]: The new releases.
        """
        state.release_start_date = ensure_timezone(state.release_start_date, UTC)
        state.release_end_date = ensure_timezone(state.release_end_date, UTC)

        parsed = self.api_client.get_parsed_file()
        current_release_dates = self._get_current_releases_in_db(state.dataset.id)

        new_releases = []
        for _label, release_date in parsed.vintages:
            if self._is_new_release(
                release_date, current_release_dates
            ) and self._is_wanted_release(
                release_date, state.release_start_date, state.release_end_date
            ):
                new_releases.append(
                    Release(dataset=state.dataset, release_date=release_date)
                )

        logger.info(
            f"Found {len(new_releases)} new RTDSM release(s) out of "
            f"{len(parsed.vintages)} vintage(s)"
        )
        return new_releases

    def fetch_new_release_dimensions(
        self, state: UpdateState
    ) -> List[ReleaseDimension]:
        """
        Associate each new release with the series' single dimension.

        Args:
            state: The update state.

        Returns:
            List[ReleaseDimension]: The new release-dimension associations.

        Raises:
            ValueError: If the dataset has no dimension to associate.
        """
        all_dims = self._get_all_local_dataset_dimensions(state.dataset.id)
        if not all_dims:
            raise ValueError(
                f"Dataset {state.dataset.id} has no dimensions to associate "
                f"with releases."
            )

        new_release_dimensions = []
        for release in state.new_releases:
            for dimension in all_dims:
                in_lower_bound = release.release_date >= dimension.valid_from
                in_upper_bound = (
                    dimension.valid_to is None
                    or release.release_date <= dimension.valid_to
                )
                if in_lower_bound and in_upper_bound:
                    new_release_dimensions.append(
                        ReleaseDimension(release=release, dimension=dimension)
                    )

        logger.info(
            f"Created {len(new_release_dimensions)} RTDSM release-dimension "
            f"association(s)"
        )
        return new_release_dimensions

fetch_new_releases(state)

Create a Release for each vintage column not already stored.

The whole vintage history is downloaded, so releases are filtered client-side against any requested vintage window and against what is already in the database.

Parameters:

Name	Type	Description	Default
state	UpdateState	The update state.	required

Returns:

Type	Description
List[Release]	List[Release]: The new releases.

Source code in macrotrace/sources/rtdsm.py

def fetch_new_releases(self, state: UpdateState) -> List[Release]:
    """
    Create a Release for each vintage column not already stored.

    The whole vintage history is downloaded, so releases are filtered
    client-side against any requested vintage window and against what is
    already in the database.

    Args:
        state: The update state.

    Returns:
        List[Release]: The new releases.
    """
    state.release_start_date = ensure_timezone(state.release_start_date, UTC)
    state.release_end_date = ensure_timezone(state.release_end_date, UTC)

    parsed = self.api_client.get_parsed_file()
    current_release_dates = self._get_current_releases_in_db(state.dataset.id)

    new_releases = []
    for _label, release_date in parsed.vintages:
        if self._is_new_release(
            release_date, current_release_dates
        ) and self._is_wanted_release(
            release_date, state.release_start_date, state.release_end_date
        ):
            new_releases.append(
                Release(dataset=state.dataset, release_date=release_date)
            )

    logger.info(
        f"Found {len(new_releases)} new RTDSM release(s) out of "
        f"{len(parsed.vintages)} vintage(s)"
    )
    return new_releases

fetch_new_release_dimensions(state)

Associate each new release with the series' single dimension.

Parameters:

Name	Type	Description	Default
state	UpdateState	The update state.	required

Returns:

Type	Description
List[ReleaseDimension]	List[ReleaseDimension]: The new release-dimension associations.

Raises:

Type	Description
ValueError	If the dataset has no dimension to associate.

Source code in macrotrace/sources/rtdsm.py

def fetch_new_release_dimensions(
    self, state: UpdateState
) -> List[ReleaseDimension]:
    """
    Associate each new release with the series' single dimension.

    Args:
        state: The update state.

    Returns:
        List[ReleaseDimension]: The new release-dimension associations.

    Raises:
        ValueError: If the dataset has no dimension to associate.
    """
    all_dims = self._get_all_local_dataset_dimensions(state.dataset.id)
    if not all_dims:
        raise ValueError(
            f"Dataset {state.dataset.id} has no dimensions to associate "
            f"with releases."
        )

    new_release_dimensions = []
    for release in state.new_releases:
        for dimension in all_dims:
            in_lower_bound = release.release_date >= dimension.valid_from
            in_upper_bound = (
                dimension.valid_to is None
                or release.release_date <= dimension.valid_to
            )
            if in_lower_bound and in_upper_bound:
                new_release_dimensions.append(
                    ReleaseDimension(release=release, dimension=dimension)
                )

    logger.info(
        f"Created {len(new_release_dimensions)} RTDSM release-dimension "
        f"association(s)"
    )
    return new_release_dimensions

RTDSMSeriesManager

Bases: SeriesManager

Source code in macrotrace/sources/rtdsm.py

class RTDSMSeriesManager(SeriesManager):
    def __init__(self, api_client: RTDSMAPIClient):
        super().__init__(api_client)

    def fetch_new_series_dimension_filters(self, state: UpdateState) -> List:
        """
        RTDSM series have no dimension filters.

        The ``frequency`` entry in the series key selects which spreadsheet
        (vintage cadence) to download; it is not a dataset dimension, so there
        are no SeriesDimensionFilter rows to create. The base implementation
        would try to look up a dimension named "frequency" and fail, so we
        override it to return nothing.

        Args:
            state: The update state.

        Returns:
            List: Always empty.
        """
        return []

fetch_new_series_dimension_filters(state)

RTDSM series have no dimension filters.

The frequency entry in the series key selects which spreadsheet (vintage cadence) to download; it is not a dataset dimension, so there are no SeriesDimensionFilter rows to create. The base implementation would try to look up a dimension named "frequency" and fail, so we override it to return nothing.

Parameters:

Name	Type	Description	Default
state	UpdateState	The update state.	required

Returns:

Name	Type	Description
List	List	Always empty.

Source code in macrotrace/sources/rtdsm.py

def fetch_new_series_dimension_filters(self, state: UpdateState) -> List:
    """
    RTDSM series have no dimension filters.

    The ``frequency`` entry in the series key selects which spreadsheet
    (vintage cadence) to download; it is not a dataset dimension, so there
    are no SeriesDimensionFilter rows to create. The base implementation
    would try to look up a dimension named "frequency" and fail, so we
    override it to return nothing.

    Args:
        state: The update state.

    Returns:
        List: Always empty.
    """
    return []

RTDSMObservationManager

Bases: ObservationManager

Source code in macrotrace/sources/rtdsm.py

class RTDSMObservationManager(ObservationManager):
    def __init__(self, api_client: RTDSMAPIClient):
        super().__init__(api_client)

    def fetch_new_observations(self, state: UpdateState) -> List[Observation]:
        """
        Create observations for every non-missing cell of the new releases.

        Args:
            state: The update state.

        Returns:
            List[Observation]: The new observations.
        """
        if not state.new_releases:
            logger.debug("No new RTDSM releases; no observations to create.")
            return []

        parsed = self.api_client.get_parsed_file()
        date_to_label = {release_date: label for label, release_date in parsed.vintages}

        new_observations = []
        for release in tqdm(
            state.new_releases, desc="Processing RTDSM vintages", leave=False
        ):
            label = date_to_label.get(release.release_date)
            if label is None:
                logger.warning(
                    f"No vintage column found for release "
                    f"{release.release_date}; skipping."
                )
                continue
            for obs_timestamp, value in parsed.cells.get(label, []):
                new_observations.append(
                    Observation(
                        series=state.series,
                        release=release,
                        observation_timestamp=obs_timestamp,
                        value=value,
                    )
                )

        logger.info(f"Created {len(new_observations)} new RTDSM observation(s)")
        return new_observations

fetch_new_observations(state)

Create observations for every non-missing cell of the new releases.

Parameters:

Name	Type	Description	Default
state	UpdateState	The update state.	required

Returns:

Type	Description
List[Observation]	List[Observation]: The new observations.

Source code in macrotrace/sources/rtdsm.py

def fetch_new_observations(self, state: UpdateState) -> List[Observation]:
    """
    Create observations for every non-missing cell of the new releases.

    Args:
        state: The update state.

    Returns:
        List[Observation]: The new observations.
    """
    if not state.new_releases:
        logger.debug("No new RTDSM releases; no observations to create.")
        return []

    parsed = self.api_client.get_parsed_file()
    date_to_label = {release_date: label for label, release_date in parsed.vintages}

    new_observations = []
    for release in tqdm(
        state.new_releases, desc="Processing RTDSM vintages", leave=False
    ):
        label = date_to_label.get(release.release_date)
        if label is None:
            logger.warning(
                f"No vintage column found for release "
                f"{release.release_date}; skipping."
            )
            continue
        for obs_timestamp, value in parsed.cells.get(label, []):
            new_observations.append(
                Observation(
                    series=state.series,
                    release=release,
                    observation_timestamp=obs_timestamp,
                    value=value,
                )
            )

    logger.info(f"Created {len(new_observations)} new RTDSM observation(s)")
    return new_observations

RTDSMUpdateManager

Bases: UpdateManager

Source code in macrotrace/sources/rtdsm.py

class RTDSMUpdateManager(UpdateManager):
    NATIVE_OBSERVATION_TZ = UTC

    def __init__(
        self,
        dataset_id: str,
        source: str = RTDSM_SOURCE,
        series_key: Optional[Dict] = None,
        release_start_date: Optional[datetime] = None,
        release_end_date: Optional[datetime] = None,
        db_path: Optional[str] = None,
        cache_settings: Optional[Dict[str, Any]] = None,
        cache_path: Optional[str] = None,
        excel_dir: Optional[str] = None,
    ):
        """
        Initialize an RTDSM update manager for a single series.

        Args:
            dataset_id: The series identifier (e.g. "ROUTPUT"); case-insensitive.
            source: The source name; defaults to "RTDSM".
            series_key: Optionally ``{"frequency": "Q"}`` or ``{"frequency":
                "M"}`` to select the vintage frequency. If omitted, defaults to
                the series' only vintage frequency, or quarterly when both are
                offered.
            release_start_date: Optional lower bound on vintage dates to load.
            release_end_date: Optional upper bound on vintage dates to load.
            db_path: Optional path to the SQLite database.
            cache_settings: Optional request-cache settings.
            cache_path: Optional path to the request-cache SQLite file.
            excel_dir: Optional directory in which to save the downloaded
                spreadsheet. If None, the file is parsed in memory and not kept.

        Raises:
            ValueError: If ``dataset_id`` is not a known RTDSM series, or if the
                requested vintage frequency is not offered by that series. The
                catalog is bundled, so an unknown series fails fast here rather
                than at fetch time.
        """
        dataset_id = dataset_id.upper()
        self.dataset_id = dataset_id
        self.excel_dir = excel_dir

        info = RTDSM_CATALOG.get(dataset_id)
        if info is None:
            raise ValueError(
                f"Unknown RTDSM series {dataset_id!r}. See "
                f"macrotrace.sources.rtdsm.RTDSM_CATALOG for the "
                f"{len(RTDSM_CATALOG)} supported series identifiers."
            )

        requested = series_key.get("frequency") if series_key else None
        self.vintage_freq = _resolve_vintage_freq(dataset_id, info, requested)
        self.data_freq = info.data_freq
        self.filename = _build_filename(dataset_id, self.vintage_freq, info.data_freq)
        resolved_series_key = {"frequency": self.vintage_freq}
        logger.debug(
            f"Initializing RTDSMUpdateManager for {dataset_id} "
            f"(vintage_freq={self.vintage_freq}, file={self.filename})"
        )

        super().__init__(
            dataset_id=dataset_id,
            source=source,
            series_key=resolved_series_key,
            release_start_date=release_start_date,
            release_end_date=release_end_date,
            db_path=db_path,
            cache_settings=cache_settings,
            cache_path=cache_path,
        )

    def _create_api_client(
        self,
        cache_settings: Optional[Dict[str, Any]] = None,
        cache_path: Optional[str] = None,
    ) -> RTDSMAPIClient:
        return RTDSMAPIClient(
            dataset_id=self.dataset_id,
            filename=self.filename,
            data_freq=self.data_freq,
            excel_dir=self.excel_dir,
            cache_settings=cache_settings,
            cache_path=cache_path,
        )

    def _create_dataset_manager(self) -> DatasetManager:
        return RTDSMDatasetManager(self.api_client)

    def _create_release_manager(self) -> ReleaseManager:
        return RTDSMReleaseManager(self.api_client)

    def _create_series_manager(self) -> SeriesManager:
        return RTDSMSeriesManager(self.api_client)

    def _create_observation_manager(self) -> ObservationManager:
        return RTDSMObservationManager(self.api_client)

__init__(dataset_id, source=RTDSM_SOURCE, series_key=None, release_start_date=None, release_end_date=None, db_path=None, cache_settings=None, cache_path=None, excel_dir=None)

Initialize an RTDSM update manager for a single series.

Parameters:

Name	Type	Description	Default
dataset_id	str	The series identifier (e.g. "ROUTPUT"); case-insensitive.	required
source	str	The source name; defaults to "RTDSM".	RTDSM_SOURCE
series_key	Optional[Dict]	Optionally {"frequency": "Q"} or {"frequency": "M"} to select the vintage frequency. If omitted, defaults to the series' only vintage frequency, or quarterly when both are offered.	None
release_start_date	Optional[datetime]	Optional lower bound on vintage dates to load.	None
release_end_date	Optional[datetime]	Optional upper bound on vintage dates to load.	None
db_path	Optional[str]	Optional path to the SQLite database.	None
cache_settings	Optional[Dict[str, Any]]	Optional request-cache settings.	None
cache_path	Optional[str]	Optional path to the request-cache SQLite file.	None
excel_dir	Optional[str]	Optional directory in which to save the downloaded spreadsheet. If None, the file is parsed in memory and not kept.	None

Raises:

Type	Description
ValueError	If dataset_id is not a known RTDSM series, or if the requested vintage frequency is not offered by that series. The catalog is bundled, so an unknown series fails fast here rather than at fetch time.

Source code in macrotrace/sources/rtdsm.py

def __init__(
    self,
    dataset_id: str,
    source: str = RTDSM_SOURCE,
    series_key: Optional[Dict] = None,
    release_start_date: Optional[datetime] = None,
    release_end_date: Optional[datetime] = None,
    db_path: Optional[str] = None,
    cache_settings: Optional[Dict[str, Any]] = None,
    cache_path: Optional[str] = None,
    excel_dir: Optional[str] = None,
):
    """
    Initialize an RTDSM update manager for a single series.

    Args:
        dataset_id: The series identifier (e.g. "ROUTPUT"); case-insensitive.
        source: The source name; defaults to "RTDSM".
        series_key: Optionally ``{"frequency": "Q"}`` or ``{"frequency":
            "M"}`` to select the vintage frequency. If omitted, defaults to
            the series' only vintage frequency, or quarterly when both are
            offered.
        release_start_date: Optional lower bound on vintage dates to load.
        release_end_date: Optional upper bound on vintage dates to load.
        db_path: Optional path to the SQLite database.
        cache_settings: Optional request-cache settings.
        cache_path: Optional path to the request-cache SQLite file.
        excel_dir: Optional directory in which to save the downloaded
            spreadsheet. If None, the file is parsed in memory and not kept.

    Raises:
        ValueError: If ``dataset_id`` is not a known RTDSM series, or if the
            requested vintage frequency is not offered by that series. The
            catalog is bundled, so an unknown series fails fast here rather
            than at fetch time.
    """
    dataset_id = dataset_id.upper()
    self.dataset_id = dataset_id
    self.excel_dir = excel_dir

    info = RTDSM_CATALOG.get(dataset_id)
    if info is None:
        raise ValueError(
            f"Unknown RTDSM series {dataset_id!r}. See "
            f"macrotrace.sources.rtdsm.RTDSM_CATALOG for the "
            f"{len(RTDSM_CATALOG)} supported series identifiers."
        )

    requested = series_key.get("frequency") if series_key else None
    self.vintage_freq = _resolve_vintage_freq(dataset_id, info, requested)
    self.data_freq = info.data_freq
    self.filename = _build_filename(dataset_id, self.vintage_freq, info.data_freq)
    resolved_series_key = {"frequency": self.vintage_freq}
    logger.debug(
        f"Initializing RTDSMUpdateManager for {dataset_id} "
        f"(vintage_freq={self.vintage_freq}, file={self.filename})"
    )

    super().__init__(
        dataset_id=dataset_id,
        source=source,
        series_key=resolved_series_key,
        release_start_date=release_start_date,
        release_end_date=release_end_date,
        db_path=db_path,
        cache_settings=cache_settings,
        cache_path=cache_path,
    )