From 7a17b1328b1a683d0a502495f20deb1066f324d1 Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Thu, 1 Oct 2026 13:08:27 +1000 Subject: [PATCH 1/2] Add FXMacroData macro provider --- docs/api/providers.md | 14 + docs/data-model.md | 4 +- docs/providers.md | 9 + docs/providers/fxmacrodata.md | 118 +++++ mkdocs.yml | 1 + src/pineforge_data/__init__.py | 14 + src/pineforge_data/providers/__init__.py | 18 + src/pineforge_data/providers/fxmacrodata.py | 440 ++++++++++++++++++ .../fxmacrodata_usd_non_farm_payrolls.json | 91 ++++ tests/test_fxmacrodata.py | 248 ++++++++++ 10 files changed, 955 insertions(+), 2 deletions(-) create mode 100644 docs/providers/fxmacrodata.md create mode 100644 src/pineforge_data/providers/fxmacrodata.py create mode 100644 tests/fixtures/fxmacrodata_usd_non_farm_payrolls.json create mode 100644 tests/test_fxmacrodata.py diff --git a/docs/api/providers.md b/docs/api/providers.md index bd72350..e896d37 100644 --- a/docs/api/providers.md +++ b/docs/api/providers.md @@ -42,6 +42,12 @@ ::: pineforge_data.providers.sqlalchemy.SqlAlchemyBarProvider +::: pineforge_data.providers.fxmacrodata.FxMacroDataProvider + +::: pineforge_data.providers.fxmacrodata.FxMacroDataTransport + +::: pineforge_data.providers.fxmacrodata.FxMacroDataResponse + ## Provider errors ::: pineforge_data.providers.ccxt.CcxtError @@ -53,3 +59,11 @@ ::: pineforge_data.providers.ccxt.CcxtDataError ::: pineforge_data.providers.sqlalchemy.SqlAlchemyDependencyError + +::: pineforge_data.providers.fxmacrodata.FxMacroDataError + +::: pineforge_data.providers.fxmacrodata.FxMacroDataHTTPError + +::: pineforge_data.providers.fxmacrodata.FxMacroDataDataError + +::: pineforge_data.providers.fxmacrodata.FxMacroDataAccessWarning diff --git a/docs/data-model.md b/docs/data-model.md index d9ae952..6c62aac 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -117,8 +117,8 @@ The enforced ordering is `period_end_ms <= released_at_ms <= vintage_at_ms`. Backtests must align on availability/vintage time rather than inserting today's revised value into historical periods. -`MacroDataProvider` and `MacroRequest` define the public contract; the bootstrap -package does not yet ship a built-in macro provider. +`MacroDataProvider` and `MacroRequest` define the public contract. +[`FxMacroDataProvider`](providers/fxmacrodata.md) is the built-in macro provider. ## Low-level engine streaming diff --git a/docs/providers.md b/docs/providers.md index d302054..5e16be3 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -21,6 +21,15 @@ guide for each built-in provider. | `sqlite` | Local SQLite table or view | reflected catalog, historical bars | none | [SQLite](providers/sqlite.md) | | `sqlalchemy` | SQLAlchemy-supported database table or view | reflected catalog, historical bars | `database` plus a dialect driver | [SQLAlchemy](providers/sqlalchemy.md) | +## Macro providers + +| Provider | Data source | Capabilities | Install extra | API guide | +|---|---|---|---|---| +| `FxMacroDataProvider` | FXMacroData economic announcements | macro observations with release and vintage times | none | [FXMacroData](providers/fxmacrodata.md) | + +Macro providers implement `MacroDataProvider` and are constructed directly; +they are not registry entries for the backtest harness. + CSV, SQLite, and SQLAlchemy share runtime schema discovery and arbitrary column mapping. Read the [tabular schema mapping](providers/tabular-schema.md) guide before configuring one of them. diff --git a/docs/providers/fxmacrodata.md b/docs/providers/fxmacrodata.md new file mode 100644 index 0000000..30b34ac --- /dev/null +++ b/docs/providers/fxmacrodata.md @@ -0,0 +1,118 @@ +# FXMacroData macro provider API + +`FxMacroDataProvider` implements `MacroDataProvider`. It reads economic +announcements (CPI, policy rates, payrolls, GDP and so on) from the +[FXMacroData API](https://fxmacrodata.com/documentation/reference?utm_source=github&utm_medium=referral&utm_campaign=pineforge-data&utm_content=docs) +and returns `MacroObservation` records with release and vintage timestamps. It +has no market catalog and no bars, so it is not a `MarketDataProvider` and is +not registered for `pineforge-backtest --provider`. + +FXMacroData is a commercial API. Without a key only USD is available, limited +to the most recent 90 days and delayed by 15 minutes. Other currencies and full +history need an API key. + +## Install + +```bash +pip install pineforge-data +``` + +No extra is needed: the default transport uses the standard library. + +## Construct the provider + +```python +import os + +from pineforge_data import FxMacroDataProvider, MacroRequest + + +async def us_payrolls(): + async with FxMacroDataProvider(api_key=os.environ.get("FXMACRODATA_API_KEY")) as provider: + return await provider.fetch_observations( + MacroRequest( + key="non_farm_payrolls", + currency="USD", + start_ms=1_775_001_600_000, # 2026-04-01 + end_ms=1_790_812_800_000, # 2026-10-01 + ) + ) +``` + +`key` is an FXMacroData indicator slug, such as `inflation`, `policy_rate`, +`non_farm_payrolls` or `gdp`. The per-currency list is served by +`GET /v1/data_catalogue/{currency}`. `currency` is a currency code (`USD`, +`EUR`, `JPY` and so on). + +The key is sent only in the `X-API-Key` header. It is not placed in URLs, +exception messages or the provider's `repr`. + +## Timestamps and vintages + +| Field | Taken from | +|---|---| +| `period_end_ms` | UTC midnight of the record's `date` | +| `released_at_ms` | the record's `announcement_datetime` | +| `vintage_at_ms` | when that value became available (below) | + +The provider requests `revisions=all` and emits one observation each time a +period's value changed: + +- a source vintage (`vintage_status: source_vintage`) is dated by its + publication time (`publication_at_ns`, else `epoch`); +- a captured snapshot (`vintage_status: captured_snapshot`) is dated by + `observed_at_ns`, the time FXMacroData read it from the publisher. Its + `epoch` repeats the original release time, so using it would date a later + revision back to the first print; +- a vintage is never earlier than `released_at_ms`, and a vintage that repeats + the previous value is dropped. + +Some history was collected after publication and has only a captured snapshot. +Those values are dated by when they were collected, so a backtest set before +that date does not see them. This is deliberate: the alternative is +lookahead. + +## Records that are skipped + +`MacroObservation` requires a release time, and the provider does not invent +one. These records are left out and counted in `provider.last_skipped` for the +most recent fetch: + +| Reason | Meaning | +|---|---| +| `unknown_release_time` | `announcement_datetime` is null (`publication_time_status: unknown`) | +| `assumed_release_time` | the time was derived by FXMacroData from the series' usual publication lag, not captured (`release_time_assumed` or `publication_time_status: assumed_historical`). Pass `include_assumed_release_times=True` to keep them | +| `released_before_period_end` | the release precedes the period date, for example a flash estimate, which `MacroObservation` cannot represent | +| `revision_without_vintage_time` | a revision has no usable availability time | +| `missing_value` | a record or revision has a null value | + +## Constructor reference + +| Argument | Default | Meaning | +|---|---|---| +| `api_key` | `None` | FXMacroData API key; `None` uses the keyless USD tier | +| `api_url` | `https://api.fxmacrodata.com/v1` | API root | +| `timeout_seconds` | `20.0` | per-request timeout of the default transport | +| `max_retries` | `2` | retries after HTTP 429, 500, 502, 503 or 504 | +| `retry_backoff_seconds` | `1.0` | first retry delay, doubled each attempt; `Retry-After` takes precedence | +| `include_assumed_release_times` | `False` | keep records whose release time was derived rather than captured | +| `transport` | `None` | an `FxMacroDataTransport`; inject one for offline tests or a custom HTTP client | + +Pages hold at most 100 records; the provider follows `pagination.has_more` and +`next_offset` until the request is complete. + +## Errors and limitations + +| Error | Meaning | +|---|---| +| `FxMacroDataHTTPError` | a non-success status after retries; `.status` holds the code (403 for a currency or history the key does not cover) | +| `FxMacroDataDataError` | a response or record cannot be normalized safely | +| `FxMacroDataError` | base class; also raised for network failures | +| `FxMacroDataAccessWarning` | a warning, not an error: the keyless tier cut the requested history short or withheld a release from the last 15 minutes | + +- Units come from the response's `value_metadata.source_unit`; `unknown` is + used when it is absent. +- Responses are not cached. Snapshot them at your job boundary when exact + replay matters. +- Only the announcements endpoint is used. Release calendars, consensus + forecasts and FX prices are out of scope for this provider. diff --git a/mkdocs.yml b/mkdocs.yml index 6bb8215..fcb39fc 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -81,6 +81,7 @@ nav: - CSV: providers/csv.md - SQLite: providers/sqlite.md - SQLAlchemy: providers/sqlalchemy.md + - FXMacroData: providers/fxmacrodata.md - Backtesting: - Harness and reports: backtesting.md - FastAPI server: server.md diff --git a/src/pineforge_data/__init__.py b/src/pineforge_data/__init__.py index 7617ce2..872d11a 100644 --- a/src/pineforge_data/__init__.py +++ b/src/pineforge_data/__init__.py @@ -34,6 +34,13 @@ CcxtError, CcxtProvider, CsvBarProvider, + FxMacroDataAccessWarning, + FxMacroDataDataError, + FxMacroDataError, + FxMacroDataHTTPError, + FxMacroDataProvider, + FxMacroDataResponse, + FxMacroDataTransport, HistoricalBarProvider, LiveTradeProvider, MacroDataProvider, @@ -84,6 +91,13 @@ "EngineStreamError", "EngineStreamSink", "FastApiBacktestClient", + "FxMacroDataAccessWarning", + "FxMacroDataDataError", + "FxMacroDataError", + "FxMacroDataHTTPError", + "FxMacroDataProvider", + "FxMacroDataResponse", + "FxMacroDataTransport", "HistoricalBarProvider", "Instrument", "LiveTradeProvider", diff --git a/src/pineforge_data/providers/__init__.py b/src/pineforge_data/providers/__init__.py index e960ab3..ec2931a 100644 --- a/src/pineforge_data/providers/__init__.py +++ b/src/pineforge_data/providers/__init__.py @@ -15,6 +15,16 @@ CcxtError, CcxtProvider, ) +from .fxmacrodata import ( + FXMACRODATA_API_URL, + FxMacroDataAccessWarning, + FxMacroDataDataError, + FxMacroDataError, + FxMacroDataHTTPError, + FxMacroDataProvider, + FxMacroDataResponse, + FxMacroDataTransport, +) from .local import CsvBarProvider, SqliteBarProvider from .registry import ( ENTRY_POINT_GROUP, @@ -38,6 +48,7 @@ __all__ = [ "ENTRY_POINT_GROUP", + "FXMACRODATA_API_URL", "BarColumnMapping", "CcxtCapabilityError", "CcxtDataError", @@ -45,6 +56,13 @@ "CcxtError", "CcxtProvider", "CsvBarProvider", + "FxMacroDataAccessWarning", + "FxMacroDataDataError", + "FxMacroDataError", + "FxMacroDataHTTPError", + "FxMacroDataProvider", + "FxMacroDataResponse", + "FxMacroDataTransport", "HistoricalBarProvider", "LiveTradeProvider", "MacroDataProvider", diff --git a/src/pineforge_data/providers/fxmacrodata.py b/src/pineforge_data/providers/fxmacrodata.py new file mode 100644 index 0000000..9fd9335 --- /dev/null +++ b/src/pineforge_data/providers/fxmacrodata.py @@ -0,0 +1,440 @@ +"""FXMacroData adapter for release- and vintage-aware macro observations.""" + +from __future__ import annotations + +import asyncio +import json +import warnings +from collections import Counter +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import UTC, datetime +from email.message import Message +from math import isfinite +from typing import Protocol +from urllib.error import HTTPError, URLError +from urllib.parse import quote, urlencode +from urllib.request import Request, urlopen + +from ..models import MacroObservation +from ..requests import MacroRequest + +FXMACRODATA_API_URL = "https://api.fxmacrodata.com/v1" +_PAGE_LIMIT = 100 +_RETRY_STATUSES = frozenset({429, 500, 502, 503, 504}) + + +class FxMacroDataError(RuntimeError): + """Base error raised by the FXMacroData macro provider.""" + + +class FxMacroDataHTTPError(FxMacroDataError): + """FXMacroData answered with a non-success HTTP status.""" + + def __init__(self, status: int, message: str) -> None: + super().__init__(f"FXMacroData returned HTTP {status}: {message}") + self.status = status + + +class FxMacroDataDataError(FxMacroDataError): + """An FXMacroData response cannot be normalized safely.""" + + +class FxMacroDataAccessWarning(UserWarning): + """The response was limited by the keyless access tier.""" + + +@dataclass(frozen=True, slots=True) +class FxMacroDataResponse: + """Status, decoded JSON body, and optional ``Retry-After`` delay of one response.""" + + status: int + payload: object + retry_after_seconds: float | None = None + + +class FxMacroDataTransport(Protocol): + """Async GET transport; inject one for offline tests or a custom HTTP stack.""" + + async def get( + self, + url: str, + *, + params: Mapping[str, str | int], + headers: Mapping[str, str], + ) -> FxMacroDataResponse: ... + + +def _retry_after(headers: Message | None) -> float | None: + value = None if headers is None else headers.get("Retry-After") + try: + return None if value is None else max(0.0, float(value)) + except ValueError: + return None + + +def _decode(body: bytes) -> object: + try: + return json.loads(body.decode("utf-8")) + except (UnicodeDecodeError, ValueError): + return None + + +class UrllibTransport: + """Standard-library transport: no optional dependency is required.""" + + def __init__(self, timeout_seconds: float) -> None: + self.timeout_seconds = timeout_seconds + + def _get(self, url: str, headers: Mapping[str, str]) -> FxMacroDataResponse: + request = Request(url, headers=dict(headers), method="GET") + try: + with urlopen(request, timeout=self.timeout_seconds) as response: + return FxMacroDataResponse(response.status, _decode(response.read())) + except HTTPError as exc: + return FxMacroDataResponse(exc.code, _decode(exc.read()), _retry_after(exc.headers)) + except (URLError, TimeoutError) as exc: + raise FxMacroDataError(f"FXMacroData request failed: {exc}") from exc + + async def get( + self, + url: str, + *, + params: Mapping[str, str | int], + headers: Mapping[str, str], + ) -> FxMacroDataResponse: + query = urlencode({key: str(value) for key, value in params.items()}) + return await asyncio.to_thread(self._get, f"{url}?{query}", headers) + + +def _text(value: object, field: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise FxMacroDataDataError(f"{field} must be a non-empty string") + return value.strip() + + +def _optional_number(value: object, field: str) -> float | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise FxMacroDataDataError(f"{field} must be numeric") + normalized = float(value) + if not isfinite(normalized): + raise FxMacroDataDataError(f"{field} must be finite") + return normalized + + +def _optional_epoch_ms(value: object, field: str, *, divisor: int) -> int | None: + """Convert epoch seconds (``divisor=1``) or nanoseconds to Unix milliseconds.""" + + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise FxMacroDataDataError(f"{field} must be a non-negative integer epoch") + return value * 1_000 if divisor == 1 else value // divisor + + +def _period_end_ms(value: object) -> int: + date_text = _text(value, "record.date") + try: + day = datetime.strptime(date_text, "%Y-%m-%d").replace(tzinfo=UTC) + except ValueError as exc: + raise FxMacroDataDataError("record.date must be an ISO YYYY-MM-DD date") from exc + return int(day.timestamp() * 1_000) + + +def _request_date(timestamp_ms: int) -> str: + return datetime.fromtimestamp(timestamp_ms / 1_000, tz=UTC).date().isoformat() + + +def _available_at_ms(entry: Mapping[str, object], field: str) -> int | None: + """When the value in ``entry`` is known to have been available. + + A ``captured_snapshot`` was read from the publisher at ``observed_at_ns``; + its ``epoch`` repeats the original release time, so using it would date a + later revision back to the first release. Source vintages carry their own + publication time. + """ + + if entry.get("vintage_status") == "captured_snapshot": + return _optional_epoch_ms( + entry.get("observed_at_ns"), f"{field}.observed_at_ns", divisor=1_000_000 + ) + published = _optional_epoch_ms( + entry.get("publication_at_ns"), f"{field}.publication_at_ns", divisor=1_000_000 + ) + if published is not None: + return published + return _optional_epoch_ms(entry.get("epoch"), f"{field}.epoch", divisor=1) + + +def _payload_records(payload: Mapping[str, object]) -> Sequence[Mapping[str, object]]: + records = payload.get("data") + if not isinstance(records, list): + raise FxMacroDataDataError("response.data must be a list") + normalized: list[Mapping[str, object]] = [] + for record in records: + if not isinstance(record, Mapping): + raise FxMacroDataDataError("response.data must contain only objects") + normalized.append(record) + return normalized + + +def _next_offset(payload: Mapping[str, object], current: int, returned: int) -> int | None: + pagination = payload.get("pagination") + if pagination is None: + return None + if not isinstance(pagination, Mapping): + raise FxMacroDataDataError("response.pagination must be an object") + if pagination.get("has_more") is not True: + return None + next_offset = pagination.get("next_offset", current + returned) + if isinstance(next_offset, bool) or not isinstance(next_offset, int) or next_offset <= current: + raise FxMacroDataDataError("response.pagination.next_offset must advance the cursor") + return next_offset + + +def _error_message(payload: object) -> str: + if isinstance(payload, Mapping): + for field in ("detail", "message", "error"): + value = payload.get(field) + if isinstance(value, str) and value.strip(): + return value.strip() + if isinstance(value, Mapping): + nested = value.get("message") + if isinstance(nested, str) and nested.strip(): + return nested.strip() + return "no error message" + + +class FxMacroDataProvider: + """Fetch FXMacroData announcements as ``MacroObservation`` vintages. + + ``request.key`` is an FXMacroData indicator slug (``inflation``, + ``policy_rate``, ``non_farm_payrolls``) and ``request.currency`` a currency + code. The provider asks for ``revisions=all`` and emits one observation per + distinct value a period has had, dated by when that value became available. + + Records without a publication time are skipped rather than given one, and so + are records whose time FXMacroData derived instead of captured, unless + ``include_assumed_release_times`` is set. ``last_skipped`` counts the + records and revisions left out of the most recent fetch, by reason. + """ + + def __init__( + self, + *, + api_key: str | None = None, + api_url: str = FXMACRODATA_API_URL, + timeout_seconds: float = 20.0, + max_retries: int = 2, + retry_backoff_seconds: float = 1.0, + include_assumed_release_times: bool = False, + transport: FxMacroDataTransport | None = None, + ) -> None: + if not api_url.strip(): + raise ValueError("api_url must not be empty") + if timeout_seconds <= 0: + raise ValueError("timeout_seconds must be positive") + if max_retries < 0: + raise ValueError("max_retries must be non-negative") + if retry_backoff_seconds < 0: + raise ValueError("retry_backoff_seconds must be non-negative") + if api_key is not None and not api_key.strip(): + raise ValueError("api_key must not be empty when supplied") + + self.name = "fxmacrodata" + self._api_key = api_key + self.api_url = api_url.rstrip("/") + self.timeout_seconds = timeout_seconds + self.max_retries = max_retries + self.retry_backoff_seconds = retry_backoff_seconds + self.include_assumed_release_times = include_assumed_release_times + self._transport: FxMacroDataTransport = transport or UrllibTransport(timeout_seconds) + self.last_skipped: Mapping[str, int] = {} + + def __repr__(self) -> str: + key_state = "set" if self._api_key else "unset" + return f"FxMacroDataProvider(api_url={self.api_url!r}, api_key={key_state})" + + def _headers(self) -> dict[str, str]: + headers = { + "Accept": "application/json", + "User-Agent": "pineforge-data-fxmacrodata", + } + if self._api_key is not None: + headers["X-API-Key"] = self._api_key + return headers + + async def _get_page(self, url: str, params: Mapping[str, str | int]) -> Mapping[str, object]: + attempt = 0 + while True: + response = await self._transport.get(url, params=params, headers=self._headers()) + if response.status in _RETRY_STATUSES and attempt < self.max_retries: + delay = response.retry_after_seconds + if delay is None: + delay = self.retry_backoff_seconds * 2**attempt + await asyncio.sleep(delay) + attempt += 1 + continue + if not 200 <= response.status < 300: + raise FxMacroDataHTTPError(response.status, _error_message(response.payload)) + if not isinstance(response.payload, Mapping): + raise FxMacroDataDataError("response must be a JSON object") + return response.payload + + def _release_ms(self, record: Mapping[str, object], skipped: Counter[str]) -> int | None: + released_at_ms = _optional_epoch_ms( + record.get("announcement_datetime"), "record.announcement_datetime", divisor=1 + ) + if released_at_ms is None: + skipped["unknown_release_time"] += 1 + return None + assumed = ( + record.get("release_time_assumed") is True + or record.get("publication_time_status") == "assumed_historical" + ) + if assumed and not self.include_assumed_release_times: + skipped["assumed_release_time"] += 1 + return None + return released_at_ms + + @staticmethod + def _vintages( + record: Mapping[str, object], released_at_ms: int, skipped: Counter[str] + ) -> list[tuple[int, float]]: + revisions = record.get("revisions") + if revisions is None: + entries: list[tuple[str, Mapping[str, object]]] = [("record", record)] + elif isinstance(revisions, list): + entries = [] + for index, revision in enumerate(revisions): + if not isinstance(revision, Mapping): + raise FxMacroDataDataError(f"record.revisions[{index}] must be an object") + entries.append((f"record.revisions[{index}]", revision)) + else: + raise FxMacroDataDataError("record.revisions must be a list when supplied") + + candidates: set[tuple[int, float]] = set() + for field, entry in entries: + value = _optional_number(entry.get("val"), f"{field}.val") + if value is None: + skipped["missing_value"] += 1 + continue + if field == "record" and entry.get("vintage_status") != "captured_snapshot": + available_at_ms: int | None = released_at_ms + else: + available_at_ms = _available_at_ms(entry, field) + if available_at_ms is None: + skipped["revision_without_vintage_time"] += 1 + continue + candidates.add((max(released_at_ms, available_at_ms), value)) + + # Keep a vintage only when the value differs from the one before it. + vintages: list[tuple[int, float]] = [] + for vintage_at_ms, value in sorted(candidates): + if vintages and vintages[-1][1] == value: + continue + vintages.append((vintage_at_ms, value)) + return vintages + + def _warn_on_access_limits(self, payload: Mapping[str, object], request: MacroRequest) -> None: + window = payload.get("freemium_window") + if isinstance(window, Mapping) and window.get("applied") is True: + cutoff = window.get("cutoff_date") + if isinstance(cutoff, str) and _request_date(request.start_ms) < cutoff: + warnings.warn( + FxMacroDataAccessWarning( + f"keyless access returns {request.currency} {request.key} only from " + f"{cutoff}; supply an API key for earlier periods" + ), + stacklevel=3, + ) + delay = payload.get("freemium_delay") + if isinstance(delay, Mapping): + withheld = delay.get("withheld_count") + if isinstance(withheld, int) and not isinstance(withheld, bool) and withheld > 0: + warnings.warn( + FxMacroDataAccessWarning( + f"{withheld} release(s) published in the last 15 minutes were " + "withheld by the keyless tier" + ), + stacklevel=3, + ) + + async def fetch_observations(self, request: MacroRequest) -> Sequence[MacroObservation]: + """Fetch every vintage for periods in ``[start_ms, end_ms)``.""" + + currency = request.currency.strip().upper() + key = request.key.strip().lower() + url = f"{self.api_url}/announcements/{quote(currency)}/{quote(key)}" + params: dict[str, str | int] = { + "start_date": _request_date(request.start_ms), + "end_date": _request_date(request.end_ms - 1), + "limit": _PAGE_LIMIT, + "offset": 0, + "revisions": "all", + } + skipped: Counter[str] = Counter() + observations: dict[tuple[int, int], MacroObservation] = {} + first_page = True + + while True: + payload = await self._get_page(url, params) + if first_page: + self._warn_on_access_limits(payload, request) + first_page = False + records = _payload_records(payload) + value_metadata = payload.get("value_metadata") + page_unit = ( + value_metadata.get("source_unit") if isinstance(value_metadata, Mapping) else None + ) + page_source = payload.get("source") + + for record in records: + period_end_ms = _period_end_ms(record.get("date")) + if not request.start_ms <= period_end_ms < request.end_ms: + continue + released_at_ms = self._release_ms(record, skipped) + if released_at_ms is None: + continue + if released_at_ms < period_end_ms: + # e.g. a flash estimate published before the period closed + skipped["released_before_period_end"] += 1 + continue + record_unit = record.get("unit", page_unit) + unit = "unknown" if record_unit is None else _text(record_unit, "unit") + record_source = record.get("source", page_source) + source = self.name if record_source is None else _text(record_source, "source") + for vintage_at_ms, value in self._vintages(record, released_at_ms, skipped): + observations[(period_end_ms, vintage_at_ms)] = MacroObservation( + key=key, + currency=currency, + period_end_ms=period_end_ms, + released_at_ms=released_at_ms, + vintage_at_ms=vintage_at_ms, + value=value, + unit=unit, + source=source, + ) + + offset = int(params["offset"]) + next_offset = _next_offset(payload, offset, len(records)) + if next_offset is None or not records: + break + params["offset"] = next_offset + + self.last_skipped = dict(skipped) + return sorted( + observations.values(), + key=lambda observation: (observation.period_end_ms, observation.vintage_at_ms), + ) + + async def close(self) -> None: + """Release resources; the default transport holds no open connection.""" + + async def __aenter__(self) -> FxMacroDataProvider: + return self + + async def __aexit__(self, *_exc: object) -> None: + await self.close() diff --git a/tests/fixtures/fxmacrodata_usd_non_farm_payrolls.json b/tests/fixtures/fxmacrodata_usd_non_farm_payrolls.json new file mode 100644 index 0000000..a99f754 --- /dev/null +++ b/tests/fixtures/fxmacrodata_usd_non_farm_payrolls.json @@ -0,0 +1,91 @@ +{ + "_comment": "Two pages of /v1/announcements/USD/non_farm_payrolls?revisions=all, trimmed to the fields the provider reads.", + "pages": [ + { + "currency": "USD", + "indicator": "non_farm_payrolls", + "source": "BLS", + "value_metadata": {"source_unit": "Persons", "normalization_applied": true}, + "pagination": {"limit": 100, "offset": 0, "returned_count": 2, "has_more": true, "next_offset": 2}, + "data": [ + { + "date": "2026-06-30", + "val": 158936000.0, + "announcement_datetime": 1783513800, + "publication_time_status": "unverified" + }, + { + "date": "2026-07-31", + "val": 158913000.0, + "source": "BLS", + "announcement_datetime": 1786105800, + "publication_time_status": "unverified", + "vintage_status": "captured_snapshot", + "observed_at_ns": 1789364950289573500, + "revisions": [ + { + "epoch": 1786105800, + "val": 158858000.0, + "publication_time_status": "confirmed", + "vintage_status": "source_vintage", + "is_first_release": true + }, + { + "epoch": 1786105800, + "val": 158913000.0, + "observed_at_ns": 1789364950289573500, + "vintage_status": "captured_snapshot", + "is_first_release": false + } + ] + } + ] + }, + { + "currency": "USD", + "indicator": "non_farm_payrolls", + "source": "BLS", + "value_metadata": {"source_unit": "Persons", "normalization_applied": true}, + "pagination": {"limit": 100, "offset": 2, "returned_count": 3, "has_more": false}, + "data": [ + { + "date": "2026-08-31", + "val": 159075000.0, + "announcement_datetime": 1788525000, + "publication_time_status": "unverified", + "vintage_status": "captured_snapshot", + "observed_at_ns": 1789104351424651147, + "revisions": [ + { + "epoch": 1788525000, + "val": 159075000.0, + "publication_time_status": "confirmed", + "vintage_status": "source_vintage", + "is_first_release": true + }, + { + "epoch": 1788525000, + "val": 159075000.0, + "observed_at_ns": 1789104351424651147, + "vintage_status": "captured_snapshot", + "is_first_release": false + } + ] + }, + { + "date": "2026-05-31", + "val": 158900000.0, + "announcement_datetime": null, + "publication_time_status": "unknown" + }, + { + "date": "2026-04-30", + "val": 158700000.0, + "announcement_datetime": 1778247000, + "publication_time_status": "assumed_historical", + "release_time_assumed": true + } + ] + } + ] +} diff --git a/tests/test_fxmacrodata.py b/tests/test_fxmacrodata.py new file mode 100644 index 0000000..2a7770a --- /dev/null +++ b/tests/test_fxmacrodata.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import asyncio +import json +from collections.abc import Mapping +from pathlib import Path + +import pytest + +from pineforge_data import ( + FxMacroDataAccessWarning, + FxMacroDataDataError, + FxMacroDataHTTPError, + FxMacroDataProvider, + FxMacroDataResponse, + MacroDataProvider, + MacroObservation, + MacroRequest, +) + +FIXTURE = Path(__file__).parent / "fixtures" / "fxmacrodata_usd_non_farm_payrolls.json" +START_MS = 1_775_001_600_000 # 2026-04-01 +END_MS = 1_790_812_800_000 # 2026-10-01 + + +def fixture_pages() -> list[object]: + pages: list[object] = json.loads(FIXTURE.read_text(encoding="utf-8"))["pages"] + return pages + + +class FakeTransport: + def __init__(self, responses: list[FxMacroDataResponse]) -> None: + self._responses = responses + self.calls: list[tuple[str, dict[str, str | int], dict[str, str]]] = [] + + async def get( + self, + url: str, + *, + params: Mapping[str, str | int], + headers: Mapping[str, str], + ) -> FxMacroDataResponse: + self.calls.append((url, dict(params), dict(headers))) + return self._responses.pop(0) + + +def ok(payload: object) -> FxMacroDataResponse: + return FxMacroDataResponse(200, payload) + + +def fetch( + provider: FxMacroDataProvider, start_ms: int = START_MS, end_ms: int = END_MS +) -> list[MacroObservation]: + request = MacroRequest( + key="non_farm_payrolls", currency="usd", start_ms=start_ms, end_ms=end_ms + ) + return list(asyncio.run(provider.fetch_observations(request))) + + +def test_provider_implements_macro_protocol() -> None: + assert isinstance(FxMacroDataProvider(transport=FakeTransport([])), MacroDataProvider) + + +def test_fixture_pages_become_release_and_vintage_observations() -> None: + transport = FakeTransport([ok(page) for page in fixture_pages()]) + provider = FxMacroDataProvider(transport=transport) + + observations = fetch(provider) + + assert [ + (o.period_end_ms, o.released_at_ms, o.vintage_at_ms, o.value) for o in observations + ] == [ + # June: no revisions, one vintage at the release time + (1_782_777_600_000, 1_783_513_800_000, 1_783_513_800_000, 158_936_000.0), + # July: first print at release, the revised value only from when it was observed + (1_785_456_000_000, 1_786_105_800_000, 1_786_105_800_000, 158_858_000.0), + (1_785_456_000_000, 1_786_105_800_000, 1_789_364_950_289, 158_913_000.0), + # August: the snapshot repeats the first print, so it adds no vintage + (1_788_134_400_000, 1_788_525_000_000, 1_788_525_000_000, 159_075_000.0), + ] + assert {(o.key, o.currency, o.unit, o.source) for o in observations} == { + ("non_farm_payrolls", "USD", "Persons", "BLS") + } + assert provider.last_skipped == {"unknown_release_time": 1, "assumed_release_time": 1} + + first_url, first_params, _ = transport.calls[0] + assert first_url == "https://api.fxmacrodata.com/v1/announcements/USD/non_farm_payrolls" + assert first_params == { + "start_date": "2026-04-01", + "end_date": "2026-09-30", + "limit": 100, + "offset": 0, + "revisions": "all", + } + assert transport.calls[1][1]["offset"] == 2 + + +def test_missing_release_time_is_never_fabricated() -> None: + page = { + "data": [ + { + "date": "2026-05-31", + "val": 1.0, + "announcement_datetime": None, + "official_planned_release_datetime": 1_780_000_000, + "publication_time_status": "unknown", + } + ] + } + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + assert fetch(provider) == [] + assert provider.last_skipped == {"unknown_release_time": 1} + + +def test_assumed_release_times_are_opt_in() -> None: + pages = fixture_pages() + provider = FxMacroDataProvider( + include_assumed_release_times=True, + transport=FakeTransport([ok(page) for page in pages]), + ) + + observations = fetch(provider) + + assert (1_777_507_200_000, 1_778_247_000_000, 158_700_000.0) in { + (o.period_end_ms, o.released_at_ms, o.value) for o in observations + } + assert provider.last_skipped == {"unknown_release_time": 1} + + +def test_snapshot_without_observation_time_is_skipped() -> None: + page = { + "data": [ + { + "date": "2026-07-31", + "val": 2.0, + "announcement_datetime": 1_786_105_800, + "revisions": [ + {"epoch": 1_786_105_800, "val": 1.0, "vintage_status": "source_vintage"}, + {"epoch": 1_786_105_800, "val": 2.0, "vintage_status": "captured_snapshot"}, + ], + } + ] + } + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + assert [o.value for o in fetch(provider)] == [1.0] + assert provider.last_skipped == {"revision_without_vintage_time": 1} + + +def test_release_before_period_end_is_skipped_not_reordered() -> None: + page = {"data": [{"date": "2026-07-31", "val": 50.1, "announcement_datetime": 1_785_000_000}]} + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + assert fetch(provider) == [] + assert provider.last_skipped == {"released_before_period_end": 1} + + +def test_periods_outside_the_request_are_dropped() -> None: + pages = fixture_pages() + provider = FxMacroDataProvider(transport=FakeTransport([ok(page) for page in pages])) + + observations = fetch(provider, start_ms=1_785_456_000_000, end_ms=1_788_134_400_000) + + assert {o.period_end_ms for o in observations} == {1_785_456_000_000} + + +def test_api_key_is_sent_only_in_header() -> None: + transport = FakeTransport([ok({"data": []})]) + provider = FxMacroDataProvider(api_key="placeholder-key", transport=transport) + + fetch(provider) + + url, params, headers = transport.calls[0] + assert headers["X-API-Key"] == "placeholder-key" + assert "placeholder-key" not in url + assert "placeholder-key" not in json.dumps(params) + assert "placeholder-key" not in repr(provider) + + +def test_keyless_requests_send_no_key_header() -> None: + transport = FakeTransport([ok({"data": []})]) + + fetch(FxMacroDataProvider(transport=transport)) + + assert "X-API-Key" not in transport.calls[0][2] + + +def test_keyless_history_window_warns() -> None: + page = { + "freemium_window": {"applied": True, "max_days": 90, "cutoff_date": "2026-07-03"}, + "data": [], + } + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + with pytest.warns(FxMacroDataAccessWarning, match="only from 2026-07-03"): + fetch(provider) + + +def test_retries_rate_limits_then_raises_http_errors() -> None: + transport = FakeTransport( + [ + FxMacroDataResponse(429, {"detail": "slow down"}, retry_after_seconds=0.0), + ok({"data": []}), + ] + ) + assert fetch(FxMacroDataProvider(transport=transport)) == [] + assert len(transport.calls) == 2 + + denied = FakeTransport([FxMacroDataResponse(403, {"detail": "subscription required"})]) + with pytest.raises(FxMacroDataHTTPError, match="HTTP 403: subscription required") as exc: + fetch(FxMacroDataProvider(api_key="placeholder-key", transport=denied)) + assert exc.value.status == 403 + assert "placeholder-key" not in str(exc.value) + + exhausted = FakeTransport( + [FxMacroDataResponse(503, None, retry_after_seconds=0.0) for _ in range(2)] + ) + with pytest.raises(FxMacroDataHTTPError, match="HTTP 503"): + fetch(FxMacroDataProvider(max_retries=1, transport=exhausted)) + + +def test_pagination_must_advance() -> None: + page = { + "pagination": {"has_more": True, "next_offset": 0}, + "data": [{"date": "2026-07-31", "val": 1.0, "announcement_datetime": 1_786_105_800}], + } + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + with pytest.raises(FxMacroDataDataError, match="advance"): + fetch(provider) + + +def test_malformed_records_raise() -> None: + page = {"data": [{"date": "July 2026", "val": 1.0, "announcement_datetime": 1_786_105_800}]} + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + with pytest.raises(FxMacroDataDataError, match=r"record\.date"): + fetch(provider) + + +def test_constructor_validation() -> None: + with pytest.raises(ValueError, match="api_key"): + FxMacroDataProvider(api_key=" ") + with pytest.raises(ValueError, match="timeout_seconds"): + FxMacroDataProvider(timeout_seconds=0) + with pytest.raises(ValueError, match="max_retries"): + FxMacroDataProvider(max_retries=-1) From 1b904c28e70a60e3b1ce02d92268147468560b3d Mon Sep 17 00:00:00 2001 From: Robert Tidball Date: Thu, 1 Oct 2026 13:56:28 +1000 Subject: [PATCH 2/2] Address review: refuse redirects, cap pages and Retry-After, date legacy snapshots by observation --- docs/api/providers.md | 2 + docs/providers.md | 8 +- docs/providers/fxmacrodata.md | 47 +++-- src/pineforge_data/providers/fxmacrodata.py | 83 ++++++-- tests/test_fxmacrodata.py | 199 +++++++++++++++++++- 5 files changed, 307 insertions(+), 32 deletions(-) diff --git a/docs/api/providers.md b/docs/api/providers.md index e896d37..65ee8d2 100644 --- a/docs/api/providers.md +++ b/docs/api/providers.md @@ -48,6 +48,8 @@ ::: pineforge_data.providers.fxmacrodata.FxMacroDataResponse +::: pineforge_data.providers.fxmacrodata.UrllibTransport + ## Provider errors ::: pineforge_data.providers.ccxt.CcxtError diff --git a/docs/providers.md b/docs/providers.md index 5e16be3..ff5f2f5 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -21,6 +21,10 @@ guide for each built-in provider. | `sqlite` | Local SQLite table or view | reflected catalog, historical bars | none | [SQLite](providers/sqlite.md) | | `sqlalchemy` | SQLAlchemy-supported database table or view | reflected catalog, historical bars | `database` plus a dialect driver | [SQLAlchemy](providers/sqlalchemy.md) | +CSV, SQLite, and SQLAlchemy share runtime schema discovery and arbitrary column +mapping. Read the [tabular schema mapping](providers/tabular-schema.md) guide +before configuring one of them. + ## Macro providers | Provider | Data source | Capabilities | Install extra | API guide | @@ -30,10 +34,6 @@ guide for each built-in provider. Macro providers implement `MacroDataProvider` and are constructed directly; they are not registry entries for the backtest harness. -CSV, SQLite, and SQLAlchemy share runtime schema discovery and arbitrary column -mapping. Read the [tabular schema mapping](providers/tabular-schema.md) guide -before configuring one of them. - ## Shared provider lifecycle Create a provider by registry name when configuration comes from a CLI, service, diff --git a/docs/providers/fxmacrodata.md b/docs/providers/fxmacrodata.md index 30b34ac..d25361a 100644 --- a/docs/providers/fxmacrodata.md +++ b/docs/providers/fxmacrodata.md @@ -1,8 +1,10 @@ # FXMacroData macro provider API +[Provider catalog](../providers.md) ยท [Normalized data model](../data-model.md) + `FxMacroDataProvider` implements `MacroDataProvider`. It reads economic announcements (CPI, policy rates, payrolls, GDP and so on) from the -[FXMacroData API](https://fxmacrodata.com/documentation/reference?utm_source=github&utm_medium=referral&utm_campaign=pineforge-data&utm_content=docs) +[FXMacroData API](https://fxmacrodata.com/documentation/reference) and returns `MacroObservation` records with release and vintage timestamps. It has no market catalog and no bars, so it is not a `MarketDataProvider` and is not registered for `pineforge-backtest --provider`. @@ -54,16 +56,26 @@ exception messages or the provider's `repr`. | `period_end_ms` | UTC midnight of the record's `date` | | `released_at_ms` | the record's `announcement_datetime` | | `vintage_at_ms` | when that value became available (below) | +| `source` | `fxmacrodata:`, for example `fxmacrodata:BLS` | + +`date` is the end of the reference period for periodic series: the last day +of the month for monthly data, the week-ending date for weekly data. For +event series such as `policy_rate` it is the decision date, and for daily +series such as government bond yields it is the observation date. The provider requests `revisions=all` and emits one observation each time a period's value changed: -- a source vintage (`vintage_status: source_vintage`) is dated by its - publication time (`publication_at_ns`, else `epoch`); -- a captured snapshot (`vintage_status: captured_snapshot`) is dated by - `observed_at_ns`, the time FXMacroData read it from the publisher. Its - `epoch` repeats the original release time, so using it would date a later - revision back to the first print; +- a snapshot (`vintage_status: captured_snapshot` or `legacy_snapshot`) is + dated by `observed_at_ns`, the time FXMacroData read it from the publisher, + and skipped when that is absent. Its `epoch` repeats the original release + time, so using it would date a later revision back to the first print; +- any other revision is dated by `publication_at_ns`, else `epoch`. On + revisions without a `vintage_status`, `epoch` is either the publication time + of that vintage or the time FXMacroData collected it; either way it is not + earlier than the value was available; +- a record whose `revisions` is null or empty is treated as a single vintage + of its own `val`; - a vintage is never earlier than `released_at_ms`, and a vintage that repeats the previous value is dropped. @@ -72,7 +84,13 @@ Those values are dated by when they were collected, so a backtest set before that date does not see them. This is deliberate: the alternative is lookahead. -## Records that are skipped +## Records that are kept or skipped + +Records with `publication_time_status: unverified` are kept. Their +`announcement_datetime` is a real timestamp whose origin the row does not +evidence; only `confirmed` is evidence of when a value became public. Where a +release time came from a release calendar, the planned time can precede the +actual release, so treat unverified times as approximate. `MacroObservation` requires a release time, and the provider does not invent one. These records are left out and counted in `provider.last_skipped` for the @@ -94,18 +112,25 @@ most recent fetch: | `api_url` | `https://api.fxmacrodata.com/v1` | API root | | `timeout_seconds` | `20.0` | per-request timeout of the default transport | | `max_retries` | `2` | retries after HTTP 429, 500, 502, 503 or 504 | -| `retry_backoff_seconds` | `1.0` | first retry delay, doubled each attempt; `Retry-After` takes precedence | +| `retry_backoff_seconds` | `1.0` | first retry delay, doubled each attempt | +| `max_retry_after_seconds` | `60.0` | a server `Retry-After` up to this is honored; above it the backoff schedule is used | +| `max_pages` | `1000` | pages followed per fetch before `FxMacroDataDataError` is raised | | `include_assumed_release_times` | `False` | keep records whose release time was derived rather than captured | | `transport` | `None` | an `FxMacroDataTransport`; inject one for offline tests or a custom HTTP client | Pages hold at most 100 records; the provider follows `pagination.has_more` and -`next_offset` until the request is complete. +`next_offset` until the request is complete, up to `max_pages` pages. + +`key` must be an indicator slug (lowercase letters, digits and underscores) and +`currency` a three-letter code; anything else raises `ValueError` before a +request is made. The default transport does not follow redirects, so the API +key is only sent to `api_url`; a 3xx response raises `FxMacroDataHTTPError`. ## Errors and limitations | Error | Meaning | |---|---| -| `FxMacroDataHTTPError` | a non-success status after retries; `.status` holds the code (403 for a currency or history the key does not cover) | +| `FxMacroDataHTTPError` | a non-success status after retries, including a refused redirect; `.status` holds the code (403 for a currency or history the key does not cover). The key is redacted from the server's message | | `FxMacroDataDataError` | a response or record cannot be normalized safely | | `FxMacroDataError` | base class; also raised for network failures | | `FxMacroDataAccessWarning` | a warning, not an error: the keyless tier cut the requested history short or withheld a release from the last 15 minutes | diff --git a/src/pineforge_data/providers/fxmacrodata.py b/src/pineforge_data/providers/fxmacrodata.py index 9fd9335..37ddeb7 100644 --- a/src/pineforge_data/providers/fxmacrodata.py +++ b/src/pineforge_data/providers/fxmacrodata.py @@ -4,17 +4,19 @@ import asyncio import json +import re import warnings from collections import Counter from collections.abc import Mapping, Sequence from dataclasses import dataclass from datetime import UTC, datetime from email.message import Message +from http.client import HTTPMessage from math import isfinite -from typing import Protocol +from typing import IO, Protocol from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode -from urllib.request import Request, urlopen +from urllib.request import HTTPRedirectHandler, Request, build_opener from ..models import MacroObservation from ..requests import MacroRequest @@ -22,6 +24,11 @@ FXMACRODATA_API_URL = "https://api.fxmacrodata.com/v1" _PAGE_LIMIT = 100 _RETRY_STATUSES = frozenset({429, 500, 502, 503, 504}) +# Snapshot entries repeat the original release time in ``epoch``; only the time +# FXMacroData observed them says when their value was available. +_SNAPSHOT_STATUSES = frozenset({"captured_snapshot", "legacy_snapshot"}) +_CURRENCY = re.compile(r"[A-Z]{3}") +_INDICATOR = re.compile(r"[a-z0-9_]+") class FxMacroDataError(RuntimeError): @@ -80,16 +87,36 @@ def _decode(body: bytes) -> object: return None +class _RefuseRedirects(HTTPRedirectHandler): + """Surface a 3xx as an HTTP error instead of re-sending headers elsewhere.""" + + def redirect_request( + self, + req: Request, + fp: IO[bytes], + code: int, + msg: str, + headers: HTTPMessage, + newurl: str, + ) -> Request | None: + return None + + class UrllibTransport: - """Standard-library transport: no optional dependency is required.""" + """Standard-library transport: no optional dependency is required. + + Redirects are not followed, so the ``X-API-Key`` header is only ever sent to + the configured host; a 3xx response becomes an ``FxMacroDataHTTPError``. + """ def __init__(self, timeout_seconds: float) -> None: self.timeout_seconds = timeout_seconds + self._opener = build_opener(_RefuseRedirects()) def _get(self, url: str, headers: Mapping[str, str]) -> FxMacroDataResponse: request = Request(url, headers=dict(headers), method="GET") try: - with urlopen(request, timeout=self.timeout_seconds) as response: + with self._opener.open(request, timeout=self.timeout_seconds) as response: return FxMacroDataResponse(response.status, _decode(response.read())) except HTTPError as exc: return FxMacroDataResponse(exc.code, _decode(exc.read()), _retry_after(exc.headers)) @@ -150,13 +177,13 @@ def _request_date(timestamp_ms: int) -> str: def _available_at_ms(entry: Mapping[str, object], field: str) -> int | None: """When the value in ``entry`` is known to have been available. - A ``captured_snapshot`` was read from the publisher at ``observed_at_ns``; - its ``epoch`` repeats the original release time, so using it would date a - later revision back to the first release. Source vintages carry their own - publication time. + A ``captured_snapshot`` or ``legacy_snapshot`` repeats the original release + time in ``epoch``, so using it would date a later revision back to the first + release; only ``observed_at_ns`` is used for those. Other vintages are dated + by ``publication_at_ns``, else ``epoch``. """ - if entry.get("vintage_status") == "captured_snapshot": + if entry.get("vintage_status") in _SNAPSHOT_STATUSES: return _optional_epoch_ms( entry.get("observed_at_ns"), f"{field}.observed_at_ns", divisor=1_000_000 ) @@ -215,6 +242,7 @@ class FxMacroDataProvider: code. The provider asks for ``revisions=all`` and emits one observation per distinct value a period has had, dated by when that value became available. + ``source`` is ``fxmacrodata:``, for example ``fxmacrodata:BLS``. Records without a publication time are skipped rather than given one, and so are records whose time FXMacroData derived instead of captured, unless ``include_assumed_release_times`` is set. ``last_skipped`` counts the @@ -229,6 +257,8 @@ def __init__( timeout_seconds: float = 20.0, max_retries: int = 2, retry_backoff_seconds: float = 1.0, + max_retry_after_seconds: float = 60.0, + max_pages: int = 1_000, include_assumed_release_times: bool = False, transport: FxMacroDataTransport | None = None, ) -> None: @@ -240,6 +270,10 @@ def __init__( raise ValueError("max_retries must be non-negative") if retry_backoff_seconds < 0: raise ValueError("retry_backoff_seconds must be non-negative") + if max_retry_after_seconds < 0: + raise ValueError("max_retry_after_seconds must be non-negative") + if max_pages < 1: + raise ValueError("max_pages must be at least 1") if api_key is not None and not api_key.strip(): raise ValueError("api_key must not be empty when supplied") @@ -249,6 +283,8 @@ def __init__( self.timeout_seconds = timeout_seconds self.max_retries = max_retries self.retry_backoff_seconds = retry_backoff_seconds + self.max_retry_after_seconds = max_retry_after_seconds + self.max_pages = max_pages self.include_assumed_release_times = include_assumed_release_times self._transport: FxMacroDataTransport = transport or UrllibTransport(timeout_seconds) self.last_skipped: Mapping[str, int] = {} @@ -272,13 +308,16 @@ async def _get_page(self, url: str, params: Mapping[str, str | int]) -> Mapping[ response = await self._transport.get(url, params=params, headers=self._headers()) if response.status in _RETRY_STATUSES and attempt < self.max_retries: delay = response.retry_after_seconds - if delay is None: + if delay is None or delay > self.max_retry_after_seconds: delay = self.retry_backoff_seconds * 2**attempt await asyncio.sleep(delay) attempt += 1 continue if not 200 <= response.status < 300: - raise FxMacroDataHTTPError(response.status, _error_message(response.payload)) + message = _error_message(response.payload) + if self._api_key is not None: + message = message.replace(self._api_key, "[redacted]") + raise FxMacroDataHTTPError(response.status, message) if not isinstance(response.payload, Mapping): raise FxMacroDataDataError("response must be a JSON object") return response.payload @@ -304,7 +343,7 @@ def _vintages( record: Mapping[str, object], released_at_ms: int, skipped: Counter[str] ) -> list[tuple[int, float]]: revisions = record.get("revisions") - if revisions is None: + if revisions is None or revisions == []: entries: list[tuple[str, Mapping[str, object]]] = [("record", record)] elif isinstance(revisions, list): entries = [] @@ -321,7 +360,7 @@ def _vintages( if value is None: skipped["missing_value"] += 1 continue - if field == "record" and entry.get("vintage_status") != "captured_snapshot": + if field == "record" and entry.get("vintage_status") not in _SNAPSHOT_STATUSES: available_at_ms: int | None = released_at_ms else: available_at_ms = _available_at_ms(entry, field) @@ -367,7 +406,11 @@ async def fetch_observations(self, request: MacroRequest) -> Sequence[MacroObser currency = request.currency.strip().upper() key = request.key.strip().lower() - url = f"{self.api_url}/announcements/{quote(currency)}/{quote(key)}" + if not _CURRENCY.fullmatch(currency): + raise ValueError(f"currency must be a three-letter code, got {request.currency!r}") + if not _INDICATOR.fullmatch(key): + raise ValueError(f"key must be an FXMacroData indicator slug, got {request.key!r}") + url = f"{self.api_url}/announcements/{quote(currency, safe='')}/{quote(key, safe='')}" params: dict[str, str | int] = { "start_date": _request_date(request.start_ms), "end_date": _request_date(request.end_ms - 1), @@ -378,8 +421,14 @@ async def fetch_observations(self, request: MacroRequest) -> Sequence[MacroObser skipped: Counter[str] = Counter() observations: dict[tuple[int, int], MacroObservation] = {} first_page = True + pages = 0 while True: + pages += 1 + if pages > self.max_pages: + raise FxMacroDataDataError( + f"response.pagination did not finish within max_pages={self.max_pages}" + ) payload = await self._get_page(url, params) if first_page: self._warn_on_access_limits(payload, request) @@ -405,7 +454,11 @@ async def fetch_observations(self, request: MacroRequest) -> Sequence[MacroObser record_unit = record.get("unit", page_unit) unit = "unknown" if record_unit is None else _text(record_unit, "unit") record_source = record.get("source", page_source) - source = self.name if record_source is None else _text(record_source, "source") + source = ( + self.name + if record_source is None + else f"{self.name}:{_text(record_source, 'source')}" + ) for vintage_at_ms, value in self._vintages(record, released_at_ms, skipped): observations[(period_end_ms, vintage_at_ms)] = MacroObservation( key=key, diff --git a/tests/test_fxmacrodata.py b/tests/test_fxmacrodata.py index 2a7770a..d1de504 100644 --- a/tests/test_fxmacrodata.py +++ b/tests/test_fxmacrodata.py @@ -2,8 +2,11 @@ import asyncio import json -from collections.abc import Mapping +import threading +from collections.abc import Iterator, Mapping +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path +from typing import ClassVar import pytest @@ -17,6 +20,7 @@ MacroObservation, MacroRequest, ) +from pineforge_data.providers.fxmacrodata import UrllibTransport FIXTURE = Path(__file__).parent / "fixtures" / "fxmacrodata_usd_non_farm_payrolls.json" START_MS = 1_775_001_600_000 # 2026-04-01 @@ -79,7 +83,7 @@ def test_fixture_pages_become_release_and_vintage_observations() -> None: (1_788_134_400_000, 1_788_525_000_000, 1_788_525_000_000, 159_075_000.0), ] assert {(o.key, o.currency, o.unit, o.source) for o in observations} == { - ("non_farm_payrolls", "USD", "Persons", "BLS") + ("non_farm_payrolls", "USD", "Persons", "fxmacrodata:BLS") } assert provider.last_skipped == {"unknown_release_time": 1, "assumed_release_time": 1} @@ -246,3 +250,194 @@ def test_constructor_validation() -> None: FxMacroDataProvider(timeout_seconds=0) with pytest.raises(ValueError, match="max_retries"): FxMacroDataProvider(max_retries=-1) + + +def test_empty_revisions_fall_back_to_the_record_value() -> None: + page = { + "data": [ + { + "date": "2026-07-31", + "val": 4.1, + "announcement_datetime": 1_786_105_800, + "revisions": [], + } + ] + } + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + [observation] = fetch(provider) + + assert (observation.vintage_at_ms, observation.value) == (1_786_105_800_000, 4.1) + assert provider.last_skipped == {} + + +def test_legacy_snapshot_is_not_dated_by_its_epoch() -> None: + page = { + "data": [ + { + "date": "2026-07-31", + "val": 2.0, + "announcement_datetime": 1_786_105_800, + "revisions": [ + {"epoch": 1_786_105_800, "val": 1.0}, + {"epoch": 1_786_105_800, "val": 2.0, "vintage_status": "legacy_snapshot"}, + { + "epoch": 1_786_105_800, + "val": 3.0, + "vintage_status": "legacy_snapshot", + "observed_at_ns": 1_789_000_000_000_000_000, + }, + ], + } + ] + } + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + assert [(o.vintage_at_ms, o.value) for o in fetch(provider)] == [ + (1_786_105_800_000, 1.0), + (1_789_000_000_000, 3.0), + ] + assert provider.last_skipped == {"revision_without_vintage_time": 1} + + +def test_page_ceiling_raises() -> None: + page = { + "pagination": {"has_more": True}, + "data": [{"date": "2026-07-31", "val": 1.0, "announcement_datetime": 1_786_105_800}], + } + pages = [ok({**page, "pagination": {"has_more": True, "next_offset": n}}) for n in (1, 2, 3)] + provider = FxMacroDataProvider(max_pages=2, transport=FakeTransport(pages)) + + with pytest.raises(FxMacroDataDataError, match="max_pages=2"): + fetch(provider) + + +def test_long_retry_after_falls_back_to_backoff(monkeypatch: pytest.MonkeyPatch) -> None: + delays: list[float] = [] + + async def fake_sleep(delay: float) -> None: + delays.append(delay) + + monkeypatch.setattr(asyncio, "sleep", fake_sleep) + transport = FakeTransport( + [ + FxMacroDataResponse(429, None, retry_after_seconds=86_400.0), + FxMacroDataResponse(503, None, retry_after_seconds=5.0), + ok({"data": []}), + ] + ) + + fetch(FxMacroDataProvider(retry_backoff_seconds=0.5, transport=transport)) + + assert delays == [0.5, 5.0] + + +def test_key_is_redacted_from_server_error_text() -> None: + transport = FakeTransport([FxMacroDataResponse(401, {"detail": "invalid key placeholder-key"})]) + + with pytest.raises(FxMacroDataHTTPError) as exc: + fetch(FxMacroDataProvider(api_key="placeholder-key", transport=transport)) + + assert "placeholder-key" not in str(exc.value) + assert "[redacted]" in str(exc.value) + + +@pytest.mark.parametrize( + ("currency", "key"), + [("US/D", "inflation"), ("USD", "../latest"), ("USD", "a/b"), ("USDX", "inflation")], +) +def test_currency_and_key_must_be_plain_identifiers(currency: str, key: str) -> None: + transport = FakeTransport([]) + provider = FxMacroDataProvider(transport=transport) + request = MacroRequest(key=key, currency=currency, start_ms=START_MS, end_ms=END_MS) + + with pytest.raises(ValueError): + asyncio.run(provider.fetch_observations(request)) + assert transport.calls == [] + + +def test_withheld_releases_warn() -> None: + page = {"freemium_delay": {"applied": True, "withheld_count": 2}, "data": []} + provider = FxMacroDataProvider(transport=FakeTransport([ok(page)])) + + with pytest.warns(FxMacroDataAccessWarning, match="2 release"): + fetch(provider) + + +class _Handler(BaseHTTPRequestHandler): + seen: ClassVar[list[tuple[str, dict[str, str]]]] = [] + + def log_message(self, format: str, *args: object) -> None: + return None + + def _send(self, status: int, body: bytes, headers: Mapping[str, str]) -> None: + self.send_response(status) + for name, value in headers.items(): + self.send_header(name, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + _Handler.seen.append((self.path, {k.lower(): v for k, v in self.headers.items()})) + if self.path.startswith("/v1/ok"): + self._send(200, b'{"data": [1]}', {"Content-Type": "application/json"}) + elif self.path.startswith("/v1/moved"): + self._send(302, b"", {"Location": "/elsewhere"}) + elif self.path.startswith("/v1/busy"): + self._send(429, b'{"detail": "busy"}', {"Retry-After": "7"}) + elif self.path.startswith("/v1/text"): + self._send(200, b"not json", {"Content-Type": "text/plain"}) + else: + self._send(404, b"{}", {}) + + +@pytest.fixture +def local_server() -> Iterator[str]: + _Handler.seen = [] + server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}/v1" + finally: + server.shutdown() + server.server_close() + + +def test_urllib_transport_sends_headers_and_decodes_json(local_server: str) -> None: + transport = UrllibTransport(timeout_seconds=5) + + response = asyncio.run( + transport.get( + f"{local_server}/ok", + params={"limit": 100, "offset": 0}, + headers={"X-API-Key": "placeholder-key", "Accept": "application/json"}, + ) + ) + + assert (response.status, response.payload) == (200, {"data": [1]}) + path, headers = _Handler.seen[0] + assert path == "/v1/ok?limit=100&offset=0" + assert headers["x-api-key"] == "placeholder-key" + + +def test_urllib_transport_refuses_redirects(local_server: str) -> None: + provider = FxMacroDataProvider(api_key="placeholder-key", api_url=f"{local_server}/moved") + request = MacroRequest(key="inflation", currency="USD", start_ms=START_MS, end_ms=END_MS) + + with pytest.raises(FxMacroDataHTTPError, match="HTTP 302"): + asyncio.run(provider.fetch_observations(request)) + assert [path.split("?")[0] for path, _ in _Handler.seen] == [ + "/v1/moved/announcements/USD/inflation" + ] + + +def test_urllib_transport_reads_retry_after_and_tolerates_non_json(local_server: str) -> None: + transport = UrllibTransport(timeout_seconds=5) + + busy = asyncio.run(transport.get(f"{local_server}/busy", params={}, headers={})) + text = asyncio.run(transport.get(f"{local_server}/text", params={}, headers={})) + + assert (busy.status, busy.payload, busy.retry_after_seconds) == (429, {"detail": "busy"}, 7.0) + assert (text.status, text.payload) == (200, None)