Skip to content

Added a method to retrieve the next scheduled release date for a series #75

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
scottynomad wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Added a method to retrieve the next scheduled release date for a series #75

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b65d73f
Added a couple more tests.
scottynomad Apr 4, 2025
03269fd
Docstring update.
scottynomad Apr 4, 2025
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
64 changes: 64 additions & 0 deletions fredapi/fred.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
83 changes: 66 additions & 17 deletions fredapi/tests/test_fred.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,15 @@
if sys.version_info[0] >= 3:
unicode = str

from datetime import datetime
import io
import unittest
if sys.version_info < (3, 3):
import mock # pylint: disable=import-error
else:
from unittest import mock # pylint: disable=import-error
import textwrap
import pandas as pd
import fredapi
import fredapi.fred

Expand Down Expand Up @@ -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('''\
<?xml version="1.0" encoding="utf-8" ?>
<observations realtime_start="2015-06-28" realtime_end="2015-06-28"
observation_start="2014-09-02"
observation_end="2014-09-05" units="lin"
<observations realtime_start="2025-04-03" realtime_end="2025-04-03"
observation_start="2024-09-02"
observation_end="2024-09-05" units="lin"
output_type="1" file_type="xml"
order_by="observation_date" sort_order="asc"
count="4" offset="0" limit="100000">
<observation realtime_start="2015-06-28" realtime_end="2015-06-28"
date="2014-09-02" value="2002.28"/>
<observation realtime_start="2015-06-28" realtime_end="2015-06-28"
date="2014-09-03" value="2000.72"/>
<observation realtime_start="2015-06-28" realtime_end="2015-06-28"
date="2014-09-04" value="1997.65"/>
<observation realtime_start="2015-06-28" realtime_end="2015-06-28"
date="2014-09-05" value="2007.71"/>
<observation realtime_start="2025-04-03" realtime_end="2025-04-03"
date="2024-09-02" value="."/>
<observation realtime_start="2025-04-03" realtime_end="2025-04-03"
date="2024-09-03" value="5528.93"/>
<observation realtime_start="2025-04-03" realtime_end="2025-04-03"
date="2024-09-04" value="5520.07"/>
<observation realtime_start="2025-04-03" realtime_end="2025-04-03"
date="2024-09-05" value="5503.41"/>
</observations>'''))
search_call = HTTPCall('release/series?release_id=175&' +
'order_by=series_id&sort_order=asc',
Expand Down Expand Up @@ -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="..." />
</seriess>'''))

series_release_pce_call = HTTPCall('series/release?series_id=PCE',
response=textwrap.dedent('''\
<?xml version="1.0" encoding="utf-8" ?>
<releases realtime_start="2025-04-03" realtime_end="2025-04-03">
<release id="54" realtime_start="2025-04-03" realtime_end="2025-04-03" name="Personal Income and Outlays" press_release="true" link="https://www.bea.gov/data/income-saving/personal-income"/>
</releases>'''))
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('''\
<?xml version="1.0" encoding="utf-8" ?>
<release_dates realtime_start="2025-04-04" realtime_end="9999-12-31" order_by="release_date" sort_order="asc" count="9" offset="0" limit="1">
<release_date release_id="54">2025-04-30</release_date>
</release_dates>'''))
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('''\
<?xml version="1.0" encoding="utf-8" ?>
<release_dates realtime_start="2099-04-04" realtime_end="9999-12-31" order_by="release_date" sort_order="asc" count="0" offset="0" limit="1">
</release_dates>\n\n\n\n'''))

class TestFred(unittest.TestCase):

Expand Down Expand Up @@ -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')
Expand Down Expand Up @@ -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()