Monitor official economic releases

"""Customer-side FXMacroData workflow demonstrations.

Run on your own machine with requests available and FXMD_API_KEY in your
environment. Prints JSON; does not place orders or create alert subscriptions.
"""

import argparse
import json
import os
from datetime import date, datetime, timezone
import re

import requests


BASE = "https://api.fxmacrodata.com"


def get_json(path, params):
    key = os.environ.get("FXMD_API_KEY", "")
    if not key:
        raise RuntimeError("Set FXMD_API_KEY for these subscriber workflows.")
    response = requests.get(BASE + path, params=params, headers={"X-API-Key": key}, timeout=30)
    # Avoid exceptions containing an authenticated URL or response body.
    if response.status_code != 200:
        raise RuntimeError(f"Request failed with HTTP {response.status_code}; review access and rate limits.")
    return response.json()


def history(currency, indicator, start, end, *, as_of=None):
    """Retain complete response pages and stop if a dataset changes mid-read."""
    pages, offset, version = [], 0, None
    while True:
        params = {"start_date": start, "end_date": end, "limit": 100, "offset": offset,
                  "value_mode": "source", "series_mode": "raw", "revisions": "all"}
        if version:
            params["dataset_version"] = version
        if as_of:
            params.update(as_of=as_of, availability="public", revisions="latest")
        page = get_json(f"/v1/announcements/{currency}/{indicator}", params)
        current = page.get("dataset_version")
        if not isinstance(current, str) or not current:
            raise RuntimeError("Dataset-version evidence is missing; do not combine pages.")
        if version is not None and current != version:
            raise RuntimeError("Dataset changed during pagination; restart the download.")
        version = current
        if as_of and page.get("data") and page.get("data_quality", {}).get("point_in_time_safe") is not True:
            raise RuntimeError("Selected history is not verified for point-in-time replay.")
        if page.get("data") and (page.get("value_mode") != "source" or page.get("value_metadata", {}).get("normalization_applied") is not False):
            raise RuntimeError("Source-unit values were not confirmed.")
        pages.append(page)
        pagination = page.get("pagination")
        if not isinstance(pagination, dict) or not isinstance(pagination.get("has_more"), bool):
            raise RuntimeError("Pagination evidence is missing; do not treat this download as complete.")
        if not pagination["has_more"]:
            break
        next_offset = pagination.get("next_offset")
        if not isinstance(next_offset, int) or isinstance(next_offset, bool) or next_offset <= offset:
            raise RuntimeError("Pagination did not advance.")
        offset = next_offset
    return {"currency": currency, "indicator": indicator, "as_of": as_of,
            "downloaded_at": datetime.now(timezone.utc).isoformat(),
            "row_count": sum(len(p.get("data", [])) for p in pages), "pages": pages}


def inflation_history(currencies, start, end):
    # Keep national measures separate: matching slugs do not establish comparability.
    return [history(currency, "inflation", start, end) for currency in currencies]


def point_in_time(currency, indicator, start, end, as_of):
    if not as_of:
        raise ValueError("Supply an explicit timezone-aware decision cutoff with --as-of.")
    cutoff = datetime.fromisoformat(as_of.replace("Z", "+00:00"))
    if cutoff.tzinfo is None:
        raise ValueError("The decision cutoff must include a timezone.")
    return history(currency, indicator, start, end, as_of=cutoff.isoformat())


def monitor_release(currency, indicator, start, end):
    # One bounded read, intended to be run by your scheduler. No inferred next dates.
    result = history(currency, indicator, start, end)
    return {"checked_at": datetime.now(timezone.utc).isoformat(),
            "records": result,
            "interpretation": "Inspect announcement_id, retained vintages and publication evidence. An empty range is not proof that no release occurred; this read is not a latency measurement."}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("workflow", choices=["inflation-history", "point-in-time", "release-monitoring"])
    parser.add_argument("--currencies", default="usd,eur,gbp")
    parser.add_argument("--indicator", default="inflation")
    parser.add_argument("--start", required=True)
    parser.add_argument("--end", required=True)
    parser.add_argument("--as-of")
    args = parser.parse_args()
    currencies = [code.strip().lower() for code in args.currencies.split(",")]
    if any(len(code) != 3 or not code.isascii() or not code.isalpha() for code in currencies):
        parser.error("Use three-letter currency codes separated by commas.")
    try:
        if date.fromisoformat(args.start) > date.fromisoformat(args.end):
            parser.error("The observation start must be on or before the end.")
    except ValueError:
        parser.error("Use valid YYYY-MM-DD observation dates.")
    if not re.fullmatch(r"[a-z][a-z0-9_]{0,79}", args.indicator):
        parser.error("Choose a valid indicator slug.")
    if args.workflow != "inflation-history" and len(currencies) != 1:
        parser.error("Choose one currency for a decision cutoff or monitoring read.")
    if args.workflow == "inflation-history":
        result = inflation_history(currencies, args.start, args.end)
    elif args.workflow == "point-in-time":
        result = point_in_time(currencies[0], args.indicator, args.start, args.end, args.as_of)
    else:
        result = monitor_release(currencies[0], args.indicator, args.start, args.end)
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()