feat(loop): add --run-dir isolation, heartbeat, ntfy-on-crash; scaffold start command

Closes jepa-fx-risk#11 Phase A.

- scripts/autoresearch_start.py: scaffold runs/<rq-id>/ from Council backlog leaf;
  fail-closed on non-autoresearch-ready; strips candidate_metric; writes program.md
  + run.json (provenance) + train.py copy. 19 TDD tests.
- loop.py: --run-dir flag redirects STATUS.md / metrics.json / HEARTBEAT / train.py
  into the run dir; METRICS_OUT env var passed to train subprocess so it writes
  metrics.json to the run dir; heartbeat file written each iter phase; ntfy-on-crash
  via NTFY_URL env var (best-effort).
- train.py: METRICS_OUT env var overrides metrics.json path (default unchanged).

Launch: LITELLM_KEY=xxx python loop.py --run-dir runs/rq-04

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

STATUS.md

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

Taskfile.yml

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

loop.py

@@ -4,7 +4,7 @@ Agent (on iguana/berget — NOT koala, whose GPU is reserved for train.py) reads
 program.md + train.py + STATUS.md, proposes ONE change to train.py, we run it,
 keep if val_vol_r2 improved else git-revert. Appends per-iter record to STATUS.md.
 
-    LITELLM_KEY=xxx python loop.py [--iters N] [--model MODEL]
+    LITELLM_KEY=xxx python loop.py [--iters N] [--model MODEL] [--run-dir runs/rq-04]
 
 Env:
   LITELLM_KEY      — LiteLLM master key (required)
@@ -12,6 +12,7 @@ Env:
   LOOP_MODEL       — default berget/gemma4-31b (non-thinking; iguana/berget only)
   LOOP_ITERS       — default 3
   TRAIN_TIMEOUT    — seconds per train.py run, default 120
+  NTFY_URL         — optional: POST crash/stall alerts here (e.g. ntfy.sh/<topic>)
 """
 import argparse
 import json
@@ -29,9 +30,14 @@ LITELLM_KEY  = os.environ.get("LITELLM_KEY", "")
 LOOP_MODEL    = os.environ.get("LOOP_MODEL", "berget/gemma4-31b")
 LOOP_ITERS    = int(os.environ.get("LOOP_ITERS", "3"))
 TRAIN_TIMEOUT = int(os.environ.get("TRAIN_TIMEOUT", "120"))
+NTFY_URL      = os.environ.get("NTFY_URL", "")
+
+# Resolved by main() once --run-dir is parsed.
+RUN_DIR      = Path(".")
 STATUS_MD    = Path("STATUS.md")
 METRICS_JSON = Path("metrics.json")
 TRAIN_PY     = Path("train.py")
+HEARTBEAT    = Path("HEARTBEAT")
 
 AGENT_SYSTEM = textwrap.dedent("""\
     You are the autoresearch agent for jepa-fx-risk. Your job: propose ONE small,
@@ -64,7 +70,7 @@ def gpu_snapshot() -> str:
         return "gpu=N/A"
 
 
-def read_metric() -> float | None:
+def read_metric() -> "float | None":
     if not METRICS_JSON.exists():
         return None
     try:
@@ -73,14 +79,15 @@ def read_metric() -> float | None:
         return None
 
 
-def run_train() -> tuple[float | None, float, str]:
-    """Run train.py. Returns (val_vol_r2 or None, wall_secs, stderr_tail)."""
+def run_train() -> "tuple[float | None, float, str]":
+    """Run train.py from project root with METRICS_OUT pointing into the run dir."""
     t0 = time.time()
-    gpu_before = gpu_snapshot()
+    env = dict(os.environ)
+    env["METRICS_OUT"] = str(METRICS_JSON.resolve())
     try:
         r = subprocess.run(
-            [sys.executable, "train.py"],
-            capture_output=True, text=True, timeout=TRAIN_TIMEOUT,
+            [sys.executable, str(TRAIN_PY.resolve())],
+            capture_output=True, text=True, timeout=TRAIN_TIMEOUT, env=env,
         )
         elapsed = time.time() - t0
         if r.returncode != 0:
@@ -91,10 +98,10 @@ def run_train() -> tuple[float | None, float, str]:
         return None, TRAIN_TIMEOUT, "TIMEOUT"
 
 
