From b10ccdfda20f0171b05e3737c1cf16d0495967c2 Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Wed, 8 Jul 2026 19:55:34 +1000 Subject: [PATCH 1/4] Add FXMacroData macro calendar integration --- src/cutebacktests/providers/__init__.py | 8 ++- src/cutebacktests/providers/fxmacrodata.py | 75 ++++++++++++++++++++++ 2 files changed, 82 insertions(+), 1 deletion(-) create mode 100644 src/cutebacktests/providers/fxmacrodata.py diff --git a/src/cutebacktests/providers/__init__.py b/src/cutebacktests/providers/__init__.py index 5be6f7b..ca782c0 100644 --- a/src/cutebacktests/providers/__init__.py +++ b/src/cutebacktests/providers/__init__.py @@ -2,5 +2,11 @@ from .alpaca import AlpacaDataProvider, AlpacaPaperBroker from .cutemarkets import CuteMarketsProvider +from .fxmacrodata import FXMacroDataProvider -__all__ = ["CuteMarketsProvider", "AlpacaDataProvider", "AlpacaPaperBroker"] +__all__ = [ + "CuteMarketsProvider", + "AlpacaDataProvider", + "AlpacaPaperBroker", + "FXMacroDataProvider", +] diff --git a/src/cutebacktests/providers/fxmacrodata.py b/src/cutebacktests/providers/fxmacrodata.py new file mode 100644 index 0000000..5b32249 --- /dev/null +++ b/src/cutebacktests/providers/fxmacrodata.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional + +import requests + + +class FXMacroDataProvider: + """Fetch macroeconomic release calendars from FXMacroData.""" + + def __init__( + self, + api_key: Optional[str] = None, + base_url: str = "https://fxmacrodata.com/api/v1", + timeout: int = 30, + session: Optional[requests.Session] = None, + ) -> None: + self.api_key = api_key + self.base_url = base_url.rstrip("/") + self.timeout = timeout + self._session = session or requests.Session() + + def fetch_calendar( + self, + currency: str = "usd", + start_date: Optional[str] = None, + end_date: Optional[str] = None, + top_tier_only: bool = False, + ) -> List[Dict[str, Any]]: + params: Dict[str, Any] = {} + if self.api_key: + params["api_key"] = self.api_key + if start_date: + params["start_date"] = start_date + if end_date: + params["end_date"] = end_date + response = self._session.get( + f"{self.base_url}/calendar/{currency.lower()}", + params=params, + headers={"Accept": "application/json"}, + timeout=self.timeout, + ) + response.raise_for_status() + rows = list((response.json() or {}).get("data") or []) + if top_tier_only: + rows = [row for row in rows if row.get("top_tier_for_currency") or row.get("market_tier") == 1] + return rows + + def upcoming_events( + self, + currency: str = "usd", + now: Optional[datetime] = None, + lookahead: timedelta = timedelta(days=7), + top_tier_only: bool = True, + ) -> List[Dict[str, Any]]: + now = now or datetime.now(timezone.utc) + if now.tzinfo is None: + now = now.replace(tzinfo=timezone.utc) + end = now + lookahead + rows = self.fetch_calendar( + currency=currency, + start_date=now.date().isoformat(), + end_date=end.date().isoformat(), + top_tier_only=top_tier_only, + ) + return [row for row in rows if (event_time := _event_time(row)) is not None and now <= event_time <= end] + + +def _event_time(row: Dict[str, Any]) -> Optional[datetime]: + text = str(row.get("announcement_datetime_utc") or row.get("announcement_datetime_local") or "") + if not text: + return None + return datetime.fromisoformat(text.replace("Z", "+00:00")) + From 3dfc8ab7990d8f26f8bb19aadb56a8fbdf04432b Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Wed, 26 Aug 2026 12:11:38 +1000 Subject: [PATCH 2/4] Use the documented FXMacroData API host Switches the base URL from the undocumented fxmacrodata.com/api/v1 alias to the published api.fxmacrodata.com/v1 host, so the client matches the public API reference. --- src/cutebacktests/providers/fxmacrodata.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cutebacktests/providers/fxmacrodata.py b/src/cutebacktests/providers/fxmacrodata.py index 5b32249..c75dc34 100644 --- a/src/cutebacktests/providers/fxmacrodata.py +++ b/src/cutebacktests/providers/fxmacrodata.py @@ -12,7 +12,7 @@ class FXMacroDataProvider: def __init__( self, api_key: Optional[str] = None, - base_url: str = "https://fxmacrodata.com/api/v1", + base_url: str = "https://api.fxmacrodata.com/v1", timeout: int = 30, session: Optional[requests.Session] = None, ) -> None: From fc698acc59bf7e11a65180d6d91cde17bf9cd22f Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Mon, 28 Sep 2026 12:50:09 +1000 Subject: [PATCH 3/4] fix: send the FXMacroData API key in the X-API-Key header --- src/cutebacktests/providers/fxmacrodata.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/cutebacktests/providers/fxmacrodata.py b/src/cutebacktests/providers/fxmacrodata.py index c75dc34..1727454 100644 --- a/src/cutebacktests/providers/fxmacrodata.py +++ b/src/cutebacktests/providers/fxmacrodata.py @@ -29,8 +29,9 @@ def fetch_calendar( top_tier_only: bool = False, ) -> List[Dict[str, Any]]: params: Dict[str, Any] = {} + headers = {"Accept": "application/json"} if self.api_key: - params["api_key"] = self.api_key + headers["X-API-Key"] = self.api_key if start_date: params["start_date"] = start_date if end_date: @@ -38,7 +39,7 @@ def fetch_calendar( response = self._session.get( f"{self.base_url}/calendar/{currency.lower()}", params=params, - headers={"Accept": "application/json"}, + headers=headers, timeout=self.timeout, ) response.raise_for_status() From afe074d4bf6faa023202d92f3abf56c0ecadfac8 Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Mon, 5 Oct 2026 20:57:31 +1100 Subject: [PATCH 4/4] fix: keep the FXMacroData key off redirects and reject malformed calendar responses --- src/cutebacktests/providers/fxmacrodata.py | 21 ++++++-- tests/test_fxmacrodata_provider.py | 61 ++++++++++++++++++++++ 2 files changed, 79 insertions(+), 3 deletions(-) create mode 100644 tests/test_fxmacrodata_provider.py diff --git a/src/cutebacktests/providers/fxmacrodata.py b/src/cutebacktests/providers/fxmacrodata.py index 1727454..f8f2f4f 100644 --- a/src/cutebacktests/providers/fxmacrodata.py +++ b/src/cutebacktests/providers/fxmacrodata.py @@ -16,7 +16,9 @@ def __init__( timeout: int = 30, session: Optional[requests.Session] = None, ) -> None: - self.api_key = api_key + self.api_key = (api_key or "").strip() or None + if self.api_key and any(ch.isspace() for ch in self.api_key): + raise ValueError("FXMacroData api_key must not contain whitespace") self.base_url = base_url.rstrip("/") self.timeout = timeout self._session = session or requests.Session() @@ -36,14 +38,27 @@ def fetch_calendar( params["start_date"] = start_date if end_date: params["end_date"] = end_date + url = f"{self.base_url}/calendar/{currency.lower()}" + # Redirects are not followed so the key header is never replayed to another host. response = self._session.get( - f"{self.base_url}/calendar/{currency.lower()}", + url, params=params, headers=headers, timeout=self.timeout, + allow_redirects=False, ) + if response.is_redirect: + raise RuntimeError(f"FXMacroData returned an unexpected redirect (HTTP {response.status_code}) for {url}") response.raise_for_status() - rows = list((response.json() or {}).get("data") or []) + try: + payload = response.json() + except ValueError as exc: + raise RuntimeError(f"FXMacroData returned a non-JSON response for {url}") from exc + data = payload.get("data") if isinstance(payload, dict) else None + if not isinstance(data, list): + detail = payload.get("detail") if isinstance(payload, dict) else None + raise RuntimeError(f"FXMacroData returned an unexpected response for {url}" + (f": {detail}" if detail else "")) + rows = [row for row in data if isinstance(row, dict)] if top_tier_only: rows = [row for row in rows if row.get("top_tier_for_currency") or row.get("market_tier") == 1] return rows diff --git a/tests/test_fxmacrodata_provider.py b/tests/test_fxmacrodata_provider.py new file mode 100644 index 0000000..4a4985c --- /dev/null +++ b/tests/test_fxmacrodata_provider.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import json +import unittest +from unittest.mock import MagicMock + +import requests + +from cutebacktests.providers.fxmacrodata import FXMacroDataProvider + + +def _response(status: int, body: object = None, headers: dict | None = None) -> requests.Response: + response = requests.Response() + response.status_code = status + response.headers.update(headers or {}) + response._content = b"" if body is None else json.dumps(body).encode() + response.url = "https://api.fxmacrodata.com/v1/calendar/usd" + return response + + +def _provider(response: requests.Response, api_key: str = "test-key") -> tuple[FXMacroDataProvider, MagicMock]: + session = MagicMock() + session.get.return_value = response + return FXMacroDataProvider(api_key=api_key, session=session), session + + +class FXMacroDataProviderTests(unittest.TestCase): + def test_returns_rows_and_sends_key_header(self) -> None: + provider, session = _provider(_response(200, {"data": [{"release": "cpi", "market_tier": 1}]})) + + rows = provider.fetch_calendar("USD") + + self.assertEqual(rows, [{"release": "cpi", "market_tier": 1}]) + kwargs = session.get.call_args.kwargs + self.assertEqual(kwargs["headers"]["X-API-Key"], "test-key") + self.assertIs(kwargs["allow_redirects"], False) + + def test_redirect_raises_instead_of_forwarding_key(self) -> None: + provider, session = _provider(_response(302, headers={"Location": "https://other.example/calendar"})) + + with self.assertRaises(RuntimeError) as ctx: + provider.fetch_calendar("usd") + + self.assertIn("redirect", str(ctx.exception)) + self.assertEqual(session.get.call_count, 1) + + def test_key_with_whitespace_is_rejected_without_echo(self) -> None: + with self.assertRaises(ValueError) as ctx: + FXMacroDataProvider(api_key="test key") + self.assertNotIn("test key", str(ctx.exception)) + self.assertEqual(FXMacroDataProvider(api_key=" test-key\n").api_key, "test-key") + + def test_error_body_or_wrong_shape_raises_clean_error(self) -> None: + for body in ({"detail": "Invalid API key"}, [], {"data": {"release": "cpi"}}): + provider, _ = _provider(_response(200, body)) + with self.assertRaises(RuntimeError): + provider.fetch_calendar("usd") + + +if __name__ == "__main__": + unittest.main()