Add FXMacroData macro calendar integration

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",
+]

src/cutebacktests/providers/fxmacrodata.py

@@ -0,0 +1,91 @@
+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://api.fxmacrodata.com/v1",
+        timeout: int = 30,
+        session: Optional[requests.Session] = None,
+    ) -> None:
+        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()
+
+    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] = {}
+        headers = {"Accept": "application/json"}
+        if self.api_key:
+            headers["X-API-Key"] = self.api_key
+        if start_date:
+            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(
+            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()
+        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
+
+    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"))
+

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()