-def call_agent(iteration: int, best_so_far: float | None) -> str:
+def call_agent(iteration: int, best_so_far: "float | None") -> str:
     """Ask the LLM agent to edit train.py. Returns new train.py content."""
     context = "\n\n".join([
-        "# program.md\n" + read_file(Path("program.md")),
+        "# program.md\n" + read_file(RUN_DIR / "program.md"),
         "# train.py (current)\n" + read_file(TRAIN_PY),
         "# STATUS.md (history)\n" + read_file(STATUS_MD)[-2000:],
         "# metrics.json (last run)\n" + read_file(METRICS_JSON),
@@ -132,37 +139,96 @@ def append_status(line: str):
         f.write(line + "\n")
 
 
+def write_heartbeat(iteration: int, status: str = "alive"):
+    """Update HEARTBEAT so watchdogs can detect stalls."""
+    HEARTBEAT.write_text("%s  iter=%d  ts=%.0f\n" % (status, iteration, time.time()))
+
+
+def ntfy(msg: str):
+    """POST an alert to NTFY_URL (best-effort; silently ignored on any error)."""
+    if not NTFY_URL:
+        return
+    try:
+        req = urllib.request.Request(
+            NTFY_URL, data=msg.encode(), method="POST",
+            headers={"Content-Type": "text/plain"},
+        )
+        urllib.request.urlopen(req, timeout=5)
+    except Exception:
+        pass
+
+
 def main():
+    global RUN_DIR, STATUS_MD, METRICS_JSON, TRAIN_PY, HEARTBEAT
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--iters", type=int, default=LOOP_ITERS)
+    parser.add_argument("--model", default=LOOP_MODEL)
+    parser.add_argument(
+        "--run-dir", default=None,
+        help="run dir scaffolded by autoresearch_start.py; "
+             "STATUS.md, metrics.json, HEARTBEAT, and train.py live here",
+    )
+    args = parser.parse_args()
+
+    loop_iters = args.iters
+    loop_model = args.model
+
+    if args.run_dir:
+        RUN_DIR = Path(args.run_dir)
+        if not RUN_DIR.is_dir():
+            print("ERROR: run dir not found:", RUN_DIR); sys.exit(1)
+
+    STATUS_MD    = RUN_DIR / "STATUS.md"
+    METRICS_JSON = RUN_DIR / "metrics.json"
+    TRAIN_PY     = RUN_DIR / "train.py"
+    HEARTBEAT    = RUN_DIR / "HEARTBEAT"
+
     if not LITELLM_KEY:
         print("ERROR: set LITELLM_KEY"); sys.exit(1)
 
     if not STATUS_MD.exists():
-        STATUS_MD.write_text("# Autoresearch STATUS\n\n| iter | val_vol_r2 | delta | action | secs | gpu | change |\n|------|-----------|-------|--------|------|-----|--------|\n")
+        STATUS_MD.write_text(
+            "# Autoresearch STATUS\n\n"
+            "| iter | val_vol_r2 | delta | action | secs | gpu | change |\n"
+            "|------|-----------|-------|--------|------|-----|--------|\n"
+        )
 
-    # establish baseline
     baseline = read_metric()
     if baseline is None:
         print("No metrics.json — running train.py for baseline...")
         m, secs, err = run_train()
         if m is None:
-            print("Baseline run failed:", err); sys.exit(1)
+            msg = "Baseline run failed: " + err
+            print(msg)
+            ntfy("[jepa-fx-risk] loop CRASH — " + msg)
+            sys.exit(1)
         baseline = m
         print("Baseline: val_vol_r2 = %.4f (%.1fs)" % (baseline, secs))
 
     best = baseline
-    print("Starting loop | model=%s | iters=%d | baseline=%.4f" % (LOOP_MODEL, LOOP_ITERS, best))
+    print("Starting loop | model=%s | iters=%d | baseline=%.4f" % (loop_model, loop_iters, best))
+    if args.run_dir:
+        print("  run-dir:", RUN_DIR)
 
-    for i in range(1, LOOP_ITERS + 1):
-        print("\n--- iter %d/%d ---" % (i, LOOP_ITERS))
+    iter_index = 0
+    try:
+        for i in range(1, loop_iters + 1):
+            iter_index = i
+            write_heartbeat(i, "agent-call")
+            print("\n--- iter %d/%d ---" % (i, loop_iters))
             original = TRAIN_PY.read_text()
 
-        print("  calling agent (%s)..." % LOOP_MODEL)
+            print("  calling agent (%s)..." % loop_model)
             t_agent = time.time()
             try:
                 new_code = call_agent(i, best)
             except Exception as e:
-            print("  agent call failed:", e)
-            append_status("| %d | ERR | — | agent-fail | — | — | %s |" % (i, str(e)[:60]))
+                msg = str(e)
+                print("  agent call failed:", msg)
+                append_status("| %d | ERR | — | agent-fail | — | — | %s |" % (i, msg[:60]))
+                write_heartbeat(i, "agent-fail")
+                ntfy("[jepa-fx-risk] iter %d agent FAIL — %s" % (i, msg[:80]))
                 continue
             agent_secs = time.time() - t_agent
             print("  agent replied in %.1fs" % agent_secs)
@@ -174,6 +240,7 @@ def main():
 
             TRAIN_PY.write_text(new_code)
 
+            write_heartbeat(i, "training")
             gpu = gpu_snapshot()
             print("  running train.py [%s]..." % gpu)
             metric, secs, err = run_train()
@@ -182,6 +249,8 @@ def main():
                 print("  train.py FAILED — reverting. err:", err[:100])
                 revert_train(original)
                 append_status("| %d | FAIL | — | revert | %.0fs | %s | run error |" % (i, secs, gpu))
+                write_heartbeat(i, "train-fail")
+                ntfy("[jepa-fx-risk] iter %d train FAIL — %s" % (i, err[:80]))
                 continue
 
             delta = metric - best
@@ -195,10 +264,19 @@ def main():
             summary = "| %d | %.4f | %+.4f | %s | %.0fs | %s | iter%d |" % (
                 i, metric, delta, action, secs, gpu, i)
             append_status(summary)
+            write_heartbeat(i, "done")
             print("  val_vol_r2=%.4f  delta=%+.4f  action=%s  [%.0fs]" % (metric, delta, action, secs))
 
+    except Exception as e:
+        msg = "loop CRASH at iter %d: %s" % (iter_index, e)
+        print("FATAL:", msg)
+        ntfy("[jepa-fx-risk] " + msg)
+        raise
+
     print("\nDone. Best val_vol_r2 = %.4f (baseline was %.4f, delta %+.4f)" % (best, baseline, best - baseline))
     print("STATUS.md updated.")
+    write_heartbeat(loop_iters, "done")
+    ntfy("[jepa-fx-risk] loop done. best val_vol_r2=%.4f (delta %+.4f)" % (best, best - baseline))
 
 
 if __name__ == "__main__":

scripts/autoresearch_start.py

@@ -0,0 +1,155 @@
+"""autoresearch start — scaffold a run dir from an Autoresearch Council backlog leaf.
+
+Usage:
+    python scripts/autoresearch_start.py <backlog.json> <rq-id>
+
+Reads the Council backlog JSON (from agentsquad autoresearch_pipe.py Stage-3 output),
+finds the node by rq-id, validates it is autoresearch-ready (fail-closed), then
+scaffolds runs/<rq-id>/ with:
+
+  program.md  — hypothesis, single metric (stripped), agent search-space seam
+  run.json    — provenance (strategic_question + council_node) + config
+  train.py    — copy of project train.py (the loop edits this, keeps history clean)
+
+Launch:
+  LITELLM_KEY=xxx python loop.py --run-dir runs/<rq-id>
+
+Refs: jepa-fx-risk#11, agentsquad#44
+"""
+
+import json
+import shutil
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def load_backlog(path: str) -> dict:
+    try:
+        with open(path) as f:
+            return json.load(f)
+    except FileNotFoundError:
+        print(f"error: backlog file not found: {path}", file=sys.stderr)
+        raise
+
+
+def scaffold_run(
+    backlog_path_or_dict,
+    rq_id: str,
+    run_dir: Path,
+    train_py_src: Path,
+) -> None:
+    """Scaffold a run dir. Raises SystemExit on any validation failure."""
+    if isinstance(backlog_path_or_dict, (str, Path)):
+        backlog = load_backlog(str(backlog_path_or_dict))
+    else:
+        backlog = backlog_path_or_dict
+
+    # Find node
+    nodes_by_id = {n["id"]: n for n in backlog.get("nodes", [])}
+    if rq_id not in nodes_by_id:
+        print(f"error: rq-id {rq_id!r} not found in backlog", file=sys.stderr)
+        sys.exit(1)
+    node = nodes_by_id[rq_id]
+
+    # Fail-closed: only autoresearch-ready nodes may be scaffolded
+    status = node.get("status", "")
+    if status != "autoresearch-ready":
+        print(
+            f"error: {rq_id} has status {status!r}, not 'autoresearch-ready' — refusing to scaffold",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    # Guard against overwriting an existing run
+    if run_dir.exists():
+        print(
+            f"error: {run_dir} already exists — remove it first to re-scaffold",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+
+    metric = (node.get("candidate_metric") or "").strip()
+    strategic_q = backlog.get("strategic_question", "")
+    council_node = node["id"]
+    generated_at = datetime.now(timezone.utc).isoformat()
+
+    run_dir.mkdir(parents=True)
+
+    # --- program.md ---
+    program_md = f"""# program.md — {council_node}: {node.get("question", "")[:80]}
+
+## Provenance
+- strategic_question: {json.dumps(strategic_q)}
+- council_node: {council_node} (autoresearch-ready; Autoresearch Council backlog)
+- generated_at: {generated_at}
+
+## Hypothesis
+{node.get("question", "")}
+
+## Single validation metric (optimise this, nothing else)
+`{metric}` — see eval harness for the exact definition. Only this scalar drives
+keep/revert decisions. Report alongside but do NOT optimise:
+  - Kupiec POF p-value (calibration sanity)
+  - val_vol_r2 (representation quality guard)
+
+## What the agent MAY modify (the search space)
+- Hyperparameters in train.py (model size, LR, window, patch_len, epochs, etc.)
+- Conditioning mechanisms (e.g. JEPA_ENABLE_REGIME toggle)
+- Loss function weights and architecture depth
+
+## Frozen (do NOT touch — keeps the ablation clean)
+- Data pipeline and splits (train ≤2021, OOS ≥2022, test 2024 held out)
+- The metric definition and scoring code
+- loop.py, scripts/, tests/
+
+## Experiment loop (per Karpathy autoresearch)
+Each iter (≤ time-box): apply ONE change to train.py → run → read
+`{metric}` → keep if improved (and Kupiec p-value did not collapse), else revert.
+Stop on: target reached, max iters, or K consecutive iters with no improvement.
+"""
+    (run_dir / "program.md").write_text(program_md)
+
+    # --- run.json (provenance + config) ---
+    run_meta = {
+        "strategic_question": strategic_q,
+        "council_node": council_node,
+        "metric": metric,
+        "generated_at": generated_at,
+        "model_tier": "homelab",
+        "max_iters": 10,
+        "time_box_minutes": 5,
+    }
+    (run_dir / "run.json").write_text(json.dumps(run_meta, indent=2) + "\n")
+
+    # --- train.py (loop edits this copy; project root train.py is the template) ---
+    shutil.copy(train_py_src, run_dir / "train.py")
+
+
+def main() -> None:
+    if len(sys.argv) != 3:
+        print("usage: python scripts/autoresearch_start.py <backlog.json> <rq-id>")
+        sys.exit(1)
+
+    backlog_path, rq_id = sys.argv[1], sys.argv[2]
+
+    project_root = Path(__file__).parent.parent
+    run_dir = project_root / "runs" / rq_id
+    train_py_src = project_root / "train.py"
+
+    scaffold_run(backlog_path, rq_id, run_dir, train_py_src)
+
+    backlog = load_backlog(backlog_path)
+    nodes_by_id = {n["id"]: n for n in backlog.get("nodes", [])}
+    metric = (nodes_by_id[rq_id].get("candidate_metric") or "").strip()
+
+    print(f"✓ scaffolded {run_dir}")
+    print(f"  node:   {rq_id}")
+    print(f"  metric: {metric}")
+    print()
+    print("launch:")
+    print(f"  LITELLM_KEY=xxx python loop.py --run-dir runs/{rq_id}")
+
+
+if __name__ == "__main__":
+    main()

scripts/fetch_multipair.py

@@ -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()

scripts/prepare_hourly.py

@@ -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)
 

scripts/prepare_multipair.py

@@ -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)

tests/test_autoresearch_start.py

@@ -0,0 +1,215 @@
+"""Tests for scripts/autoresearch_start.py — jepa-fx-risk#11 Phase A scaffold.
+
+Success criterion: `autoresearch start <backlog.json> <rq-id>` scaffolds a
+runnable run dir from a ready leaf; refuses non-ready nodes; strips
+candidate_metric; records provenance.
+"""
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+# Load the module without executing main()
+_SCRIPT = Path(__file__).parent.parent / "scripts" / "autoresearch_start.py"
+
+
+def _import():
+    spec = importlib.util.spec_from_file_location("autoresearch_start", _SCRIPT)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+@pytest.fixture()
+def mod():
+    return _import()
+
+
+@pytest.fixture()
+def backlog(tmp_path):
+    data = {
+        "strategic_question": "Test strategic question?",
+        "generated_at": "2026-06-27T00:00:00Z",
+        "nodes": [
+            {
+                "id": "rq-01",
+                "question": "Does X improve Y?",
+                "case_type": "autoresearch-loop",
+                "data": "obtainable",
+                "method": "adjacent",
+                "falsifiable": "yes",
+                "candidate_metric": " val_vol_r2",  # leading space — bypass test
+                "depends_on": [],
+                "status": "autoresearch-ready",
+                "track": "autoresearch",
+                "converged": True,
+                "survived_review": True,
+            },
+            {
+                "id": "rq-02",
+                "question": "Not ready yet?",
+                "case_type": "empirical-study",
+                "data": "obtainable",
+                "method": "adjacent",
+                "falsifiable": "yes",
+                "candidate_metric": None,
+                "depends_on": [],
+                "status": "needs-metric",
+                "track": "study",
+                "converged": True,
+                "survived_review": True,
+            },
+            {
+                "id": "rq-03",
+                "question": "A spike.",
+                "case_type": "spike",
+                "data": "have",
+                "method": "yes-named",
+                "falsifiable": "yes",
+                "candidate_metric": None,
+                "depends_on": [],
+                "status": "spike-ready",
+                "track": "spike",
+                "converged": True,
+                "survived_review": True,
+            },
+        ],
+    }
+    p = tmp_path / "backlog.json"
+    p.write_text(json.dumps(data))
+    return p
+
+
+@pytest.fixture()
+def fake_train_py(tmp_path):
+    """Minimal train.py placeholder for scaffold tests."""
+    src = tmp_path / "train_template.py"
+    src.write_text("# train.py placeholder\n")
+    return src
+
+
+# ---------------------------------------------------------------------------
+# fail-closed: refuse non-autoresearch-ready nodes
+# ---------------------------------------------------------------------------
+
+class TestRefuseNonReady:
+    def test_refuses_needs_metric(self, mod, backlog, fake_train_py, tmp_path):
+        run_dir = tmp_path / "runs" / "rq-02"
+        with pytest.raises(SystemExit) as exc:
+            mod.scaffold_run(backlog, "rq-02", run_dir, fake_train_py)
+        assert exc.value.code != 0
+
+    def test_refuses_spike_ready(self, mod, backlog, fake_train_py, tmp_path):
+        run_dir = tmp_path / "runs" / "rq-03"
+        with pytest.raises(SystemExit) as exc:
+            mod.scaffold_run(backlog, "rq-03", run_dir, fake_train_py)
+        assert exc.value.code != 0
+
+    def test_refuses_missing_rq_id(self, mod, backlog, fake_train_py, tmp_path):
+        run_dir = tmp_path / "runs" / "rq-99"
+        with pytest.raises(SystemExit) as exc:
+            mod.scaffold_run(backlog, "rq-99", run_dir, fake_train_py)
+        assert exc.value.code != 0
+
+    def test_refuses_existing_run_dir(self, mod, backlog, fake_train_py, tmp_path):
+        run_dir = tmp_path / "runs" / "rq-01"
+        run_dir.mkdir(parents=True)
+        with pytest.raises(SystemExit) as exc:
+            mod.scaffold_run(backlog, "rq-01", run_dir, fake_train_py)
+        assert exc.value.code != 0
+
+
+# ---------------------------------------------------------------------------
+# scaffold structure: correct files created
+# ---------------------------------------------------------------------------
+
+class TestScaffoldStructure:
+    @pytest.fixture(autouse=True)
+    def _scaffold(self, mod, backlog, fake_train_py, tmp_path):
+        self.run_dir = tmp_path / "runs" / "rq-01"
+        mod.scaffold_run(backlog, "rq-01", self.run_dir, fake_train_py)
+
+    def test_run_dir_created(self):
+        assert self.run_dir.is_dir()
+
+    def test_program_md_created(self):
+        assert (self.run_dir / "program.md").exists()
+
+    def test_run_json_created(self):
+        assert (self.run_dir / "run.json").exists()
+
+    def test_train_py_copied(self):
+        assert (self.run_dir / "train.py").exists()
+        assert (self.run_dir / "train.py").read_text() == "# train.py placeholder\n"
+
+
+# ---------------------------------------------------------------------------
+# program.md content
+# ---------------------------------------------------------------------------
+
+class TestProgramMd:
+    @pytest.fixture(autouse=True)
+    def _scaffold(self, mod, backlog, fake_train_py, tmp_path):
+        self.run_dir = tmp_path / "runs" / "rq-01"
+        mod.scaffold_run(backlog, "rq-01", self.run_dir, fake_train_py)
+        self.content = (self.run_dir / "program.md").read_text()
+
+    def test_contains_hypothesis(self):
+        assert "Does X improve Y?" in self.content
+
+    def test_metric_key_stripped(self):
+        # candidate_metric had leading space " val_vol_r2" — must be stripped
+        assert "`val_vol_r2`" in self.content
+        assert "` val_vol_r2`" not in self.content
+
+    def test_contains_strategic_question(self):
+        assert "Test strategic question?" in self.content
+
+    def test_contains_council_node(self):
+        assert "rq-01" in self.content
+
+
+# ---------------------------------------------------------------------------
+# run.json provenance
+# ---------------------------------------------------------------------------
+
+class TestRunJson:
+    @pytest.fixture(autouse=True)
+    def _scaffold(self, mod, backlog, fake_train_py, tmp_path):
+        self.run_dir = tmp_path / "runs" / "rq-01"
+        mod.scaffold_run(backlog, "rq-01", self.run_dir, fake_train_py)
+        self.run = json.loads((self.run_dir / "run.json").read_text())
+
+    def test_strategic_question_in_provenance(self):
+        assert self.run["strategic_question"] == "Test strategic question?"
+
+    def test_council_node_in_provenance(self):
+        assert self.run["council_node"] == "rq-01"
+
+    def test_metric_stripped_in_provenance(self):
+        assert self.run["metric"] == "val_vol_r2"
+        assert self.run["metric"] == self.run["metric"].strip()
+
+    def test_generated_at_present(self):
+        assert "generated_at" in self.run
+
+    def test_max_iters_present(self):
+        assert "max_iters" in self.run
+
+
+# ---------------------------------------------------------------------------
+# load_backlog helper
+# ---------------------------------------------------------------------------
+
+class TestLoadBacklog:
+    def test_loads_json(self, mod, backlog):
+        data = mod.load_backlog(str(backlog))
+        assert data["strategic_question"] == "Test strategic question?"
+        assert len(data["nodes"]) == 3
+
+    def test_missing_file_raises(self, mod, tmp_path):
+        with pytest.raises((FileNotFoundError, SystemExit)):
+            mod.load_backlog(str(tmp_path / "nonexistent.json"))

tests/test_hepa.py

@@ -227,3 +227,51 @@ def test_hpo_sweep_configs():
     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"

tests/test_multipair.py

@@ -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]}"
+    )

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,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]}"

