fix(train): mini-batch training to avoid GPU OOM on hourly dataset

BATCH_SIZE=512 per step; batched embed() at eval + export time.
78k hourly windows can't fit in GPU in one shot (was fine at 877 daily).

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

Taskfile.yml

@@ -18,6 +18,26 @@ tasks:
     deps: [generate]
     cmds: [go test ./... -race]
 
+  data:fetch:
+    desc: "Download EUR/USD M1 from histdata (set YEARS env var)"
+    cmds: [.venv/bin/python scripts/fetch_data.py]
+  data:fetch:historical:
+    desc: "Download EUR/USD M1 2008-2018 from histdata"
+    cmds:
+      - YEARS=2008,2009,2010,2011,2012,2013,2014,2015,2016,2017,2018 .venv/bin/python scripts/fetch_data.py
+  data:prepare:daily:
+    desc: "Rebuild eurusd_daily.parquet from all M1 zips"
+    cmds: [.venv/bin/python scripts/prepare_data.py]
+  data:prepare:hourly:
+    desc: "Build eurusd_hourly.parquet from all M1 zips"
+    cmds: [.venv/bin/python scripts/prepare_hourly.py]
+  data:prepare:all:
+    desc: "Build both daily and hourly parquets"
+    deps: [data:prepare:daily, data:prepare:hourly]
+  data:test:
+    desc: "Run Python data pipeline tests"
+    cmds: [.venv/bin/python -m pytest tests/test_prepare_hourly.py tests/test_hepa.py -v]
+
   eval:probe:
     desc: "Run linear-probe (val_vol_r2) on embeddings from metrics.json"
     cmds: [./bin/eval -metric probe]

scripts/prepare_hourly.py

