diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 5148bddd00..56f7c4d0a4 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,6 +1,6 @@ exclude: 'Stock_NeurIPS2018.py' ci: - skip: [flake8] # remove this eventually + skip: [flake8, reorder-python-imports] # remove this eventually repos: - repo: https://github.com/pre-commit/pre-commit-hooks rev: v6.0.0 @@ -16,7 +16,7 @@ repos: - id: reorder-python-imports args: [--py37-plus, --add-import, "from __future__ import annotations"] - repo: https://github.com/asottile/pyupgrade - rev: v3.20.0 + rev: v3.21.2 hooks: - id: pyupgrade args: [--py37-plus] diff --git a/README.md b/README.md index ccdb6882b9..50e026b2c8 100755 --- a/README.md +++ b/README.md @@ -219,6 +219,7 @@ FinRL |[Binance](https://binance-docs.github.io/apidocs/spot/en/#public-api-definitions)| Cryptocurrency| API-specific, 1s, 1min| API-specific| Tick-level daily aggregated trades, OHLCV| Prices&Indicators| |[CCXT](https://docs.ccxt.com/en/latest/manual.html)| Cryptocurrency| API-specific, 1min| API-specific| OHLCV| Prices&Indicators| |[EODhistoricaldata](https://eodhistoricaldata.com/financial-apis/)| US Securities| Frequency-specific, 1min| API-specific | OHLCV | Prices&Indicators| +|[FXMacroData](https://fxmacrodata.com/)| FX spot rates and macro announcements| Daily and event-time | Account-specific | FX rates, release calendars, forecasts, official macro announcements | OHLCV-shaped FX rates and macro event features | |[IEXCloud](https://iexcloud.io/docs/api/)| NMS US securities|1970-now, 1 day|100 per second per IP|OHLCV| Prices&Indicators| |[JoinQuant](https://www.joinquant.com/)| CN Securities| 2005-now, 1min| 3 requests each time| OHLCV| Prices&Indicators| |[QuantConnect](https://www.quantconnect.com/docs/v2)| US Securities| 1998-now, 1s| NA| OHLCV| Prices&Indicators| diff --git a/docs/source/start/quick_start.rst b/docs/source/start/quick_start.rst index 02303853a9..76e155b020 100644 --- a/docs/source/start/quick_start.rst +++ b/docs/source/start/quick_start.rst @@ -147,3 +147,10 @@ Run the library: python main.py --mode=trade # if trade. Users should input your alpaca parameters in config.py Choices for ``--mode``: start mode, train, download_data, backtest + +FXMacroData can be used as a daily FX spot data source by setting +``data_source="fxmacrodata"`` and passing pairs such as ``["EURUSD"]`` as the +ticker list. It also exposes macro announcement, release-calendar, and forecast +data through the FXMacroData processor so event features can be joined into an +RL state frame. API keys can be passed in ``kwargs`` with ``api_key`` or supplied +through the ``FXMACRODATA_API_KEY`` or ``FXMD_API_KEY`` environment variables. diff --git a/docs/source/tutorial/stocktrading/1-data.rst b/docs/source/tutorial/stocktrading/1-data.rst index e94df75914..cc5c64c903 100644 --- a/docs/source/tutorial/stocktrading/1-data.rst +++ b/docs/source/tutorial/stocktrading/1-data.rst @@ -50,6 +50,31 @@ In FinRL's YahooDownloader, we modified the data frame to the form that convenie end_date = '2020-01-31', ticker_list = ['aapl']).fetch_data() +**using FXMacroData for daily FX spot data and macro events** + +FXMacroData provides daily FX spot rates for currency pairs such as EUR/USD. +The downloader maps each daily rate to FinRL's OHLCV shape, with open, high, +low, and close set to the FX spot rate and volume set to 0. FXMacroData also +provides official macro announcements, release-calendar rows, and forecast +groups that can be joined into a trading state. + +.. code-block:: python + from finrl.meta.preprocessor.fxmacrodatadownloader import FXMacroDataDownloader + from finrl.meta.data_processors.processor_fxmacrodata import FXMacroDataProcessor + + eurusd_df_finrl = FXMacroDataDownloader(start_date = '2020-01-01', + end_date = '2020-01-31', + ticker_list = ['EURUSD']).fetch_data() + + processor = FXMacroDataProcessor() + macro_df = processor.download_macro_data(currency = 'usd', + indicator_list = ['inflation', + 'policy_rate'], + start_date = '2020-01-01', + end_date = '2020-01-31') + eurusd_with_macro = processor.add_macro_features(eurusd_df_finrl, macro_df, + date_column = 'date') + Data for the chosen ticker ---------------------------------------- .. code-block:: python diff --git a/finrl/meta/data_processor.py b/finrl/meta/data_processor.py index ae60e39ea5..3b262a94df 100644 --- a/finrl/meta/data_processor.py +++ b/finrl/meta/data_processor.py @@ -4,6 +4,9 @@ import pandas as pd from finrl.meta.data_processors.processor_alpaca import AlpacaProcessor as Alpaca +from finrl.meta.data_processors.processor_fxmacrodata import ( + FXMacroDataProcessor as FXMacroData, +) from finrl.meta.data_processors.processor_wrds import WrdsProcessor as Wrds from finrl.meta.data_processors.processor_yahoofinance import ( YahooFinanceProcessor as YahooFinance, @@ -28,6 +31,14 @@ def __init__(self, data_source, tech_indicator=None, vix=None, **kwargs): elif data_source == "yahoofinance": self.processor = YahooFinance() + elif data_source == "fxmacrodata": + self.processor = FXMacroData( + api_key=kwargs.get("API_KEY") or kwargs.get("api_key"), + base_url=kwargs.get("BASE_URL") + or kwargs.get("base_url") + or "https://api.fxmacrodata.com/v1", + ) + else: raise ValueError("Data source input is NOT supported yet.") diff --git a/finrl/meta/data_processors/processor_fxmacrodata.py b/finrl/meta/data_processors/processor_fxmacrodata.py new file mode 100644 index 0000000000..1b48a6da0d --- /dev/null +++ b/finrl/meta/data_processors/processor_fxmacrodata.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +from typing import List + +import numpy as np +import pandas as pd +from stockstats import StockDataFrame as Sdf + +from finrl.meta.preprocessor.fxmacrodatadownloader import FXMacroDataDownloader +from finrl.meta.preprocessor.fxmacrodatadownloader import FXMacroDataMacroDownloader + + +class FXMacroDataProcessor: + """Provides daily FX spot data from FXMacroData for FinRL processors.""" + + def __init__(self, api_key=None, base_url="https://api.fxmacrodata.com/v1"): + self.api_key = api_key + self.base_url = base_url + + def download_data( + self, + ticker_list: list[str], + start_date: str, + end_date: str, + time_interval: str, + ) -> pd.DataFrame: + if time_interval.lower() not in {"1d", "1day"}: + raise ValueError("FXMacroDataProcessor supports daily data only.") + + self.start = start_date + self.end = end_date + self.time_interval = "1d" + data_df = FXMacroDataDownloader( + start_date=start_date, + end_date=end_date, + ticker_list=ticker_list, + api_key=self.api_key, + base_url=self.base_url, + ).fetch_data() + data_df["timestamp"] = pd.to_datetime(data_df["date"]) + return data_df[["timestamp", "open", "high", "low", "close", "volume", "tic"]] + + def download_macro_data( + self, + currency: str, + indicator_list: list[str], + start_date: str = None, + end_date: str = None, + dataset: str = "announcements", + ) -> pd.DataFrame: + downloader = FXMacroDataMacroDownloader( + currency=currency, + indicator_list=indicator_list, + start_date=start_date, + end_date=end_date, + api_key=self.api_key, + base_url=self.base_url, + ) + if dataset == "announcements": + return downloader.fetch_announcements() + if dataset == "calendar": + return downloader.fetch_calendar() + if dataset == "predictions": + return downloader.fetch_predictions() + raise ValueError("dataset must be announcements, calendar, or predictions") + + def download_release_calendar( + self, + currency: str, + indicator_list: list[str] = None, + start_date: str = None, + end_date: str = None, + ) -> pd.DataFrame: + return FXMacroDataMacroDownloader( + currency=currency, + indicator_list=indicator_list, + start_date=start_date, + end_date=end_date, + api_key=self.api_key, + base_url=self.base_url, + ).fetch_calendar() + + def download_predictions( + self, + currency: str, + indicator_list: list[str], + start_date: str = None, + end_date: str = None, + ) -> pd.DataFrame: + return FXMacroDataMacroDownloader( + currency=currency, + indicator_list=indicator_list, + start_date=start_date, + end_date=end_date, + api_key=self.api_key, + base_url=self.base_url, + ).fetch_predictions() + + def add_macro_features( + self, + data: pd.DataFrame, + macro_data: pd.DataFrame, + date_column: str = "timestamp", + ) -> pd.DataFrame: + if macro_data.empty: + return data.copy() + + df = data.copy() + macro = macro_data.copy() + df["_fxmacrodata_date"] = pd.to_datetime(df[date_column]).dt.normalize() + macro["_fxmacrodata_date"] = pd.to_datetime(macro["date"]).dt.normalize() + + for (currency, indicator), group in macro.groupby(["currency", "indicator"]): + prefix = f"macro_{currency}_{indicator}" + feature_cols = [ + "value", + "actual", + "consensus", + "forecast", + "surprise", + "prediction", + "announcement_datetime", + ] + selected = group[["_fxmacrodata_date"] + feature_cols].copy() + selected = selected.sort_values("_fxmacrodata_date") + selected = selected.drop_duplicates("_fxmacrodata_date", keep="last") + selected[f"{prefix}_event"] = 1.0 + selected = selected.rename( + columns={col: f"{prefix}_{col}" for col in feature_cols} + ) + df = df.merge(selected, on="_fxmacrodata_date", how="left") + event_col = f"{prefix}_event" + df[event_col] = df[event_col].fillna(0.0) + fill_cols = [col for col in selected.columns if col.startswith(prefix)] + fill_cols = [col for col in fill_cols if col != event_col] + if "tic" in df.columns: + df[fill_cols] = df.groupby("tic", group_keys=False)[fill_cols].ffill() + else: + df[fill_cols] = df[fill_cols].ffill() + + return df.drop(columns=["_fxmacrodata_date"]) + + def clean_data(self, df: pd.DataFrame) -> pd.DataFrame: + df = df.copy() + df["timestamp"] = pd.to_datetime(df["timestamp"]) + df = df.dropna() + df = df.sort_values(["timestamp", "tic"]).reset_index(drop=True) + return df + + def add_technical_indicator( + self, data: pd.DataFrame, tech_indicator_list: list[str] + ) -> pd.DataFrame: + df = data.copy() + df = df.sort_values(by=["tic", "timestamp"]) + stock = Sdf.retype(df.copy()) + unique_ticker = stock.tic.unique() + + for indicator in tech_indicator_list: + indicator_df = pd.DataFrame() + for tic in unique_ticker: + try: + temp_indicator = stock[stock.tic == tic][indicator] + temp_indicator = pd.DataFrame(temp_indicator) + temp_indicator["tic"] = tic + temp_indicator["timestamp"] = df[df.tic == tic][ + "timestamp" + ].to_list() + indicator_df = pd.concat( + [indicator_df, temp_indicator], ignore_index=True + ) + except Exception as exc: + print(exc) + df = df.merge( + indicator_df[["tic", "timestamp", indicator]], + on=["tic", "timestamp"], + how="left", + ) + df = df.sort_values(by=["timestamp", "tic"]).reset_index(drop=True) + return df + + def add_turbulence(self, df: pd.DataFrame) -> pd.DataFrame: + df = df.copy() + df["turbulence"] = 0 + return df + + def add_vix(self, df: pd.DataFrame) -> pd.DataFrame: + df = df.copy() + df["VIXY"] = 0 + return df + + def add_vixor(self, df: pd.DataFrame) -> pd.DataFrame: + return self.add_vix(df) + + def df_to_array( + self, df: pd.DataFrame, tech_indicator_list: list[str], if_vix: bool + ) -> list[np.ndarray]: + df = df.copy() + unique_ticker = df.tic.unique() + if_first_time = True + turbulence_array = None + for tic in unique_ticker: + if if_first_time: + price_array = df[df.tic == tic][["close"]].values + tech_array = df[df.tic == tic][tech_indicator_list].values + if if_vix: + turbulence_array = df[df.tic == tic]["VIXY"].values + else: + turbulence_array = df[df.tic == tic]["turbulence"].values + if_first_time = False + else: + price_array = np.hstack( + [price_array, df[df.tic == tic][["close"]].values] + ) + tech_array = np.hstack( + [tech_array, df[df.tic == tic][tech_indicator_list].values] + ) + return price_array, tech_array, turbulence_array diff --git a/finrl/meta/preprocessor/fxmacrodatadownloader.py b/finrl/meta/preprocessor/fxmacrodatadownloader.py new file mode 100644 index 0000000000..99b2b62d22 --- /dev/null +++ b/finrl/meta/preprocessor/fxmacrodatadownloader.py @@ -0,0 +1,358 @@ +from __future__ import annotations + +import json +import os +from typing import List +from urllib.parse import urlencode +from urllib.request import Request +from urllib.request import urlopen + +import pandas as pd + +DEFAULT_BASE_URL = "https://api.fxmacrodata.com/v1" +API_KEY_ENV_VARS = ("FXMACRODATA_API_KEY", "FXMD_API_KEY") + + +class FXMacroDataDownloader: + """Provides methods for retrieving daily FX spot data from FXMacroData. + + Parameters + ---------- + start_date : str + Start date of the data in ``YYYY-MM-DD`` format. + end_date : str + End date of the data in ``YYYY-MM-DD`` format. + ticker_list : list + FX pairs such as ``EURUSD`` or ``EUR/USD``. + api_key : str, optional + FXMacroData API key. If omitted, ``FXMACRODATA_API_KEY`` and + ``FXMD_API_KEY`` environment variables are checked. + base_url : str + FXMacroData API base URL. + timeout : float + HTTP timeout in seconds. + """ + + def __init__( + self, + start_date: str, + end_date: str, + ticker_list: list[str], + api_key: str = None, + base_url: str = DEFAULT_BASE_URL, + timeout: float = 30, + ): + self.start_date = start_date + self.end_date = end_date + self.ticker_list = ticker_list + self.api_key = api_key or get_env_api_key() + self.base_url = base_url.rstrip("/") + self.timeout = timeout + + def fetch_data(self) -> pd.DataFrame: + """Fetches daily FX spot data from FXMacroData. + + Returns + ------- + pd.DataFrame + Columns: date, open, high, low, close, volume, tic, day. + FX spot rates are mapped to OHLC using the same daily rate and + ``volume`` is set to 0. + """ + data_df = pd.DataFrame() + failures = 0 + for ticker in self.ticker_list: + base, quote = self._parse_pair(ticker) + rows = self._request_rows(base, quote) + temp_df = self._rows_to_dataframe(ticker, rows) + if len(temp_df) > 0: + data_df = pd.concat([data_df, temp_df], axis=0) + else: + failures += 1 + + if failures == len(self.ticker_list): + raise ValueError("no data is fetched.") + + data_df = data_df.dropna().reset_index(drop=True) + data_df = data_df.sort_values(by=["date", "tic"]).reset_index(drop=True) + print("Shape of DataFrame: ", data_df.shape) + return data_df + + @staticmethod + def _parse_pair(ticker: str): + pair = ticker.strip().upper() + if pair.endswith("=X"): + pair = pair[:-2] + pair = pair.replace("/", "").replace("-", "").replace("_", "") + if len(pair) != 6 or not pair.isalpha(): + raise ValueError( + "FXMacroData tickers must be currency pairs such as " + "'EURUSD' or 'EUR/USD'." + ) + return pair[:3].lower(), pair[3:].lower() + + def _request_rows(self, base: str, quote: str): + return request_rows( + self.base_url, + f"forex/{base}/{quote}", + {"start_date": self.start_date, "end_date": self.end_date}, + self.api_key, + self.timeout, + ) + + @staticmethod + def _rows_to_dataframe(ticker: str, rows: list) -> pd.DataFrame: + data = [] + for row in rows: + date = row.get("date") + rate = FXMacroDataDownloader._get_rate(row) + if date is None or rate is None: + continue + data.append( + { + "date": pd.to_datetime(date), + "open": rate, + "high": rate, + "low": rate, + "close": rate, + "volume": 0.0, + "tic": ticker, + } + ) + if not data: + return pd.DataFrame( + columns=[ + "date", + "open", + "high", + "low", + "close", + "volume", + "tic", + "day", + ] + ) + data_df = pd.DataFrame(data) + data_df["day"] = data_df["date"].dt.dayofweek + data_df["date"] = data_df.date.apply(lambda x: x.strftime("%Y-%m-%d")) + return data_df + + @staticmethod + def _get_rate(row: dict): + for key in ("val", "value", "close", "rate"): + value = row.get(key) + if value is not None: + return float(value) + return None + + +class FXMacroDataMacroDownloader: + """Download macro announcements, release calendars, and forecast groups.""" + + def __init__( + self, + currency: str, + indicator_list: list[str] = None, + start_date: str = None, + end_date: str = None, + api_key: str = None, + base_url: str = DEFAULT_BASE_URL, + timeout: float = 30, + ): + self.currency = currency.lower() + self.indicator_list = [ + indicator.lower() for indicator in (indicator_list or []) + ] + self.start_date = start_date + self.end_date = end_date + self.api_key = api_key or get_env_api_key() + self.base_url = base_url.rstrip("/") + self.timeout = timeout + + def fetch_catalogue(self, include_coverage: bool = True) -> dict: + return request_json( + self.base_url, + f"data_catalogue/{self.currency}", + {"include_coverage": str(include_coverage).lower()}, + self.api_key, + self.timeout, + ) + + def fetch_announcements(self) -> pd.DataFrame: + frames = [] + for indicator in self.indicator_list: + rows = request_rows( + self.base_url, + f"announcements/{self.currency}/{indicator}", + self._date_params(), + self.api_key, + self.timeout, + ) + frames.append(self._rows_to_frame(indicator, rows, "announcements")) + return concat_frames(frames) + + def fetch_calendar(self) -> pd.DataFrame: + indicators = self.indicator_list or [None] + frames = [] + for indicator in indicators: + params = self._date_params() + if indicator: + params["indicator"] = indicator + rows = request_rows( + self.base_url, + f"calendar/{self.currency}", + params, + self.api_key, + self.timeout, + ) + frames.append(self._rows_to_frame(indicator, rows, "calendar")) + return concat_frames(frames) + + def fetch_predictions(self) -> pd.DataFrame: + frames = [] + for indicator in self.indicator_list: + rows = request_rows( + self.base_url, + f"predictions/{self.currency}/{indicator}", + self._date_params(), + self.api_key, + self.timeout, + ) + frames.append(self._rows_to_frame(indicator, rows, "predictions")) + return concat_frames(frames) + + def _date_params(self) -> dict: + return {"start_date": self.start_date, "end_date": self.end_date} + + def _rows_to_frame(self, indicator: str, rows: list, dataset: str) -> pd.DataFrame: + data = [] + for row in rows: + release = indicator or row.get("release") or row.get("indicator") + date = row.get("date") or row.get("release_date") + if date is None: + continue + prediction, prediction_count = prediction_summary(row) + actual = number(row.get("actual")) + value = number(row.get("val") or row.get("value")) + if value is None: + value = actual + data.append( + { + "date": pd.to_datetime(date), + "currency": self.currency, + "indicator": release, + "dataset": dataset, + "announcement_datetime": int(row.get("announcement_datetime") or 0), + "value": value, + "actual": actual if actual is not None else value, + "previous": number(row.get("previous")), + "revised_previous": number(row.get("revised_previous")), + "consensus": number( + row.get("consensus") + or row.get("expected") + or row.get("estimate") + ), + "forecast": number(row.get("forecast")), + "surprise": number(row.get("surprise")), + "surprise_zscore": number(row.get("surprise_zscore")), + "prediction": prediction, + "prediction_count": prediction_count, + "is_future": bool( + row.get("announcement_timing") == "future" + or row.get("actual_available") is False + ), + "source": row.get("source"), + "announcement_id": row.get("announcement_id"), + } + ) + if not data: + return empty_macro_frame() + frame = pd.DataFrame(data) + frame["date"] = frame["date"].dt.strftime("%Y-%m-%d") + return frame.sort_values(["date", "currency", "indicator"]).reset_index( + drop=True + ) + + +def request_json(base_url: str, path: str, params: dict, api_key: str, timeout: float): + query = urlencode( + {key: value for key, value in params.items() if value is not None} + ) + url = f"{base_url.rstrip('/')}/{path.lstrip('/')}" + if query: + url = f"{url}?{query}" + request = Request(url) + if api_key: + request.add_header("X-API-Key", api_key) + with urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + + +def request_rows(base_url: str, path: str, params: dict, api_key: str, timeout: float): + payload = request_json(base_url, path, params, api_key, timeout) + if isinstance(payload, dict): + data = payload.get("data", []) + return data if isinstance(data, list) else [] + if isinstance(payload, list): + return payload + return [] + + +def get_env_api_key(): + for name in API_KEY_ENV_VARS: + api_key = os.getenv(name) + if api_key: + return api_key + return None + + +def number(value): + if value is None: + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def prediction_summary(row: dict): + predictions = row.get("predictions") + if isinstance(predictions, list) and predictions: + return number(predictions[0].get("predicted_value")), len(predictions) + for key in ("forecast_prediction", "consensus_prediction"): + prediction = row.get(key) + if isinstance(prediction, dict): + return number(prediction.get("predicted_value")), 1 + return None, 0 + + +def empty_macro_frame() -> pd.DataFrame: + return pd.DataFrame( + columns=[ + "date", + "currency", + "indicator", + "dataset", + "announcement_datetime", + "value", + "actual", + "previous", + "revised_previous", + "consensus", + "forecast", + "surprise", + "surprise_zscore", + "prediction", + "prediction_count", + "is_future", + "source", + "announcement_id", + ] + ) + + +def concat_frames(frames: list[pd.DataFrame]) -> pd.DataFrame: + frames = [frame for frame in frames if frame is not None and not frame.empty] + if not frames: + return empty_macro_frame() + return pd.concat(frames, ignore_index=True) diff --git a/unit_tests/downloaders/test_fxmacrodata_downloader.py b/unit_tests/downloaders/test_fxmacrodata_downloader.py new file mode 100644 index 0000000000..97e9d6a45a --- /dev/null +++ b/unit_tests/downloaders/test_fxmacrodata_downloader.py @@ -0,0 +1,268 @@ +from __future__ import annotations + +import json + +import pandas as pd +import pytest + +from finrl.meta.data_processors.processor_fxmacrodata import FXMacroDataProcessor +from finrl.meta.preprocessor.fxmacrodatadownloader import FXMacroDataDownloader +from finrl.meta.preprocessor.fxmacrodatadownloader import FXMacroDataMacroDownloader + +API_KEY = "api_key" + + +class FXMacroDataResponse: + def __init__(self, payload): + self.payload = payload + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def read(self): + return json.dumps(self.payload).encode("utf-8") + + +def test_fxmacrodata_downloader_fetch_data(monkeypatch): + requests = [] + + def mock_urlopen(request, timeout): + requests.append((request, timeout)) + return FXMacroDataResponse( + { + "data": [ + {"date": "2024-01-02", "val": "1.101"}, + {"date": "2024-01-03", "val": 1.102}, + ] + } + ) + + monkeypatch.setattr( + "finrl.meta.preprocessor.fxmacrodatadownloader.urlopen", mock_urlopen + ) + + data = FXMacroDataDownloader( + start_date="2024-01-02", + end_date="2024-01-03", + ticker_list=["EUR/USD"], + api_key=API_KEY, + base_url="https://example.com/v1", + ).fetch_data() + + request, timeout = requests[0] + assert request.full_url == ( + "https://example.com/v1/forex/eur/usd?" + "start_date=2024-01-02&end_date=2024-01-03" + ) + assert dict(request.header_items())["X-api-key"] == API_KEY + assert timeout == 30 + assert list(data.columns) == [ + "date", + "open", + "high", + "low", + "close", + "volume", + "tic", + "day", + ] + assert data["date"].tolist() == ["2024-01-02", "2024-01-03"] + assert data["tic"].tolist() == ["EUR/USD", "EUR/USD"] + assert data["open"].tolist() == [1.101, 1.102] + assert data["high"].tolist() == [1.101, 1.102] + assert data["low"].tolist() == [1.101, 1.102] + assert data["close"].tolist() == [1.101, 1.102] + assert data["volume"].tolist() == [0.0, 0.0] + + +def test_fxmacrodata_processor_download_data(monkeypatch): + monkeypatch.setattr( + FXMacroDataDownloader, + "fetch_data", + lambda self: pd.DataFrame( + { + "date": ["2024-01-02"], + "open": [1.101], + "high": [1.101], + "low": [1.101], + "close": [1.101], + "volume": [0.0], + "tic": ["EURUSD"], + "day": [1], + } + ), + ) + + data = FXMacroDataProcessor( + api_key=API_KEY, base_url="https://example.com/v1" + ).download_data(["EURUSD"], "2024-01-02", "2024-01-03", "1D") + + assert list(data.columns) == [ + "timestamp", + "open", + "high", + "low", + "close", + "volume", + "tic", + ] + assert data.loc[0, "timestamp"] == pd.Timestamp("2024-01-02") + assert data.loc[0, "close"] == 1.101 + + +def test_fxmacrodata_downloader_invalid_pair(): + with pytest.raises(ValueError, match="currency pairs"): + FXMacroDataDownloader("2024-01-02", "2024-01-03", ["EUR"]).fetch_data() + + +def test_fxmacrodata_processor_daily_only(): + with pytest.raises(ValueError, match="daily data only"): + FXMacroDataProcessor().download_data( + ["EURUSD"], "2024-01-02", "2024-01-03", "1Min" + ) + + +def test_fxmacrodata_macro_downloader_fetches_announcements(monkeypatch): + requests = [] + + def mock_urlopen(request, timeout): + requests.append((request, timeout)) + return FXMacroDataResponse( + { + "data": [ + { + "announcement_id": "usd_inflation_2026-05-31", + "date": "2026-05-31", + "val": 4.2, + "announcement_datetime": 1781094600, + "consensus": 3.9, + } + ] + } + ) + + monkeypatch.setattr( + "finrl.meta.preprocessor.fxmacrodatadownloader.urlopen", mock_urlopen + ) + + data = FXMacroDataMacroDownloader( + currency="USD", + indicator_list=["inflation"], + start_date="2026-01-01", + end_date="2026-06-30", + api_key=API_KEY, + base_url="https://example.com/v1", + ).fetch_announcements() + + request, timeout = requests[0] + assert request.full_url == ( + "https://example.com/v1/announcements/usd/inflation?" + "start_date=2026-01-01&end_date=2026-06-30" + ) + assert dict(request.header_items())["X-api-key"] == API_KEY + assert timeout == 30 + assert data.loc[0, "currency"] == "usd" + assert data.loc[0, "indicator"] == "inflation" + assert data.loc[0, "dataset"] == "announcements" + assert data.loc[0, "value"] == 4.2 + assert data.loc[0, "consensus"] == 3.9 + assert data.loc[0, "announcement_datetime"] == 1781094600 + + +def test_fxmacrodata_macro_downloader_fetches_calendar_and_predictions(monkeypatch): + payloads = [] + + def mock_urlopen(request, timeout): + payloads.append(request.full_url) + if "/calendar/" in request.full_url: + return FXMacroDataResponse( + { + "data": [ + { + "release": "policy_rate", + "date": "2026-07-29", + "announcement_datetime": 1785330000, + "forecast": 4.25, + "actual_available": False, + } + ] + } + ) + return FXMacroDataResponse( + { + "data": [ + { + "announcement_id": "usd_inflation_2026-07-31", + "date": "2026-07-31", + "announcement_datetime": 1786537800, + "announcement_timing": "future", + "predictions": [ + { + "predicted_value": 3.81, + "prediction_type": "fxmacrodata", + } + ], + } + ] + } + ) + + monkeypatch.setattr( + "finrl.meta.preprocessor.fxmacrodatadownloader.urlopen", mock_urlopen + ) + + downloader = FXMacroDataMacroDownloader( + currency="usd", + indicator_list=["policy_rate"], + api_key=API_KEY, + base_url="https://example.com/v1", + ) + calendar = downloader.fetch_calendar() + assert calendar.loc[0, "dataset"] == "calendar" + assert calendar.loc[0, "forecast"] == 4.25 + assert bool(calendar.loc[0, "is_future"]) is True + + predictions = FXMacroDataMacroDownloader( + currency="usd", + indicator_list=["inflation"], + api_key=API_KEY, + base_url="https://example.com/v1", + ).fetch_predictions() + assert predictions.loc[0, "dataset"] == "predictions" + assert predictions.loc[0, "prediction"] == 3.81 + assert predictions.loc[0, "prediction_count"] == 1 + assert len(payloads) == 2 + + +def test_fxmacrodata_processor_adds_macro_features(): + price_data = pd.DataFrame( + { + "timestamp": pd.to_datetime(["2026-05-30", "2026-05-31", "2026-06-01"]), + "tic": ["EURUSD", "EURUSD", "EURUSD"], + "close": [1.1, 1.2, 1.3], + } + ) + macro_data = pd.DataFrame( + { + "date": ["2026-05-31"], + "currency": ["usd"], + "indicator": ["inflation"], + "value": [4.2], + "actual": [4.2], + "consensus": [3.9], + "forecast": [4.0], + "surprise": [0.3], + "prediction": [4.1], + "announcement_datetime": [1781094600], + } + ) + + data = FXMacroDataProcessor().add_macro_features(price_data, macro_data) + + assert data["macro_usd_inflation_event"].tolist() == [0.0, 1.0, 0.0] + assert pd.isna(data.loc[0, "macro_usd_inflation_value"]) + assert data.loc[1, "macro_usd_inflation_value"] == 4.2 + assert data.loc[2, "macro_usd_inflation_value"] == 4.2