diff --git a/fredapi/fred.py b/fredapi/fred.py
index c6c7cb3..78722ee 100644
--- a/fredapi/fred.py
+++ b/fredapi/fred.py
@@ -12,6 +12,7 @@
import urllib2 as url_error
import pandas as pd
+from datetime import date, timedelta
urlopen = url_request.urlopen
quote_plus = url_parse.quote_plus
@@ -299,6 +300,69 @@ def get_series_vintage_dates(self, series_id):
dates.append(self._parse(child.text))
return dates
+ def get_series_release_publications(self, series_id):
+ """
+ Get details of the release publication the series data is sourced from.
+
+ Parameters
+ ----------
+ series_id : str
+ Fred series id such as 'CPIAUCSL'
+
+ Returns
+ -------
+ {
+ "id": 21,
+ "realtime_start": "2013-08-14",
+ "realtime_end": "2013-08-14",
+ "name": "H.6 Money Stock Measures",
+ "press_release": true,
+ "link": "http://www.federalreserve.gov/releases/h6/"
+ }
+ """
+ url = "%s/series/release?series_id=%s" % (self.root_url, series_id)
+ root = self.__fetch_data(url)
+ if root is None:
+ raise ValueError('No release found for series id: ' + series_id)
+ for child in root:
+ return {
+ 'id': child.get('id'),
+ 'name': child.get('name'),
+ 'realtime_start': self._parse(child.get('realtime_start')),
+ 'realtime_end': self._parse(child.get('realtime_end')),
+ 'press_release': child.get('press_release') == 'true',
+ 'link': child.get('link'),
+ }
+
+ def get_series_next_release_date(self, series_id, realtime_start=None):
+ """
+ Get the next scheduled release date for a series.
+
+ Parameters
+ ----------
+ series_id : str
+ Fred series id such as 'CPIAUCSL'
+ realtime_start : str, optional
+ specifies the realtime_start value used in the query, defaults to tomorrow
+
+ Returns
+ -------
+ datetime
+ next release date
+ """
+ if realtime_start is None:
+ realtime_start = (date.today() + timedelta(days=1)).strftime('%Y-%m-%d')
+ release_id = self.get_series_release_publications(series_id)['id']
+ if release_id is None:
+ raise ValueError('No release id found for series id: ' + series_id)
+ url = "%s/release/dates?release_id=%s&realtime_start=%s&include_release_dates_with_no_data=true&limit=1" % (
+ self.root_url, release_id, realtime_start)
+ root = self.__fetch_data(url)
+ if root is None:
+ raise ValueError('No release dates found for series id: ' + series_id)
+ for release_date in root:
+ return self._parse(release_date.text)
+
def __do_series_search(self, url):
"""
helper function for making one HTTP request for data, and parsing the returned results into a DataFrame
diff --git a/fredapi/tests/test_fred.py b/fredapi/tests/test_fred.py
index 317e9fb..8bc1a9a 100644
--- a/fredapi/tests/test_fred.py
+++ b/fredapi/tests/test_fred.py
@@ -3,6 +3,7 @@
if sys.version_info[0] >= 3:
unicode = str
+from datetime import datetime
import io
import unittest
if sys.version_info < (3, 3):
@@ -10,6 +11,7 @@
else:
from unittest import mock # pylint: disable=import-error
import textwrap
+import pandas as pd
import fredapi
import fredapi.fred
@@ -45,24 +47,24 @@ def __init__(self, rel_url, response=None, side_effect=None):
sp500_obs_call = HTTPCall('series/observations?series_id=SP500&{}&{}'.
- format('observation_start=2014-09-02',
- 'observation_end=2014-09-05'),
+ format('observation_start=2024-09-02',
+ 'observation_end=2024-09-05'),
response=textwrap.dedent('''\
-
-
-
-
-
+
+
+
+
'''))
search_call = HTTPCall('release/series?release_id=175&' +
'order_by=series_id&sort_order=asc',
@@ -112,7 +114,23 @@ def __init__(self, rel_url, response=None, side_effect=None):
last_updated="2015-06-05 08:47:20-05"
popularity="86" notes="..." />
'''))
-
+series_release_pce_call = HTTPCall('series/release?series_id=PCE',
+ response=textwrap.dedent('''\
+
+
+
+'''))
+release_dates_pce_call = HTTPCall('release/dates?release_id=54&realtime_start=2025-04-04&include_release_dates_with_no_data=true&limit=1',
+ response=textwrap.dedent('''\
+
+
+ 2025-04-30
+'''))
+release_dates_no_release_call = HTTPCall('release/dates?release_id=54&realtime_start=2025-04-04&include_release_dates_with_no_data=true&limit=1',
+ response=textwrap.dedent('''\
+
+
+\n\n\n\n'''))
class TestFred(unittest.TestCase):
@@ -161,10 +179,11 @@ def test_get_series(self, urlopen):
"""Test retrieval of series for SP500."""
self.prepare_urlopen(urlopen,
http_response=sp500_obs_call.response)
- serie = self.fred.get_series('SP500', observation_start='9/2/2014',
- observation_end='9/5/2014')
+ serie = self.fred.get_series('SP500', observation_start='9/2/2024',
+ observation_end='9/5/2024')
urlopen.assert_called_with(sp500_obs_call.url)
- self.assertEqual(serie.loc['9/2/2014'], 2002.28)
+ self.assertTrue(pd.isna(serie.loc['9/2/2024']))
+ self.assertEqual(serie.loc['9/3/2024'], 5528.93)
self.assertEqual(len(serie), 4)
@mock.patch('fredapi.fred.urlopen')
@@ -248,6 +267,36 @@ def test_search(self, urlopen):
for aline, eline in zip(actual.split('\n'), expected.split('\n')):
self.assertEqual(aline.strip(), eline.strip())
+ @mock.patch('fredapi.fred.urlopen')
+ def test_get_series_release_publication(self, urlopen):
+ """Test retrieval of release publication details for a series."""
+ self.prepare_urlopen(urlopen, http_response=series_release_pce_call.response)
+ pub_info = self.fred.get_series_release_publications('PCE')
+ urlopen.assert_called_with(series_release_pce_call.url)
+ self.assertEqual(pub_info['id'], '54')
+ self.assertEqual(pub_info['name'], 'Personal Income and Outlays')
+ self.assertEqual(pub_info['press_release'], True)
+ self.assertEqual(pub_info['link'], 'https://www.bea.gov/data/income-saving/personal-income')
+
+ @mock.patch('fredapi.fred.urlopen')
+ def test_get_series_next_release_date_no_release(self, urlopen):
+ """Test retrieval of next release date for a series with no release."""
+ responses = [release_dates_no_release_call.response, series_release_pce_call.response]
+ self.prepare_urlopen(urlopen, side_effect=lambda: responses.pop())
+ next_release_date = self.fred.get_series_next_release_date('PCE', realtime_start='2099-04-04')
+ self.assertIsNone(next_release_date)
+
+ @mock.patch('fredapi.fred.urlopen')
+ def test_get_series_next_release_date(self, urlopen):
+ """Test retrieval of next release date for a series."""
+ responses = [release_dates_pce_call.response, series_release_pce_call.response]
+ self.prepare_urlopen(urlopen, side_effect=lambda: responses.pop())
+ next_release_date = self.fred.get_series_next_release_date('PCE', realtime_start='2025-04-04')
+ urlopen.assert_has_calls([
+ mock.call(series_release_pce_call.url),
+ mock.call(release_dates_pce_call.url)
+ ], any_order=True)
+ self.assertEqual(next_release_date, datetime.fromisoformat('2025-04-30'))
if __name__ == '__main__':
unittest.main()