@@ -0,0 +1,108 @@
+"""Prepare EUR/USD hourly OHLCV + realized vol from histdata M1 zips.
+
+Aggregates all M1 bars in data/raw/DAT_ASCII_EURUSD_M1_*.zip to hourly.
+Realized vol per hour = sqrt(sum(log-return²)) over the constituent M1 bars.
+Weekend hours are naturally absent (FX market closed Sat/Sun); NO interpolation.
+Hours with fewer than MIN_BARS M1 bars are dropped (holidays, thin sessions).
+
+Output: data/processed/eurusd_hourly.parquet
+Columns: datetime (UTC, tz-naive), close, ret (log), realized_vol
+
+    python scripts/prepare_hourly.py
+    RAW=data/raw OUT=data/processed/eurusd_hourly.parquet python scripts/prepare_hourly.py
+"""
+import glob
+import os
+import zipfile
+
+import numpy as np
+import pandas as pd
+
+RAW_DEFAULT = "data/raw"
+OUT_DEFAULT = "data/processed/eurusd_hourly.parquet"
+MIN_BARS    = 30   # drop hours thinner than this (holidays, DST boundary artefacts)
+
+
+# ── Core transformation ──────────────────────────────────────────────────────
+
+def resample_to_hourly(m1: pd.DataFrame) -> pd.DataFrame:
+    """Aggregate M1 DataFrame to hourly bars.
+
+    Args:
+        m1: DataFrame with columns ['ts' (datetime), 'close' (float)]
+
+    Returns:
+        DataFrame with columns ['datetime', 'close', 'ret', 'realized_vol']
+        sorted by datetime; hours with fewer than MIN_BARS M1 ticks dropped.
+    """
+    m1 = m1.sort_values("ts").copy()
+    m1["log_r"] = np.log(m1["close"]).diff()
+    m1["hour"]  = m1["ts"].dt.floor("h")
+
+    agg = m1.groupby("hour").agg(
+        close       = ("close",  "last"),
+        realized_vol= ("log_r",  lambda x: np.sqrt(np.nansum(x.values ** 2))),
+        n_bars      = ("log_r",  "count"),
+    ).reset_index()
+
+    agg = agg[agg["n_bars"] >= MIN_BARS].copy()
+    agg["ret"] = np.log(agg["close"]).diff()
+    agg = agg.dropna(subset=["ret"]).reset_index(drop=True)
+    agg = agg.rename(columns={"hour": "datetime"})
+    return agg[["datetime", "close", "ret", "realized_vol"]]
+
+
+def load_m1_from_zips(raw_dir: str) -> pd.DataFrame:
+    """Load and concatenate all M1 zips from raw_dir (histdata format)."""
+    pattern = os.path.join(raw_dir, "DAT_ASCII_EURUSD_M1_*.zip")
+    zips = sorted(glob.glob(pattern))
+    if not zips:
+        raise FileNotFoundError(f"No M1 zips found at {pattern}")
+    frames = []
+    for zp in zips:
+        with zipfile.ZipFile(zp) as z:
+            csv = [n for n in z.namelist() if n.endswith(".csv")][0]
+            with z.open(csv) as f:
+                df = pd.read_csv(
+                    f, sep=";", header=None,
+                    names=["dt", "open", "high", "low", "close", "vol"],
+                )
+        df["ts"] = pd.to_datetime(df["dt"], format="%Y%m%d %H%M%S")
+        frames.append(df[["ts", "close"]])
+        print(f"  loaded {os.path.basename(zp)}: {len(df):,} rows")
+    return pd.concat(frames).sort_values("ts").reset_index(drop=True)
+
+
+def build_hourly_parquet(
+    raw_dir: str = RAW_DEFAULT,
+    out_path: str = OUT_DEFAULT,
+) -> pd.DataFrame:
+    """Full pipeline: load all M1 zips → hourly parquet. Returns the DataFrame."""
+    print(f"Loading M1 zips from {raw_dir}...")
+    m1 = load_m1_from_zips(raw_dir)
+    print(f"Total M1 bars: {len(m1):,}  ({m1['ts'].min().date()} → {m1['ts'].max().date()})")
+
+    print("Resampling to hourly...")
+    hourly = resample_to_hourly(m1)
+    print(f"Hourly rows: {len(hourly):,}  ({hourly['datetime'].min()} → {hourly['datetime'].max()})")
+
+    # Sanity: COVID crash (Mar 2020) should show realized vol spike if data covers it
+    if hourly["datetime"].dt.year.isin([2020]).any():
+        rv = hourly.set_index("datetime")["realized_vol"]
+        try:
+            mar20 = rv["2020-03-01":"2020-03-31"].max()
+            typ   = rv["2019-01-01":"2019-12-31"].median()
+            print(f"Sanity — median 2019 RV: {typ:.6f} | max Mar-2020 RV: {mar20:.6f} | spike ×{mar20/typ:.1f}")
+        except Exception:
+            pass
+
+    os.makedirs(os.path.dirname(os.path.abspath(out_path)), exist_ok=True)
+    hourly.to_parquet(out_path, index=False)
+    print(f"Written: {out_path}")
+    return hourly
+
+
+if __name__ == "__main__":
+    raw_dir  = os.environ.get("RAW",  RAW_DEFAULT)
+    out_path = os.environ.get("OUT",  OUT_DEFAULT)
+    build_hourly_parquet(raw_dir=raw_dir, out_path=out_path)

tests/test_hepa.py

@@ -92,11 +92,21 @@ def test_jepa_step_end_to_end(train_mod):
     assert loss.item() < 100, "loss exploded"
 
 
-# 6. build() still returns year-based OOS split (2022-2023)
+# 6. build() returns year-based OOS split (2022-2023); hourly gives many more windows
 def test_build_year_split(train_mod):
     (Xtr, ytr), (Xte, yte) = train_mod.build()
     assert Xtr.shape[1] == train_mod.WINDOW
     assert Xte.shape[1] == train_mod.WINDOW
     assert len(Xtr) > 0 and len(Xte) > 0
-    # OOS set should be ~600 windows (2 years of daily data)
+    # OOS: daily ≈ 600; hourly ≈ 17,000 (2 years × ~8,500 trading hours/year)
-    assert 400 < len(Xte) < 900, f"OOS size unexpected: {len(Xte)}"
+    assert len(Xte) > 400, f"OOS too small: {len(Xte)}"
+
+
+# 7. hourly build gives > 10× more training windows than daily
+def test_build_hourly_more_windows(train_mod):
+    import os
+    if not os.path.exists("data/processed/eurusd_hourly.parquet"):
+        pytest.skip("eurusd_hourly.parquet not present — run data:prepare:hourly first")
+    (Xtr, _), _ = train_mod.build()
+    # Daily had ~877 train windows; hourly with 2008-2021 should have > 50,000
+    assert len(Xtr) > 10_000, f"expected >10k hourly train windows, got {len(Xtr)}"

tests/test_prepare_hourly.py

