Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
# Changelog

## v1.3.0 (2025-2-05)
- Made function available to fetch latest holdings and allocations date
- Improved handling of unavailable dates

## v1.2.0 (2024-12-20)
- Add support for fetching Special Drawing Rights (SDR) data

Expand All @@ -18,5 +22,4 @@
- Basic functionality for accessing WEO data for initial testing

## v0.0.1 (2024-05-17)

- First release of `imf-reader`
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ SDRs holdings and allocations are published at a monthly frequency. The function
default. Check the latest available date

```python
sdr.get_latest_allocations_holdings_date()
sdr.fetch_latest_allocations_holdings_date()
```

To retrieve SDR holdings and allocations for a specific month and year, eg April 2021, pass the year and month as a tuple
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "imf-reader"
version = "1.2.0"
version = "1.3.0"
description = "A package to access imf data"
authors = ["The ONE Campaign"]
license = "MIT"
Expand Down
4 changes: 2 additions & 2 deletions src/imf_reader/sdr/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
default. Check the latest available date

```python
sdr.get_latest_allocations_holdings_date()
sdr.fetch_latest_allocations_holdings_date()
```

To retrieve SDR holdings and allocations for a specific month and year, eg April 2021, pass the year and month as a tuple
Expand Down Expand Up @@ -62,6 +62,6 @@
from imf_reader.sdr.read_exchange_rate import fetch_exchange_rates
from imf_reader.sdr.read_announcements import (
fetch_allocations_holdings,
get_latest_allocations_holdings_date,
fetch_latest_allocations_holdings_date,
)
from imf_reader.sdr.clear_cache import clear_cache
4 changes: 2 additions & 2 deletions src/imf_reader/sdr/clear_cache.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from imf_reader.sdr.read_announcements import (
get_holdings_and_allocations_data,
get_latest_allocations_holdings_date,
fetch_latest_allocations_holdings_date,
)
from imf_reader.sdr.read_exchange_rate import fetch_exchange_rates
from imf_reader.sdr.read_interest_rate import fetch_interest_rates
Expand All @@ -12,7 +12,7 @@ def clear_cache():

# clear cache from read_announcements module
get_holdings_and_allocations_data.cache_clear()
get_latest_allocations_holdings_date.cache_clear()
fetch_latest_allocations_holdings_date.cache_clear()

# clear cache from read_exchange_rate module
fetch_exchange_rates.cache_clear()
Expand Down
30 changes: 23 additions & 7 deletions src/imf_reader/sdr/read_announcements.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
import calendar
from bs4 import BeautifulSoup
from datetime import datetime
import logging

from imf_reader.utils import make_request
from imf_reader.config import logger
Expand All @@ -19,13 +20,13 @@


def read_tsv(url: str) -> pd.DataFrame:
"""Read a tsv file from a url and return a dataframe"""
"""Read a tsv file from 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")
raise ValueError("SDR data not available for this date")


def clean_df(df: pd.DataFrame) -> pd.DataFrame:
Expand Down Expand Up @@ -73,13 +74,13 @@ def get_holdings_and_allocations_data(


@lru_cache
def get_latest_allocations_holdings_date() -> tuple[int, int]:
def fetch_latest_allocations_holdings_date() -> tuple[int, int]:
"""
Get the latest available SDR allocation holdings date.

Returns:
tuple[int, int]: A tuple containing the year and month of the latest SDR data.
"""

logger.info("Fetching latest date")

response = make_request(MAIN_PAGE_URL)
Expand All @@ -94,16 +95,31 @@ def get_latest_allocations_holdings_date() -> tuple[int, int]:


def fetch_allocations_holdings(date: tuple[int, int] | None = None) -> pd.DataFrame:
"""Fetch SDR holdings and allocations data for a given date
"""
Fetch SDR holdings and allocations data for a given date. If date is not specified, it fetches data for the latest date

Args:
date: The year and month to get allocations and holdings data for. e.g. (2024, 11) for November 2024. If None, the latest announcements released are fetched
date: tuple[int, int]. The year and month to get allocations and holdings data for. e.g. (2024, 11) for November 2024.
If None, the latest announcements released are fetched.

returns:
A dataframe with the SDR allocations and holdings data
"""

if date is None:
date = get_latest_allocations_holdings_date()
date = fetch_latest_allocations_holdings_date()
else:
# Temporarily disable logging while calling fetch_latest_allocations_holdings_date()
original_logger_level = logger.level
logger.setLevel(logging.WARNING)
latest_date = fetch_latest_allocations_holdings_date()
logger.setLevel(original_logger_level)

date_obj = datetime(date[0], date[1], 1)
latest_date_obj = datetime(latest_date[0], latest_date[1], 1)
if date_obj > latest_date_obj:
raise ValueError(
f"SDR data unavailable for: ({date[0]}, {date[1]}).\nLatest available: ({latest_date[0]}, {latest_date[1]})"
)

