diff --git a/scripts/prepare_hourly.py b/scripts/prepare_hourly.py index 8552da5..ff9b055 100644 --- a/scripts/prepare_hourly.py +++ b/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) diff --git a/tests/test_prepare_hourly.py b/tests/test_prepare_hourly.py index caba777..68a3881 100644 --- a/tests/test_prepare_hourly.py +++ b/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]}" diff --git a/train.py b/train.py index 519b20f..803c0e0 100644 --- a/train.py +++ b/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):