Add FXMacroData data collector

scripts/data_collector/README.md

@@ -5,6 +5,7 @@
 Scripts for data collection
 
 - yahoo: get *US/CN* stock data from *Yahoo Finance*
+- fxmacrodata: get daily *FX spot rates* from *FXMacroData*
 - fund: get fund data from *http://fund.eastmoney.com*
 - cn_index: get *CN index* from *http://www.csindex.com.cn*, *CSI300*/*CSI100*
 - us_index: get *US index* from *https://en.wikipedia.org/wiki*, *SP500*/*NASDAQ100*/*DJIA*/*SP400*
@@ -57,4 +58,4 @@ Scripts for data collection
   | Component      | required data                                     |
   |---------------------------------------------------|--------------------------------|
   | Data retrieval | Features, Calendar, Instrument                    |
-  | Backtest       | **Features[Price/Volume]**, Calendar, Instruments |
+  | Backtest       | **Features[Price/Volume]**, Calendar, Instruments |

scripts/data_collector/fxmacrodata/README.md

@@ -0,0 +1,65 @@
+# Collect Data From FXMacroData
+
+FXMacroData provides daily FX spot-rate series for currency pairs such as `EUR/USD`.
+This collector downloads those series into qlib-compatible CSV files and then uses
+qlib's existing `dump_bin.py` script to convert them into qlib binary data.
+
+## Requirements
+
+```bash
+pip install -r scripts/data_collector/fxmacrodata/requirements.txt
+```
+
+Set an API key when you need authenticated access:
+
+```bash
+export FXMACRODATA_API_KEY="<your key>"
+```
+
+`FXMD_API_KEY` is also supported. You can also pass `--api_key` to the collector.
+
+## Download FX Data
+
+```bash
+python scripts/data_collector/fxmacrodata/collector.py download_data \
+  --source_dir ~/.qlib/fxmacrodata/source \
+  --start 2024-01-01 \
+  --end 2024-03-01 \
+  --pairs EURUSD,GBPUSD,USDJPY
+```
+
+Supported pair formats include `EURUSD`, `EUR/USD`, `EUR-USD`, `EUR_USD`, and
+Yahoo-style `EURUSD=X`. The collector currently supports daily data only.
+
+## Normalize Data
+
+```bash
+python scripts/data_collector/fxmacrodata/collector.py normalize_data \
+  --source_dir ~/.qlib/fxmacrodata/source \
+  --normalize_dir ~/.qlib/fxmacrodata/normalize
+```
+
+FX spot data is shaped with `open`, `high`, `low`, and `close` equal to the daily
+spot rate. `volume` is set to `0`, `factor` is set to `1`, and `change` is the
+daily percentage change in `close`.
+
+## Dump To qlib Format
+
+```bash
+python scripts/dump_bin.py dump_all \
+  --data_path ~/.qlib/fxmacrodata/normalize \
+  --qlib_dir ~/.qlib/qlib_data/fxmacrodata \
+  --freq day \
+  --exclude_fields date,symbol \
+  --file_suffix .csv
+```
+
+## Use The Data
+
+```python
+import qlib
+from qlib.data import D
+
+qlib.init(provider_uri="~/.qlib/qlib_data/fxmacrodata", region="us")
+df = D.features(["eurusd", "gbpusd"], ["$close", "$change"], freq="day")
+```

scripts/data_collector/fxmacrodata/collector.py

