jepa-fx-risk/tests/test_prepare_hourly.py

"""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):
    import zipfile, io
    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


# ── 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 2 feature channels (HPO: hl_range/ret_intrabar redundant)
def test_build_uses_2_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")
    spec = importlib.util.spec_from_file_location("train_2ch", "train.py")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    (Xtr, _), _ = mod.build()
    assert Xtr.shape[2] == 2, f"expected 2 channels, got {Xtr.shape[2]}"