From 754010e169b19ff3c8837c462672f3278fba21be Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Wed, 8 Jul 2026 23:45:45 +1000 Subject: [PATCH 1/5] Add FXMacroData release calendar integration --- pypfopt/__init__.py | 3 ++ pypfopt/fxmacrodata.py | 81 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 84 insertions(+) create mode 100644 pypfopt/fxmacrodata.py diff --git a/pypfopt/__init__.py b/pypfopt/__init__.py index 249f0e6d..155064d3 100755 --- a/pypfopt/__init__.py +++ b/pypfopt/__init__.py @@ -11,6 +11,7 @@ EfficientFrontier, EfficientSemivariance, ) +from .fxmacrodata import event_window_mask, get_release_calendar from .hierarchical_portfolio import HRPOpt from .risk_models import CovarianceShrinkage @@ -27,6 +28,8 @@ "EfficientSemivariance", "EfficientCVaR", "EfficientCDaR", + "event_window_mask", + "get_release_calendar", "HRPOpt", "CovarianceShrinkage", ] diff --git a/pypfopt/fxmacrodata.py b/pypfopt/fxmacrodata.py new file mode 100644 index 00000000..c3b6508a --- /dev/null +++ b/pypfopt/fxmacrodata.py @@ -0,0 +1,81 @@ +"""FXMacroData helpers for portfolio event-risk analysis.""" + +from __future__ import annotations + +import json +import os +from urllib.parse import urlencode +from urllib.request import urlopen + +import pandas as pd + +FXMACRODATA_BASE_URL = "https://fxmacrodata.com/api/v1" + + +def get_release_calendar( + currency: str = "usd", + limit: int = 100, + min_tier: int | None = None, + api_key: str | None = None, + base_url: str = FXMACRODATA_BASE_URL, +) -> pd.DataFrame: + """Fetch an FXMacroData economic release calendar as a DataFrame.""" + limit = max(1, int(limit)) + params = {"limit": limit} + token = api_key or os.environ.get("FXMACRODATA_API_KEY") + if token: + params["api_key"] = token + + url = f"{base_url.rstrip('/')}/calendar/{currency.lower()}?{urlencode(params)}" + with urlopen(url, timeout=30) as response: # nosec B310 + payload = json.loads(response.read().decode("utf-8")) + + events = payload.get("data", []) + if min_tier is not None: + events = [ + event + for event in events + if int(event.get("market_tier") or 99) <= int(min_tier) + ] + + frame = pd.DataFrame(events[:limit]) + if frame.empty: + return frame + + if "date" in frame.columns: + frame["date"] = pd.to_datetime(frame["date"], errors="coerce") + frame = frame.set_index("date").sort_index() + if "announcement_datetime" in frame.columns: + frame["announcement_datetime"] = pd.to_datetime( + frame["announcement_datetime"], unit="s", utc=True, errors="coerce" + ) + if "announcement_datetime_utc" in frame.columns: + frame["announcement_datetime_utc"] = pd.to_datetime( + frame["announcement_datetime_utc"], utc=True, errors="coerce" + ) + + return frame + + +def event_window_mask( + index: pd.DatetimeIndex, + events: pd.DataFrame, + days_before: int = 1, + days_after: int = 1, +) -> pd.Series: + """Return a boolean mask for dates that fall near release-calendar events.""" + normalized_index = pd.DatetimeIndex(index).normalize() + mask = pd.Series(False, index=index) + if events.empty: + return mask + + event_dates = ( + pd.DatetimeIndex(events.index) + if isinstance(events.index, pd.DatetimeIndex) + else pd.to_datetime(events["date"], errors="coerce") + ) + for event_date in event_dates.dropna().normalize(): + window_start = event_date - pd.Timedelta(days=days_before) + window_end = event_date + pd.Timedelta(days=days_after) + mask |= (normalized_index >= window_start) & (normalized_index <= window_end) + return mask From 12583d762f5473c7147e79013b2cc1fb9fcd28db Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Thu, 9 Jul 2026 00:28:22 +1000 Subject: [PATCH 2/5] Expand FXMacroData read endpoint coverage --- pypfopt/fxmacrodata.py | 538 ++++++++++++++++++++++++++++++++++++----- 1 file changed, 482 insertions(+), 56 deletions(-) diff --git a/pypfopt/fxmacrodata.py b/pypfopt/fxmacrodata.py index c3b6508a..5bbd1922 100644 --- a/pypfopt/fxmacrodata.py +++ b/pypfopt/fxmacrodata.py @@ -1,81 +1,507 @@ -"""FXMacroData helpers for portfolio event-risk analysis.""" +"""FXMacroData client and pandas helpers for macro, FX, and event data.""" from __future__ import annotations import json import os from urllib.parse import urlencode -from urllib.request import urlopen +from urllib.request import Request, urlopen import pandas as pd FXMACRODATA_BASE_URL = "https://fxmacrodata.com/api/v1" +FXMACRODATA_API_KEY_ENV_VARS = ("FXMACRODATA_API_KEY", "FXMD_API_KEY") +FXMACRODATA_ENDPOINTS = { + "data_catalogue": ( + "data_catalogue/{currency}", + {"include_capabilities", "include_coverage", "indicator"}, + ), + "announcements": ( + "announcements/{currency}/{indicator}", + { + "start_date", + "end_date", + "series_mode", + "limit", + "offset", + "page", + "seasonality", + "frequency", + "revisions", + "basis", + "official_only", + }, + ), + "latest_announcements": ("announcements/{currency}/latest", set()), + "announcement_changes": ( + "announcements/changes", + {"currencies", "indicators", "since", "limit", "payload"}, + ), + "predictions": ( + "predictions/{currency}/{indicator}", + { + "prediction_type", + "prediction_source", + "start_date", + "end_date", + "limit", + "offset", + "page", + }, + ), + "calendar": ( + "calendar/{currency}", + {"indicator", "start_date", "end_date", "timezone"}, + ), + "forex": ( + "forex/{base}/{quote}", + {"start_date", "end_date", "limit", "offset", "page", "indicators"}, + ), + "cot": ("cot/{currency}", {"start_date", "end_date", "limit", "offset", "page"}), + "commodity": ( + "commodities/{indicator}", + {"start_date", "end_date", "limit", "offset", "page"}, + ), + "commodities_latest": ("commodities/latest", set()), + "curves": ("curves/{currency}", {"curve_family", "metric", "date"}), + "curve_proxies": ("curve_proxies/{currency}", {"curve_family", "date"}), + "forward_curves": ("forward_curves/{currency}", {"curve_family", "method", "date"}), + "rate_differentials": ( + "rate_differentials/{base}/{quote}", + {"measure", "start_date", "end_date", "limit", "offset"}, + ), + "forward_differentials": ( + "forward_differentials/{base}/{quote}", + { + "curve_family", + "start_tenor_years", + "end_tenor_years", + "start_date", + "end_date", + "limit", + "offset", + }, + ), + "market_sessions": ("market_sessions", {"at"}), + "risk_sentiment": ("risk_sentiment", {"start_date", "end_date", "limit", "offset"}), + "news": ("news/{currency}", {"limit", "offset"}), + "press_releases": ("press-releases/{currency}", {"limit", "offset"}), +} +FXMACRODATA_DATASET_ALIASES = { + "catalogue": "data_catalogue", + "macro": "announcements", + "macro_indicators": "announcements", + "release_calendar": "calendar", + "calendar": "calendar", + "changes": "announcement_changes", + "latest": "latest_announcements", + "fx": "forex", + "fx_spot": "forex", + "commodities": "commodity", + "press-releases": "press_releases", +} -def get_release_calendar( - currency: str = "usd", - limit: int = 100, - min_tier: int | None = None, - api_key: str | None = None, - base_url: str = FXMACRODATA_BASE_URL, -) -> pd.DataFrame: - """Fetch an FXMacroData economic release calendar as a DataFrame.""" - limit = max(1, int(limit)) - params = {"limit": limit} - token = api_key or os.environ.get("FXMACRODATA_API_KEY") - if token: - params["api_key"] = token - - url = f"{base_url.rstrip('/')}/calendar/{currency.lower()}?{urlencode(params)}" - with urlopen(url, timeout=30) as response: # nosec B310 - payload = json.loads(response.read().decode("utf-8")) - - events = payload.get("data", []) - if min_tier is not None: - events = [ - event - for event in events - if int(event.get("market_tier") or 99) <= int(min_tier) - ] - - frame = pd.DataFrame(events[:limit]) +def _env_api_key(): + for name in FXMACRODATA_API_KEY_ENV_VARS: + value = os.environ.get(name) + if value: + return value + return None + + +def _dataset_name(dataset): + normalized = dataset.lower().replace("-", "_") + return FXMACRODATA_DATASET_ALIASES.get(normalized, normalized) + + +def _clean_params(params): + cleaned = {} + for key, value in (params or {}).items(): + if value is None: + continue + if isinstance(value, (list, tuple, set)): + value = ",".join(str(item) for item in value) + cleaned[key] = value + return cleaned + + +def _format_path(path_template, kwargs): + values = { + key: str(kwargs[key]).lower() + for key in ("currency", "base", "quote") + if key in kwargs + } + if "indicator" in kwargs: + values["indicator"] = str(kwargs["indicator"]) + try: + return path_template.format(**values) + except KeyError as exc: + raise ValueError( + "missing required FXMacroData parameter: %s" % exc.args[0] + ) from exc + + +def _payload_rows(payload): + if isinstance(payload, list): + return payload + if not isinstance(payload, dict): + return [] + data = payload.get("data") + if isinstance(data, list): + return data + if isinstance(data, dict): + rows = [] + for key, value in sorted(data.items()): + if isinstance(value, dict): + row = {"indicator": key} + row.update(value) + else: + row = {"indicator": key, "value": value} + rows.append(row) + return rows + for key in ("events", "results", "items", "rows"): + value = payload.get(key) + if isinstance(value, list): + return value + return [payload] + + +def _frame_from_payload(payload, limit=None, index=True): + frame = pd.DataFrame(_payload_rows(payload)) if frame.empty: return frame + for column in ("date", "timestamp", "release_date"): + if column in frame.columns: + frame[column] = pd.to_datetime(frame[column], errors="coerce") + for column in ("announcement_datetime", "announcement_datetime_utc", "time"): + if column not in frame.columns: + continue + if pd.api.types.is_numeric_dtype(frame[column]): + frame[column] = pd.to_datetime( + frame[column], unit="s", utc=True, errors="coerce" + ) + else: + frame[column] = pd.to_datetime(frame[column], utc=True, errors="coerce") + if index: + for column in ("date", "timestamp", "release_date", "time"): + if column in frame.columns: + frame = frame.dropna(subset=[column]).set_index(column).sort_index() + break + if limit is not None: + frame = frame.head(max(1, int(limit))) + return frame + + +def _filter_market_tier(frame, min_tier): + if min_tier is None or frame.empty or "market_tier" not in frame.columns: + return frame + tier = pd.to_numeric(frame["market_tier"], errors="coerce").fillna(99) + return frame.loc[tier <= int(min_tier)] + + +class FXMacroDataClient: + """Small client for the public FXMacroData read/data API surface.""" + + def __init__(self, api_key=None, base_url=FXMACRODATA_BASE_URL, timeout=30): + self.api_key = api_key or _env_api_key() + self.base_url = base_url.rstrip("/") + self.timeout = timeout - if "date" in frame.columns: - frame["date"] = pd.to_datetime(frame["date"], errors="coerce") - frame = frame.set_index("date").sort_index() - if "announcement_datetime" in frame.columns: - frame["announcement_datetime"] = pd.to_datetime( - frame["announcement_datetime"], unit="s", utc=True, errors="coerce" + def fetch_dataset(self, dataset, **kwargs): + dataset = _dataset_name(dataset) + if dataset not in FXMACRODATA_ENDPOINTS: + raise ValueError( + "dataset must be one of %s" % ", ".join(sorted(FXMACRODATA_ENDPOINTS)) + ) + path_template, query_keys = FXMACRODATA_ENDPOINTS[dataset] + path = _format_path(path_template, kwargs) + query = _clean_params({key: kwargs.get(key) for key in query_keys}) + if self.api_key and "api_key" not in query: + query["api_key"] = self.api_key + url = "%s/%s" % (self.base_url, path.lstrip("/")) + if query: + url = "%s?%s" % (url, urlencode(query)) + request = Request(url, headers={"User-Agent": "fxmacrodata-integration"}) + with urlopen(request, timeout=self.timeout) as response: # nosec B310 + return json.loads(response.read().decode("utf-8")) + + def graphql(self, query, variables=None): + body = json.dumps({"query": query, "variables": variables or {}}).encode( + "utf-8" ) - if "announcement_datetime_utc" in frame.columns: - frame["announcement_datetime_utc"] = pd.to_datetime( - frame["announcement_datetime_utc"], utc=True, errors="coerce" + request = Request( + "%s/graphql" % self.base_url, + data=body, + headers={ + "Content-Type": "application/json", + "User-Agent": "fxmacrodata-integration", + }, + method="POST", ) + with urlopen(request, timeout=self.timeout) as response: # nosec B310 + return json.loads(response.read().decode("utf-8")) - return frame + def to_dataframe(self, payload, limit=None, index=True): + return _frame_from_payload(payload, limit=limit, index=index) + + def data_catalogue(self, currency="usd", **kwargs): + return self.fetch_dataset("data_catalogue", currency=currency, **kwargs) + + def announcements(self, currency, indicator, **kwargs): + return self.fetch_dataset( + "announcements", currency=currency, indicator=indicator, **kwargs + ) + + macro_indicators = announcements + + def latest_announcements(self, currency="usd"): + return self.fetch_dataset("latest_announcements", currency=currency) + + def announcement_changes(self, **kwargs): + return self.fetch_dataset("announcement_changes", **kwargs) + + def predictions(self, currency, indicator, **kwargs): + return self.fetch_dataset( + "predictions", currency=currency, indicator=indicator, **kwargs + ) + + def release_calendar(self, currency="usd", **kwargs): + return self.fetch_dataset("calendar", currency=currency, **kwargs) + + def forex(self, base, quote, **kwargs): + return self.fetch_dataset("forex", base=base, quote=quote, **kwargs) + + def cot(self, currency, **kwargs): + return self.fetch_dataset("cot", currency=currency, **kwargs) + + def commodity(self, indicator, **kwargs): + return self.fetch_dataset("commodity", indicator=indicator, **kwargs) + + commodities = commodity + + def commodities_latest(self): + return self.fetch_dataset("commodities_latest") + + def curves(self, currency, **kwargs): + return self.fetch_dataset("curves", currency=currency, **kwargs) + + def curve_proxies(self, currency, **kwargs): + return self.fetch_dataset("curve_proxies", currency=currency, **kwargs) + + def forward_curves(self, currency, **kwargs): + return self.fetch_dataset("forward_curves", currency=currency, **kwargs) + + def rate_differentials(self, base, quote, **kwargs): + return self.fetch_dataset( + "rate_differentials", base=base, quote=quote, **kwargs + ) + + def forward_differentials(self, base, quote, **kwargs): + return self.fetch_dataset( + "forward_differentials", base=base, quote=quote, **kwargs + ) + + def market_sessions(self, **kwargs): + return self.fetch_dataset("market_sessions", **kwargs) + + def risk_sentiment(self, **kwargs): + return self.fetch_dataset("risk_sentiment", **kwargs) + + def news(self, currency, **kwargs): + return self.fetch_dataset("news", currency=currency, **kwargs) + def press_releases(self, currency, **kwargs): + return self.fetch_dataset("press_releases", currency=currency, **kwargs) -def event_window_mask( - index: pd.DatetimeIndex, - events: pd.DataFrame, - days_before: int = 1, - days_after: int = 1, -) -> pd.Series: - """Return a boolean mask for dates that fall near release-calendar events.""" + def dataframe(self, dataset, limit=None, min_tier=None, index=True, **kwargs): + api_limit = None if _dataset_name(dataset) == "calendar" else limit + payload = self.fetch_dataset( + dataset, **{**kwargs, **({"limit": api_limit} if api_limit else {})} + ) + frame = self.to_dataframe(payload, limit=limit, index=index) + return _filter_market_tier(frame, min_tier) + + +def load_fxmacrodata_dataset( + dataset, api_key=None, base_url=FXMACRODATA_BASE_URL, timeout=30, **kwargs +): + """Load any public FXMacroData read dataset into a pandas DataFrame.""" + return FXMacroDataClient( + api_key=api_key, base_url=base_url, timeout=timeout + ).dataframe(dataset, **kwargs) + + +def get_fxmacrodata_dataset(*args, **kwargs): + return load_fxmacrodata_dataset(*args, **kwargs) + + +def load_fxmacrodata_catalogue( + currency="usd", + include_capabilities=False, + include_coverage=False, + indicator=None, + **kwargs, +): + return load_fxmacrodata_dataset( + "data_catalogue", + currency=currency, + include_capabilities=include_capabilities, + include_coverage=include_coverage, + indicator=indicator, + **kwargs, + ) + + +def load_fxmacrodata_announcements( + currency, indicator, start_date=None, end_date=None, limit=20, **kwargs +): + return load_fxmacrodata_dataset( + "announcements", + currency=currency, + indicator=indicator, + start_date=start_date, + end_date=end_date, + limit=limit, + **kwargs, + ) + + +def load_fxmacrodata_latest_announcements(currency="usd", **kwargs): + return load_fxmacrodata_dataset("latest_announcements", currency=currency, **kwargs) + + +def load_fxmacrodata_announcement_changes( + currencies=None, indicators=None, since=None, limit=100, **kwargs +): + return load_fxmacrodata_dataset( + "announcement_changes", + currencies=currencies, + indicators=indicators, + since=since, + payload=kwargs.pop("payload", "compact"), + limit=limit, + **kwargs, + ) + + +def load_fxmacrodata_predictions( + currency, indicator, start_date=None, end_date=None, limit=20, **kwargs +): + return load_fxmacrodata_dataset( + "predictions", + currency=currency, + indicator=indicator, + start_date=start_date, + end_date=end_date, + limit=limit, + **kwargs, + ) + + +def load_fxmacrodata_release_calendar( + currency="usd", + indicator=None, + start_date=None, + end_date=None, + timezone=None, + limit=100, + min_tier=None, + **kwargs, +): + return load_fxmacrodata_dataset( + "calendar", + currency=currency, + indicator=indicator, + start_date=start_date, + end_date=end_date, + timezone=timezone, + limit=limit, + min_tier=min_tier, + **kwargs, + ) + + +def load_fxmacrodata_forex( + base, quote, start_date=None, end_date=None, limit=20, indicators=None, **kwargs +): + return load_fxmacrodata_dataset( + "forex", + base=base, + quote=quote, + start_date=start_date, + end_date=end_date, + limit=limit, + indicators=indicators, + **kwargs, + ) + + +def load_fxmacrodata_cot(currency, start_date=None, end_date=None, limit=20, **kwargs): + return load_fxmacrodata_dataset( + "cot", + currency=currency, + start_date=start_date, + end_date=end_date, + limit=limit, + **kwargs, + ) + + +def load_fxmacrodata_commodity( + indicator, start_date=None, end_date=None, limit=20, **kwargs +): + return load_fxmacrodata_dataset( + "commodity", + indicator=indicator, + start_date=start_date, + end_date=end_date, + limit=limit, + **kwargs, + ) + + +def load_fxmacrodata_commodities_latest(**kwargs): + return load_fxmacrodata_dataset("commodities_latest", **kwargs) + + +def merge_fxmacrodata_features(left, features, tolerance="7D", direction="backward"): + """As-of merge a price/factor frame with FXMacroData macro features.""" + if left.empty or features.empty: + return left.copy() + return pd.merge_asof( + left.sort_index().copy(), + features.sort_index().copy(), + left_index=True, + right_index=True, + direction=direction, + tolerance=pd.Timedelta(tolerance) if tolerance is not None else None, + ) + + +def fxmacrodata_event_windows(index, events, days_before=1, days_after=1): + """Return a boolean Series for observations near FXMacroData release events.""" normalized_index = pd.DatetimeIndex(index).normalize() mask = pd.Series(False, index=index) if events.empty: return mask - - event_dates = ( - pd.DatetimeIndex(events.index) - if isinstance(events.index, pd.DatetimeIndex) - else pd.to_datetime(events["date"], errors="coerce") - ) - for event_date in event_dates.dropna().normalize(): - window_start = event_date - pd.Timedelta(days=days_before) - window_end = event_date + pd.Timedelta(days=days_after) - mask |= (normalized_index >= window_start) & (normalized_index <= window_end) + if isinstance(events.index, pd.DatetimeIndex): + event_dates = pd.DatetimeIndex(events.index) + else: + event_dates = pd.to_datetime(events.get("date"), errors="coerce") + for event_date in pd.DatetimeIndex(event_dates).dropna().normalize(): + start = event_date - pd.Timedelta(days=days_before) + end = event_date + pd.Timedelta(days=days_after) + mask |= (normalized_index >= start) & (normalized_index <= end) return mask + + +get_data_catalogue = load_fxmacrodata_catalogue +get_macro_indicators = load_fxmacrodata_announcements +get_release_calendar = load_fxmacrodata_release_calendar +get_forex = load_fxmacrodata_forex +fxmacrodata_release_calendar = load_fxmacrodata_release_calendar +load_release_calendar = load_fxmacrodata_release_calendar +load_fxmacrodata_calendar = load_fxmacrodata_release_calendar +event_window_mask = fxmacrodata_event_windows +event_window_filter = fxmacrodata_event_windows From ec3756598fa1fb12bdfe7d0f66214aa00c45f1d0 Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Wed, 26 Aug 2026 12:13:15 +1000 Subject: [PATCH 3/5] 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. --- pypfopt/fxmacrodata.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pypfopt/fxmacrodata.py b/pypfopt/fxmacrodata.py index 5bbd1922..37fea4b4 100644 --- a/pypfopt/fxmacrodata.py +++ b/pypfopt/fxmacrodata.py @@ -9,7 +9,7 @@ import pandas as pd -FXMACRODATA_BASE_URL = "https://fxmacrodata.com/api/v1" +FXMACRODATA_BASE_URL = "https://api.fxmacrodata.com/v1" FXMACRODATA_API_KEY_ENV_VARS = ("FXMACRODATA_API_KEY", "FXMD_API_KEY") FXMACRODATA_ENDPOINTS = { "data_catalogue": ( From 2fb54a67f215bf71fa67fa597298c2fc84832605 Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Mon, 28 Sep 2026 14:18:47 +1000 Subject: [PATCH 4/5] [ENH] Send the FXMacroData key as a header and page long windows The key now goes in the X-API-Key header instead of the query string, so it no longer ends up in URLs or logs. List endpoints return at most 100 rows per request, so history windows are read with limit/offset until pagination.has_more is false. --- pypfopt/fxmacrodata.py | 77 ++++++++++++++++++++++---- tests/test_fxmacrodata.py | 110 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 176 insertions(+), 11 deletions(-) create mode 100644 tests/test_fxmacrodata.py diff --git a/pypfopt/fxmacrodata.py b/pypfopt/fxmacrodata.py index 37fea4b4..d629da0e 100644 --- a/pypfopt/fxmacrodata.py +++ b/pypfopt/fxmacrodata.py @@ -11,6 +11,19 @@ FXMACRODATA_BASE_URL = "https://api.fxmacrodata.com/v1" FXMACRODATA_API_KEY_ENV_VARS = ("FXMACRODATA_API_KEY", "FXMD_API_KEY") +# List endpoints return at most 100 rows per request, newest first. +FXMACRODATA_PAGE_SIZE = 100 +FXMACRODATA_MAX_PAGES = 100 +FXMACRODATA_PAGED_DATASETS = { + "announcements", + "predictions", + "forex", + "cot", + "commodity", + "rate_differentials", + "forward_differentials", + "risk_sentiment", +} FXMACRODATA_ENDPOINTS = { "data_catalogue": ( "data_catalogue/{currency}", @@ -208,6 +221,12 @@ def __init__(self, api_key=None, base_url=FXMACRODATA_BASE_URL, timeout=30): self.base_url = base_url.rstrip("/") self.timeout = timeout + def _headers(self): + headers = {"User-Agent": "fxmacrodata-integration"} + if self.api_key: + headers["X-API-Key"] = self.api_key + return headers + def fetch_dataset(self, dataset, **kwargs): dataset = _dataset_name(dataset) if dataset not in FXMACRODATA_ENDPOINTS: @@ -217,12 +236,12 @@ def fetch_dataset(self, dataset, **kwargs): path_template, query_keys = FXMACRODATA_ENDPOINTS[dataset] path = _format_path(path_template, kwargs) query = _clean_params({key: kwargs.get(key) for key in query_keys}) - if self.api_key and "api_key" not in query: - query["api_key"] = self.api_key + if dataset in FXMACRODATA_PAGED_DATASETS and "limit" in query: + query["limit"] = min(int(query["limit"]), FXMACRODATA_PAGE_SIZE) url = "%s/%s" % (self.base_url, path.lstrip("/")) if query: url = "%s?%s" % (url, urlencode(query)) - request = Request(url, headers={"User-Agent": "fxmacrodata-integration"}) + request = Request(url, headers=self._headers()) with urlopen(request, timeout=self.timeout) as response: # nosec B310 return json.loads(response.read().decode("utf-8")) @@ -233,10 +252,7 @@ def graphql(self, query, variables=None): request = Request( "%s/graphql" % self.base_url, data=body, - headers={ - "Content-Type": "application/json", - "User-Agent": "fxmacrodata-integration", - }, + headers={"Content-Type": "application/json", **self._headers()}, method="POST", ) with urlopen(request, timeout=self.timeout) as response: # nosec B310 @@ -314,11 +330,50 @@ def news(self, currency, **kwargs): def press_releases(self, currency, **kwargs): return self.fetch_dataset("press_releases", currency=currency, **kwargs) + def fetch_rows( + self, dataset, limit=None, max_pages=FXMACRODATA_MAX_PAGES, **kwargs + ): + """Page through a list endpoint and return up to ``limit`` rows. + + Rows come back newest first, at most 100 per request, so the window is + read with ``offset`` until ``pagination.has_more`` is false. With + ``limit=None`` the whole window is read, up to ``max_pages`` requests. + """ + kwargs.pop("page", None) + offset = int(kwargs.pop("offset", None) or 0) + limit = None if limit is None else max(1, int(limit)) + rows = [] + for _ in range(max(1, int(max_pages))): + page_size = FXMACRODATA_PAGE_SIZE + if limit is not None: + page_size = min(page_size, limit - len(rows)) + payload = self.fetch_dataset( + dataset, limit=page_size, offset=offset, **kwargs + ) + page = payload.get("data") if isinstance(payload, dict) else payload + if not isinstance(page, list) or not page: + break + rows.extend(page) + if limit is not None and len(rows) >= limit: + break + pagination = ( + payload.get("pagination") if isinstance(payload, dict) else None + ) + if not isinstance(pagination, dict) or not pagination.get("has_more"): + break + next_offset = pagination.get("next_offset") + offset = int(next_offset) if next_offset is not None else offset + len(page) + return rows + def dataframe(self, dataset, limit=None, min_tier=None, index=True, **kwargs): - api_limit = None if _dataset_name(dataset) == "calendar" else limit - payload = self.fetch_dataset( - dataset, **{**kwargs, **({"limit": api_limit} if api_limit else {})} - ) + name = _dataset_name(dataset) + if name in FXMACRODATA_PAGED_DATASETS: + payload = {"data": self.fetch_rows(name, limit=limit, **kwargs)} + else: + api_limit = None if name == "calendar" else limit + payload = self.fetch_dataset( + name, **{**kwargs, **({"limit": api_limit} if api_limit else {})} + ) frame = self.to_dataframe(payload, limit=limit, index=index) return _filter_market_tier(frame, min_tier) diff --git a/tests/test_fxmacrodata.py b/tests/test_fxmacrodata.py new file mode 100644 index 00000000..b4b6d87e --- /dev/null +++ b/tests/test_fxmacrodata.py @@ -0,0 +1,110 @@ +from datetime import date, timedelta +import json +from urllib.parse import parse_qs, urlparse + +import pytest + +from pypfopt import fxmacrodata + + +class _Response: + def __init__(self, payload): + self._body = json.dumps(payload).encode("utf-8") + + def read(self): + return self._body + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + +def _rows(total): + # Newest first, like the API. + start = date(2024, 1, 1) + return [ + {"date": (start + timedelta(days=i)).isoformat(), "val": float(i)} + for i in reversed(range(total)) + ] + + +@pytest.fixture +def fake_api(monkeypatch): + for name in fxmacrodata.FXMACRODATA_API_KEY_ENV_VARS: + monkeypatch.delenv(name, raising=False) + requests = [] + rows = _rows(250) + + def urlopen(request, timeout=None): + requests.append(request) + query = parse_qs(urlparse(request.full_url).query) + limit = int(query.get("limit", ["20"])[0]) + offset = int(query.get("offset", ["0"])[0]) + page = rows[offset : offset + limit] + has_more = offset + len(page) < len(rows) + return _Response( + { + "data": page, + "pagination": { + "limit": limit, + "offset": offset, + "returned_count": len(page), + "total_count": len(rows), + "has_more": has_more, + "next_offset": offset + len(page) if has_more else None, + }, + } + ) + + monkeypatch.setattr(fxmacrodata, "urlopen", urlopen) + return requests + + +def test_api_key_is_sent_as_header(fake_api): + client = fxmacrodata.FXMacroDataClient(api_key="test-key") + client.fetch_dataset("forex", base="eur", quote="usd", limit=5) + request = fake_api[0] + assert request.get_header("X-api-key") == "test-key" + assert "test-key" not in request.full_url + assert "api_key" not in request.full_url + + +def test_no_api_key_header_without_key(fake_api): + client = fxmacrodata.FXMacroDataClient() + client.fetch_dataset("announcements", currency="usd", indicator="inflation") + request = fake_api[0] + assert request.get_header("X-api-key") is None + assert "api_key" not in request.full_url + + +def test_dataframe_pages_through_window(fake_api): + client = fxmacrodata.FXMacroDataClient(api_key="test-key") + frame = client.dataframe( + "forex", base="eur", quote="usd", start_date="2024-01-01", end_date="2024-12-31" + ) + assert len(frame) == 250 + assert frame.index.is_monotonic_increasing + queries = [parse_qs(urlparse(r.full_url).query) for r in fake_api] + assert [q["offset"] for q in queries] == [["0"], ["100"], ["200"]] + assert all(q["limit"] == ["100"] for q in queries) + assert all("api_key" not in r.full_url for r in fake_api) + + +def test_dataframe_limit_caps_rows_and_page_size(fake_api): + client = fxmacrodata.FXMacroDataClient() + frame = client.dataframe("cot", currency="eur", limit=150) + assert len(frame) == 150 + queries = [parse_qs(urlparse(r.full_url).query) for r in fake_api] + assert [(q["limit"], q["offset"]) for q in queries] == [ + (["100"], ["0"]), + (["50"], ["100"]), + ] + assert frame.index.max().isoformat().startswith("2024-09-06") + + +def test_fetch_dataset_never_sends_limit_above_100(fake_api): + client = fxmacrodata.FXMacroDataClient() + client.fetch_dataset("commodity", indicator="gold", limit=500) + assert parse_qs(urlparse(fake_api[0].full_url).query)["limit"] == ["100"] From 91799919245e5bd10ff8b2c023617a97c8651970 Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Mon, 5 Oct 2026 20:54:12 +1100 Subject: [PATCH 5/5] [ENH] Keep the FXMacroData key off redirects and reject error bodies cleanly --- pypfopt/fxmacrodata.py | 36 ++++++++++++++++++++++++++++-------- tests/test_fxmacrodata.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 8 deletions(-) diff --git a/pypfopt/fxmacrodata.py b/pypfopt/fxmacrodata.py index d629da0e..9dad01cf 100644 --- a/pypfopt/fxmacrodata.py +++ b/pypfopt/fxmacrodata.py @@ -206,6 +206,17 @@ def _frame_from_payload(payload, limit=None, index=True): return frame +def _check_payload(payload): + """Raise a clean error for an error body or an unexpected JSON shape.""" + if isinstance(payload, list): + return payload + if not isinstance(payload, dict): + raise ValueError("FXMacroData returned an unexpected response shape") + if "detail" in payload and "data" not in payload: + raise ValueError("FXMacroData returned an error: %s" % payload["detail"]) + return payload + + def _filter_market_tier(frame, min_tier): if min_tier is None or frame.empty or "market_tier" not in frame.columns: return frame @@ -217,15 +228,26 @@ class FXMacroDataClient: """Small client for the public FXMacroData read/data API surface.""" def __init__(self, api_key=None, base_url=FXMACRODATA_BASE_URL, timeout=30): - self.api_key = api_key or _env_api_key() + self.api_key = (api_key or _env_api_key() or "").strip() or None + if self.api_key and any( + ch.isspace() or not ch.isprintable() for ch in self.api_key + ): + raise ValueError( + "FXMacroData API key contains whitespace or control characters" + ) self.base_url = base_url.rstrip("/") self.timeout = timeout def _headers(self): - headers = {"User-Agent": "fxmacrodata-integration"} + return {"User-Agent": "fxmacrodata-integration"} + + def _open(self, request): if self.api_key: - headers["X-API-Key"] = self.api_key - return headers + # Unredirected headers are not copied onto a redirected request, + # so the key is never forwarded to another host. + request.add_unredirected_header("X-API-Key", self.api_key) + with urlopen(request, timeout=self.timeout) as response: # nosec B310 + return json.loads(response.read().decode("utf-8")) def fetch_dataset(self, dataset, **kwargs): dataset = _dataset_name(dataset) @@ -242,8 +264,7 @@ def fetch_dataset(self, dataset, **kwargs): if query: url = "%s?%s" % (url, urlencode(query)) request = Request(url, headers=self._headers()) - with urlopen(request, timeout=self.timeout) as response: # nosec B310 - return json.loads(response.read().decode("utf-8")) + return _check_payload(self._open(request)) def graphql(self, query, variables=None): body = json.dumps({"query": query, "variables": variables or {}}).encode( @@ -255,8 +276,7 @@ def graphql(self, query, variables=None): headers={"Content-Type": "application/json", **self._headers()}, method="POST", ) - with urlopen(request, timeout=self.timeout) as response: # nosec B310 - return json.loads(response.read().decode("utf-8")) + return self._open(request) def to_dataframe(self, payload, limit=None, index=True): return _frame_from_payload(payload, limit=limit, index=index) diff --git a/tests/test_fxmacrodata.py b/tests/test_fxmacrodata.py index b4b6d87e..07e28305 100644 --- a/tests/test_fxmacrodata.py +++ b/tests/test_fxmacrodata.py @@ -108,3 +108,34 @@ def test_fetch_dataset_never_sends_limit_above_100(fake_api): client = fxmacrodata.FXMacroDataClient() client.fetch_dataset("commodity", indicator="gold", limit=500) assert parse_qs(urlparse(fake_api[0].full_url).query)["limit"] == ["100"] + + +def test_api_key_is_not_forwarded_on_redirect(fake_api): + from urllib.request import HTTPRedirectHandler + + client = fxmacrodata.FXMacroDataClient(api_key="test-key") + client.fetch_dataset("forex", base="eur", quote="usd", limit=5) + redirected = HTTPRedirectHandler().redirect_request( + fake_api[0], None, 302, "Found", {}, "https://elsewhere.example/v1/forex" + ) + assert redirected.get_header("X-api-key") is None + assert "test-key" not in str(redirected.header_items()) + + +def test_api_key_with_control_characters_is_rejected_without_echo(monkeypatch): + for name in fxmacrodata.FXMACRODATA_API_KEY_ENV_VARS: + monkeypatch.delenv(name, raising=False) + with pytest.raises(ValueError) as excinfo: + fxmacrodata.FXMacroDataClient(api_key="test-key\nInjected: 1") + assert "test-key" not in str(excinfo.value) + assert fxmacrodata.FXMacroDataClient(api_key=" test-key ").api_key == "test-key" + + +@pytest.mark.parametrize("payload", [{"detail": "Invalid API key"}, "oops"]) +def test_error_body_raises_clean_error(monkeypatch, payload): + monkeypatch.setattr( + fxmacrodata, "urlopen", lambda request, timeout=None: _Response(payload) + ) + client = fxmacrodata.FXMacroDataClient(api_key="test-key") + with pytest.raises(ValueError, match="FXMacroData returned"): + client.fetch_dataset("cot", currency="eur")