Merge pull request #8 from ONEcampaign/7-sdr-data

7 sdr data

src/imf_reader/sdr/__init__.py

@@ -0,0 +1,52 @@
+"""Special Drawing Rights (SDR) reader module.
+
+This module offers access to the IMF's Special Drawing Rights (SDR) data.
+The SDR is an international reserve asset created by the IMF in 1969.
+It is not a currency, but the holder of SDRs can exchange them for usable currencies in times of need.
+
+Read more about SDRs at: https://www.imf.org/en/About/Factsheets/Sheets/2023/special-drawing-rights-sdr
+
+
+Usage:
+
+Import the module
+
+```python
+from imf_reader import sdr
+```
+
+Read allocations and holdings data
+
+```python
+sdr.fetch_allocations_holdings()
+```
+SDRs holdings and allocations are published at a monthly frequency. The function fetches the latest data available.
+
+To retrieve SDR holdings and allocations for a specific month and year, eg April 2021, pass the year and month as a tuple
+
+```python
+sdr.fetch_allocations_holdings((2021, 4))
+```
+
+Read interest rates
+
+```python
+sdr.fetch_interest_rates()
+```
+
+Read exchange rates
+
+```python
+sdr.fetch_exchange_rates()
+```
+By default, the exchange rate is in USDs per 1 SDR. To get the exchange rate in SDRs per 1 USD, pass the unit basis as "USD"
+
+```python
+sdr.fetch_exchange_rates("USD")
+```
+
+"""
+
+from imf_reader.sdr.read_interest_rate import fetch_interest_rates
+from imf_reader.sdr.read_exchange_rate import fetch_exchange_rates
+from imf_reader.sdr.read_announcements import fetch_allocations_holdings

src/imf_reader/sdr/read_announcements.py

@@ -0,0 +1,103 @@
+"""Module to get SDR data from the IMF website
+
+
+info: https://www.imf.org/en/About/Factsheets/Sheets/2023/special-drawing-rights-sdr
+
+"""
+
+from functools import lru_cache
+import pandas as pd
+import calendar
+from bs4 import BeautifulSoup
+from datetime import datetime
+
+from imf_reader.utils import make_request
+from imf_reader.config import logger, NoDataError
+
+BASE_URL = "https://www.imf.org/external/np/fin/tad/"
+MAIN_PAGE_URL = "https://www.imf.org/external/np/fin/tad/extsdr1.aspx"
+
+
+def read_tsv(url: str) -> pd.DataFrame:
+    """Read a tsv file from a url and return a dataframe"""
+
+    try:
+        return pd.read_csv(url, delimiter="/t", engine="python")
+
+    except pd.errors.ParserError:
+        raise ValueError("SDR _data not available for this date")
+
+
+def clean_df(df: pd.DataFrame) -> pd.DataFrame:
+    """Clean the SDR dataframe"""
+
+    df = df.iloc[3:, 0].str.split("\t", expand=True)
+    df.columns = ["entity", "holdings", "allocations"]
+
+    return df.assign(
+        holdings=lambda d: pd.to_numeric(
+            d.holdings.str.replace(r"[^\d.]", "", regex=True), errors="coerce"
+        ),
+        allocations=lambda d: pd.to_numeric(
+            d.allocations.str.replace(r"[^\d.]", "", regex=True), errors="coerce"
+        ),
+    ).melt(
+        id_vars="entity", value_vars=["holdings", "allocations"], var_name="indicator"
+    )
+
+
+def format_date(month: int, year: int) -> str:
+    """Return a date as year-month-day where day is the last day in the month"""
+
+    last_day = calendar.monthrange(year, month)[1]
+    return f"{year}-{month}-{last_day}"
+
+
+@lru_cache
+def get_holdings_and_allocations_data(year: int, month: int):
+    """Get sdr allocations and holdings data for a given month and year"""
+
+    date = format_date(month, year)
+    url = f"{BASE_URL}extsdr2.aspx?date1key={date}&tsvflag=Y"
+
+    logger.info(f"Fetching SDR data for date: {date}")
+
+    df = read_tsv(url)
+    df = clean_df(df)
+    df["date"] = pd.to_datetime(date)
+
+    return df
+
+
+@lru_cache
+def get_latest_date() -> tuple[int, int]:
+    """Get the latest date for which SDR data is available"""
+
+    logger.info("Fetching latest date")
+
+    response = make_request(MAIN_PAGE_URL)
+    soup = BeautifulSoup(response.content, "html.parser")
+    table = soup.find_all("table")[4]
+    row = table.find_all("tr")[1]
+
+    date = row.td.text.strip()
+    date = datetime.strptime(date, "%B %d, %Y")
+
+    # Extract the year and month as a tuple
+    return date.year, date.month
+
+
+def fetch_allocations_holdings(date: tuple[int, int] | None = None) -> pd.DataFrame:
+    """Fetch SDR holdings and allocations data for a given date
+
+    Args:
+        date: A tuple of year and month e.g (2024, 11). If None, the latest data announcements released are fetched
+
+    returns:
+        pd.DataFrame: A dataframe with the SDR data
+    """
+
+    if date is None:
+        date = get_latest_date()
+
+    return get_holdings_and_allocations_data(*date)

