feat(features): add hl_range + ret_intrabar OHLCV features (4-channel input)

- prepare_hourly.py: keep O/H/L columns from M1 zips; compute per-hour
  hl_range=log(H/L) and ret_intrabar=log(close/open); backward-compat
  (falls back to 4-col output only when O/H/L present in input)
- train.py build(): auto-detect extra features from parquet columns
  (FEAT_COLS = [ret, realized_vol] + [hl_range, ret_intrabar] if present)
- 5 new tests (9 total in test_prepare_hourly); 24/24 pass

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

scripts/prepare_hourly.py

@@ -29,27 +29,45 @@ def resample_to_hourly(m1: pd.DataFrame) -> pd.DataFrame:
     """Aggregate M1 DataFrame to hourly bars.
 
     Args:
-        m1: DataFrame with columns ['ts' (datetime), 'close' (float)]
+        m1: DataFrame with columns ['ts', 'open', 'high', 'low', 'close']
+            ('open'/'high'/'low' optional — omit for close-only data).
 
     Returns:
-        DataFrame with columns ['datetime', 'close', 'ret', 'realized_vol']
-        sorted by datetime; hours with fewer than MIN_BARS M1 ticks dropped.
+        DataFrame with columns ['datetime', 'close', 'ret', 'realized_vol',
+        'hl_range', 'ret_intrabar'] sorted by datetime.
+        Hours with fewer than MIN_BARS M1 ticks are 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()
+    has_ohlc = all(c in m1.columns for c in ("open", "high", "low"))
+
+    agg_dict = dict(
+        close        = ("close", "last"),
+        realized_vol = ("log_r", lambda x: np.sqrt(np.nansum(x.values ** 2))),
+        n_bars       = ("log_r", "count"),
+    )
+    if has_ohlc:
+        agg_dict["high"]  = ("high",  "max")
+        agg_dict["low"]   = ("low",   "min")
+        agg_dict["open_"] = ("open",  "first")
+
+    agg = m1.groupby("hour").agg(**agg_dict).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"]]
+
+    if has_ohlc:
+        agg["hl_range"]    = np.log(agg["high"] / agg["low"])
+        agg["ret_intrabar"]= np.log(agg["close"] / agg["open_"])
+        cols = ["datetime", "close", "ret", "realized_vol", "hl_range", "ret_intrabar"]
+    else:
+        cols = ["datetime", "close", "ret", "realized_vol"]
+
+    return agg[cols]
 
 
 def load_m1_from_zips(raw_dir: str) -> pd.DataFrame:
@@ -68,7 +86,7 @@ def load_m1_from_zips(raw_dir: str) -> pd.DataFrame:
                     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"]])
+        frames.append(df[["ts", "open", "high", "low", "close"]])
         print(f"  loaded {os.path.basename(zp)}: {len(df):,} rows")
     return pd.concat(frames).sort_values("ts").reset_index(drop=True)
 

tests/test_prepare_hourly.py

@@ -102,9 +102,7 @@ def test_thin_hours_dropped(ph):
 
 # 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):
@@ -124,3 +122,89 @@ def test_output_schema_from_zips(ph, tmp_path):
     df = pd.read_parquet(out_path)
     assert set(["datetime", "close", "ret", "realized_vol"]).issubset(df.columns)
     assert len(df) > 0
+
+
+# ── New OHLCV-derived features ────────────────────────────────────────────────
+
+def _make_m1_ohlcv(n_hours: int = 4, price: float = 1.1) -> pd.DataFrame:
+    """Synthetic M1 with distinct O, H, L, C so hl_range and ret_intrabar are nonzero."""
+    rng = np.random.default_rng(7)
+    ts = pd.date_range("2020-01-06 00:00", periods=n_hours * 60, freq="min")
+    closes = price + np.cumsum(rng.normal(0, 0.0002, len(ts)))
+    highs  = closes + rng.uniform(0.0001, 0.0005, len(ts))
+    lows   = closes - rng.uniform(0.0001, 0.0005, len(ts))
+    opens  = np.roll(closes, 1); opens[0] = price
+    return pd.DataFrame({"ts": ts, "open": opens, "high": highs, "low": lows, "close": closes})
+
+
+# 6. resample_to_hourly produces hl_range column
+def test_hourly_has_hl_range(ph):
+    m1 = _make_m1_ohlcv()
+    hourly = ph.resample_to_hourly(m1)
+    assert "hl_range" in hourly.columns, f"missing hl_range; cols={hourly.columns.tolist()}"
+    assert (hourly["hl_range"] > 0).all(), "hl_range should be positive"
+
+
+# 7. resample_to_hourly produces ret_intrabar column
+def test_hourly_has_ret_intrabar(ph):
+    m1 = _make_m1_ohlcv()
+    hourly = ph.resample_to_hourly(m1)
+    assert "ret_intrabar" in hourly.columns, f"missing ret_intrabar; cols={hourly.columns.tolist()}"
+
+
+# 8. hl_range = log(hourly_high / hourly_low)
+def test_hl_range_formula(ph):
+    # Two hours; second has known H=1.105, L=1.095
+    ts0 = pd.date_range("2020-01-06 00:00", periods=60, freq="min")
+    ts1 = pd.date_range("2020-01-06 01:00", periods=60, freq="min")
+    closes = np.full(120, 1.1)
+    highs  = np.full(120, 1.1)
+    lows   = np.full(120, 1.1)
+    # second hour: known spread
+    highs[60:] = 1.105
+    lows[60:]  = 1.095
+    m1 = pd.DataFrame({
+        "ts": np.concatenate([ts0, ts1]),
+        "open": closes, "high": highs, "low": lows, "close": closes,
+    })
+    hourly = ph.resample_to_hourly(m1)
+    assert len(hourly) >= 1
+    hl = hourly.iloc[-1]["hl_range"]
+    expected = float(np.log(1.105 / 1.095))
+    assert abs(hl - expected) < 1e-6, f"hl_range={hl:.8f}, expected={expected:.8f}"
+
+
+# 9. ret_intrabar = log(hourly_last_close / hourly_first_open)
+def test_ret_intrabar_formula(ph):
+    ts0 = pd.date_range("2020-01-06 00:00", periods=60, freq="min")
+    ts1 = pd.date_range("2020-01-06 01:00", periods=60, freq="min")
+    closes = np.full(120, 1.1)
+    opens  = np.full(120, 1.1)
+    # second hour: open=1.09, close=1.11
+    opens[60]   = 1.09
+    closes[119] = 1.11
+    m1 = pd.DataFrame({
+        "ts": np.concatenate([ts0, ts1]),
+        "open": opens, "high": closes + 0.001, "low": closes - 0.001, "close": closes,
+    })
+    hourly = ph.resample_to_hourly(m1)
+    assert len(hourly) >= 1
+    rib = hourly.iloc[-1]["ret_intrabar"]
+    expected = float(np.log(1.11 / 1.09))
+    assert abs(rib - expected) < 1e-6, f"ret_intrabar={rib:.8f}, expected={expected:.8f}"
+
+
+# 10. build() in train.py uses 4 feature channels when hl_range + ret_intrabar present
+def test_build_uses_4_channels(tmp_path):
+    import importlib.util, os
+    hourly_path = "data/processed/eurusd_hourly.parquet"
+    if not os.path.exists(hourly_path):
+        pytest.skip("eurusd_hourly.parquet not present")
+    df = pd.read_parquet(hourly_path)
+    if "hl_range" not in df.columns:
+        pytest.skip("eurusd_hourly.parquet lacks hl_range — rebuild first")
+    spec = importlib.util.spec_from_file_location("train_4ch", "train.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    (Xtr, _), _ = mod.build()
+    assert Xtr.shape[2] == 4, f"expected 4 channels, got {Xtr.shape[2]}"

train.py

@@ -154,7 +154,11 @@ def build():
     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)
+    # Use OHLCV-derived features when available; fall back to 2-channel
+    base_feats = ["ret", "realized_vol"]
+    extra_feats = [c for c in ["hl_range", "ret_intrabar"] if c in df.columns]
+    FEAT_COLS = base_feats + extra_feats
+    feats  = df[FEAT_COLS].to_numpy(np.float32)
     target = df["realized_vol"].to_numpy(np.float32)
     tr_idx = df.index[df["date"].dt.year <= 2021].tolist()
     te_idx = df.index[df["date"].dt.year >= 2022].tolist()
@@ -268,7 +272,9 @@ def main():
             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)
+        base2 = ["ret", "realized_vol"]
+        extra2 = [c for c in ["hl_range", "ret_intrabar"] if c in df2.columns]
+        feats2 = df2[base2 + extra2].to_numpy(np.float32)
         mu2 = feats2[tr_mask].mean(0); sd2 = feats2[tr_mask].std(0) + 1e-8
         fn2 = (feats2 - mu2) / sd2
         def _export_windows(year_mask):