sdmxabs.fetch_gdp

  1"""Fetch GDP data from the ABS SDMX API with specific seasonality and price measure options."""
  2
  3from typing import Literal, Unpack
  4
  5import pandas as pd
  6
  7from sdmxabs.download_cache import GetFileKwargs
  8from sdmxabs.fetch_selection import MatchType as Mt
  9from sdmxabs.fetch_selection import fetch_selection
 10
 11# --- constants
 12PRICE_MAP = {"cvm": "Chain volume measures", "cp": "Current prices"}  # MEASURE
 13SEAS_MAP = {"o": "Original", "s": "Seasonally Adjusted", "t": "Trend"}  # TSEST
 14
 15
 16# --- public functions
 17def fetch_gdp(
 18    seasonality: Literal["o", "s", "t"] = "o",
 19    price_measure: Literal["cp", "cvm"] = "cp",
 20    parameters: dict[str, str] | None = None,
 21    *,
 22    validate: bool = False,
 23    **kwargs: Unpack[GetFileKwargs],
 24) -> tuple[pd.DataFrame, pd.DataFrame]:
 25    """Fetch quarterly GDP data in $ from the ABS SDMX API.
 26
 27    Args:
 28        seasonality (str): Type of seasonal adjustment to apply:
 29            - "o": Original data without seasonal adjustment (default)
 30            - "s": Seasonally adjusted data
 31            - "t": Trend data
 32        price_measure (str): Price measure type:
 33            - "cp": Current prices (default)
 34            - "cvm": Chain volume measures
 35        parameters (dict[str, str] | None): Additional parameters for the API request,
 36            such as 'startPeriod'.
 37        validate (bool, optional): If True, validate the selection against the flow's
 38            required dimensions when generating the URL key. Defaults to False.
 39        **kwargs: Additional arguments passed to the fetch_selection() function
 40
 41    Returns:
 42        tuple[pd.DataFrame, pd.DataFrame]: A tuple containing the GDP data and metadata
 43
 44    Raises:
 45        ValueError: If invalid seasonality or price_measure values are provided
 46
 47    """
 48    # report the parameters used if requested
 49    verbose = kwargs.get("verbose", False)
 50    if verbose:
 51        print(f"fetch_gdp(): {seasonality=}, {price_measure=} {validate=} {kwargs=}")
 52
 53    # Validate inputs
 54    if seasonality not in SEAS_MAP:
 55        error = f"Invalid '{seasonality=}'. Must be one of: {list(SEAS_MAP.keys())}"
 56        raise ValueError(error)
 57    if price_measure not in PRICE_MAP:
 58        error = f"Invalid '{price_measure=}'. Must be one of: {list(PRICE_MAP.keys())}"
 59        raise ValueError(error)
 60
 61    # build a selection criteria
 62    selection_criteria = [
 63        (SEAS_MAP[seasonality], "TSEST", Mt.EXACT),
 64        (PRICE_MAP[price_measure], "MEASURE", Mt.EXACT),
 65        ("Gross domestic product", "DATA_ITEM", Mt.EXACT),
 66    ]
 67    # return the data
 68    flow_id = "ANA_AGG"
 69    return fetch_selection(flow_id, selection_criteria, parameters, validate=validate, **kwargs)
 70
 71
 72if __name__ == "__main__":
 73
 74    def test_fetch_gdp() -> None:
 75        """Test function to fetch GDP data.
 76
 77        NOTE: The trend data is not available after 2019-Q1???
 78              Not sure why? [Report: as at 12 July 2025]
 79
 80        """
 81        failed = False
 82        parameters = {"startPeriod": "2019-Q1"}
 83        print(f"Testing with {parameters=}")
 84        for seasonality in ("o", "s", "t"):
 85            for price_measure in ("cp", "cvm"):
 86                gdp_data, metadata = fetch_gdp(
 87                    seasonality=seasonality,
 88                    price_measure=price_measure,
 89                    parameters=parameters,
 90                    verbose=False,
 91                )
 92                if gdp_data.empty or metadata.empty:
 93                    print(f"Test FAILED for {seasonality=} {price_measure=}")
 94                    failed = True
 95                    continue
 96                if len(metadata) != 1:
 97                    print(f"Test FAILED for {seasonality=} {price_measure=}")
 98                    failed = True
 99                if metadata.iloc[0]["TSEST"].lower()[0] != seasonality:
100                    print(f"Test FAILED for {seasonality=}")
101                    failed = True
102                if metadata.iloc[0]["MEASURE"] != PRICE_MAP[price_measure]:
103                    print(f"Test FAILED for {price_measure=}")
104                    failed = True
105                print(f"Test passed for {seasonality=} {price_measure=} ==> {len(gdp_data)} rows")
106        if not failed:
107            print("Test passed for all combinations of seasonality and price_measure")
108
109    test_fetch_gdp()