diff --git a/Taskfile.yml b/Taskfile.yml index e09a71d..bee3606 100644 --- a/Taskfile.yml +++ b/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] diff --git a/scripts/prepare_hourly.py b/scripts/prepare_hourly.py new file mode 100644 index 0000000..8552da5 --- /dev/null +++ b/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) diff --git a/tests/test_hepa.py b/tests/test_hepa.py index 9c69a01..9ec7893 100644 --- a/tests/test_hepa.py +++ b/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) - assert 400 < len(Xte) < 900, f"OOS size unexpected: {len(Xte)}" + # OOS: daily ≈ 600; hourly ≈ 17,000 (2 years × ~8,500 trading hours/year) + 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)}" diff --git a/tests/test_prepare_hourly.py b/tests/test_prepare_hourly.py new file mode 100644 index 0000000..caba777 --- /dev/null +++ b/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 diff --git a/train.py b/train.py index dbd2c81..710f814 100644 --- a/train.py +++ b/train.py @@ -18,8 +18,9 @@ import torch.nn as nn import torch.nn.functional as F # --- agent-tunable knobs --- -WINDOW = 60 -PATCH_LEN = 10 # non-overlapping patches (6 tokens per window) +USE_HOURLY = True # prefer eurusd_hourly.parquet when available +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 @@ -120,9 +121,20 @@ class HorizonPredictor(nn.Module): # ── Data ───────────────────────────────────────────────────────────────────── def build(): - """Year-based split: encoder trains on 2019-2021; probe evaluates on 2022-2023 OOS.""" - df = pd.read_parquet("data/processed/eurusd_daily.parquet").reset_index(drop=True) - df["date"] = pd.to_datetime(df["date"]) + """Year-based split: encoder trains on ≤2021; probe evaluates on ≥2022 OOS. + + 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) tr_idx = df.index[df["date"].dt.year <= 2021].tolist() @@ -195,8 +207,14 @@ 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) - df2["date"] = pd.to_datetime(df2["date"]) + 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) mu2 = feats2[tr_mask].mean(0); sd2 = feats2[tr_mask].std(0) + 1e-8