@@ -0,0 +1,126 @@
+"""Failing tests for scripts/prepare_hourly.py.
+
+Tests the M1 → hourly aggregation logic using synthetic data before touching
+real downloads.
+
+Run: cd ~/dev/AI/jepa-fx-risk && .venv/bin/python -m pytest tests/test_prepare_hourly.py -v
+"""
+import numpy as np
+import pandas as pd
+import pytest
+import importlib.util, sys, os
+
+
+def _import():
+    spec = importlib.util.spec_from_file_location(
+        "prepare_hourly", "scripts/prepare_hourly.py"
+    )
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+@pytest.fixture(scope="module")
+def ph():
+    return _import()
+
+
+def _make_m1(n_days: int = 3, price: float = 1.1000, noise: float = 0.0005) -> pd.DataFrame:
+    """Synthetic M1 DataFrame starting 2020-01-06 (Monday), 390 ticks/day."""
+    rng = np.random.default_rng(42)
+    # generate full trading hours: Mon-Fri 00:00-23:59 (FX is 24h weekday)
+    start = pd.Timestamp("2020-01-06 00:00:00")  # Monday
+    periods = n_days * 24 * 60
+    ts = pd.date_range(start, periods=periods, freq="min")
+    # remove weekends
+    ts = ts[ts.day_of_week < 5]
+    prices = price + np.cumsum(rng.normal(0, noise, len(ts)))
+    return pd.DataFrame({"ts": ts, "close": prices})
+
+
+# 1. resample_to_hourly: DataFrame has correct columns
+def test_columns(ph):
+    m1 = _make_m1()
+    hourly = ph.resample_to_hourly(m1)
+    assert set(["datetime", "close", "ret", "realized_vol"]).issubset(hourly.columns), \
+        f"missing columns: {hourly.columns.tolist()}"
+
+
+# 2. No cross-weekend interpolation: gap between Friday 23:xx and Sunday/Monday must remain
+def test_no_weekend_interpolation(ph):
+    # Make 2 days: Friday + Monday (skip Saturday/Sunday)
+    fri = pd.date_range("2020-01-10 00:00", "2020-01-10 23:59", freq="min")  # Friday
+    mon = pd.date_range("2020-01-13 00:00", "2020-01-13 23:59", freq="min")  # Monday
+    ts = fri.append(mon)
+    prices = 1.1 + np.cumsum(np.random.default_rng(0).normal(0, 0.0001, len(ts)))
+    m1 = pd.DataFrame({"ts": ts, "close": prices})
+    hourly = ph.resample_to_hourly(m1)
+    dates = pd.DatetimeIndex(hourly["datetime"]).date
+    import datetime
+    sat = datetime.date(2020, 1, 11)
+    sun = datetime.date(2020, 1, 12)
+    assert sat not in dates and sun not in dates, "weekend rows found in hourly output"
+
+
+# 3. Realized vol = sqrt(sum(r²)) over minute returns in each hour
+def test_realized_vol_formula(ph):
+    # Two hours: anchor gives 10:00 a valid ret; measurement hour has one known log-return.
+    ts0 = pd.date_range("2020-01-06 09:00", periods=60, freq="min")
+    ts1 = pd.date_range("2020-01-06 10:00", periods=60, freq="min")
+    prices0 = np.ones(60) * 1.0
+    # price jumps at minute 1 and STAYS (no reversion) → one non-zero log-return
+    prices1 = np.full(60, np.exp(0.01))
+    prices1[0] = 1.0  # only first tick is at 1.0; jump happens at tick 1
+    m1 = pd.DataFrame({
+        "ts": np.concatenate([ts0, ts1]),
+        "close": np.concatenate([prices0, prices1]),
+    })
+    hourly = ph.resample_to_hourly(m1)
+    assert len(hourly) >= 1, "no rows after resample"
+    rv = hourly.iloc[-1]["realized_vol"]
+    expected = np.sqrt(0.01 ** 2)
+    assert abs(rv - expected) < 1e-6, f"realized_vol={rv:.8f}, expected≈{expected:.8f}"
+
+
+# 4. Only hours with ≥ 30 M1 bars are kept (thin hours dropped)
+def test_thin_hours_dropped(ph):
+    # 4 hours: pre-anchor gives 09:00 a valid ret; full survives; thin (11:00) is dropped.
+    # pre-anchor (08:00): gives 09:00 a valid ret
+    # anchor (09:00): 60 bars, valid ret  → kept
+    # full  (10:00): 60 bars, valid ret  → kept
+    # thin  (11:00): 10 bars             → dropped
+    # Result: 3 hourly candidates, first (pre-anchor) gets NaN ret → dropped → 2 rows
+    pre   = pd.date_range("2020-01-06 08:00", periods=60, freq="min")
+    anchor= pd.date_range("2020-01-06 09:00", periods=60, freq="min")
+    full  = pd.date_range("2020-01-06 10:00", periods=60, freq="min")
+    thin  = pd.date_range("2020-01-06 11:00", periods=10, freq="min")
+    ts = pre.append(anchor).append(full).append(thin)
+    m1 = pd.DataFrame({"ts": ts, "close": np.ones(len(ts)) * 1.1})
+    hourly = ph.resample_to_hourly(m1)
+    assert len(hourly) == 2, f"expected 2 rows (pre-anchor NaN ret dropped + thin dropped), got {len(hourly)}"
+
+
+# 5. Output parquet path and schema (integration — reads actual M1 zips if present)
+def test_output_schema_from_zips(ph, tmp_path):
+    # Build a minimal fake zip structure
+    import zipfile, io
+    # synthetic M1 CSV (histdata format: YYYYMMDD HHMMSS;O;H;L;C;V)
+    rows = []
+    for h in range(24):
+        for m in range(60):
+            rows.append(f"20200106 {h:02d}{m:02d}00;1.10000;1.10100;1.09900;1.10000;100")
+    csv_content = "\n".join(rows).encode()
+    zip_buf = io.BytesIO()
+    with zipfile.ZipFile(zip_buf, "w") as zf:
+        zf.writestr("DAT_ASCII_EURUSD_M1_2020.csv", csv_content)
+    zip_buf.seek(0)
+    raw_dir = tmp_path / "raw"
+    raw_dir.mkdir()
+    (raw_dir / "DAT_ASCII_EURUSD_M1_2020.zip").write_bytes(zip_buf.read())
+
+    out_path = str(tmp_path / "eurusd_hourly.parquet")
+    ph.build_hourly_parquet(raw_dir=str(raw_dir), out_path=out_path)
+    assert os.path.exists(out_path), "output parquet not created"
+    df = pd.read_parquet(out_path)
+    assert set(["datetime", "close", "ret", "realized_vol"]).issubset(df.columns)
+    assert len(df) > 0

