Add FXMacroData return helpers

quantstats/utils.py

@@ -18,8 +18,13 @@
 
 import io as _io
 import datetime as _dt
+import json as _json
+import os as _os
 import pandas as _pd
 import numpy as _np
+from urllib.error import HTTPError as _HTTPError
+from urllib.parse import urlencode as _urlencode
+from urllib.request import Request as _Request, urlopen as _urlopen
 from ._compat import safe_yfinance_download
 from ._compat import safe_concat, safe_resample
 import inspect
@@ -699,6 +704,140 @@ def download_returns(ticker, period="max", proxy=None):
     return df
 
 
+def _split_fxmacrodata_pair(pair):
+    normalized = "".join(char for char in str(pair).upper() if char.isalpha())
+    if len(normalized) != 6:
+        raise ValueError("FXMacroData pairs must look like 'EURUSD' or 'EUR/USD'")
+    return normalized[:3], normalized[3:]
+
+
+def _format_fxmacrodata_date(value):
+    if value is None:
+        return None
+    return _pd.Timestamp(value).date().isoformat()
+
+
+def _read_fxmacrodata_error(error):
+    try:
+        body = error.read().decode("utf-8").strip()
+    except Exception:
+        body = ""
+    message = f"FXMacroData API error {error.code}"
+    if body:
+        message = f"{message}: {body}"
+    return message
+
+
+def download_fxmacrodata_prices(
+    pair,
+    start=None,
+    end=None,
+    api_key=None,
+    base_url="https://fxmacrodata.com/api/v1",
+    timeout=30,
+):
+    """
+    Download daily FX spot/reference prices from FXMacroData.
+
+    Parameters
+    ----------
+    pair : str
+        Six-letter FX pair such as ``"EURUSD"`` or separated pair such as
+        ``"EUR/USD"``.
+    start : str or datetime-like, optional
+        Start date.
+    end : str or datetime-like, optional
+        End date.
+    api_key : str, optional
+        FXMacroData API key. If omitted, ``FXMACRODATA_API_KEY`` or
+        ``FXMD_API_KEY`` will be used when present.
+    base_url : str, default "https://fxmacrodata.com/api/v1"
+        FXMacroData API base URL.
+    timeout : int or float, default 30
+        Request timeout in seconds.
+
+    Returns
+    -------
+    pd.Series
+        Daily FX price series indexed by date.
+    """
+    base_currency, quote_currency = _split_fxmacrodata_pair(pair)
+    params = {}
+    start_date = _format_fxmacrodata_date(start)
+    end_date = _format_fxmacrodata_date(end)
+    if start_date:
+        params["start_date"] = start_date
+    if end_date:
+        params["end_date"] = end_date
+
+    api_key = api_key or _os.getenv("FXMACRODATA_API_KEY") or _os.getenv("FXMD_API_KEY")
+    if api_key:
+        params["api_key"] = api_key
+
+    url = f"{base_url.rstrip('/')}/forex/{base_currency.lower()}/{quote_currency.lower()}"
+    query = _urlencode(params)
+    if query:
+        url = f"{url}?{query}"
+
+    request = _Request(url, headers={"Accept": "application/json"})
+    try:
+        with _urlopen(request, timeout=timeout) as response:
+            payload = _json.loads(response.read().decode("utf-8"))
+    except _HTTPError as error:
+        raise ValueError(_read_fxmacrodata_error(error)) from error
+
+    rows = payload.get("data") if isinstance(payload, dict) else None
+    if not isinstance(rows, list):
+        raise ValueError("FXMacroData response did not include a data list")
+
+    records = []
+    for row in rows:
+        if not isinstance(row, dict):
+            continue
+        date_value = row.get("date")
+        value = row.get("val")
+        if date_value is None or value is None:
+            continue
+        records.append((_pd.Timestamp(date_value), value))
+
+    if not records:
+        raise ValueError(f"FXMacroData response did not include dated values for {pair}")
+
+    series = _pd.Series(
+        data=[value for _, value in records],
+        index=_pd.DatetimeIndex([date for date, _ in records]),
+        name=f"{base_currency}{quote_currency}",
+    )
+    return _pd.to_numeric(series, errors="coerce").dropna().sort_index().tz_localize(None)
+
+
+def download_fxmacrodata_returns(
+    pair,
+    start=None,
+    end=None,
+    api_key=None,
+    base_url="https://fxmacrodata.com/api/v1",
+    timeout=30,
+):
+    """
+    Download FXMacroData daily FX prices and convert them to returns.
+
+    Returns
+    -------
+    pd.Series
+        Daily percentage returns for the requested FX pair.
+    """
+    prices = download_fxmacrodata_prices(
+        pair,
+        start=start,
+        end=end,
+        api_key=api_key,
+        base_url=base_url,
+        timeout=timeout,
+    )
+    return prices.pct_change(fill_method=None).fillna(0)
+
+
 def _prepare_benchmark(benchmark=None, period="max", rf=0.0, prepare_returns=True):
     """
     Fetch benchmark if ticker is provided, and pass through

tests/test_utils.py

@@ -5,6 +5,7 @@
 import pytest
 import pandas as pd
 import numpy as np
+import json
 
 import quantstats as qs
 from quantstats import utils
@@ -64,6 +65,79 @@ def test_to_returns_from_prices(self, sample_prices):
         assert result.dropna().abs().max() < 1
 
 
+class TestFXMacroDataDownloads:
+    """Test FXMacroData price and returns helpers."""
+
+    class FakeResponse:
+        def __init__(self, payload):
+            self.payload = payload
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def read(self):
+            return self.payload.encode("utf-8")
+
+    def test_download_fxmacrodata_prices(self, monkeypatch):
+        captured = {}
+
+        def fake_urlopen(request, timeout):
+            captured["url"] = request.full_url
+            captured["accept"] = request.headers["Accept"]
+            captured["timeout"] = timeout
+            return self.FakeResponse(
+                json.dumps(
+                    {
+                        "data": [
+                            {"date": "2024-01-03", "val": 1.0920},
+                            {"date": "2024-01-01", "val": "1.1038"},
+                        ]
+                    }
+                )
+            )
+
+        monkeypatch.setattr(utils, "_urlopen", fake_urlopen)
+        actual = utils.download_fxmacrodata_prices(
+            "eur/usd",
+            start="2024-01-01",
+            end="2024-01-31",
+            api_key="test-key",
+            timeout=12,
+        )
+
+        expected = pd.Series(
+            [1.1038, 1.092],
+            index=pd.to_datetime(["2024-01-01", "2024-01-03"]),
+            name="EURUSD",
+        )
+        pd.testing.assert_series_equal(actual, expected)
+        assert captured == {
+            "url": "https://fxmacrodata.com/api/v1/forex/eur/usd?start_date=2024-01-01&end_date=2024-01-31&api_key=test-key",
+            "accept": "application/json",
+            "timeout": 12,
+        }
+
+    def test_download_fxmacrodata_returns(self, monkeypatch):
+        prices = pd.Series(
+            [1.10, 1.21],
+            index=pd.to_datetime(["2024-01-01", "2024-01-02"]),
+            name="EURUSD",
+        )
+        monkeypatch.setattr(utils, "download_fxmacrodata_prices", lambda *args, **kwargs: prices)
+
+        actual = utils.download_fxmacrodata_returns("EURUSD")
+
+        expected = pd.Series(
+            [0.0, 0.1],
+            index=pd.to_datetime(["2024-01-01", "2024-01-02"]),
+            name="EURUSD",
+        )
+        pd.testing.assert_series_equal(actual, expected)
+
+
 class TestToPrices:
     """Test to_prices function."""