feat(multipair): G10 multi-pair pipeline + USE_MULTIPAIR knob (Option C)

- prepare_hourly.py: parameterize PAIR env var; OUT_DEFAULT per-pair; load_m1_from_zips(pair=)
- prepare_multipair.py: inner-join 5-pair hourly parquets on datetime → wide parquet
  cols: datetime, {pair}_ret, {pair}_rv × n_pairs; eurusd_rv = target
- fetch_multipair.py: download GBPUSD/USDJPY/USDCHF/AUDUSD M1 2008-2023 from histdata
- train.py: USE_MULTIPAIR knob (JEPA_USE_MULTIPAIR=1); build() reads multipair parquet
  with n_channels = n_pairs × 2; target = eurusd_rv
- Taskfile: data:fetch:multipair, data:prepare:pair, data:prepare:multipair, data:test updated
- 7 new tests in test_multipair.py; 34/35 pass (1 SKIP until multipair parquet built)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Taskfile.yml

@@ -34,9 +34,20 @@ tasks:
   data:prepare:all:
     desc: "Build both daily and hourly parquets"
     deps: [data:prepare:daily, data:prepare:hourly]
+  data:fetch:multipair:
+    desc: "Download G10 M1 data (GBPUSD/USDJPY/USDCHF/AUDUSD) 2008-2023 from histdata"
+    cmds: [.venv/bin/python scripts/fetch_multipair.py]
+  data:prepare:pair:
+    desc: "Build {PAIR}_hourly.parquet from data/raw/{PAIR}/ (e.g. PAIR=gbpusd)"
+    cmds: [PAIR={{.PAIR}} .venv/bin/python scripts/prepare_hourly.py {{.EXTRA_ARGS}}]
+    vars:
+      PAIR: '{{default "eurusd" .PAIR}}'
+  data:prepare:multipair:
+    desc: "Merge 5-pair hourly parquets into eurusd_multipair.parquet"
+    cmds: [.venv/bin/python scripts/prepare_multipair.py]
   data:test:
     desc: "Run Python data pipeline tests"
-    cmds: [.venv/bin/python -m pytest tests/test_prepare_hourly.py tests/test_hepa.py -v]
+    cmds: [.venv/bin/python -m pytest tests/test_prepare_hourly.py tests/test_hepa.py tests/test_multipair.py -v]
 
   eval:probe:
     desc: "Run linear-probe (val_vol_r2) on embeddings from metrics.json"

scripts/fetch_multipair.py

@@ -0,0 +1,48 @@
+"""Fetch G10 FX M1 data from histdata.com for all pairs except EURUSD (already fetched).
+
+Each pair's zips go into data/raw/{pair}/ to avoid collisions.
+Output: data/raw/gbpusd/DAT_ASCII_GBPUSD_M1_YYYY.zip etc.
+
+    python scripts/fetch_multipair.py
+    PAIRS=gbpusd,usdjpy YEARS=2020,2021 python scripts/fetch_multipair.py
+"""
+import os
+import time
+
+from histdata import download_hist_data
+from histdata.api import Platform as P, TimeFrame as T
+
+PAIRS_DEFAULT = ["gbpusd", "usdjpy", "usdchf", "audusd"]
+YEARS_DEFAULT = list(range(2008, 2024))
+
+
+def main():
+    pairs_env = os.environ.get("PAIRS", "")
+    pairs = [p.strip() for p in pairs_env.split(",")] if pairs_env else PAIRS_DEFAULT
+
+    years_env = os.environ.get("YEARS", "")
+    years = [int(y.strip()) for y in years_env.split(",")] if years_env else YEARS_DEFAULT
+
+    for pair in pairs:
+        out_dir = f"data/raw/{pair}"
+        os.makedirs(out_dir, exist_ok=True)
+        print(f"\n=== {pair.upper()} ===")
+        for yr in years:
+            out_path = os.path.join(out_dir, f"DAT_ASCII_{pair.upper()}_M1_{yr}.zip")
+            if os.path.exists(out_path):
+                print(f"  {yr} already present, skip")
+                continue
+            try:
+                f = download_hist_data(
+                    year=str(yr), month=None, pair=pair,
+                    platform=P.GENERIC_ASCII, time_frame=T.ONE_MINUTE,
+                    output_directory=out_dir,
+                )
+                print(f"  fetched {yr} → {f}")
+            except Exception as e:
+                print(f"  {yr} FAILED: {e}")
+            time.sleep(2)
+
+
+if __name__ == "__main__":
+    main()