train.py

@@ -18,13 +18,15 @@ import torch.nn as nn
 import torch.nn.functional as F
 
 # --- agent-tunable knobs ---
-WINDOW      = 60
+USE_HOURLY  = True   # prefer eurusd_hourly.parquet when available
-PATCH_LEN   = 10    # non-overlapping patches (6 tokens per window)
+WINDOW      = 240    # hourly: 10 trading days; if USE_HOURLY=False reset to 60
+PATCH_LEN   = 24     # hourly: 1-day patches (10 tokens); if USE_HOURLY=False reset to 10
 D_MODEL     = 128
 DEPTH       = 2
 N_HEADS     = 4
 ALPHA       = 0.1   # VICReg mixing weight (fixed at 0.1 in HEPA paper)
 DELTA_T_MAX = 3     # max prediction horizon in patches (1..min(DELTA_T_MAX, N-1-c))
+BATCH_SIZE  = 512   # mini-batch per step (hourly dataset is too large for full-batch)
 EPOCHS      = 300
 LR          = 3e-4
 SEED        = 0
@@ -120,8 +122,19 @@ class HorizonPredictor(nn.Module):
 # ── Data ─────────────────────────────────────────────────────────────────────
 
 def build():
-    """Year-based split: encoder trains on 2019-2021; probe evaluates on 2022-2023 OOS."""
+    """Year-based split: encoder trains on ≤2021; probe evaluates on ≥2022 OOS.
-    df = pd.read_parquet("data/processed/eurusd_daily.parquet").reset_index(drop=True)
+
+    Uses eurusd_hourly.parquet when USE_HOURLY=True and the file exists;
+    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):
+        df = pd.read_parquet(hourly_path).reset_index(drop=True)
+        df["date"] = pd.to_datetime(df["datetime"])
+    else:
+        df = pd.read_parquet(daily_path).reset_index(drop=True)
         df["date"] = pd.to_datetime(df["date"])
     feats  = df[["ret", "realized_vol"]].to_numpy(np.float32)
     target = df["realized_vol"].to_numpy(np.float32)
@@ -145,29 +158,37 @@ def main():
     (Xtr, ytr), (Xte, yte) = build()
     n_feats   = Xtr.shape[2]
     n_patches = WINDOW // PATCH_LEN
-    Xtr_t     = torch.tensor(Xtr, device=dev)
+    N_tr      = len(Xtr)
+    bs        = min(BATCH_SIZE, N_tr)
 
     enc  = CausalEncoder(n_feats, PATCH_LEN, D_MODEL, N_HEADS, DEPTH).to(dev)
     pred = HorizonPredictor(D_MODEL).to(dev)
     opt  = torch.optim.AdamW(list(enc.parameters()) + list(pred.parameters()), lr=LR)
 
     for ep in range(EPOCHS):
-        # Sample random context position and horizon; Δt log-biased toward short
+        # Random mini-batch (avoids OOM on large hourly dataset)
+        idx_b = torch.randperm(N_tr)[:bs]
+        Xb    = torch.tensor(Xtr[idx_b.numpy()], device=dev)
+
+        # Sample random context position and horizon
         c  = torch.randint(0, n_patches - 1, ()).item()
         dt = torch.randint(1, max(2, min(DELTA_T_MAX, n_patches - 1 - c) + 1), ()).item()
 
-        tokens = enc(Xtr_t)                     # (B, N, D)
+        tokens = enc(Xb)                        # (bs, N, D)
         h_ctx  = tokens[:, c, :]               # context embedding
         h_tgt  = tokens[:, c + dt, :]          # target embedding (joint training)
-        h_hat  = pred(h_ctx, torch.full((len(Xtr),), float(dt), device=dev))
+        h_hat  = pred(h_ctx, torch.full((bs,), float(dt), device=dev))
         loss   = vicreg_loss(h_hat, h_tgt, alpha=ALPHA)
         opt.zero_grad(); loss.backward(); opt.step()
 
     enc.eval()
     with torch.no_grad():
         def embed(X_np):
-            t = torch.tensor(X_np, device=dev)
+            chunks = []
-            return enc(t)[:, -1, :].cpu().numpy()   # last token = full-context summary
+            for i in range(0, len(X_np), bs):
+                t = torch.tensor(X_np[i:i+bs], device=dev)
+                chunks.append(enc(t)[:, -1, :].cpu().numpy())
+            return np.concatenate(chunks, axis=0)
 
         Etr = embed(Xtr)
         Ete = embed(Xte)
@@ -195,7 +216,13 @@ def main():
     # Set EXPORT_EMBEDDINGS=1 to write embeddings.json for the Go eval harness.
     import os
     if os.environ.get("EXPORT_EMBEDDINGS") == "1":
-        df2 = pd.read_parquet("data/processed/eurusd_daily.parquet").reset_index(drop=True)
+        hourly_path2 = "data/processed/eurusd_hourly.parquet"
+        daily_path2  = "data/processed/eurusd_daily.parquet"
+        if USE_HOURLY and os.path.exists(hourly_path2):
+            df2 = pd.read_parquet(hourly_path2).reset_index(drop=True)
+            df2["date"] = pd.to_datetime(df2["datetime"])
+        else:
+            df2 = pd.read_parquet(daily_path2).reset_index(drop=True)
             df2["date"] = pd.to_datetime(df2["date"])
         tr_mask = df2["date"].dt.year <= 2021
         feats2 = df2[["ret", "realized_vol"]].to_numpy(np.float32)
@@ -211,8 +238,12 @@ def main():
                     rvs.append(float(df2["realized_vol"].iloc[t]))
             if not Xs:
                 return [], [], []
+            Xa = np.stack(Xs)
+            chunks = []
             with torch.no_grad():
-                E = enc(torch.tensor(np.stack(Xs), device=dev))[:, -1, :].cpu().numpy().tolist()
+                for i in range(0, len(Xa), bs):
+                    chunks.append(enc(torch.tensor(Xa[i:i+bs], device=dev))[:, -1, :].cpu().numpy())
+            E = np.concatenate(chunks, axis=0).tolist()
             return E, dates, rvs
         Etr2, dates_tr, rv_tr = _export_windows(tr_mask)
         Eoos, dates_oos, rv_oos = _export_windows(df2["date"].dt.year >= 2022)