Add FXMacroData OHLCV utility

quanttrader/util/util_func.py

@@ -3,16 +3,22 @@
 import os
 import pickle
 from datetime import datetime
+import json
+from urllib.parse import urlencode
+from urllib.request import urlopen
 
 import pandas as pd
 
 __all__ = [
     "read_ohlcv_csv",
+    "read_fxmacrodata_ohlcv",
     "read_intraday_bar_pickle",
     "read_tick_data_txt",
     "save_one_run_results",
 ]
 
+FXMACRODATA_API_ROOT = "https://fxmacrodata.com/api/v1"
+
 
 def read_ohlcv_csv(
     filepath: str, adjust: bool = True, tz: str = "America/New_York"
@@ -33,6 +39,63 @@ def read_ohlcv_csv(
     return df
 
 
+def _split_fx_pair(pair: str) -> tuple[str, str]:
+    pair = pair.upper().replace("/", "").replace("-", "").replace("_", "")
+    if len(pair) != 6:
+        raise ValueError("FX pair must be formatted like 'EURUSD' or 'EUR/USD'")
+    return pair[:3], pair[3:]
+
+
+def read_fxmacrodata_ohlcv(
+    pair: str,
+    start_date: str,
+    end_date: str,
+    api_key: str | None = None,
+    api_root: str = FXMACRODATA_API_ROOT,
+    tz: str = "UTC",
+) -> pd.DataFrame:
+    """Read FXMacroData daily FX reference rates as OHLCV bars.
+
+    FXMacroData publishes one official reference value per currency pair and
+    date. The value is mapped to Open, High, Low, and Close with zero Volume so
+    the result can be passed to BacktestEngine.add_data.
+    """
+    base, quote = _split_fx_pair(pair)
+    params = {
+        "start_date": start_date,
+        "end_date": end_date,
+        "limit": 5000,
+    }
+    if api_key:
+        params["api_key"] = api_key
+
+    url = "{}/forex/{}/{}?{}".format(
+        api_root.rstrip("/"),
+        base,
+        quote,
+        urlencode(params),
+    )
+    with urlopen(url, timeout=30) as response:
+        payload = json.loads(response.read().decode("utf-8"))
+
+    records = []
+    for row in payload.get("data", []):
+        value = float(row["val"])
+        records.append((row["date"], value, value, value, value, 0.0))
+
+    df = pd.DataFrame.from_records(
+        records,
+        columns=["Date", "Open", "High", "Low", "Close", "Volume"],
+    )
+    if df.empty:
+        return pd.DataFrame(columns=["Open", "High", "Low", "Close", "Volume"])
+
+    df["Date"] = pd.to_datetime(df["Date"])
+    df = df.sort_values("Date").set_index("Date")
+    df.index = df.index.tz_localize(tz)
+    return df[["Open", "High", "Low", "Close", "Volume"]]
+
+
 def read_intraday_bar_pickle(
     filepath: str, syms: list[str], tz: str = "America/New_York"
 ) -> dict[str, pd.DataFrame]:

tests/test_fxmacrodata_util.py

@@ -0,0 +1,65 @@
+import importlib.util
+import json
+from pathlib import Path
+import unittest
+from urllib.parse import parse_qs, urlparse
+
+
+def _load_util_func_module():
+    path = Path(__file__).parents[1] / "quanttrader" / "util" / "util_func.py"
+    spec = importlib.util.spec_from_file_location("quanttrader_util_func", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class TestFXMacroDataUtil(unittest.TestCase):
+
+    def test_read_fxmacrodata_ohlcv(self):
+        util_func = _load_util_func_module()
+
+        class MockResponse:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+            def read(self):
+                return json.dumps({
+                    "data": [
+                        {"date": "2026-01-02", "val": 1.2},
+                        {"date": "2026-01-01", "val": 1.1},
+                    ]
+                }).encode("utf-8")
+
+        calls = {}
+
+        def mock_urlopen(url, timeout):
+            calls["url"] = url
+            calls["timeout"] = timeout
+            return MockResponse()
+
+        original_urlopen = util_func.urlopen
+        try:
+            util_func.urlopen = mock_urlopen
+            df = util_func.read_fxmacrodata_ohlcv(
+                "EUR/USD",
+                "2026-01-01",
+                "2026-01-02",
+                api_key="test-key",
+                api_root="https://example.test/api/v1",
+            )
+        finally:
+            util_func.urlopen = original_urlopen
+
+        parsed = urlparse(calls["url"])
+        params = parse_qs(parsed.query)
+        self.assertEqual(parsed.path, "/api/v1/forex/EUR/USD")
+        self.assertEqual(params["api_key"], ["test-key"])
+        self.assertEqual(list(df["Close"]), [1.1, 1.2])
+        self.assertEqual(list(df.columns), ["Open", "High", "Low", "Close", "Volume"])
+
+
+if __name__ == "__main__":
+    unittest.main()