scripts/prepare_hourly.py

@@ -18,8 +18,9 @@ import zipfile
 import numpy as np
 import pandas as pd
 
+PAIR        = os.environ.get("PAIR", "EURUSD").upper()
 RAW_DEFAULT = "data/raw"
-OUT_DEFAULT = "data/processed/eurusd_hourly.parquet"
+OUT_DEFAULT = f"data/processed/{PAIR.lower()}_hourly.parquet"
 MIN_BARS    = 30   # drop hours thinner than this (holidays, DST boundary artefacts)
 
 
@@ -70,9 +71,10 @@ def resample_to_hourly(m1: pd.DataFrame) -> pd.DataFrame:
     return agg[cols]
 
 
-def load_m1_from_zips(raw_dir: str) -> pd.DataFrame:
+def load_m1_from_zips(raw_dir: str, pair: str = None) -> pd.DataFrame:
     """Load and concatenate all M1 zips from raw_dir (histdata format)."""
-    pattern = os.path.join(raw_dir, "DAT_ASCII_EURUSD_M1_*.zip")
+    p = (pair or PAIR).upper()
+    pattern = os.path.join(raw_dir, f"DAT_ASCII_{p}_M1_*.zip")
     zips = sorted(glob.glob(pattern))
     if not zips:
         raise FileNotFoundError(f"No M1 zips found at {pattern}")

scripts/prepare_multipair.py

@@ -0,0 +1,72 @@
+"""Merge per-pair hourly parquets into a single wide multipair parquet.
+
+Each pair contributes two features: {pair}_ret and {pair}_rv (realized vol).
+The merge is an INNER JOIN on datetime — only hours present in ALL pairs are kept.
+The target for train.py remains eurusd_rv.
+
+Output: data/processed/eurusd_multipair.parquet
+
+    python scripts/prepare_multipair.py
+    PROCESSED=data/processed python scripts/prepare_multipair.py
+"""
+import os
+import pandas as pd
+
+PAIRS = ["eurusd", "gbpusd", "usdjpy", "usdchf", "audusd"]
+
+PROCESSED_DEFAULT = "data/processed"
+OUT_DEFAULT       = "data/processed/eurusd_multipair.parquet"
+
+
+def merge_pair_parquets(pair_dfs: dict) -> pd.DataFrame:
+    """Inner-join hourly DataFrames from multiple pairs on datetime.
+
+    Args:
+        pair_dfs: dict mapping pair name (e.g. "eurusd") to hourly DataFrame
+                  with columns [datetime, close, ret, realized_vol, ...].
+    Returns:
+        Wide DataFrame with columns:
+          datetime, {pair}_ret, {pair}_rv  for each pair.
+    """
+    merged = None
+    for pair, df in pair_dfs.items():
+        sub = df[["datetime", "ret", "realized_vol"]].copy()
+        sub = sub.rename(columns={"ret": f"{pair}_ret", "realized_vol": f"{pair}_rv"})
+        sub = sub.set_index("datetime")
+        if merged is None:
+            merged = sub
+        else:
+            merged = merged.join(sub, how="inner")
+
+    return merged.reset_index()
+
+
+def build_multipair_parquet(
+    processed_dir: str = PROCESSED_DEFAULT,
+    out_path: str = OUT_DEFAULT,
+    pairs: list = None,
+) -> None:
+    if pairs is None:
+        pairs = PAIRS
+    pair_dfs = {}
+    for pair in pairs:
+        path = os.path.join(processed_dir, f"{pair}_hourly.parquet")
+        if not os.path.exists(path):
+            raise FileNotFoundError(
+                f"{pair}_hourly.parquet not found at {path} — run prepare_hourly.py for this pair first"
+            )
+        df = pd.read_parquet(path)
+        pair_dfs[pair] = df
+
+    merged = merge_pair_parquets(pair_dfs)
+    merged.to_parquet(out_path, index=False)
+    n_pairs = len(pairs)
+    n_ch    = n_pairs * 2
+    print(f"Multipair parquet: {len(merged):,} rows × {n_ch} feature channels ({n_pairs} pairs)")
+    print(f"Date range: {merged['datetime'].min()} → {merged['datetime'].max()}")
+    print(f"Written: {out_path}")
+
+
+if __name__ == "__main__":
+    processed_dir = os.environ.get("PROCESSED", PROCESSED_DEFAULT)
+    build_multipair_parquet(processed_dir=processed_dir)

