diff --git a/docs/api/providers.md b/docs/api/providers.md index bd72350..65ee8d2 100644 --- a/docs/api/providers.md +++ b/docs/api/providers.md @@ -42,6 +42,14 @@ ::: pineforge_data.providers.sqlalchemy.SqlAlchemyBarProvider +::: pineforge_data.providers.fxmacrodata.FxMacroDataProvider + +::: pineforge_data.providers.fxmacrodata.FxMacroDataTransport + +::: pineforge_data.providers.fxmacrodata.FxMacroDataResponse + +::: pineforge_data.providers.fxmacrodata.UrllibTransport + ## Provider errors ::: pineforge_data.providers.ccxt.CcxtError @@ -53,3 +61,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..ff5f2f5 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -25,6 +25,15 @@ 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 | +|---|---|---|---|---| +| `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. + ## 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 new file mode 100644 index 0000000..d25361a --- /dev/null +++ b/docs/providers/fxmacrodata.md @@ -0,0 +1,143 @@ +# 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) +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) | +| `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 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. + +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 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 +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 | +| `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, 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, 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 | + +- 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..37ddeb7 --- /dev/null +++ b/src/pineforge_data/providers/fxmacrodata.py @@ -0,0 +1,493 @@ +"""FXMacroData adapter for release- and vintage-aware macro observations.""" + +from __future__ import annotations + +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 IO, Protocol +from urllib.error import HTTPError, URLError +from urllib.parse import quote, urlencode +from urllib.request import HTTPRedirectHandler, Request, build_opener + +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}) +# 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): + """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 _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. + + 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 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)) + 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`` 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") in _SNAPSHOT_STATUSES: + 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. + + ``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 + 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, + max_retry_after_seconds: float = 60.0, + max_pages: int = 1_000, + 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 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") + + 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.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] = {} + + 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 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: + 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 + + 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 or revisions == []: + 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") not in _SNAPSHOT_STATUSES: + 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() + 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), + "limit": _PAGE_LIMIT, + "offset": 0, + "revisions": "all", + } + 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) + 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 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, + 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..d1de504 --- /dev/null +++ b/tests/test_fxmacrodata.py @@ -0,0 +1,443 @@ +from __future__ import annotations + +import asyncio +import json +import threading +from collections.abc import Iterator, Mapping +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import ClassVar + +import pytest + +from pineforge_data import ( + FxMacroDataAccessWarning, + FxMacroDataDataError, + FxMacroDataHTTPError, + FxMacroDataProvider, + FxMacroDataResponse, + MacroDataProvider, + 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 +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", "fxmacrodata: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) + + +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)