feat(data): EUR/USD hourly pipeline + 2008-2023 M1 dataset (#2)

- scripts/prepare_hourly.py: M1→hourly aggregation (realized_vol = sqrt(Σr²),
  MIN_BARS=30 threshold, no weekend rows, year-based split preserved)
- tests/test_prepare_hourly.py: 5 TDD tests, all green
- train.py: USE_HOURLY=True, WINDOW=240 (10-day), PATCH_LEN=24 (1-day patches);
  build() prefers eurusd_hourly.parquet, falls back to daily; EXPORT BLOCK updated
- Taskfile.yml: data:fetch:historical, data:prepare:hourly, data:prepare:all, data:test
- 98,591 hourly rows (2008-2023) covering GFC, Euro crisis, Brexit, COVID, Fed cycle

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

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)