return get_holdings_and_allocations_data(*date)
2 changes: 1 addition & 1 deletion tests/test_sdr/test_clear_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
"imf_reader.sdr.read_announcements.get_holdings_and_allocations_data.cache_clear"
)
@patch(
"imf_reader.sdr.read_announcements.get_latest_allocations_holdings_date.cache_clear"
"imf_reader.sdr.read_announcements.fetch_latest_allocations_holdings_date.cache_clear"
)
@patch("imf_reader.sdr.read_exchange_rate.fetch_exchange_rates.cache_clear")
@patch("imf_reader.sdr.read_interest_rate.fetch_interest_rates.cache_clear")
Expand Down
55 changes: 29 additions & 26 deletions tests/test_sdr/test_read_announcements.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,14 @@
from unittest.mock import patch, Mock
import pytest
import pandas as pd
import re
from imf_reader import sdr
from imf_reader.sdr.read_announcements import (
read_tsv,
clean_df,
format_date,
get_holdings_and_allocations_data,
get_latest_allocations_holdings_date,
fetch_latest_allocations_holdings_date,
fetch_allocations_holdings,
BASE_URL,
MAIN_PAGE_URL,
Expand Down Expand Up @@ -50,7 +51,7 @@ def test_read_tsv_success(self, mock_read_csv):
def test_read_tsv_failure(self, mock_read_csv):
"""Test read_tsv raises ValueError on malformed data."""
mock_read_csv.side_effect = pd.errors.ParserError
with pytest.raises(ValueError, match="SDR _data not available for this date"):
with pytest.raises(ValueError, match="SDR data not available for this date"):
read_tsv("mock_url")

def test_clean_df_correct_format(self, input_df):
Expand Down Expand Up @@ -128,11 +129,11 @@ def test_get_holdings_and_allocations_data_success(
def test_get_holdings_and_allocations_data_failure(self, mock_read_tsv):
"""Test get_holdings_and_allocations_data raises ValueError when read_tsv fails."""
with pytest.raises(ValueError, match="Data not available"):
get_holdings_and_allocations_data(2024, 2)
get_holdings_and_allocations_data(1800, 1)

@patch("imf_reader.sdr.read_announcements.make_request")
def test_get_latest_date_success(self, mock_make_request):
"""Test get_latest_allocations_holdings_date successfully returns the latest date."""
"""Test fetch_latest_allocations_holdings_date successfully returns the latest date."""

# Mock HTML content
mock_html_content = """
Expand All @@ -159,15 +160,15 @@ def test_get_latest_date_success(self, mock_make_request):
mock_make_request.return_value = mock_response

# Call the function
result = get_latest_allocations_holdings_date()
result = fetch_latest_allocations_holdings_date()

# Assertions
mock_make_request.assert_called_once_with(MAIN_PAGE_URL)
assert result == (2023, 11) # Expected year and month

@patch("imf_reader.sdr.read_announcements.make_request")
def test_get_latest_date_invalid_html(self, mock_make_request):
"""Test get_latest_allocations_holdings_date raises an error when HTML parsing fails."""
"""Test fetch_latest_allocations_holdings_date raises an error when HTML parsing fails."""

# Mock malformed HTML content
mock_response = Mock()
Expand All @@ -176,9 +177,9 @@ def test_get_latest_date_invalid_html(self, mock_make_request):

# Call the function and expect an IndexError
with pytest.raises(IndexError):
get_latest_allocations_holdings_date()
fetch_latest_allocations_holdings_date()

@patch("imf_reader.sdr.read_announcements.get_latest_allocations_holdings_date")
@patch("imf_reader.sdr.read_announcements.fetch_latest_allocations_holdings_date")
@patch("imf_reader.sdr.read_announcements.clean_df")
@patch("imf_reader.sdr.read_announcements.read_tsv")
@patch("imf_reader.sdr.read_announcements.get_holdings_and_allocations_data")
Expand All @@ -191,8 +192,8 @@ def test_fetch_allocations_holdings_default_date(
input_df,
):
"""Test fetch_allocations_holdings when no date is provided."""
# Mock get_latest_allocations_holdings_date to return a specific date
mock_get_latest_date.return_value = (2, 2024)
# Mock fetch_latest_allocations_holdings_date to return a specific date
mock_get_latest_date.return_value = (2024, 2)

# Mock read_tsv
mock_read_tsv.return_value = input_df
Expand All @@ -215,26 +216,28 @@ def test_fetch_allocations_holdings_default_date(
result = fetch_allocations_holdings()

# Assertions
mock_get_latest_date.assert_called_once() # Ensure get_latest_allocations_holdings_date was called
mock_get_latest_date.assert_called_once() # Ensure fetch_latest_allocations_holdings_date was called
mock_get_holdings_and_allocations_data.assert_called_once_with(
2, 2024
2024, 2
) # Ensure the correct call
pd.testing.assert_frame_equal(
result, cleaned_df
) # Ensure the result matches the cleaned data

@patch("imf_reader.sdr.read_announcements.get_holdings_and_allocations_data")
@patch("imf_reader.sdr.read_announcements.get_latest_allocations_holdings_date")
def test_fetch_allocations_holdings_failure(
self, mock_get_latest_date, mock_get_holdings_data
):
"""Test fetch_allocations_holdings raises ValueError when data fetching fails."""
# Mock get_latest_allocations_holdings_date to return a date
mock_get_latest_date.return_value = (2, 2024)

# Simulate a failure in get_holdings_and_allocations_data
mock_get_holdings_data.side_effect = ValueError("Data not available")

# Assertions
with pytest.raises(ValueError, match="Data not available"):
fetch_allocations_holdings()
@patch("imf_reader.sdr.read_announcements.get_holdings_and_allocations_data")
@patch(
"imf_reader.sdr.read_announcements.fetch_latest_allocations_holdings_date",
return_value=(2024, 1),
)
def test_fetch_allocations_holdings_future_date(
mock_get_latest_date, mock_get_holdings_and_allocations_data
):
"""Test fetch_allocations_holdings when unavailable future date is provided."""
with pytest.raises(
ValueError,
match=re.escape(
"SDR data unavailable for: (2025, 1).\nLatest available: (2024, 1)"
),
):
fetch_allocations_holdings((2025, 1))