generated from mathias/template-go-web
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d4b67943fb | ||
|
|
bd8962f997 | ||
|
|
b2bc01ba9e | ||
|
|
de19bfeada | ||
|
|
caccd1aa7b | ||
|
|
e635a641a4 | ||
|
|
48c7e3bd02 | ||
|
|
3aded95ef1 | ||
|
|
e739f84afd | ||
|
|
d282571c96 | ||
|
|
1a17a4c88e |
@@ -16,3 +16,7 @@
|
||||
| 3 | -0.0716 | +0.0487 | KEEP | 4s | gpu=0% vram=10054/12227MiB temp=36°C | iter3 |
|
||||
| 4 | 0.0590 | +0.1306 | KEEP | 5s | gpu=0% vram=10054/12227MiB temp=36°C | iter4 |
|
||||
| 5 | 0.0599 | +0.0009 | KEEP | 5s | gpu=0% vram=10054/12227MiB temp=37°C | iter5 |
|
||||
| 1 | 0.0563 | -0.0036 | revert | 5s | gpu=0% vram=10054/12227MiB temp=35°C | iter1 |
|
||||
| 2 | 0.0577 | -0.0022 | revert | 5s | gpu=0% vram=10054/12227MiB temp=36°C | iter2 |
|
||||
| 3 | 0.0563 | -0.0036 | revert | 5s | gpu=0% vram=10054/12227MiB temp=36°C | iter3 |
|
||||
| 4 | -0.1613 | -0.2212 | revert | 5s | gpu=0% vram=10054/12227MiB temp=37°C | iter4 |
|
||||
|
||||
+16
-1
@@ -34,9 +34,24 @@ tasks:
|
||||
data:prepare:all:
|
||||
desc: "Build both daily and hourly parquets"
|
||||
deps: [data:prepare:daily, data:prepare:hourly]
|
||||
train:multipair:
|
||||
desc: "Train 5-pair G10 HEPA (D=256, best config, phase1_r2≈0.44)"
|
||||
cmds: [JEPA_USE_MULTIPAIR=1 JEPA_D_MODEL=256 .venv/bin/python train.py]
|
||||
|
||||
data:fetch:multipair:
|
||||
desc: "Download G10 M1 data (GBPUSD/USDJPY/USDCHF/AUDUSD) 2008-2023 from histdata"
|
||||
cmds: [.venv/bin/python scripts/fetch_multipair.py]
|
||||
data:prepare:pair:
|
||||
desc: "Build {PAIR}_hourly.parquet from data/raw/{PAIR}/ (e.g. PAIR=gbpusd)"
|
||||
cmds: [PAIR={{.PAIR}} .venv/bin/python scripts/prepare_hourly.py {{.EXTRA_ARGS}}]
|
||||
vars:
|
||||
PAIR: '{{default "eurusd" .PAIR}}'
|
||||
data:prepare:multipair:
|
||||
desc: "Merge 5-pair hourly parquets into eurusd_multipair.parquet"
|
||||
cmds: [.venv/bin/python scripts/prepare_multipair.py]
|
||||
data:test:
|
||||
desc: "Run Python data pipeline tests"
|
||||
cmds: [.venv/bin/python -m pytest tests/test_prepare_hourly.py tests/test_hepa.py -v]
|
||||
cmds: [.venv/bin/python -m pytest tests/test_prepare_hourly.py tests/test_hepa.py tests/test_multipair.py -v]
|
||||
|
||||
eval:probe:
|
||||
desc: "Run linear-probe (val_vol_r2) on embeddings from metrics.json"
|
||||
|
||||
+8
-8
@@ -1,14 +1,14 @@
|
||||
{
|
||||
"val_vol_r2": 0.05988483092470609,
|
||||
"n_test": 263,
|
||||
"val_vol_r2": 0.3641397896593044,
|
||||
"phase1_r2": 0.3908407688140869,
|
||||
"n_test": 11641,
|
||||
"knobs": {
|
||||
"WINDOW": 60,
|
||||
"PATCH_LEN": 5,
|
||||
"STRIDE": 5,
|
||||
"D_MODEL": 64,
|
||||
"WINDOW": 120,
|
||||
"PATCH_LEN": 24,
|
||||
"D_MODEL": 128,
|
||||
"DEPTH": 2,
|
||||
"MASK_FRAC": 0.5,
|
||||
"SIGREG_LAM": 0.01,
|
||||
"ALPHA": 0.1,
|
||||
"DELTA_T_MAX": 3,
|
||||
"EPOCHS": 300
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
{"config": {"JEPA_D_MODEL": 64, "JEPA_DEPTH": 2, "JEPA_WINDOW": 120}, "val_vol_r2": 0.302411480667525, "phase1_r2": 0.35809940099716187, "stdout_last": "val_vol_r2 = 0.3024 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:37:37.028406"}
|
||||
{"config": {"JEPA_D_MODEL": 64, "JEPA_DEPTH": 2, "JEPA_WINDOW": 240}, "val_vol_r2": 0.29654798431244755, "phase1_r2": 0.35618388652801514, "stdout_last": "val_vol_r2 = 0.2965 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:37:49.822762"}
|
||||
{"config": {"JEPA_D_MODEL": 64, "JEPA_DEPTH": 2, "JEPA_WINDOW": 480}, "val_vol_r2": 0.31050360040290237, "phase1_r2": 0.36530405282974243, "stdout_last": "val_vol_r2 = 0.3105 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:38:02.966176"}
|
||||
{"config": {"JEPA_D_MODEL": 64, "JEPA_DEPTH": 4, "JEPA_WINDOW": 120}, "val_vol_r2": 0.2925057399716364, "phase1_r2": 0.3467639684677124, "stdout_last": "val_vol_r2 = 0.2925 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:38:16.195552"}
|
||||
{"config": {"JEPA_D_MODEL": 64, "JEPA_DEPTH": 4, "JEPA_WINDOW": 240}, "val_vol_r2": 0.29334667623516786, "phase1_r2": 0.35872191190719604, "stdout_last": "val_vol_r2 = 0.2933 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:38:31.372602"}
|
||||
{"config": {"JEPA_D_MODEL": 64, "JEPA_DEPTH": 4, "JEPA_WINDOW": 480}, "val_vol_r2": 0.3123527205416422, "phase1_r2": 0.3572431206703186, "stdout_last": "val_vol_r2 = 0.3124 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:38:46.672203"}
|
||||
{"config": {"JEPA_D_MODEL": 128, "JEPA_DEPTH": 2, "JEPA_WINDOW": 120}, "val_vol_r2": 0.3641397896593044, "phase1_r2": 0.3908407688140869, "stdout_last": "val_vol_r2 = 0.3641 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:39:00.793878"}
|
||||
{"config": {"JEPA_D_MODEL": 128, "JEPA_DEPTH": 2, "JEPA_WINDOW": 240}, "val_vol_r2": 0.35845865364171503, "phase1_r2": 0.3737195134162903, "stdout_last": "val_vol_r2 = 0.3585 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:39:13.321428"}
|
||||
{"config": {"JEPA_D_MODEL": 128, "JEPA_DEPTH": 2, "JEPA_WINDOW": 480}, "val_vol_r2": 0.35310115657814645, "phase1_r2": 0.35306859016418457, "stdout_last": "val_vol_r2 = 0.3531 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:39:26.402249"}
|
||||
{"config": {"JEPA_D_MODEL": 128, "JEPA_DEPTH": 4, "JEPA_WINDOW": 120}, "val_vol_r2": 0.3655629727960601, "phase1_r2": 0.371029257774353, "stdout_last": "val_vol_r2 = 0.3656 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:39:39.786248"}
|
||||
{"config": {"JEPA_D_MODEL": 128, "JEPA_DEPTH": 4, "JEPA_WINDOW": 240}, "val_vol_r2": 0.36109622605593217, "phase1_r2": 0.3666273355484009, "stdout_last": "val_vol_r2 = 0.3611 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:39:53.194111"}
|
||||
{"config": {"JEPA_D_MODEL": 128, "JEPA_DEPTH": 4, "JEPA_WINDOW": 480}, "val_vol_r2": 0.362228341965093, "phase1_r2": 0.3590735197067261, "stdout_last": "val_vol_r2 = 0.3622 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:40:06.991680"}
|
||||
{"config": {"JEPA_D_MODEL": 256, "JEPA_DEPTH": 2, "JEPA_WINDOW": 120}, "val_vol_r2": 0.3749483295047378, "phase1_r2": 0.3801569938659668, "stdout_last": "val_vol_r2 = 0.3749 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:40:21.512210"}
|
||||
{"config": {"JEPA_D_MODEL": 256, "JEPA_DEPTH": 2, "JEPA_WINDOW": 240}, "val_vol_r2": 0.3765593861479334, "phase1_r2": 0.38416117429733276, "stdout_last": "val_vol_r2 = 0.3766 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:40:34.959919"}
|
||||
{"config": {"JEPA_D_MODEL": 256, "JEPA_DEPTH": 2, "JEPA_WINDOW": 480}, "val_vol_r2": 0.3653399117639956, "phase1_r2": 0.3685130476951599, "stdout_last": "val_vol_r2 = 0.3653 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:40:48.806135"}
|
||||
{"config": {"JEPA_D_MODEL": 256, "JEPA_DEPTH": 4, "JEPA_WINDOW": 120}, "val_vol_r2": 0.375961424966925, "phase1_r2": 0.37861257791519165, "stdout_last": "val_vol_r2 = 0.3760 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:41:04.056939"}
|
||||
{"config": {"JEPA_D_MODEL": 256, "JEPA_DEPTH": 4, "JEPA_WINDOW": 240}, "val_vol_r2": 0.37841726893098504, "phase1_r2": 0.3781360387802124, "stdout_last": "val_vol_r2 = 0.3784 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:41:18.693263"}
|
||||
{"config": {"JEPA_D_MODEL": 256, "JEPA_DEPTH": 4, "JEPA_WINDOW": 480}, "val_vol_r2": 0.37118530199441635, "phase1_r2": 0.3651617765426636, "stdout_last": "val_vol_r2 = 0.3712 (n_test=11641, dev=cuda)", "ts": "2026-06-26T10:41:34.796157"}
|
||||
@@ -0,0 +1,48 @@
|
||||
"""Fetch G10 FX M1 data from histdata.com for all pairs except EURUSD (already fetched).
|
||||
|
||||
Each pair's zips go into data/raw/{pair}/ to avoid collisions.
|
||||
Output: data/raw/gbpusd/DAT_ASCII_GBPUSD_M1_YYYY.zip etc.
|
||||
|
||||
python scripts/fetch_multipair.py
|
||||
PAIRS=gbpusd,usdjpy YEARS=2020,2021 python scripts/fetch_multipair.py
|
||||
"""
|
||||
import os
|
||||
import time
|
||||
|
||||
from histdata import download_hist_data
|
||||
from histdata.api import Platform as P, TimeFrame as T
|
||||
|
||||
PAIRS_DEFAULT = ["gbpusd", "usdjpy", "usdchf", "audusd"]
|
||||
YEARS_DEFAULT = list(range(2008, 2024))
|
||||
|
||||
|
||||
def main():
|
||||
pairs_env = os.environ.get("PAIRS", "")
|
||||
pairs = [p.strip() for p in pairs_env.split(",")] if pairs_env else PAIRS_DEFAULT
|
||||
|
||||
years_env = os.environ.get("YEARS", "")
|
||||
years = [int(y.strip()) for y in years_env.split(",")] if years_env else YEARS_DEFAULT
|
||||
|
||||
for pair in pairs:
|
||||
out_dir = f"data/raw/{pair}"
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
print(f"\n=== {pair.upper()} ===")
|
||||
for yr in years:
|
||||
out_path = os.path.join(out_dir, f"DAT_ASCII_{pair.upper()}_M1_{yr}.zip")
|
||||
if os.path.exists(out_path):
|
||||
print(f" {yr} already present, skip")
|
||||
continue
|
||||
try:
|
||||
f = download_hist_data(
|
||||
year=str(yr), month=None, pair=pair,
|
||||
platform=P.GENERIC_ASCII, time_frame=T.ONE_MINUTE,
|
||||
output_directory=out_dir,
|
||||
)
|
||||
print(f" fetched {yr} → {f}")
|
||||
except Exception as e:
|
||||
print(f" {yr} FAILED: {e}")
|
||||
time.sleep(2)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,97 @@
|
||||
"""HPO sweep for jepa-fx-risk HEPA backbone.
|
||||
|
||||
Runs train.py with different JEPA_* env overrides, logs results to
|
||||
results/hpo/hpo_results.jsonl. Each config writes its metrics.json then
|
||||
the result is appended to the JSONL.
|
||||
|
||||
Usage:
|
||||
python scripts/hpo_sweep.py
|
||||
python scripts/hpo_sweep.py --dry-run # print configs, don't train
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from itertools import product
|
||||
from pathlib import Path
|
||||
|
||||
# ── Search space ──────────────────────────────────────────────────────────────
|
||||
SEARCH_SPACE = {
|
||||
"JEPA_D_MODEL": [64, 128, 256],
|
||||
"JEPA_DEPTH": [2, 4],
|
||||
"JEPA_WINDOW": [120, 240, 480],
|
||||
}
|
||||
# Fixed: PATCH_LEN=24 (1-day patches), N_HEADS=4, EPOCHS=300, PHASE1_EPOCHS=200
|
||||
|
||||
PYTHON = str(Path(sys.executable))
|
||||
OUT_DIR = Path("results/hpo")
|
||||
|
||||
|
||||
def configs():
|
||||
"""Yield all configs as dicts of JEPA_* env overrides."""
|
||||
keys = list(SEARCH_SPACE.keys())
|
||||
for vals in product(*SEARCH_SPACE.values()):
|
||||
yield dict(zip(keys, vals))
|
||||
|
||||
|
||||
def run_config(cfg: dict, metrics_path: str = "metrics.json") -> dict:
|
||||
env = {**os.environ, **{k: str(v) for k, v in cfg.items()}}
|
||||
result = subprocess.run(
|
||||
[PYTHON, "train.py"],
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return {"config": cfg, "error": result.stderr[-500:]}
|
||||
stdout_last = result.stdout.strip().split("\n")[-1]
|
||||
with open(metrics_path) as f:
|
||||
m = json.load(f)
|
||||
return {
|
||||
"config": cfg,
|
||||
"val_vol_r2": m.get("val_vol_r2"),
|
||||
"phase1_r2": m.get("phase1_r2"),
|
||||
"stdout_last": stdout_last,
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--dry-run", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
out_file = OUT_DIR / "hpo_results.jsonl"
|
||||
|
||||
all_cfgs = list(configs())
|
||||
print(f"HPO sweep: {len(all_cfgs)} configs")
|
||||
for i, cfg in enumerate(all_cfgs):
|
||||
label = " ".join(f"{k.replace('JEPA_','')}={v}" for k, v in cfg.items())
|
||||
print(f"\n[{i+1}/{len(all_cfgs)}] {label}")
|
||||
if args.dry_run:
|
||||
continue
|
||||
ts = datetime.utcnow().isoformat()
|
||||
row = run_config(cfg)
|
||||
row["ts"] = ts
|
||||
with open(out_file, "a") as f:
|
||||
f.write(json.dumps(row) + "\n")
|
||||
if "error" in row:
|
||||
print(f" ERROR: {row['error'][:200]}")
|
||||
else:
|
||||
print(f" val_vol_r2={row['val_vol_r2']:.4f} phase1_r2={row['phase1_r2']:.4f}")
|
||||
|
||||
if not args.dry_run:
|
||||
# Print leaderboard
|
||||
rows = [json.loads(l) for l in open(out_file) if l.strip()]
|
||||
rows = [r for r in rows if "error" not in r]
|
||||
rows.sort(key=lambda r: r.get("phase1_r2", -999), reverse=True)
|
||||
print("\n── Leaderboard (by phase1_r2) ─────────────────────────")
|
||||
for r in rows[:5]:
|
||||
cfg_str = " ".join(f"{k.replace('JEPA_','')}={v}" for k,v in r["config"].items())
|
||||
print(f" {r['phase1_r2']:.4f} {cfg_str}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+31
-11
@@ -18,8 +18,9 @@ import zipfile
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
PAIR = os.environ.get("PAIR", "EURUSD").upper()
|
||||
RAW_DEFAULT = "data/raw"
|
||||
OUT_DEFAULT = "data/processed/eurusd_hourly.parquet"
|
||||
OUT_DEFAULT = f"data/processed/{PAIR.lower()}_hourly.parquet"
|
||||
MIN_BARS = 30 # drop hours thinner than this (holidays, DST boundary artefacts)
|
||||
|
||||
|
||||
@@ -29,32 +30,51 @@ 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(
|
||||
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))),
|
||||
realized_vol = ("log_r", lambda x: np.sqrt(np.nansum(x.values ** 2))),
|
||||
n_bars = ("log_r", "count"),
|
||||
).reset_index()
|
||||
)
|
||||
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:
|
||||
def load_m1_from_zips(raw_dir: str, pair: str = None) -> pd.DataFrame:
|
||||
"""Load and concatenate all M1 zips from raw_dir (histdata format)."""
|
||||
pattern = os.path.join(raw_dir, "DAT_ASCII_EURUSD_M1_*.zip")
|
||||
p = (pair or PAIR).upper()
|
||||
pattern = os.path.join(raw_dir, f"DAT_ASCII_{p}_M1_*.zip")
|
||||
zips = sorted(glob.glob(pattern))
|
||||
if not zips:
|
||||
raise FileNotFoundError(f"No M1 zips found at {pattern}")
|
||||
@@ -68,7 +88,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)
|
||||
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Merge per-pair hourly parquets into a single wide multipair parquet.
|
||||
|
||||
Each pair contributes two features: {pair}_ret and {pair}_rv (realized vol).
|
||||
The merge is an INNER JOIN on datetime — only hours present in ALL pairs are kept.
|
||||
The target for train.py remains eurusd_rv.
|
||||
|
||||
Output: data/processed/eurusd_multipair.parquet
|
||||
|
||||
python scripts/prepare_multipair.py
|
||||
PROCESSED=data/processed python scripts/prepare_multipair.py
|
||||
"""
|
||||
import os
|
||||
import pandas as pd
|
||||
|
||||
PAIRS = ["eurusd", "gbpusd", "usdjpy", "usdchf", "audusd"]
|
||||
|
||||
PROCESSED_DEFAULT = "data/processed"
|
||||
OUT_DEFAULT = "data/processed/eurusd_multipair.parquet"
|
||||
|
||||
|
||||
def merge_pair_parquets(pair_dfs: dict) -> pd.DataFrame:
|
||||
"""Inner-join hourly DataFrames from multiple pairs on datetime.
|
||||
|
||||
Args:
|
||||
pair_dfs: dict mapping pair name (e.g. "eurusd") to hourly DataFrame
|
||||
with columns [datetime, close, ret, realized_vol, ...].
|
||||
Returns:
|
||||
Wide DataFrame with columns:
|
||||
datetime, {pair}_ret, {pair}_rv for each pair.
|
||||
"""
|
||||
merged = None
|
||||
for pair, df in pair_dfs.items():
|
||||
sub = df[["datetime", "ret", "realized_vol"]].copy()
|
||||
sub = sub.rename(columns={"ret": f"{pair}_ret", "realized_vol": f"{pair}_rv"})
|
||||
sub = sub.set_index("datetime")
|
||||
if merged is None:
|
||||
merged = sub
|
||||
else:
|
||||
merged = merged.join(sub, how="inner")
|
||||
|
||||
return merged.reset_index()
|
||||
|
||||
|
||||
def build_multipair_parquet(
|
||||
processed_dir: str = PROCESSED_DEFAULT,
|
||||
out_path: str = OUT_DEFAULT,
|
||||
pairs: list = None,
|
||||
) -> None:
|
||||
if pairs is None:
|
||||
pairs = PAIRS
|
||||
pair_dfs = {}
|
||||
for pair in pairs:
|
||||
path = os.path.join(processed_dir, f"{pair}_hourly.parquet")
|
||||
if not os.path.exists(path):
|
||||
raise FileNotFoundError(
|
||||
f"{pair}_hourly.parquet not found at {path} — run prepare_hourly.py for this pair first"
|
||||
)
|
||||
df = pd.read_parquet(path)
|
||||
pair_dfs[pair] = df
|
||||
|
||||
merged = merge_pair_parquets(pair_dfs)
|
||||
merged.to_parquet(out_path, index=False)
|
||||
n_pairs = len(pairs)
|
||||
n_ch = n_pairs * 2
|
||||
print(f"Multipair parquet: {len(merged):,} rows × {n_ch} feature channels ({n_pairs} pairs)")
|
||||
print(f"Date range: {merged['datetime'].min()} → {merged['datetime'].max()}")
|
||||
print(f"Written: {out_path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
processed_dir = os.environ.get("PROCESSED", PROCESSED_DEFAULT)
|
||||
build_multipair_parquet(processed_dir=processed_dir)
|
||||
+169
-4
@@ -1,9 +1,10 @@
|
||||
"""Failing tests for HEPA backbone in train.py.
|
||||
"""Failing tests for HEPA backbone + Phase-1 supervised head + HPO in train.py.
|
||||
|
||||
Run: cd ~/dev/AI/jepa-fx-risk && .venv/bin/python -m pytest tests/test_hepa.py -v
|
||||
These tests define what the new backbone must satisfy BEFORE implementation.
|
||||
These tests define what the backbone and head must satisfy BEFORE implementation.
|
||||
"""
|
||||
import math
|
||||
import os
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
import pytest
|
||||
@@ -12,11 +13,24 @@ import pytest
|
||||
# They will fail until train.py implements: CausalEncoder, HorizonPredictor, vicreg_loss
|
||||
|
||||
|
||||
def _import():
|
||||
def _import(env_overrides=None):
|
||||
import importlib.util, sys
|
||||
spec = importlib.util.spec_from_file_location("train", "train.py")
|
||||
saved = {}
|
||||
if env_overrides:
|
||||
for k, v in env_overrides.items():
|
||||
saved[k] = os.environ.get(k)
|
||||
os.environ[k] = str(v)
|
||||
# Force fresh module load (env vars must be read at import time)
|
||||
name = f"train_{id(env_overrides)}"
|
||||
spec = importlib.util.spec_from_file_location(name, "train.py")
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
if env_overrides:
|
||||
for k, orig in saved.items():
|
||||
if orig is None:
|
||||
os.environ.pop(k, None)
|
||||
else:
|
||||
os.environ[k] = orig
|
||||
return mod
|
||||
|
||||
|
||||
@@ -110,3 +124,154 @@ def test_build_hourly_more_windows(train_mod):
|
||||
(Xtr, _), _ = train_mod.build()
|
||||
# Daily had ~877 train windows; hourly with 2008-2021 should have > 50,000
|
||||
assert len(Xtr) > 10_000, f"expected >10k hourly train windows, got {len(Xtr)}"
|
||||
|
||||
|
||||
# ── Phase-1: supervised head ──────────────────────────────────────────────────
|
||||
|
||||
# 8. SupervisedHead exists and maps (B, D) → (B,)
|
||||
def test_supervised_head_shape(train_mod):
|
||||
D = 128
|
||||
head = train_mod.SupervisedHead(D)
|
||||
x = torch.randn(16, D)
|
||||
out = head(x)
|
||||
assert out.shape == (16,), f"expected (16,), got {out.shape}"
|
||||
|
||||
|
||||
# 9. SupervisedHead gradient flows (not frozen)
|
||||
def test_supervised_head_backward(train_mod):
|
||||
head = train_mod.SupervisedHead(64)
|
||||
x = torch.randn(8, 64)
|
||||
loss = head(x).mean()
|
||||
loss.backward()
|
||||
for name, p in head.named_parameters():
|
||||
assert p.grad is not None, f"no grad on {name}"
|
||||
|
||||
|
||||
# 10. Phase-1 beats linear on nonlinear synthetic signal
|
||||
def test_phase1_beats_linear_on_nonlinear(train_mod):
|
||||
"""MLP head should outperform ridge regression on data with nonlinear structure."""
|
||||
import numpy as np
|
||||
torch.manual_seed(0); np.random.seed(0)
|
||||
N, D = 1000, 32
|
||||
# target = |h|² (quadratic — linear can't fit well)
|
||||
Etr = np.random.randn(N, D).astype(np.float32)
|
||||
ytr = (Etr ** 2).sum(axis=1)
|
||||
Ete = np.random.randn(200, D).astype(np.float32)
|
||||
yte = (Ete ** 2).sum(axis=1)
|
||||
|
||||
# Ridge baseline
|
||||
A = np.hstack([Etr, np.ones((N, 1))])
|
||||
w = np.linalg.solve(A.T @ A + 1e-3 * np.eye(A.shape[1]), A.T @ ytr)
|
||||
pred_lin = np.hstack([Ete, np.ones((200, 1))]) @ w
|
||||
r2_lin = float(1 - ((yte - pred_lin) ** 2).sum() / ((yte - yte.mean()) ** 2).sum())
|
||||
|
||||
# MLP head
|
||||
head = train_mod.SupervisedHead(D)
|
||||
opt = torch.optim.Adam(head.parameters(), lr=1e-2)
|
||||
Xtr_t = torch.tensor(Etr); ytr_t = torch.tensor(ytr)
|
||||
for _ in range(300):
|
||||
loss = nn.functional.mse_loss(head(Xtr_t), ytr_t)
|
||||
opt.zero_grad(); loss.backward(); opt.step()
|
||||
|
||||
head.eval()
|
||||
with torch.no_grad():
|
||||
pred_mlp = head(torch.tensor(Ete)).numpy()
|
||||
r2_mlp = float(1 - ((yte - pred_mlp) ** 2).sum() / ((yte - yte.mean()) ** 2).sum())
|
||||
|
||||
assert r2_mlp > r2_lin + 0.05, (
|
||||
f"MLP R²={r2_mlp:.3f} should beat ridge R²={r2_lin:.3f} by >0.05 on quadratic target"
|
||||
)
|
||||
|
||||
|
||||
# 11. main() returns phase1_r2 in metrics.json (integration — needs real data)
|
||||
def test_metrics_json_has_phase1_r2(train_mod):
|
||||
import json
|
||||
if not os.path.exists("metrics.json"):
|
||||
pytest.skip("metrics.json not present — run train.py first")
|
||||
with open("metrics.json") as f:
|
||||
m = json.load(f)
|
||||
assert "phase1_r2" in m, f"phase1_r2 missing from metrics.json: {list(m.keys())}"
|
||||
assert m["phase1_r2"] > m["val_vol_r2"], (
|
||||
f"MLP head phase1_r2={m['phase1_r2']:.4f} should beat linear probe "
|
||||
f"val_vol_r2={m['val_vol_r2']:.4f}"
|
||||
)
|
||||
|
||||
|
||||
# ── HPO: env-var knob overrides ───────────────────────────────────────────────
|
||||
|
||||
# 12. JEPA_WINDOW env var overrides WINDOW at import time
|
||||
def test_env_override_window():
|
||||
mod = _import({"JEPA_WINDOW": "48"})
|
||||
assert mod.WINDOW == 48, f"expected WINDOW=48, got {mod.WINDOW}"
|
||||
|
||||
|
||||
# 13. JEPA_D_MODEL and JEPA_DEPTH env vars work
|
||||
def test_env_override_d_model_depth():
|
||||
mod = _import({"JEPA_D_MODEL": "64", "JEPA_DEPTH": "4"})
|
||||
assert mod.D_MODEL == 64, f"expected D_MODEL=64, got {mod.D_MODEL}"
|
||||
assert mod.DEPTH == 4, f"expected DEPTH=4, got {mod.DEPTH}"
|
||||
|
||||
|
||||
# 14. hpo_sweep.py exists and generates correct config list
|
||||
def test_hpo_sweep_configs():
|
||||
import importlib.util
|
||||
sweep_path = "scripts/hpo_sweep.py"
|
||||
if not os.path.exists(sweep_path):
|
||||
pytest.fail(f"{sweep_path} not found — implement it")
|
||||
spec = importlib.util.spec_from_file_location("hpo_sweep", sweep_path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
cfgs = list(mod.configs())
|
||||
assert len(cfgs) > 0, "configs() returned empty list"
|
||||
# Every config must have at least D_MODEL, DEPTH, WINDOW keys
|
||||
required = {"JEPA_D_MODEL", "JEPA_DEPTH", "JEPA_WINDOW"}
|
||||
for cfg in cfgs:
|
||||
assert required.issubset(cfg.keys()), f"config missing required keys: {cfg}"
|
||||
|
||||
|
||||
# ── Option B: joint encoder fine-tuning in phase-1 ───────────────────────────
|
||||
|
||||
# 15. PHASE1_JOINT and PHASE1_ENCODER_LR knobs exist at module level
|
||||
def test_joint_phase1_knobs():
|
||||
mod = _import({"JEPA_PHASE1_JOINT": "1", "JEPA_PHASE1_ENCODER_LR": "1e-5"})
|
||||
assert hasattr(mod, "PHASE1_JOINT"), "PHASE1_JOINT knob missing from train.py"
|
||||
assert hasattr(mod, "PHASE1_ENCODER_LR"), "PHASE1_ENCODER_LR knob missing from train.py"
|
||||
assert mod.PHASE1_JOINT is True
|
||||
assert abs(mod.PHASE1_ENCODER_LR - 1e-5) < 1e-12
|
||||
|
||||
|
||||
# 16. PHASE1_JOINT defaults to True (joint mode on by default)
|
||||
def test_joint_phase1_default_on():
|
||||
mod = _import()
|
||||
assert hasattr(mod, "PHASE1_JOINT"), "PHASE1_JOINT knob missing"
|
||||
assert mod.PHASE1_JOINT is True, f"PHASE1_JOINT default should be True, got {mod.PHASE1_JOINT}"
|
||||
|
||||
|
||||
# 17. JEPA_PHASE1_JOINT=0 disables joint (env override works)
|
||||
def test_joint_phase1_can_disable():
|
||||
mod = _import({"JEPA_PHASE1_JOINT": "0"})
|
||||
assert mod.PHASE1_JOINT is False, f"expected False, got {mod.PHASE1_JOINT}"
|
||||
|
||||
|
||||
# 18. Encoder receives non-zero gradients when joint-training with the head
|
||||
def test_joint_encoder_grad_flows(train_mod):
|
||||
"""Gradient must flow into encoder when using two-param-group joint optimizer."""
|
||||
import torch.nn.functional as F
|
||||
enc = train_mod.CausalEncoder(n_channels=2, patch_len=8, d_model=16, n_heads=2, depth=1)
|
||||
head = train_mod.SupervisedHead(16)
|
||||
enc.train(); head.train()
|
||||
opt = torch.optim.Adam([
|
||||
{"params": head.parameters(), "lr": 1e-3},
|
||||
{"params": enc.parameters(), "lr": 1e-5},
|
||||
], weight_decay=1e-4)
|
||||
# Tiny batch: 4 windows of length 16 (= 2 patches of patch_len=8)
|
||||
X = torch.randn(4, 16, 2)
|
||||
y = torch.randn(4)
|
||||
tokens = enc(X) # (4, 2, 16)
|
||||
h = tokens[:, -1, :] # (4, 16) — last token
|
||||
pred = head(h)
|
||||
loss = F.mse_loss(pred, y)
|
||||
loss.backward()
|
||||
enc_grads = [p.grad for p in enc.parameters() if p.grad is not None]
|
||||
assert len(enc_grads) > 0, "no encoder params received gradients"
|
||||
assert any(g.abs().max().item() > 0 for g in enc_grads), "all encoder grads are zero"
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
"""Tests for multi-pair G10 pipeline (Option C).
|
||||
|
||||
Tests the prepare_multipair.py merge logic and train.py multipair build().
|
||||
Run: cd ~/dev/AI/jepa-fx-risk && .venv/bin/python -m pytest tests/test_multipair.py -v
|
||||
"""
|
||||
import importlib.util
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
import pytest
|
||||
import os
|
||||
|
||||
|
||||
def _import_mp():
|
||||
spec = importlib.util.spec_from_file_location("prepare_multipair", "scripts/prepare_multipair.py")
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def mp():
|
||||
return _import_mp()
|
||||
|
||||
|
||||
def _pair_df(start: str, n_hours: int, seed: int) -> pd.DataFrame:
|
||||
"""Synthetic single-pair hourly parquet (same schema as prepare_hourly output)."""
|
||||
rng = np.random.default_rng(seed)
|
||||
dts = pd.date_range(start, periods=n_hours, freq="h")
|
||||
closes = 1.1 + np.cumsum(rng.normal(0, 0.001, n_hours))
|
||||
return pd.DataFrame({
|
||||
"datetime": dts,
|
||||
"close": closes,
|
||||
"ret": rng.normal(0, 0.001, n_hours),
|
||||
"realized_vol": np.abs(rng.normal(0.0005, 0.0001, n_hours)),
|
||||
})
|
||||
|
||||
|
||||
# 1. merge_pair_parquets returns inner join on datetime
|
||||
def test_merge_inner_join(mp):
|
||||
eur = _pair_df("2020-01-01 00:00", 100, seed=1) # t0 to t0+99h
|
||||
gbp = _pair_df("2020-01-01 20:00", 60, seed=2) # t0+20 to t0+79h → 60 common
|
||||
result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
|
||||
assert len(result) == 60, f"expected 60 (inner join), got {len(result)}"
|
||||
|
||||
|
||||
# 2. merge_pair_parquets prefixes columns with pair name
|
||||
def test_merge_column_prefixes(mp):
|
||||
eur = _pair_df("2020-01-01 00:00", 50, seed=1)
|
||||
gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
|
||||
result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
|
||||
assert "datetime" in result.columns, "datetime column missing"
|
||||
assert "eurusd_ret" in result.columns
|
||||
assert "eurusd_rv" in result.columns
|
||||
assert "gbpusd_ret" in result.columns
|
||||
assert "gbpusd_rv" in result.columns
|
||||
# raw pair columns should not leak through unprefixed
|
||||
assert "ret" not in result.columns
|
||||
assert "realized_vol" not in result.columns
|
||||
|
||||
|
||||
# 3. No NaN in merged output
|
||||
def test_merge_no_nan(mp):
|
||||
eur = _pair_df("2020-01-01 00:00", 50, seed=1)
|
||||
gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
|
||||
result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
|
||||
nan_count = result.isnull().sum().sum()
|
||||
assert nan_count == 0, f"{nan_count} NaN values in merged output"
|
||||
|
||||
|
||||
# 4. PAIRS constant is a non-empty list starting with eurusd
|
||||
def test_pairs_constant(mp):
|
||||
assert hasattr(mp, "PAIRS"), "PAIRS constant missing from prepare_multipair.py"
|
||||
assert len(mp.PAIRS) >= 2, "PAIRS must have at least 2 pairs"
|
||||
assert mp.PAIRS[0] == "eurusd", "first pair must be eurusd (target pair)"
|
||||
|
||||
|
||||
# 5. merge target column is eurusd_rv (for build() target selection)
|
||||
def test_merge_has_eurusd_rv_as_target(mp):
|
||||
eur = _pair_df("2020-01-01 00:00", 50, seed=1)
|
||||
gbp = _pair_df("2020-01-01 00:00", 50, seed=2)
|
||||
result = mp.merge_pair_parquets({"eurusd": eur, "gbpusd": gbp})
|
||||
assert "eurusd_rv" in result.columns, "eurusd_rv (target) missing from merged output"
|
||||
assert (result["eurusd_rv"] > 0).all(), "eurusd_rv should be positive"
|
||||
|
||||
|
||||
# 6. train.py recognises JEPA_USE_MULTIPAIR env var
|
||||
def test_use_multipair_knob():
|
||||
import importlib.util as ilu
|
||||
spec = ilu.spec_from_file_location(f"train_mp_{id(None)}", "train.py")
|
||||
mod = ilu.module_from_spec(spec)
|
||||
saved = os.environ.get("JEPA_USE_MULTIPAIR")
|
||||
os.environ["JEPA_USE_MULTIPAIR"] = "1"
|
||||
try:
|
||||
spec.loader.exec_module(mod)
|
||||
finally:
|
||||
if saved is None:
|
||||
os.environ.pop("JEPA_USE_MULTIPAIR", None)
|
||||
else:
|
||||
os.environ["JEPA_USE_MULTIPAIR"] = saved
|
||||
assert hasattr(mod, "USE_MULTIPAIR"), "USE_MULTIPAIR knob missing from train.py"
|
||||
assert mod.USE_MULTIPAIR is True
|
||||
|
||||
|
||||
# 7. build() uses n_pairs*2 channels when multipair parquet present
|
||||
def test_build_uses_multipair_channels():
|
||||
import importlib.util as ilu
|
||||
multipair_path = "data/processed/eurusd_multipair.parquet"
|
||||
if not os.path.exists(multipair_path):
|
||||
pytest.skip("eurusd_multipair.parquet not present — run data:prepare:multipair first")
|
||||
saved = os.environ.get("JEPA_USE_MULTIPAIR")
|
||||
os.environ["JEPA_USE_MULTIPAIR"] = "1"
|
||||
try:
|
||||
spec = ilu.spec_from_file_location(f"train_mp2_{id(None)}", "train.py")
|
||||
mod = ilu.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
(Xtr, _), _ = mod.build()
|
||||
finally:
|
||||
if saved is None:
|
||||
os.environ.pop("JEPA_USE_MULTIPAIR", None)
|
||||
else:
|
||||
os.environ["JEPA_USE_MULTIPAIR"] = saved
|
||||
mp = _import_mp()
|
||||
expected_ch = len(mp.PAIRS) * 2
|
||||
assert Xtr.shape[2] == expected_ch, (
|
||||
f"expected {expected_ch} channels (n_pairs={len(mp.PAIRS)}×2), got {Xtr.shape[2]}"
|
||||
)
|
||||
@@ -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,86 @@ 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 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]}"
|
||||
|
||||
@@ -17,19 +17,26 @@ import torch
|
||||
import torch.nn as nn
|
||||
import torch.nn.functional as F
|
||||
|
||||
# --- agent-tunable knobs ---
|
||||
USE_HOURLY = True # prefer eurusd_hourly.parquet when available
|
||||
WINDOW = 240 # hourly: 10 trading days; if USE_HOURLY=False reset to 60
|
||||
PATCH_LEN = 24 # hourly: 1-day patches (10 tokens); if USE_HOURLY=False reset to 10
|
||||
D_MODEL = 128
|
||||
DEPTH = 2
|
||||
N_HEADS = 4
|
||||
ALPHA = 0.1 # VICReg mixing weight (fixed at 0.1 in HEPA paper)
|
||||
DELTA_T_MAX = 3 # max prediction horizon in patches (1..min(DELTA_T_MAX, N-1-c))
|
||||
BATCH_SIZE = 512 # mini-batch per step (hourly dataset is too large for full-batch)
|
||||
EPOCHS = 300
|
||||
LR = 3e-4
|
||||
SEED = 0
|
||||
# --- agent-tunable knobs (all overridable via JEPA_* env vars for HPO) ---
|
||||
import os as _os
|
||||
USE_HOURLY = True
|
||||
WINDOW = int(_os.environ.get("JEPA_WINDOW", 120)) # HPO winner: 5-day context
|
||||
PATCH_LEN = int(_os.environ.get("JEPA_PATCH_LEN", 24))
|
||||
D_MODEL = int(_os.environ.get("JEPA_D_MODEL", 128))
|
||||
DEPTH = int(_os.environ.get("JEPA_DEPTH", 2))
|
||||
N_HEADS = int(_os.environ.get("JEPA_N_HEADS", 4))
|
||||
ALPHA = float(_os.environ.get("JEPA_ALPHA", 0.1))
|
||||
DELTA_T_MAX = int(_os.environ.get("JEPA_DELTA_T_MAX", 3))
|
||||
BATCH_SIZE = int(_os.environ.get("JEPA_BATCH_SIZE", 512))
|
||||
EPOCHS = int(_os.environ.get("JEPA_EPOCHS", 300))
|
||||
LR = float(_os.environ.get("JEPA_LR", 3e-4))
|
||||
PHASE1_EPOCHS = int(_os.environ.get("JEPA_PHASE1_EPOCHS", 200))
|
||||
PHASE1_LR = float(_os.environ.get("JEPA_PHASE1_LR", 1e-3))
|
||||
PHASE1_JOINT = bool(int(_os.environ.get("JEPA_PHASE1_JOINT", 1)))
|
||||
PHASE1_JOINT_EPOCHS= int(_os.environ.get("JEPA_PHASE1_JOINT_EPOCHS", 30))
|
||||
PHASE1_ENCODER_LR = float(_os.environ.get("JEPA_PHASE1_ENCODER_LR", 3e-6))
|
||||
USE_MULTIPAIR = bool(int(_os.environ.get("JEPA_USE_MULTIPAIR", 0)))
|
||||
SEED = int(_os.environ.get("JEPA_SEED", 0))
|
||||
# ---------------------------
|
||||
|
||||
torch.manual_seed(SEED)
|
||||
@@ -119,6 +126,21 @@ class HorizonPredictor(nn.Module):
|
||||
return self.net(torch.cat([h, dt], dim=-1))
|
||||
|
||||
|
||||
# ── Phase-1 supervised head ──────────────────────────────────────────────────
|
||||
|
||||
class SupervisedHead(nn.Module):
|
||||
"""Small MLP trained on frozen HEPA embeddings to predict next-period realized vol."""
|
||||
def __init__(self, d_model: int):
|
||||
super().__init__()
|
||||
self.net = nn.Sequential(
|
||||
nn.Linear(d_model, d_model // 2), nn.GELU(),
|
||||
nn.Linear(d_model // 2, 1),
|
||||
)
|
||||
|
||||
def forward(self, h: torch.Tensor) -> torch.Tensor:
|
||||
return self.net(h).squeeze(-1)
|
||||
|
||||
|
||||
# ── Data ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
def build():
|
||||
@@ -128,16 +150,29 @@ def build():
|
||||
falls back to eurusd_daily.parquet otherwise.
|
||||
"""
|
||||
import os
|
||||
multipair_path = "data/processed/eurusd_multipair.parquet"
|
||||
hourly_path = "data/processed/eurusd_hourly.parquet"
|
||||
daily_path = "data/processed/eurusd_daily.parquet"
|
||||
if USE_HOURLY and os.path.exists(hourly_path):
|
||||
if USE_MULTIPAIR and os.path.exists(multipair_path):
|
||||
df = pd.read_parquet(multipair_path).reset_index(drop=True)
|
||||
df["date"] = pd.to_datetime(df["datetime"])
|
||||
# All {pair}_ret + {pair}_rv columns as features; eurusd_rv as target
|
||||
feat_cols = [c for c in df.columns if c.endswith("_ret") or c.endswith("_rv")]
|
||||
FEAT_COLS = feat_cols
|
||||
target_col = "eurusd_rv"
|
||||
elif USE_HOURLY and os.path.exists(hourly_path):
|
||||
df = pd.read_parquet(hourly_path).reset_index(drop=True)
|
||||
df["date"] = pd.to_datetime(df["datetime"])
|
||||
# 2-channel default (HPO: adding hl_range+ret_intrabar hurt — correlated with base feats)
|
||||
FEAT_COLS = ["ret", "realized_vol"]
|
||||
target_col = "realized_vol"
|
||||
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)
|
||||
target = df["realized_vol"].to_numpy(np.float32)
|
||||
FEAT_COLS = ["ret", "realized_vol"]
|
||||
target_col = "realized_vol"
|
||||
feats = df[FEAT_COLS].to_numpy(np.float32)
|
||||
target = df[target_col].to_numpy(np.float32)
|
||||
tr_idx = df.index[df["date"].dt.year <= 2021].tolist()
|
||||
te_idx = df.index[df["date"].dt.year >= 2022].tolist()
|
||||
mu = feats[:tr_idx[-1]+1].mean(0)
|
||||
@@ -204,8 +239,67 @@ def main():
|
||||
ss_tot = ((yte - yte.mean()) ** 2).sum()
|
||||
val_vol_r2 = float(1 - ss_res / ss_tot)
|
||||
|
||||
# Phase-1: MLP supervised head — joint or frozen-encoder path
|
||||
ytr_mu = float(ytr.mean()); ytr_sd = float(ytr.std()) + 1e-8
|
||||
ytr_z = (ytr - ytr_mu) / ytr_sd
|
||||
head = SupervisedHead(D_MODEL).to(dev)
|
||||
p1_bs = min(BATCH_SIZE, len(Etr_n))
|
||||
|
||||
# Shared tensors for the frozen-head warmup (used by both paths)
|
||||
Etr_t = torch.tensor(Etr_n, device=dev)
|
||||
ytr_z_t = torch.tensor(ytr_z, device=dev)
|
||||
Ete_t = torch.tensor(Ete_n, device=dev)
|
||||
N_tr_h = len(Etr_t)
|
||||
|
||||
# Phase 1a: warm up head on frozen embeddings (both paths run this)
|
||||
head_opt = torch.optim.Adam(head.parameters(), lr=PHASE1_LR, weight_decay=1e-4)
|
||||
for _ in range(PHASE1_EPOCHS):
|
||||
perm = torch.randperm(N_tr_h, device=dev)
|
||||
for start in range(0, N_tr_h, p1_bs):
|
||||
idx_h = perm[start:start + p1_bs]
|
||||
loss_h = F.mse_loss(head(Etr_t[idx_h]), ytr_z_t[idx_h])
|
||||
head_opt.zero_grad(); loss_h.backward(); head_opt.step()
|
||||
|
||||
if PHASE1_JOINT:
|
||||
# Phase 1b: short joint fine-tuning — encoder nudged with tiny LR.
|
||||
# Normalize live encoder output with FROZEN stats (mu_e, sd_e) so the
|
||||
# head sees the same embedding distribution it was warmed up on.
|
||||
enc.train()
|
||||
mu_e_t = torch.tensor(mu_e, device=dev)
|
||||
sd_e_t = torch.tensor(sd_e, device=dev)
|
||||
Xtr_t = torch.tensor(Xtr, device=dev)
|
||||
joint_opt = torch.optim.Adam([
|
||||
{"params": head.parameters(), "lr": PHASE1_LR * 0.1},
|
||||
{"params": enc.parameters(), "lr": PHASE1_ENCODER_LR},
|
||||
], weight_decay=1e-4)
|
||||
for _ in range(PHASE1_JOINT_EPOCHS):
|
||||
perm = torch.randperm(len(Xtr_t), device=dev)
|
||||
for start in range(0, len(Xtr_t), p1_bs):
|
||||
idx_j = perm[start:start + p1_bs]
|
||||
h_raw = enc(Xtr_t[idx_j])[:, -1, :]
|
||||
h_n = (h_raw - mu_e_t) / sd_e_t # frozen-stats normalisation
|
||||
loss_j = F.mse_loss(head(h_n), ytr_z_t[idx_j])
|
||||
joint_opt.zero_grad(); loss_j.backward(); joint_opt.step()
|
||||
enc.eval()
|
||||
# Re-extract test embeddings with fine-tuned encoder, same normalisation
|
||||
with torch.no_grad():
|
||||
chunks = []
|
||||
for i in range(0, len(Xte), p1_bs):
|
||||
t = torch.tensor(Xte[i:i+p1_bs], device=dev)
|
||||
h = enc(t)[:, -1, :]
|
||||
chunks.append(((h - mu_e_t) / sd_e_t).cpu().numpy())
|
||||
Ete_t = torch.tensor(np.concatenate(chunks), device=dev)
|
||||
|
||||
head.eval()
|
||||
with torch.no_grad():
|
||||
pred_h_z = head(Ete_t).cpu().numpy()
|
||||
|
||||
pred_h = pred_h_z * ytr_sd + ytr_mu # de-standardise
|
||||
phase1_r2 = float(1 - ((yte - pred_h) ** 2).sum() / ss_tot)
|
||||
print("phase1_r2 = %.4f (n_test=%d)" % (phase1_r2, len(yte)))
|
||||
|
||||
json.dump({
|
||||
"val_vol_r2": val_vol_r2, "n_test": len(yte),
|
||||
"val_vol_r2": val_vol_r2, "phase1_r2": phase1_r2, "n_test": len(yte),
|
||||
"knobs": {"WINDOW": WINDOW, "PATCH_LEN": PATCH_LEN,
|
||||
"D_MODEL": D_MODEL, "DEPTH": DEPTH, "ALPHA": ALPHA,
|
||||
"DELTA_T_MAX": DELTA_T_MAX, "EPOCHS": EPOCHS},
|
||||
@@ -225,17 +319,19 @@ 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):
|
||||
idx = df2.index[year_mask].tolist()
|
||||
Xs, dates, rvs = [], [], []
|
||||
for t in idx:
|
||||
if t - WINDOW >= 0:
|
||||
if t - WINDOW >= 0 and t + 1 < len(df2):
|
||||
Xs.append(fn2[t - WINDOW:t])
|
||||
dates.append(str(df2["date"].iloc[t].date()))
|
||||
rvs.append(float(df2["realized_vol"].iloc[t]))
|
||||
rvs.append(float(df2["realized_vol"].iloc[t + 1]))
|
||||
if not Xs:
|
||||
return [], [], []
|
||||
Xa = np.stack(Xs)
|
||||
|
||||
Reference in New Issue
Block a user