feat(backbone): replace TS-JEPA+SIGReg with HEPA causal JEPA
CD / Lint / Test / Vet (push) Successful in 4s
CD / Build & Import (push) Failing after 8s
CD / Deploy via GitOps (push) Has been skipped

HEPA (Petersen et al., arXiv:2605.11130, ICML 2026 Spotlight):
- CausalEncoder: non-overlapping patches + per-patch LayerNorm +
  causal Transformer (generate_square_subsequent_mask) → all tokens (B, N, D)
- HorizonPredictor: MLP(cat(h_t, Δt)) → predicted future embedding;
  Δt sampled uniformly from [1, min(DELTA_T_MAX, N-1-c)] per epoch
- vicreg_loss: (1-α)·L1(norm(ĥ), norm(h*)) + α·(L_var + L_cov);
  joint training — no stop-gradient on target encoder
- Probe: last-token embedding [:, -1, :], fit on 2019-2021, eval on OOS

Results (true OOS 2022-2023):
  val_vol_r2: -0.45 (TS-JEPA+SIGReg) → +0.243/+0.276 (HEPA)
  effective_rank: 58.9/64 → 122.3/128 (near-full-rank, no collapse)
  Phase-0 gate on val_vol_r2: PASS ✓

Tests: 6/6 green (causal masking verified with non-uniform perturbation;
per-patch LayerNorm is mean-invariant so constant shifts are absorbed)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-06-25 08:05:33 +02:00
co-authored by Claude Sonnet 4.6
parent 20aeecb971
commit bde651b0df
2 changed files with 214 additions and 95 deletions
+102
View File
@@ -0,0 +1,102 @@
"""Failing tests for HEPA backbone 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.
"""
import math
import torch
import torch.nn as nn
import pytest
# ── Tests import the classes from train.py ────────────────────────────────────
# They will fail until train.py implements: CausalEncoder, HorizonPredictor, vicreg_loss
def _import():
import importlib.util, sys
spec = importlib.util.spec_from_file_location("train", "train.py")
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture(scope="module")
def train_mod():
return _import()
# 1. CausalEncoder exists and has correct output shape
def test_causal_encoder_shape(train_mod):
enc = train_mod.CausalEncoder(n_channels=2, patch_len=10, d_model=32, n_heads=4, depth=1)
x = torch.randn(4, 60, 2)
tokens = enc(x) # should return all tokens (B, N, D) for JEPA pretraining
assert tokens.shape == (4, 6, 32), f"expected (4, 6, 32), got {tokens.shape}"
# 2. CausalEncoder is actually causal: earlier token outputs don't change when later inputs change
def test_causal_masking(train_mod):
enc = train_mod.CausalEncoder(n_channels=2, patch_len=10, d_model=32, n_heads=4, depth=2)
enc.eval()
torch.manual_seed(0)
x = torch.randn(1, 60, 2)
x_perturbed = x.clone()
# non-uniform noise (constant shift absorbed by per-patch LayerNorm; variance change is not)
torch.manual_seed(99)
x_perturbed[:, 30:, :] += torch.randn_like(x[:, 30:, :]) * 5.0
with torch.no_grad():
h1 = enc(x)
h2 = enc(x_perturbed)
# First 3 tokens must be identical (causal — don't see future patches)
assert torch.allclose(h1[:, :3, :], h2[:, :3, :], atol=1e-5), \
"causal masking broken: early tokens change when later input changes"
# Last token should differ (it can see the perturbed patches)
assert not torch.allclose(h1[:, -1, :], h2[:, -1, :], atol=1e-5), \
"last token should differ when later input changes"
# 3. HorizonPredictor exists, takes (h, delta_t_float) → same shape as h
def test_horizon_predictor_shape(train_mod):
pred = train_mod.HorizonPredictor(d_model=32)
h = torch.randn(4, 32)
dt = torch.tensor([1.0, 2.0, 3.0, 1.0])
out = pred(h, dt)
assert out.shape == (4, 32), f"expected (4, 32), got {out.shape}"
# 4. vicreg_loss is a scalar and backward doesn't error
def test_vicreg_loss_backward(train_mod):
h_pred = torch.randn(8, 32, requires_grad=True)
h_target = torch.randn(8, 32)
loss = train_mod.vicreg_loss(h_pred, h_target, alpha=0.1)
assert loss.shape == (), f"expected scalar, got {loss.shape}"
loss.backward()
assert h_pred.grad is not None
# 5. Full JEPA step: encode context, predict future, compute loss, backward
def test_jepa_step_end_to_end(train_mod):
enc = train_mod.CausalEncoder(n_channels=2, patch_len=10, d_model=32, n_heads=4, depth=1)
pred = train_mod.HorizonPredictor(d_model=32)
opt = torch.optim.SGD(list(enc.parameters()) + list(pred.parameters()), lr=1e-3)
x = torch.randn(4, 60, 2)
tokens = enc(x) # (4, 6, 32)
c, dt = 2, 2 # context position 2, horizon 2
h_ctx = tokens[:, c, :]
h_tgt = tokens[:, c + dt, :].detach()
h_hat = pred(h_ctx, torch.full((4,), float(dt)))
loss = train_mod.vicreg_loss(h_hat, h_tgt, alpha=0.1)
opt.zero_grad(); loss.backward(); opt.step()
assert loss.item() < 100, "loss exploded"
# 6. build() still returns year-based OOS split (2022-2023)
def test_build_year_split(train_mod):
(Xtr, ytr), (Xte, yte) = train_mod.build()
assert Xtr.shape[1] == train_mod.WINDOW
assert Xte.shape[1] == train_mod.WINDOW
assert len(Xtr) > 0 and len(Xte) > 0
# OOS set should be ~600 windows (2 years of daily data)
assert 400 < len(Xte) < 900, f"OOS size unexpected: {len(Xte)}"