"""LOCKED data pipeline (toy) — agent must NOT edit (brain Phase-1 contract).

Parses histdata EUR/USD M1 zips → daily series with realized volatility (the
val_vol_r2 target = 1-day realized vol from intraday squared returns).
Output: data/processed/eurusd_daily.parquet  [date, close, ret, realized_vol].
"""
import glob
import os
import zipfile

import numpy as np
import pandas as pd

RAW = "data/raw"
OUT = "data/processed/eurusd_daily.parquet"


def load_m1() -> pd.DataFrame:
    frames = []
    for zp in sorted(glob.glob(os.path.join(RAW, "DAT_ASCII_EURUSD_M1_*.zip"))):
        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"]])
    out = pd.concat(frames).sort_values("ts").reset_index(drop=True)
    return out


def main():
    m1 = load_m1()
    m1["r"] = np.log(m1["close"]).diff()
    m1["day"] = m1["ts"].dt.normalize()
    daily = m1.groupby("day").agg(
        close=("close", "last"),
        realized_vol=("r", lambda x: np.sqrt(np.nansum(x.values ** 2))),
        n_min=("r", "count"),
    ).reset_index()
    daily = daily[daily["n_min"] > 60]  # drop thin days (holidays)
    daily["ret"] = np.log(daily["close"]).diff()
    daily = daily.dropna().reset_index(drop=True)
    os.makedirs(os.path.dirname(OUT), exist_ok=True)
    daily[["day", "close", "ret", "realized_vol"]].rename(columns={"day": "date"}).to_parquet(OUT)
    print("rows:", len(daily), "| dates:", daily["day"].min().date(), "→", daily["day"].max().date())
    # sanity: the COVID crash (March 2020) must show a realized-vol spike
    rv = daily.set_index("day")["realized_vol"]
    mar20 = rv["2020-03-01":"2020-03-31"].max()
    typ = rv["2019-01-01":"2019-12-31"].median()
    print("median 2019 RV: %.5f | max Mar-2020 RV: %.5f | spike x%.1f" % (typ, mar20, mar20 / typ))


if __name__ == "__main__":
    main()