"""Walk-forward evaluation of the volatility-regime target. Fetches hourly BTCUSDT klines from Binance, builds the causal features in src/volatility.py and reports accuracy against two baselines that must both be beaten before the result means anything: * majority class - catches targets inflated by class imbalance, * persistence - "current volatility regime continues", which is strong because volatility clusters. Usage: python scripts/run_volatility_experiment.py [--folds 5] [--horizon 24] """ from __future__ import annotations import argparse import sys import time from pathlib import Path import numpy as np import pandas as pd import requests from sklearn.ensemble import HistGradientBoostingClassifier from sklearn.metrics import accuracy_score, matthews_corrcoef, roc_auc_score sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from src.volatility import add_volatility_target, build_volatility_features, purge_split HOSTS = ("api.binance.com", "api1.binance.com", "api2.binance.com") COLUMNS = ["open_time", "open", "high", "low", "close", "volume", "close_time", "quote_volume", "num_trades", "taker_buy_base", "taker_buy_quote", "ignore"] ARCHIVE = "https://data.binance.vision/data/spot" def fetch_klines_archive(start: str, symbol: str = "BTCUSDT") -> pd.DataFrame: """Hourly klines from Binance's public monthly archive. The REST API answers HTTP 451 to many cloud IP ranges, which is how a Colab run spent twenty minutes retrying a request it could never satisfy. This archive carries no such restriction, so it is tried first and the API is kept only as a fallback. Returns an empty frame rather than raising, so the caller can fall back. """ import io import zipfile def read_zip(url: str): try: response = requests.get(url, timeout=60) if response.status_code != 200: return None with zipfile.ZipFile(io.BytesIO(response.content)) as bundle: raw = bundle.read(bundle.namelist()[0]).decode() except (requests.RequestException, zipfile.BadZipFile, KeyError, IndexError): return None part = pd.read_csv(io.StringIO(raw), header=None, names=COLUMNS) # Binance switched these files from milliseconds to microseconds; detect # by magnitude rather than by date, which would need maintaining. unit = "us" if part["open_time"].iloc[0] > 10 ** 14 else "ms" part["timestamp"] = pd.to_datetime(part["open_time"], unit=unit, utc=True) return part now = pd.Timestamp.now(tz="UTC") frames, missing = [], 0 for month in pd.period_range(pd.Timestamp(start).to_period("M"), now.tz_localize(None).to_period("M"), freq="M"): part = read_zip(f"{ARCHIVE}/monthly/klines/{symbol}/1h/{symbol}-1h-{month}.zip") if part is None: missing += 1 else: frames.append(part) # Monthly files appear only after a month closes, and sometimes later, so the # most recent weeks must come from the daily files or the data arrives stale. covered = max((f["timestamp"].max() for f in frames), default=None) first_day = (covered + pd.Timedelta(hours=1)) if covered is not None else pd.Timestamp(start, tz="UTC") days = 0 for day in pd.date_range(first_day.normalize(), now.normalize(), freq="D", tz="UTC"): part = read_zip(f"{ARCHIVE}/daily/klines/{symbol}/1h/" f"{symbol}-1h-{day.date()}.zip") if part is not None: frames.append(part) days += 1 if not frames: return pd.DataFrame() frame = (pd.concat(frames).drop_duplicates("timestamp") .sort_values("timestamp").reset_index(drop=True)) print(f" archive: {len(frame):,} candles for {symbol} " f"({len(frames) - days} months + {days} days, {missing} unavailable, " f"through {frame['timestamp'].max()})") return frame def fetch_klines(start: str, cache: Path, symbol: str = "BTCUSDT") -> pd.DataFrame: """Download hourly klines, rotating hosts because Binance rate-limits by IP. `symbol` defaults to BTCUSDT so every existing caller is unaffected; the multi-asset experiment passes ETHUSDT and ADAUSDT. """ if cache.exists(): return pd.read_csv(cache, parse_dates=["timestamp"]) # The archive first: it is not geo-restricted, and one request per month # beats sixty-six paged requests against an endpoint that may refuse them. archived = fetch_klines_archive(start, symbol) if not archived.empty: for column in ("open", "high", "low", "close", "volume", "quote_volume", "num_trades", "taker_buy_base", "taker_buy_quote"): archived[column] = pd.to_numeric(archived[column]) cache.parent.mkdir(parents=True, exist_ok=True) archived.to_csv(cache, index=False) return archived rows: list = [] cursor = int(pd.Timestamp(start, tz="UTC").timestamp() * 1000) end = int(pd.Timestamp.now(tz="UTC").timestamp() * 1000) host_index = 0 # Consecutive failures are budgeted. Without this the loop retried for ever # on a response it could never satisfy -- Binance answers 451 to many cloud # IP ranges, and a Colab run sat silent for twenty minutes before anyone # suspected it was not simply slow. attempts, max_attempts = 0, len(HOSTS) * 4 last_status = None while cursor < end: if attempts >= max_attempts: hint = (" Binance refuses this network (HTTP 451); fetch the data elsewhere " "and place the CSV at the cache path." if last_status == 451 else "") raise RuntimeError( f"Gave up fetching {symbol} after {attempts} consecutive failures across " f"{len(HOSTS)} hosts; last status {last_status}.{hint}") host = HOSTS[host_index % len(HOSTS)] try: response = requests.get( f"https://{host}/api/v3/klines", params={"symbol": symbol, "interval": "1h", "startTime": cursor, "limit": 1000}, timeout=20, ) if response.status_code != 200: last_status = response.status_code host_index += 1 attempts += 1 time.sleep(1) continue batch = response.json() except requests.RequestException as exc: last_status = type(exc).__name__ host_index += 1 attempts += 1 time.sleep(1) continue if not batch: break attempts = 0 # a good page clears the budget rows.extend(batch) cursor = batch[-1][0] + 3_600_000 time.sleep(0.12) if not rows: raise RuntimeError(f"No candles returned for {symbol}; nothing to cache.") frame = pd.DataFrame(rows, columns=COLUMNS) frame["timestamp"] = pd.to_datetime(frame["open_time"], unit="ms", utc=True) for column in ("open", "high", "low", "close", "volume", "quote_volume", "num_trades", "taker_buy_base", "taker_buy_quote"): frame[column] = pd.to_numeric(frame[column]) frame = frame.drop_duplicates("timestamp").sort_values("timestamp").reset_index(drop=True) cache.parent.mkdir(parents=True, exist_ok=True) frame.to_csv(cache, index=False) return frame def fit_with_tuned_threshold(train: pd.DataFrame, features: list, horizon: int): """Fit on train, choosing the decision threshold on a held-out validation tail. A 0.5 cut-off underperforms the persistence baseline; the operating point has to be tuned, and tuning it on the test set would be leakage. """ split = int(len(train) * 0.85) fit_part, validation = train.iloc[: split - horizon], train.iloc[split:] probe = HistGradientBoostingClassifier( max_iter=400, learning_rate=0.06, max_depth=6, l2_regularization=1.0, random_state=42, ).fit(fit_part[features], fit_part["target_volatility_regime"].astype(int)) validation_probs = probe.predict_proba(validation[features])[:, 1] threshold = max( np.linspace(0.30, 0.70, 41), key=lambda t: matthews_corrcoef( validation["target_volatility_regime"].astype(int), (validation_probs >= t).astype(int), ), ) model = HistGradientBoostingClassifier( max_iter=400, learning_rate=0.06, max_depth=6, l2_regularization=1.0, random_state=42, ).fit(train[features], train["target_volatility_regime"].astype(int)) return model, threshold def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--folds", type=int, default=5) parser.add_argument("--horizon", type=int, default=24) parser.add_argument("--start", default="2019-01-01") parser.add_argument("--cache", default="outputs/volatility/btc_1h.csv") args = parser.parse_args() raw = fetch_klines(args.start, Path(args.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)} {data.timestamp.min().date()} -> {data.timestamp.max().date()}") print(f"{'fold':<5}{'test period':<26}{'n':>7}{'base':>7}" f"{'acc':>8}{'auc':>8}{'mcc':>8}{'persist':>9}") results = [] total = len(data) first = int(total * 0.40) step = (total - first) // args.folds 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 model, threshold = fit_with_tuned_threshold(train, features, args.horizon) probs = model.predict_proba(test[features])[:, 1] predictions = (probs >= threshold).astype(int) truth = test["target_volatility_regime"].astype(int) persistence = (test["vol_24"] > test["threshold_causal"]).astype(int) row = (accuracy_score(truth, predictions), roc_auc_score(truth, probs), matthews_corrcoef(truth, predictions), accuracy_score(truth, persistence)) results.append(row) period = f"{test.timestamp.min().date()}->{test.timestamp.max().date()}" print(f"{fold + 1:<5}{period:<26}{len(test):>7}{truth.mean():>7.3f}" f"{row[0]:>8.4f}{row[1]:>8.4f}{row[2]:>8.4f}{row[3]:>9.4f}") summary = np.array(results) print(f"\n{'MEAN':<5}{'':<33}{'':>7}{summary[:, 0].mean():>8.4f}" f"{summary[:, 1].mean():>8.4f}{summary[:, 2].mean():>8.4f}{summary[:, 3].mean():>9.4f}") print(f"{'STD':<5}{'':<33}{'':>7}{summary[:, 0].std():>8.4f}" f"{summary[:, 1].std():>8.4f}{summary[:, 2].std():>8.4f}{summary[:, 3].std():>9.4f}") if __name__ == "__main__": main()