"""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