train.py

@@ -32,6 +32,10 @@ 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))
 # ---------------------------
 
@@ -146,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)
@@ -222,37 +239,72 @@ def main():
     ss_tot  = ((yte - yte.mean()) ** 2).sum()
     val_vol_r2 = float(1 - ss_res / ss_tot)
 
-    # Phase-1: MLP supervised head on frozen embeddings
-    # Standardise targets so the head trains on unit-scale signals.
+    # 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)
-    head_opt = torch.optim.Adam(head.parameters(), lr=PHASE1_LR, weight_decay=1e-4)
+    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_t = torch.tensor(ytr_z, device=dev)
+    ytr_z_t = torch.tensor(ytr_z, device=dev)
     Ete_t   = torch.tensor(Ete_n, device=dev)
-    p1_bs = min(BATCH_SIZE, len(Etr_t))
     N_tr_h  = len(Etr_t)
-    # Real epoch iteration: shuffle full dataset each epoch
+
+    # 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_t[idx_h])
+            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)))
 
+    _metrics_out = os.environ.get("METRICS_OUT", "metrics.json")
     json.dump({
         "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},
-    }, open("metrics.json", "w"), indent=2)
+    }, open(_metrics_out, "w"), indent=2)
     print("val_vol_r2 = %.4f  (n_test=%d, dev=%s)" % (val_vol_r2, len(yte), dev))
 
     # ── EXPORT BLOCK — do NOT edit (agent boundary) ──────────────────────────
@@ -268,7 +320,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):