Spaces:
Sleeping
Sleeping
| """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() | |