[Feature] Add Central Bank Holdings (Federal Reserve)

openbb_platform/core/openbb_core/provider/standard_models/central_bank_holdings.py

@@ -0,0 +1,30 @@
+"""Central Bank Holdings Standard Model."""
+
+from datetime import (
+    date as dateType,
+)
+from typing import Optional
+
+from pydantic import Field
+
+from openbb_core.provider.abstract.data import Data
+from openbb_core.provider.abstract.query_params import QueryParams
+from openbb_core.provider.utils.descriptions import (
+    DATA_DESCRIPTIONS,
+    QUERY_DESCRIPTIONS,
+)
+
+
+class CentralBankHoldingsQueryParams(QueryParams):
+    """Central Bank Holdings Query."""
+
+    date: Optional[dateType] = Field(
+        default=None,
+        description=QUERY_DESCRIPTIONS.get("date", ""),
+    )
+
+
+class CentralBankHoldingsData(Data):
+    """Central Bank Holdings Data."""
+
+    date: dateType = Field(description=DATA_DESCRIPTIONS.get("date", ""))

openbb_platform/extensions/economy/integration/test_economy_api.py

@@ -657,3 +657,42 @@ def test_economy_country_profile(params, headers):
     result = requests.get(url, headers=headers, timeout=30)
     assert isinstance(result, requests.Response)
     assert result.status_code == 200
+
+
+@parametrize(
+    "params",
+    [
+        (
+            {
+                "date": None,
+                "provider": "federal_reserve",
+                "holding_type": "all_treasury",
+                "summary": False,
+                "monthly": False,
+                "cusip": None,
+                "wam": False,
+            }
+        ),
+        (
+            {
+                "date": None,
+                "provider": "federal_reserve",
+                "holding_type": "all_agency",
+                "summary": False,
+                "monthly": False,
+                "cusip": None,
+                "wam": True,
+            }
+        ),
+    ],
+)
+@pytest.mark.integration
+def test_economy_central_bank_holdings(params, headers):
+    """Test the economy central bank holdings."""
+    params = {p: v for p, v in params.items() if v}
+
+    query_str = get_querystring(params, [])
+    url = f"http://0.0.0.0:8000/api/v1/economy/central_bank_holdings?{query_str}"
+    result = requests.get(url, headers=headers, timeout=5)
+    assert isinstance(result, requests.Response)
+    assert result.status_code == 200

openbb_platform/extensions/economy/integration/test_economy_python.py

@@ -606,3 +606,41 @@ def test_economy_indicators(params, obb):
     assert result
     assert isinstance(result, OBBject)
     assert len(result.results) > 0
+
+
+@parametrize(
+    "params",
+    [
+        (
+            {
+                "date": None,
+                "provider": "federal_reserve",
+                "holding_type": "all_treasury",
+                "summary": False,
+                "monthly": False,
+                "cusip": None,
+                "wam": False,
+            }
+        ),
+        (
+            {
+                "date": None,
+                "provider": "federal_reserve",
+                "holding_type": "all_agency",
+                "summary": False,
+                "monthly": False,
+                "cusip": None,
+                "wam": True,
+            }
+        ),
+    ],
+)
+@pytest.mark.integration
+def test_economy_central_bank_holdings(params, obb):
+    """Test economy central bank holdings."""
+    params = {p: v for p, v in params.items() if v}
+
+    result = obb.economy.central_bank_holdings(**params)
+    assert result
+    assert isinstance(result, OBBject)
+    assert len(result.results) > 0

openbb_platform/extensions/economy/openbb_economy/economy_router.py

@@ -378,3 +378,35 @@ async def indicators(
 ) -> OBBject:
     """Get economic indicators by country and indicator."""
     return await OBBject.from_query(Query(**locals()))
+
+
+@router.command(
+    model="CentralBankHoldings",
+    examples=[
+        APIEx(
+            description="The default is the latest Treasury securities held by the Federal Reserve.",
+            parameters={"provider": "federal_reserve"},
+        ),
+        APIEx(
+            description="Get historical summaries of the Fed's holdings.",
+            parameters={"provider": "federal_reserve", "summary": True},
+        ),
+        APIEx(
+            description="Get the balance sheet holdings as-of a historical date.",
+            parameters={"provider": "federal_reserve", "date": "2019-05-21"},
+        ),
+        APIEx(
+            description="Use the `holding_type` parameter to select Agency securities,"
+            + " or specific categories or Treasury securities.",
+            parameters={"provider": "federal_reserve", "holding_type": "agency_debts"},
+        ),
+    ],
+)
+async def central_bank_holdings(
+    cc: CommandContext,
+    provider_choices: ProviderChoices,
+    standard_params: StandardParams,
+    extra_params: ExtraParams,
+) -> OBBject:
+    """Get the balance sheet holdings of a central bank."""
+    return await OBBject.from_query(Query(**locals()))