diff --git a/Taskfile.yml b/Taskfile.yml index bee3606..14ae0d5 100644 --- a/Taskfile.yml +++ b/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" diff --git a/scripts/fetch_multipair.py b/scripts/fetch_multipair.py new file mode 100644 index 0000000..055a53d --- /dev/null +++ b/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() diff --git a/scripts/prepare_hourly.py b/scripts/prepare_hourly.py index ff9b055..beb52fc 100644 --- a/scripts/prepare_hourly.py +++ b/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}") diff --git a/scripts/prepare_multipair.py b/scripts/prepare_multipair.py new file mode 100644 index 0000000..b996403 --- /dev/null +++ b/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) diff --git a/tests/test_multipair.py b/tests/test_multipair.py new file mode 100644 index 0000000..816535f --- /dev/null +++ b/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]}" + ) diff --git a/train.py b/train.py index a235db2..0eea9cc 100644 --- a/train.py +++ b/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)