Spaces:
Sleeping
Sleeping
| """Can an autoencoder's latent representation improve the volatility model? | |
| Two papers point the same way. Badar et al. (Mathematics 13(12):1908, 2025) use | |
| a VAE in *preprocessing* to learn features, then predict with a CNN-LSTM. Ang et | |
| al. (CMS-VAE, ICAIF '25) use a VAE to generate market data. Neither uses the | |
| autoencoder end to end, which is how this project has tested it so far -- and end | |
| to end, VAE+Transformer trails gradient-boosted trees by roughly 0.02 AUC on this | |
| target. | |
| So this tests the other arrangement: train the autoencoder unsupervised, take its | |
| latent, and hand that to the model that currently wins. | |
| Three reasons it might help, all testable here rather than assumed: | |
| more data the reconstruction objective needs no labels and no 48-hour | |
| contiguity, so the encoder can learn from every training row, | |
| including the 53% the sequence models must discard for gaps | |
| denoising corrupting inputs and reconstructing clean ones is a strong | |
| regulariser on noisy financial features | |
| compression a 24-dimensional latent may express the 33 features more usefully | |
| than the raw set, or may simply lose information -- the ablation | |
| says which | |
| The VAE is fitted on training rows only, inside each fold. Fitting it on the full | |
| series first would be transductive: the reconstruction objective uses no labels, | |
| but it would still have seen the test period's features, which is lookahead in a | |
| forecasting setting. | |
| Usage: | |
| python scripts/run_vae_feature_extraction.py | |
| python scripts/run_vae_feature_extraction.py --folds 5 --latent 24 --epochs 60 | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import warnings | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| import torch | |
| import torch.nn as nn | |
| from scipy import stats | |
| from sklearn.ensemble import HistGradientBoostingClassifier | |
| from sklearn.metrics import matthews_corrcoef, roc_auc_score | |
| from sklearn.preprocessing import StandardScaler | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT)) | |
| warnings.filterwarnings("ignore") | |
| from src.volatility import add_volatility_target, build_volatility_features, purge_split # noqa: E402 | |
| from scripts.run_volatility_experiment import fetch_klines # noqa: E402 | |
| SEED = 42 | |
| CACHE = ROOT / "outputs" / "volatility" / "btc_1h.csv" | |
| # identical to the deployed classifier | |
| CLF = dict(max_iter=400, learning_rate=0.06, max_depth=6, | |
| l2_regularization=1.0, random_state=SEED) | |
| class TabularVAE(nn.Module): | |
| """A plain VAE over one row of features -- no sequence, so no contiguity.""" | |
| def __init__(self, n_features: int, latent: int = 24, hidden: int = 96): | |
| super().__init__() | |
| self.encoder = nn.Sequential( | |
| nn.Linear(n_features, hidden), nn.GELU(), nn.LayerNorm(hidden)) | |
| self.mu = nn.Linear(hidden, latent) | |
| self.logvar = nn.Linear(hidden, latent) | |
| self.decoder = nn.Sequential( | |
| nn.Linear(latent, hidden), nn.GELU(), nn.Linear(hidden, n_features)) | |
| def encode(self, x): | |
| h = self.encoder(x) | |
| return self.mu(h), torch.clamp(self.logvar(h), -10.0, 10.0) | |
| def forward(self, x): | |
| mu, logvar = self.encode(x) | |
| z = mu + torch.randn_like(mu) * torch.exp(0.5 * logvar) if self.training else mu | |
| return self.decoder(z), mu, logvar | |
| def fit_vae(X: np.ndarray, latent: int, epochs: int, kl_beta: float, | |
| noise: float) -> TabularVAE: | |
| """Unsupervised fit on training rows. `noise` > 0 makes it denoising.""" | |
| torch.manual_seed(SEED) | |
| model = TabularVAE(X.shape[1], latent) | |
| opt = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5) | |
| data = torch.tensor(X, dtype=torch.float32) | |
| batch = 512 | |
| model.train() | |
| for _ in range(epochs): | |
| perm = torch.randperm(len(data)) | |
| for s in range(0, len(perm), batch): | |
| clean = data[perm[s: s + batch]] | |
| corrupted = clean + torch.randn_like(clean) * noise if noise > 0 else clean | |
| opt.zero_grad() | |
| recon, mu, logvar = model(corrupted) | |
| recon_loss = nn.functional.mse_loss(recon, clean) | |
| kl = -0.5 * torch.mean(1 + logvar - mu.pow(2) - logvar.exp()) | |
| (recon_loss + kl_beta * kl).backward() | |
| opt.step() | |
| model.eval() | |
| return model | |
| def encode(model: TabularVAE, X: np.ndarray) -> np.ndarray: | |
| with torch.no_grad(): | |
| mu, _ = model.encode(torch.tensor(X, dtype=torch.float32)) | |
| return mu.numpy() | |
| def score(name: str, X_tr, y_tr, X_te, y_te) -> dict: | |
| model = HistGradientBoostingClassifier(**CLF).fit(X_tr, y_tr) | |
| probs = model.predict_proba(X_te)[:, 1] | |
| preds = (probs >= 0.5).astype(int) | |
| return dict(arm=name, n_features=X_tr.shape[1], | |
| auc=float(roc_auc_score(y_te, probs)), | |
| accuracy=float((preds == y_te).mean()), | |
| mcc=float(matthews_corrcoef(y_te, preds))) | |
| def main() -> None: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--folds", type=int, default=5) | |
| ap.add_argument("--horizon", type=int, default=24) | |
| ap.add_argument("--latent", type=int, default=24) | |
| ap.add_argument("--epochs", type=int, default=60) | |
| ap.add_argument("--kl-beta", type=float, default=1e-3) | |
| ap.add_argument("--noise", type=float, default=0.15) | |
| args = ap.parse_args() | |
| raw = fetch_klines("2019-01-01", CACHE) | |
| featured, features = build_volatility_features(raw) | |
| featured = add_volatility_target(featured, args.horizon) | |
| data = featured.dropna( | |
| subset=features + ["forward_vol", "threshold_causal", "target_volatility_regime"] | |
| ).reset_index(drop=True) | |
| print(f"rows {len(data):,} features {len(features)} latent {args.latent}") | |
| print(f"the sequence models use only 30,763 contiguous rows; this uses all " | |
| f"{len(data):,}\n") | |
| total = len(data) | |
| first = int(total * 0.40) | |
| step = (total - first) // args.folds | |
| rows = [] | |
| for fold in range(args.folds): | |
| train_end = first + fold * step | |
| train, test = purge_split(data, train_end, min(train_end + step, total), args.horizon) | |
| if len(test) < 500: | |
| continue | |
| y_tr = train["target_volatility_regime"].astype(int).to_numpy() | |
| y_te = test["target_volatility_regime"].astype(int).to_numpy() | |
| scaler = StandardScaler().fit(train[features]) | |
| Xs_tr, Xs_te = scaler.transform(train[features]), scaler.transform(test[features]) | |
| raw_tr, raw_te = train[features].to_numpy(float), test[features].to_numpy(float) | |
| plain = fit_vae(Xs_tr, args.latent, args.epochs, args.kl_beta, 0.0) | |
| denoise = fit_vae(Xs_tr, args.latent, args.epochs, args.kl_beta, args.noise) | |
| z_tr, z_te = encode(plain, Xs_tr), encode(plain, Xs_te) | |
| d_tr, d_te = encode(denoise, Xs_tr), encode(denoise, Xs_te) | |
| fold_rows = [ | |
| score("features only (deployed)", raw_tr, y_tr, raw_te, y_te), | |
| score("VAE latent only", z_tr, y_tr, z_te, y_te), | |
| score("features + VAE latent", np.hstack([raw_tr, z_tr]), y_tr, | |
| np.hstack([raw_te, z_te]), y_te), | |
| score("denoising VAE latent only", d_tr, y_tr, d_te, y_te), | |
| score("features + denoising latent", np.hstack([raw_tr, d_tr]), y_tr, | |
| np.hstack([raw_te, d_te]), y_te), | |
| ] | |
| for r in fold_rows: | |
| r.update(fold=fold + 1, n_test=len(test), | |
| start=str(test.timestamp.min().date()), end=str(test.timestamp.max().date())) | |
| rows += fold_rows | |
| best = max(fold_rows, key=lambda r: r["auc"]) | |
| print(f" fold {fold + 1} {fold_rows[0]['start']} -> {fold_rows[0]['end']} " | |
| f"baseline {fold_rows[0]['auc']:.4f} best {best['auc']:.4f} ({best['arm']})") | |
| df = pd.DataFrame(rows) | |
| agg = (df.groupby("arm") | |
| .agg(folds=("fold", "count"), n_features=("n_features", "first"), | |
| auc=("auc", "mean"), auc_sd=("auc", "std"), | |
| accuracy=("accuracy", "mean"), mcc=("mcc", "mean")) | |
| .reset_index().sort_values("auc", ascending=False)) | |
| print("\n" + "=" * 96) | |
| print("AUTOENCODER FEATURES vs THE DEPLOYED FEATURE SET") | |
| print("=" * 96) | |
| print(f"{'arm':<32}{'inputs':>8}{'ROC-AUC':>10}{'sd':>8}{'accuracy':>11}{'MCC':>9}") | |
| for _, r in agg.iterrows(): | |
| print(f"{r['arm']:<32}{int(r['n_features']):>8}{r['auc']:>10.4f}{r['auc_sd']:>8.4f}" | |
| f"{r['accuracy'] * 100:>10.2f}%{r['mcc']:>+9.4f}") | |
| baseline = df[df.arm == "features only (deployed)"].set_index("fold")["auc"] | |
| print("\n" + "-" * 96) | |
| print("PAIRED AGAINST THE DEPLOYED FEATURE SET (same folds, same classifier)") | |
| print("-" * 96) | |
| summary = {} | |
| for arm in agg["arm"]: | |
| if arm == "features only (deployed)": | |
| continue | |
| other = df[df.arm == arm].set_index("fold")["auc"] | |
| delta = (other - baseline).dropna().to_numpy() | |
| if len(delta) < 3: | |
| continue | |
| t, p = stats.ttest_1samp(delta, 0) | |
| ci = stats.t.interval(0.95, len(delta) - 1, delta.mean(), stats.sem(delta)) | |
| verdict = ("helps" if p < 0.05 and delta.mean() > 0 | |
| else "hurts" if p < 0.05 and delta.mean() < 0 else "no effect") | |
| summary[arm] = dict(delta=float(delta.mean()), p=float(p), | |
| ci=[float(ci[0]), float(ci[1])], | |
| better=int((delta > 0).sum()), folds=len(delta)) | |
| print(f" {arm:<32}{delta.mean():>+9.4f} 95% CI [{ci[0]:+.4f}, {ci[1]:+.4f}]" | |
| f" p={p:.3f} {int((delta > 0).sum())}/{len(delta)} folds {verdict}") | |
| print("\n A latent that matches the raw features on its own means the autoencoder") | |
| print(" has compressed without losing anything. Only a positive delta on the") | |
| print(" combined arms means it has added something the features did not carry.") | |
| out = ROOT / "outputs" / "vae_feature_extraction.json" | |
| out.write_text(json.dumps({"folds": rows, "summary": agg.to_dict("records"), | |
| "paired": summary}, indent=2, default=str), encoding="utf-8") | |
| print(f"\nwritten: {out.relative_to(ROOT)}") | |
| if __name__ == "__main__": | |
| main() | |