@@ -0,0 +1,297 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import os
+import sys
+from pathlib import Path
+from typing import Iterable, List, Optional, Sequence, Tuple
+
+import fire
+import pandas as pd
+import requests
+
+CUR_DIR = Path(__file__).resolve().parent
+sys.path.append(str(CUR_DIR.parent.parent))
+
+from data_collector.base import BaseCollector, BaseNormalize, BaseRun
+
+DEFAULT_BASE_URL = "https://fxmacrodata.com/api/v1"
+DEFAULT_PAIRS = (
+    "EURUSD",
+    "GBPUSD",
+    "USDJPY",
+    "AUDUSD",
+    "USDCAD",
+    "USDCHF",
+    "NZDUSD",
+)
+API_KEY_ENV_VARS = ("FXMACRODATA_API_KEY", "FXMD_API_KEY")
+OUTPUT_COLUMNS = [
+    "date",
+    "symbol",
+    "open",
+    "close",
+    "high",
+    "low",
+    "volume",
+    "factor",
+    "change",
+]
+
+
+class FXMacroDataCollector(BaseCollector):
+    """Collect daily FX spot rates from FXMacroData."""
+
+    def __init__(
+        self,
+        save_dir: [str, Path],
+        start=None,
+        end=None,
+        interval="1d",
+        max_workers=1,
+        max_collector_count=2,
+        delay=0,
+        check_data_length: int = None,
+        limit_nums: int = None,
+        pairs: [str, Sequence[str]] = None,
+        api_key: Optional[str] = None,
+        base_url: str = DEFAULT_BASE_URL,
+        timeout: float = 30,
+    ):
+        if interval != self.INTERVAL_1d:
+            raise ValueError(
+                "FXMacroDataCollector supports daily data only: --interval 1d"
+            )
+
+        self.pairs = self._normalize_pairs(pairs)
+        self.api_key = api_key or self._get_env_api_key()
+        self.base_url = base_url.rstrip("/")
+        self.timeout = timeout
+
+        super(FXMacroDataCollector, self).__init__(
+            save_dir=save_dir,
+            start=start,
+            end=end,
+            interval=interval,
+            max_workers=max_workers,
+            max_collector_count=max_collector_count,
+            delay=delay,
+            check_data_length=check_data_length,
+            limit_nums=limit_nums,
+        )
+
+    def get_instrument_list(self):
+        return self.pairs
+
+    def normalize_symbol(self, symbol: str):
+        return self._normalize_pair(symbol).lower()
+
+    def get_data(
+        self,
+        symbol: str,
+        interval: str,
+        start_datetime: pd.Timestamp,
+        end_datetime: pd.Timestamp,
+    ) -> pd.DataFrame:
+        if interval != self.INTERVAL_1d:
+            raise ValueError(
+                "FXMacroDataCollector supports daily data only: --interval 1d"
+            )
+
+        pair = self._normalize_pair(symbol)
+        base, quote = self._split_pair(pair)
+        params = {
+            "start_date": self._format_date(start_datetime),
+            "end_date": self._format_date(end_datetime),
+        }
+        headers = {}
+        if self.api_key:
+            headers["X-API-Key"] = self.api_key
+
+        response = requests.get(
+            f"{self.base_url}/forex/{base}/{quote}",
+            params=params,
+            headers=headers,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        rows = self._payload_rows(response.json())
+        return self._rows_to_frame(pair, rows)
+
+    @classmethod
+    def _normalize_pairs(cls, pairs: [str, Sequence[str], None]) -> List[str]:
+        if pairs is None:
+            return list(DEFAULT_PAIRS)
+        if isinstance(pairs, str):
+            pairs = [pair.strip() for pair in pairs.split(",") if pair.strip()]
+        return [cls._normalize_pair(pair) for pair in pairs]
+
+    @staticmethod
+    def _normalize_pair(pair: str) -> str:
+        pair = pair.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 pairs must look like EURUSD or EUR/USD"
+            )
+        return pair
+
+    @staticmethod
+    def _split_pair(pair: str) -> Tuple[str, str]:
+        return pair[:3].lower(), pair[3:].lower()
+
+    @staticmethod
+    def _get_env_api_key() -> Optional[str]:
+        for name in API_KEY_ENV_VARS:
+            value = os.getenv(name)
+            if value:
+                return value
+        return None
+
+    @staticmethod
+    def _format_date(value: pd.Timestamp) -> str:
+        return pd.Timestamp(value).strftime("%Y-%m-%d")
+
+    @staticmethod
+    def _payload_rows(payload) -> list:
+        if isinstance(payload, list):
+            return payload
+        if isinstance(payload, dict):
+            data = payload.get("data", [])
+            return data if isinstance(data, list) else []
+        return []
+
+    @classmethod
+    def _rows_to_frame(cls, pair: str, rows: list) -> pd.DataFrame:
+        records = []
+        for row in rows:
+            date = row.get("date") or row.get("timestamp")
+            rate = cls._extract_rate(row)
+            if date is None or rate is None:
+                continue
+            records.append(
+                {
+                    "date": pd.Timestamp(date),
+                    "symbol": pair.lower(),
+                    "open": rate,
+                    "close": rate,
+                    "high": rate,
+                    "low": rate,
+                    "volume": 0.0,
+                    "factor": 1.0,
+                }
+            )
+        if not records:
+            return pd.DataFrame(columns=OUTPUT_COLUMNS)
+        df = pd.DataFrame(records)
+        df = (
+            df.drop_duplicates("date")
+            .sort_values("date")
+            .reset_index(drop=True)
+        )
+        df["change"] = df["close"].ffill().pct_change().fillna(0.0)
+        df["date"] = df["date"].dt.strftime("%Y-%m-%d")
+        return df[OUTPUT_COLUMNS]
+
+    @staticmethod
+    def _extract_rate(row: dict) -> Optional[float]:
+        for key in ("val", "value", "close", "rate", "fx_rate"):
+            value = row.get(key)
+            if value is not None:
+                return float(value)
+        return None
+
+
+class FXMacroDataNormalize(BaseNormalize):
+    """Normalize FXMacroData CSVs for qlib dump_bin."""
+
+    def _get_calendar_list(self) -> Iterable[pd.Timestamp]:
+        return []
+
+    def normalize(self, df: pd.DataFrame) -> pd.DataFrame:
+        if df.empty:
+            return df
+
+        df = df.copy()
+        df[self._date_field_name] = pd.to_datetime(df[self._date_field_name])
+        df = df.drop_duplicates(self._date_field_name).sort_values(
+            self._date_field_name
+        )
+        df["close"] = pd.to_numeric(df["close"], errors="coerce")
+        df = df.dropna(subset=["close"])
+
+        for column in ("open", "high", "low"):
+            if column not in df.columns:
+                df[column] = df["close"]
+            df[column] = pd.to_numeric(
+                df[column], errors="coerce"
+            ).fillna(df["close"])
+        df["volume"] = 0.0
+        df["factor"] = 1.0
+        df["change"] = df["close"].ffill().pct_change().fillna(0.0)
+        df[self._date_field_name] = df[self._date_field_name].dt.strftime(
+            "%Y-%m-%d"
+        )
+        df[self._symbol_field_name] = (
+            df[self._symbol_field_name].astype(str).str.lower()
+        )
+        return df[OUTPUT_COLUMNS]
+
+
+class Run(BaseRun):
+    def download_data(
+        self,
+        max_collector_count=2,
+        delay=0,
+        start=None,
+        end=None,
+        check_data_length: int = None,
+        limit_nums=None,
+        pairs: str = ",".join(DEFAULT_PAIRS),
+        api_key: Optional[str] = None,
+        base_url: str = DEFAULT_BASE_URL,
+        timeout: float = 30,
+    ):
+        """Download daily FX spot rates from FXMacroData."""
+
+        super(Run, self).download_data(
+            max_collector_count=max_collector_count,
+            delay=delay,
+            start=start,
+            end=end,
+            check_data_length=check_data_length,
+            limit_nums=limit_nums,
+            pairs=pairs,
+            api_key=api_key,
+            base_url=base_url,
+            timeout=timeout,
+        )
+
+    def normalize_data(
+        self, date_field_name: str = "date", symbol_field_name: str = "symbol"
+    ):
+        """Normalize FXMacroData daily data."""
+
+        if self.interval != "1d":
+            raise ValueError(
+                "FXMacroData collector supports daily data only: --interval 1d"
+            )
+        super(Run, self).normalize_data(date_field_name, symbol_field_name)
+
+    @property
+    def collector_class_name(self):
+        return "FXMacroDataCollector"
+
+    @property
+    def normalize_class_name(self):
+        return "FXMacroDataNormalize"
+
+    @property
+    def default_base_dir(self) -> [Path, str]:
+        return CUR_DIR
+
+
+if __name__ == "__main__":
+    fire.Fire(Run)

scripts/data_collector/fxmacrodata/requirements.txt

@@ -0,0 +1,6 @@
+fire
+joblib
+loguru
+pandas
+requests
+tqdm