src/imf_reader/sdr/read_exchange_rate.py

@@ -0,0 +1,121 @@
+"""Module to read exchange rate data from the IMF's Special Drawing Rights (SDR) Valuation dataset.
+
+Read about SDR valuation at: https://www.imf.org/external/np/fin/data/rms_sdrv.aspx
+"""
+
+import requests
+import pandas as pd
+import io
+from functools import lru_cache
+from typing import Literal
+
+from imf_reader.config import logger
+
+
+BASE_URL = "https://www.imf.org/external/np/fin/data/rms_sdrv.aspx"
+
+
+def get_exchange_rates_data():
+    """Read the data from the IMF website"""
+
+    data = {
+        "__EVENTTARGET": "lbnTSV",
+    }
+
+    try:
+        response = requests.post(BASE_URL, data=data)
+        response.raise_for_status()
+
+    except requests.exceptions.RequestException as e:
+        raise ConnectionError(f"Could not connect to {BASE_URL}. Error: {str(e)}")
+
+    try:
+        return pd.read_csv(
+            io.BytesIO(response.content), delimiter="/t", engine="python"
+        )
+
+    except pd.errors.ParserError as e:
+        raise ValueError(f"Could not parse data. Error: {str(e)}")
+
+
+def preprocess_data(df: pd.DataFrame):
+    """
+    Preprocess the input DataFrame by splitting columns and setting headers.
+    """
+    df = df.iloc[:, 0].str.split("\t", expand=True)
+    df.columns = df.iloc[0]
+    df = df.iloc[1:].reset_index(drop=True)
+
+    # Ensure required columns are present
+    required_columns = ["Report date"]
+    for column in required_columns:
+        if column not in df.columns:
+            raise KeyError(f"Missing required column: {column}")
+
+    return df
+
+
+def extract_exchange_series(df: pd.DataFrame, col_val: str):
+    """
+    Extract the exchange rate series for the given column value.
+    """
+    return (
+        df.loc[lambda d: d["Report date"] == col_val].iloc[:, 1].reset_index(drop=True)
+    )
+
+
+def extract_dates_series(df: pd.DataFrame):
+    """
+    Extract the dates series from the DataFrame.
+    """
+    return (
+        df.dropna(subset=df.columns[3])
+        .iloc[:, 0]
+        .drop_duplicates()
+        .reset_index(drop=True)
+    )
+
+
+def parse_data(df: pd.DataFrame, unit_basis: Literal["SDR", "USD"]):
+    """Parse the data from the IMF website"""
+
+    # Validate unit basis
+    if unit_basis == "USD":
+        col_val = "U.S.$1.00 = SDR"
+    elif unit_basis == "SDR":
+        col_val = "SDR1 = US$"
+    else:
+        raise ValueError("unit_basis must be either 'SDR' or 'USD'")
+
+    # Preprocess dataframe and extract relevant columns
+    df = preprocess_data(df)
+    exchange_series = extract_exchange_series(df, col_val)
+    dates_series = extract_dates_series(df)
+
+    return pd.DataFrame(
+        {"date": dates_series, "exchange_rate": exchange_series}
+    ).assign(
+        date=lambda d: pd.to_datetime(d.date),
+        exchange_rate=lambda d: pd.to_numeric(d.exchange_rate, errors="coerce"),
+    )
+
+
+@lru_cache
+def fetch_exchange_rates(unit_basis: Literal["SDR", "USD"] = "SDR") -> pd.DataFrame:
+    """Fetch the historic SDR exchange rates from the IMF
+
+    The currency value of the SDR is determined by summing the values in U.S. dollars, based on market exchange rates, of a basket of major currencies (the U.S. dollar, Euro, Japanese yen, pound sterling and the Chinese renminbi). The SDR currency value is calculated daily except on IMF holidays, or whenever the IMF is closed for business, or on an ad-hoc basis to facilitate unscheduled IMF operations. The SDR valuation basket is reviewed and adjusted every five years.
+
+    Read more at: https://www.imf.org/en/About/Factsheets/Sheets/2023/special-drawing-rights-sdr
+
+    Args:
+        unit_basis: The unit basis for the exchange rate. Default is "SDR" i.e. 1 SDR in USD. Other option is "USD" i.e. 1 USD in SDR
+
+    Returns:
+        A DataFrame with the exchange rate data
+    """
+
+    logger.info("Fetching exchange rate data")
+
+    df = get_exchange_rates_data()
+    return parse_data(df, unit_basis)