diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3f54913..f50155b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -185,6 +185,10 @@ jobs: working-directory: rippra run: python tools/reproduce_all.py + - name: Run temporal leakage audit + working-directory: rippra + run: python tests/test_split_leakage.py + # ─── CUDA Build ─────────────────────────────────────── cuda: name: CUDA Build Check diff --git a/AGENTS.md b/AGENTS.md index 1462415..87f74ec 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,12 +1,18 @@ # AGENTS.md — Project Memory ## CI Pipeline -- **3 jobs** (Linux C, Windows C, Python) — all green -- Linux/Windows: compile C library (`io`, `la`, `centroid`, `recon`, `rippra_api`), build tests -- Python: test_onnx_models.py, predictive_ao.py (skips torch if unavailable) +- **Multiple jobs** (Linux C, Windows C, Python, CUDA, benchmarks) — all green +- Linux/Windows: compile C library (`io`, `la`, `centroid`, `recon`, `rippra_api`, `simd`), build tests +- Python: test_onnx_models.py, predictive_ao.py, test_split_leakage.py (skips torch if unavailable) - Synthetic data generated in CI before C tests (via `synthetic_shwfs.generate_test_data`) - CI does NOT commit synthetic data — regenerates every run +## Temporal Leakage Fix (2026-07-08) +- `train_sequence.py` and `evaluate_sequence.py` previously used `random_split` on flat sample list → adjacent sliding windows leaked across train/val/test +- Fix: `SHSequenceDataset.split_by_sequence()` splits at the *sequence* level (contiguous blocks), then `check_split_leakage()` asserts no sequence ID appears in >1 split +- `test_split_leakage.py` programmatically verifies the invariant +- Documented in class docstring and split method + ## Critical Bug Fix (2026-06-27) - `rippra_zonal_reconstruct` and `rippra_modal_reconstruct` used `cfg->totlenses` instead of actual detected spot count → out-of-bounds reads - Fix: added `nspots` field to `rippra_zonal_mesh` and `rippra_modal_model` structs, stored during `_setup()`, used in `_reconstruct()` diff --git a/rippra/ml/evaluate_sequence.py b/rippra/ml/evaluate_sequence.py index aa965f9..1901faa 100644 --- a/rippra/ml/evaluate_sequence.py +++ b/rippra/ml/evaluate_sequence.py @@ -4,7 +4,7 @@ import numpy as np import torch import torch.nn as nn -from torch.utils.data import DataLoader, random_split +from torch.utils.data import DataLoader # Add current directory to sys.path sys.path.append(os.path.dirname(__file__)) @@ -42,15 +42,10 @@ def main(): print(" 1. FUTURE WAVEFRONT PREDICTION EVALUATION") print("="*80) - # Load dataset split + # Load dataset split (sequence-level to avoid temporal leakage) dataset = SHSequenceDataset(dataset_path, lookback=10, step=1, task='predict') - total_len = len(dataset) - train_len = int(0.8 * total_len) - val_len = int(0.1 * total_len) - test_len = total_len - train_len - val_len - _, _, test_set = random_split( - dataset, [train_len, val_len, test_len], - generator=torch.Generator().manual_seed(42) + _, _, test_set = SHSequenceDataset.split_by_sequence( + dataset, train_ratio=0.8, val_ratio=0.1, seed=42 ) test_loader = DataLoader(test_set, batch_size=128, shuffle=False) @@ -101,13 +96,8 @@ def main(): print("="*80) dataset = SHSequenceDataset(dataset_path, lookback=10, step=1, task='classify') - total_len = len(dataset) - train_len = int(0.8 * total_len) - val_len = int(0.1 * total_len) - test_len = total_len - train_len - val_len - _, _, test_set = random_split( - dataset, [train_len, val_len, test_len], - generator=torch.Generator().manual_seed(42) + _, _, test_set = SHSequenceDataset.split_by_sequence( + dataset, train_ratio=0.8, val_ratio=0.1, seed=42 ) test_loader = DataLoader(test_set, batch_size=128, shuffle=False) @@ -150,13 +140,8 @@ def main(): print("="*80) dataset = SHSequenceDataset(dataset_path, lookback=10, step=1, task='parameter') - total_len = len(dataset) - train_len = int(0.8 * total_len) - val_len = int(0.1 * total_len) - test_len = total_len - train_len - val_len - _, _, test_set = random_split( - dataset, [train_len, val_len, test_len], - generator=torch.Generator().manual_seed(42) + _, _, test_set = SHSequenceDataset.split_by_sequence( + dataset, train_ratio=0.8, val_ratio=0.1, seed=42 ) test_loader = DataLoader(test_set, batch_size=128, shuffle=False) diff --git a/rippra/ml/train_sequence.py b/rippra/ml/train_sequence.py index bba7c74..0132d7a 100644 --- a/rippra/ml/train_sequence.py +++ b/rippra/ml/train_sequence.py @@ -4,14 +4,36 @@ import numpy as np import torch import torch.nn as nn -from torch.utils.data import Dataset, DataLoader, random_split +from torch.utils.data import Dataset, DataLoader, Subset from sequence_models import WavefrontLSTM, TurbulenceClassifierLSTM, TurbulenceParameterEstimator +def check_split_leakage(dataset, train_indices, val_indices, test_indices): + """ + Assert that no sequence ID appears in more than one split. + Must be called after SHSequenceDataset is constructed. + """ + def seq_ids(indices): + return set(dataset.sequence_ids[i] for i in indices) + train_s = seq_ids(train_indices) + val_s = seq_ids(val_indices) + test_s = seq_ids(test_indices) + tv = train_s & val_s + tt = train_s & test_s + vt = val_s & test_s + if tv or tt or vt: + raise AssertionError( + f"Temporal leakage detected: train↔val {len(tv)}, " + f"train↔test {len(tt)}, val↔test {len(vt)} sequences overlap." + ) + class SHSequenceDataset(Dataset): """ Sequence dataset loader that slices sequence frames into sliding windows without crossing sequence boundaries (each sequence is 1000 frames). + + Splits should be performed at the *sequence* level (see split_by_sequence()) + to avoid temporal leakage from adjacent overlapping windows. """ def __init__(self, dataset_path, lookback=10, step=1, task='predict'): self.lookback = lookback @@ -29,6 +51,7 @@ def __init__(self, dataset_path, lookback=10, step=1, task='predict'): n_sequences = n_frames // self.seq_len self.samples = [] + self.sequence_ids = [] for s in range(n_sequences): seq_start = s * self.seq_len @@ -62,6 +85,7 @@ def __init__(self, dataset_path, lookback=10, step=1, task='predict'): elif self.task == 'parameter': # Target: average D_r0 of sequence self.samples.append((hist_disp, avg_dr0)) + self.sequence_ids.append(s) def __len__(self): return len(self.samples) @@ -72,6 +96,36 @@ def __getitem__(self, idx): return torch.tensor(x, dtype=torch.float32), torch.tensor(y, dtype=torch.long) else: return torch.tensor(x, dtype=torch.float32), torch.tensor(y, dtype=torch.float32) + + @staticmethod + def split_by_sequence(dataset, train_ratio=0.8, val_ratio=0.1, seed=42): + """ + Split dataset by contiguous blocks of sequences (not individual samples). + This avoids temporal leakage from adjacent overlapping windows. + + Returns (train_set, val_set, test_set) as torch Subset instances. + """ + n_seqs = max(dataset.sequence_ids) + 1 if dataset.sequence_ids else 0 + seq_indices = list(range(n_seqs)) + rng = np.random.RandomState(seed) + rng.shuffle(seq_indices) + + n_train = int(train_ratio * n_seqs) + n_val = int(val_ratio * n_seqs) + + train_seqs = set(seq_indices[:n_train]) + val_seqs = set(seq_indices[n_train:n_train + n_val]) + test_seqs = set(seq_indices[n_train + n_val:]) + + train_idx = [i for i, sid in enumerate(dataset.sequence_ids) if sid in train_seqs] + val_idx = [i for i, sid in enumerate(dataset.sequence_ids) if sid in val_seqs] + test_idx = [i for i, sid in enumerate(dataset.sequence_ids) if sid in test_seqs] + + check_split_leakage(dataset, train_idx, val_idx, test_idx) + + return (Subset(dataset, train_idx), + Subset(dataset, val_idx), + Subset(dataset, test_idx)) def train_epoch(model, loader, criterion, optimizer, device): @@ -159,15 +213,9 @@ def main(): print(f"Loading sequence dataset for task '{args.task}' (lookback={args.lookback}, step={args.step})...") full_dataset = SHSequenceDataset(args.dataset, lookback=args.lookback, step=args.step, task=args.task) - # Train / Val / Test split (80% / 10% / 10%) - total_len = len(full_dataset) - train_len = int(0.8 * total_len) - val_len = int(0.1 * total_len) - test_len = total_len - train_len - val_len - - train_set, val_set, test_set = random_split( - full_dataset, [train_len, val_len, test_len], - generator=torch.Generator().manual_seed(42) + # Train / Val / Test split (80% / 10% / 10%) at sequence level + train_set, val_set, test_set = SHSequenceDataset.split_by_sequence( + full_dataset, train_ratio=0.8, val_ratio=0.1, seed=42 ) train_loader = DataLoader(train_set, batch_size=args.batch_size, shuffle=True) diff --git a/rippra/tests/test_split_leakage.py b/rippra/tests/test_split_leakage.py new file mode 100644 index 0000000..e1bfe68 --- /dev/null +++ b/rippra/tests/test_split_leakage.py @@ -0,0 +1,85 @@ +""" +test_split_leakage.py — verify no temporal overlap between train/val/test splits. + +The SHSequenceDataset uses contiguous-block splitting at the sequence level +to prevent leakage from adjacent overlapping sliding windows. This test +confirms that invariant programmatically. +""" +import os +import sys +import numpy as np + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "ml")) + +from train_sequence import SHSequenceDataset, check_split_leakage + + +def _make_mini_dataset(path, n_seqs=5, seq_len=1000, nspots=10, nmodes=5): + """Generate a tiny in-memory dataset for testing.""" + n_frames = n_seqs * seq_len + rng = np.random.RandomState(0) + disps = rng.randn(n_frames, 2 * nspots) + coeff = rng.randn(n_frames, nmodes) + dr0 = rng.uniform(1, 10, n_frames) + np.savez(path, displacements=disps, coefficients=coeff, D_r0=dr0) + + +def test_split_by_sequence_no_leakage(): + path = "_test_leakage.npz" + try: + _make_mini_dataset(path) + ds = SHSequenceDataset(path, lookback=10, step=1, task="predict") + train, val, test = SHSequenceDataset.split_by_sequence(ds, seed=42) + # check_split_leakage is called inside split_by_sequence, so if it + # passes we already have the invariant. Double-check explicit sets. + train_s = set(ds.sequence_ids[i] for i in train.indices) + val_s = set(ds.sequence_ids[i] for i in val.indices) + test_s = set(ds.sequence_ids[i] for i in test.indices) + assert train_s.isdisjoint(val_s), "train ↔ val overlap" + assert train_s.isdisjoint(test_s), "train ↔ test overlap" + assert val_s.isdisjoint(test_s), "val ↔ test overlap" + print(f"PASS: {len(train_s)} train / {len(val_s)} val / {len(test_s)} test sequences, no leakage") + finally: + if os.path.exists(path): + os.remove(path) + + +def test_split_by_sequence_exhaustive(): + """With a 4-sequence dataset and 50/25/25 split, every sample appears in exactly one split.""" + path = "_test_leakage_exhaustive.npz" + try: + _make_mini_dataset(path, n_seqs=4) + ds = SHSequenceDataset(path, lookback=10, step=1, task="predict") + train, val, test = SHSequenceDataset.split_by_sequence( + ds, train_ratio=0.5, val_ratio=0.25, seed=42 + ) + all_idx = set(train.indices) | set(val.indices) | set(test.indices) + assert all_idx == set(range(len(ds))), f"Missing {set(range(len(ds))) - all_idx}" + print(f"PASS: {len(train)} train + {len(val)} val + {len(test)} test = {len(ds)} total") + finally: + if os.path.exists(path): + os.remove(path) + + +def test_check_split_leakage_detects_overlap(): + """check_split_leakage must raise on deliberately overlapping splits.""" + path = "_test_leakage_detect.npz" + try: + _make_mini_dataset(path, n_seqs=2) + ds = SHSequenceDataset(path, lookback=10, step=1, task="predict") + # deliberately pass overlapping indices + try: + check_split_leakage(ds, [0, 1], [1, 2], [3, 4]) + assert False, "expected AssertionError" + except AssertionError: + print("PASS: overlap correctly detected") + finally: + if os.path.exists(path): + os.remove(path) + + +if __name__ == "__main__": + test_split_by_sequence_no_leakage() + test_split_by_sequence_exhaustive() + test_check_split_leakage_detects_overlap() + print("\nAll tests PASSED")