tests/test_multipair.py

@@ -0,0 +1,126 @@
+"""Tests for multi-pair G10 pipeline (Option C).
+
+Tests the prepare_multipair.py merge logic and train.py multipair build().
+Run: cd ~/dev/AI/jepa-fx-risk && .venv/bin/python -m pytest tests/test_multipair.py -v
+"""
+import importlib.util
+import numpy as np
+import pandas as pd
+import pytest
+import os
+
+
+def _import_mp():
+    spec = importlib.util.spec_from_file_location("prepare_multipair", "scripts/prepare_multipair.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+@pytest.fixture(scope="module")
+def mp():
+    return _import_mp()
+
+
+def _pair_df(start: str, n_hours: int, seed: int) -> pd.DataFrame:
+    """Synthetic single-pair hourly parquet (same schema as prepare_hourly output)."""
+    rng = np.random.default_rng(seed)
+    dts = pd.date_range(start, periods=n_hours, freq="h")
+    closes = 1.1 + np.cumsum(rng.normal(0, 0.001, n_hours))
+    return pd.DataFrame({
+        "datetime":     dts,
+        "close":        closes,
+        "ret":          rng.normal(0, 0.001, n_hours),
+        "realized_vol": np.abs(rng.normal(0.0005, 0.0001, n_hours)),
+    })
+
+
+# 1. merge_pair_parquets returns inner join on datetime
+def test_merge_inner_join(mp):
+    eur = _pair_df("2020-01-01 00:00", 100, seed=1)  # t0 to t0+99h
+    gbp = _pair_df("2020-01-01 20:00", 60, seed=2)   # t0+20 to t0+79h → 60 common
+    result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
+    assert len(result) == 60, f"expected 60 (inner join), got {len(result)}"
+
+
+# 2. merge_pair_parquets prefixes columns with pair name
+def test_merge_column_prefixes(mp):
+    eur = _pair_df("2020-01-01 00:00", 50, seed=1)
+    gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
+    result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
+    assert "datetime" in result.columns, "datetime column missing"
+    assert "eurusd_ret" in result.columns
+    assert "eurusd_rv" in result.columns
+    assert "gbpusd_ret" in result.columns
+    assert "gbpusd_rv" in result.columns
+    # raw pair columns should not leak through unprefixed
+    assert "ret" not in result.columns
+    assert "realized_vol" not in result.columns
+
+
+# 3. No NaN in merged output
+def test_merge_no_nan(mp):
+    eur = _pair_df("2020-01-01 00:00", 50, seed=1)
+    gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
+    result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
+    nan_count = result.isnull().sum().sum()
+    assert nan_count == 0, f"{nan_count} NaN values in merged output"
+
+
+# 4. PAIRS constant is a non-empty list starting with eurusd
+def test_pairs_constant(mp):
+    assert hasattr(mp, "PAIRS"), "PAIRS constant missing from prepare_multipair.py"
+    assert len(mp.PAIRS) >= 2, "PAIRS must have at least 2 pairs"
+    assert mp.PAIRS[0] == "eurusd", "first pair must be eurusd (target pair)"
+
+
+# 5. merge target column is eurusd_rv (for build() target selection)
+def test_merge_has_eurusd_rv_as_target(mp):
+    eur = _pair_df("2020-01-01 00:00", 50, seed=1)
+    gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
+    result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
+    assert "eurusd_rv" in result.columns, "eurusd_rv (target) missing from merged output"
+    assert (result["eurusd_rv"] > 0).all(), "eurusd_rv should be positive"
+
+
+# 6. train.py recognises JEPA_USE_MULTIPAIR env var
+def test_use_multipair_knob():
+    import importlib.util as ilu
+    spec = ilu.spec_from_file_location(f"train_mp_{id(None)}", "train.py")
+    mod = ilu.module_from_spec(spec)
+    saved = os.environ.get("JEPA_USE_MULTIPAIR")
+    os.environ["JEPA_USE_MULTIPAIR"] = "1"
+    try:
+        spec.loader.exec_module(mod)
+    finally:
+        if saved is None:
+            os.environ.pop("JEPA_USE_MULTIPAIR", None)
+        else:
+            os.environ["JEPA_USE_MULTIPAIR"] = saved
+    assert hasattr(mod, "USE_MULTIPAIR"), "USE_MULTIPAIR knob missing from train.py"
+    assert mod.USE_MULTIPAIR is True
+
+
+# 7. build() uses n_pairs*2 channels when multipair parquet present
+def test_build_uses_multipair_channels():
+    import importlib.util as ilu
+    multipair_path = "data/processed/eurusd_multipair.parquet"
+    if not os.path.exists(multipair_path):
+        pytest.skip("eurusd_multipair.parquet not present — run data:prepare:multipair first")
+    saved = os.environ.get("JEPA_USE_MULTIPAIR")
+    os.environ["JEPA_USE_MULTIPAIR"] = "1"
+    try:
+        spec = ilu.spec_from_file_location(f"train_mp2_{id(None)}", "train.py")
+        mod = ilu.module_from_spec(spec)
+        spec.loader.exec_module(mod)
+        (Xtr, _), _ = mod.build()
+    finally:
+        if saved is None:
+            os.environ.pop("JEPA_USE_MULTIPAIR", None)
+        else:
+            os.environ["JEPA_USE_MULTIPAIR"] = saved
+    mp = _import_mp()
+    expected_ch = len(mp.PAIRS) * 2
+    assert Xtr.shape[2] == expected_ch, (
+        f"expected {expected_ch} channels (n_pairs={len(mp.PAIRS)}×2), got {Xtr.shape[2]}"
+    )

train.py

@@ -35,6 +35,7 @@ PHASE1_LR          = float(_os.environ.get("JEPA_PHASE1_LR",           1e-3))
 PHASE1_JOINT       = bool(int(_os.environ.get("JEPA_PHASE1_JOINT",     1)))
 PHASE1_JOINT_EPOCHS= int(_os.environ.get("JEPA_PHASE1_JOINT_EPOCHS",  30))
 PHASE1_ENCODER_LR  = float(_os.environ.get("JEPA_PHASE1_ENCODER_LR",  3e-6))
+USE_MULTIPAIR      = bool(int(_os.environ.get("JEPA_USE_MULTIPAIR",    0)))
 SEED          = int(_os.environ.get("JEPA_SEED",          0))
 # ---------------------------
 
@@ -149,19 +150,29 @@ def build():
     falls back to eurusd_daily.parquet otherwise.
     """
     import os
-    hourly_path = "data/processed/eurusd_hourly.parquet"
-    daily_path  = "data/processed/eurusd_daily.parquet"
-    if USE_HOURLY and os.path.exists(hourly_path):
+    multipair_path = "data/processed/eurusd_multipair.parquet"
+    hourly_path    = "data/processed/eurusd_hourly.parquet"
+    daily_path     = "data/processed/eurusd_daily.parquet"
+    if USE_MULTIPAIR and os.path.exists(multipair_path):
+        df = pd.read_parquet(multipair_path).reset_index(drop=True)
+        df["date"] = pd.to_datetime(df["datetime"])
+        # All {pair}_ret + {pair}_rv columns as features; eurusd_rv as target
+        feat_cols = [c for c in df.columns if c.endswith("_ret") or c.endswith("_rv")]
+        FEAT_COLS = feat_cols
+        target_col = "eurusd_rv"
+    elif USE_HOURLY and os.path.exists(hourly_path):
         df = pd.read_parquet(hourly_path).reset_index(drop=True)
         df["date"] = pd.to_datetime(df["datetime"])
+        # 2-channel default (HPO: adding hl_range+ret_intrabar hurt — correlated with base feats)
+        FEAT_COLS = ["ret", "realized_vol"]
+        target_col = "realized_vol"
     else:
         df = pd.read_parquet(daily_path).reset_index(drop=True)
         df["date"] = pd.to_datetime(df["date"])
-    # 2-channel default (HPO: adding hl_range+ret_intrabar hurt — correlated with base feats)
-    # To experiment: change to ["ret", "realized_vol", "hl_range", "ret_intrabar"]
-    FEAT_COLS = ["ret", "realized_vol"]
+        FEAT_COLS = ["ret", "realized_vol"]
+        target_col = "realized_vol"
     feats  = df[FEAT_COLS].to_numpy(np.float32)
-    target = df["realized_vol"].to_numpy(np.float32)
+    target = df[target_col].to_numpy(np.float32)
     tr_idx = df.index[df["date"].dt.year <= 2021].tolist()
     te_idx = df.index[df["date"].dt.year >= 2022].tolist()
     mu = feats[:tr_idx[-1]+1].mean(0)