"""Reproduces the numerical results in BiffCoin Research Paper 001. Paper 001: Synchronised Prices, Divergent Flows. Run from project root: python3 biffcoin/papers/001_compute.py Outputs computed values that should match the figures in the paper to the same precision reported. Requires only the standard library + `requests`. What this script computes: - Trailing 21-day pairwise correlations on log returns - Event-window (25 May 2026 UTC) correlations on both intra-bar and log-return bases - The 25 May V-bottom (21:00 and 22:00 UTC bars per asset) - Volatility-normalised V-bottom magnitudes - Annualised hourly volatility per asset The script intentionally fetches data live from MEXC's public klines endpoint rather than reading a cached file, so anyone running it against the same time window can verify the numbers independently. """ from __future__ import annotations import math import statistics from datetime import datetime, timezone import requests SYMBOLS = ["BTCUSDT", "ETHUSDT", "XRPUSDT"] EVENT_DAY_START_UTC = datetime(2026, 5, 25, 0, 0, 0, tzinfo=timezone.utc) EVENT_DAY_END_UTC = datetime(2026, 5, 26, 0, 0, 0, tzinfo=timezone.utc) V_DIP_UTC = datetime(2026, 5, 24, 21, 0, 0, tzinfo=timezone.utc) # 07:00 AEST V_BOUNCE_UTC = datetime(2026, 5, 24, 22, 0, 0, tzinfo=timezone.utc) # 08:00 AEST # Anchor for the historical pull. Set to a date shortly after the # event window so this script reproduces the paper's exact figures # at any point in the future, not just within ~125 days of run time. # To replicate v1.x results, leave this at 2026-06-01. To rerun # against a moving window (e.g. for ongoing analysis), pass # --anchor-now to fetch_klines. FETCH_ANCHOR_UTC = datetime(2026, 6, 1, 0, 0, 0, tzinfo=timezone.utc) def fetch_klines(symbol: str, n_fetches: int = 3, limit: int = 1000, anchor_utc: datetime | None = None): """Paginate MEXC public klines API to get up to n_fetches * limit bars. Walks backwards from `anchor_utc` (default: the FETCH_ANCHOR_UTC constant above, set to 2026-06-01 for paper reproducibility). Pass `anchor_utc=datetime.now(timezone.utc)` to use a live anchor instead. """ if anchor_utc is None: anchor_utc = FETCH_ANCHOR_UTC all_bars = [] end_ms = int(anchor_utc.timestamp() * 1000) for _ in range(n_fetches): r = requests.get( "https://api.mexc.com/api/v3/klines", params={ "symbol": symbol, "interval": "60m", "endTime": end_ms, "limit": limit, }, timeout=15, ) r.raise_for_status() bars = r.json() if not bars: break all_bars = bars + all_bars end_ms = int(bars[0][0]) - 1 return all_bars def align(data: dict) -> tuple[list[int], dict]: """Align bars across all symbols by open-time; return sorted common timestamps and a per-symbol {ts: close} dict.""" closes = {sym: {int(b[0]): float(b[4]) for b in data[sym]} for sym in data} ts_sets = [set(closes[sym].keys()) for sym in closes] common = sorted(set.intersection(*ts_sets)) return common, closes def log_returns(prices: list[float]) -> list[float]: return [math.log(prices[i] / prices[i - 1]) for i in range(1, len(prices))] def intra_bar_returns(bars_for_ts: dict, common: list[int]) -> list[float]: """(close - open)/open for each common bar, indexed to common[i].""" return [(b[1] - b[0]) / b[0] for b in (bars_for_ts[t] for t in common)] def pearson(x: list[float], y: list[float]) -> float: """Sample Pearson correlation coefficient. Uses the standard formula: ρ = Σ(xᵢ-x̄)(yᵢ-ȳ) / sqrt(Σ(xᵢ-x̄)² · Σ(yᵢ-ȳ)²) Bessel's correction (dividing by n-1 in the variance terms) cancels in the numerator/denominator ratio, so this expression is identical to the sample-Pearson result. For n ≈ 23 or n ≈ 499 the value is also numerically indistinguishable from Python's `statistics.correlation` (sample) — we verified this in cross-checks. Implementing from first principles here keeps the paper's math fully visible without a stdlib dependency on `statistics.correlation` (which is Python 3.10+). """ n = len(x) if n < 2: return 0.0 mx, my = sum(x) / n, sum(y) / n num = sum((x[i] - mx) * (y[i] - my) for i in range(n)) dx = math.sqrt(sum((xi - mx) ** 2 for xi in x)) dy = math.sqrt(sum((yi - my) ** 2 for yi in y)) return num / (dx * dy) if dx * dy > 0 else 0.0 def main(): print("Fetching MEXC 1h klines for BTC/ETH/XRP …") raw = {sym: fetch_klines(sym) for sym in SYMBOLS} for sym in SYMBOLS: n = len(raw[sym]) first = datetime.fromtimestamp(raw[sym][0][0] / 1000, tz=timezone.utc) last = datetime.fromtimestamp(raw[sym][-1][0] / 1000, tz=timezone.utc) print(f" {sym}: {n} bars {first.isoformat()} → {last.isoformat()}") # Align by open-time common, closes = align(raw) print(f"\nCommon bars across all 3 symbols: {len(common)}") # Build open/close per symbol indexed by ts for intra-bar work ohlc = {sym: {int(b[0]): (float(b[1]), float(b[4])) for b in raw[sym]} for sym in SYMBOLS} common_set = set(common) # ── Trailing baseline (log returns) ──────────────────────── aligned_closes = {sym: [closes[sym][t] for t in common] for sym in SYMBOLS} log_rets = {sym: log_returns(aligned_closes[sym]) for sym in SYMBOLS} n_baseline = len(log_rets["BTCUSDT"]) print(f"\n=== Trailing baseline (n={n_baseline} 1h bars ≈ " f"{n_baseline / 24:.0f} days) ===") print(f"Pairwise Pearson correlation on log returns:") for a, b in [("BTCUSDT", "ETHUSDT"), ("BTCUSDT", "XRPUSDT"), ("ETHUSDT", "XRPUSDT")]: r = pearson(log_rets[a], log_rets[b]) print(f" {a[:3]} ↔ {b[:3]}: ρ = {r:+.4f}") # ── Trailing annualised volatility ───────────────────────── print(f"\n=== Annualised hourly-log-return volatility (trailing baseline) ===") for sym in SYMBOLS: sigma_hr = statistics.stdev(log_rets[sym]) sigma_ann = sigma_hr * math.sqrt(24 * 365) * 100 print(f" {sym}: σ_hr = {sigma_hr * 100:.3f}% σ_ann = {sigma_ann:.1f}%") # ── Event window: 25 May 2026 UTC, 24h ───────────────────── start_ms = int(EVENT_DAY_START_UTC.timestamp() * 1000) end_ms = int(EVENT_DAY_END_UTC.timestamp() * 1000) event_idx = [i for i, t in enumerate(common) if start_ms <= t < end_ms] event_ret_idx = [i - 1 for i in event_idx if i > 0] event_ret_idx = [i for i in event_ret_idx if 0 <= i < n_baseline] print(f"\n=== Event window: 25 May 2026 UTC ({len(event_idx)} bars) ===") # Log-return correlations within event window print(f"Pairwise correlation on log returns (n={len(event_ret_idx)}):") for a, b in [("BTCUSDT", "ETHUSDT"), ("BTCUSDT", "XRPUSDT"), ("ETHUSDT", "XRPUSDT")]: ra = [log_rets[a][i] for i in event_ret_idx] rb = [log_rets[b][i] for i in event_ret_idx] r = pearson(ra, rb) print(f" {a[:3]} ↔ {b[:3]}: ρ = {r:+.4f}") # Intra-bar correlations within event window print(f"Pairwise correlation on intra-bar returns (n={len(event_idx)}):") intra = {} for sym in SYMBOLS: intra[sym] = [] for t in (common[i] for i in event_idx): o, c = ohlc[sym][t] intra[sym].append((c - o) / o) for a, b in [("BTCUSDT", "ETHUSDT"), ("BTCUSDT", "XRPUSDT"), ("ETHUSDT", "XRPUSDT")]: r = pearson(intra[a], intra[b]) print(f" {a[:3]} ↔ {b[:3]}: ρ = {r:+.4f}") # ── V-bottom: 21:00 and 22:00 UTC bars ───────────────────── print(f"\n=== Synchronised V-bottom (21:00 / 22:00 UTC, 24 May 2026) ===") v_dip = int(V_DIP_UTC.timestamp() * 1000) v_bnc = int(V_BOUNCE_UTC.timestamp() * 1000) for sym in SYMBOLS: if v_dip in ohlc[sym]: o, c = ohlc[sym][v_dip] r = (c - o) / o * 100 sigma_hr = statistics.stdev(log_rets[sym]) r_sigma = (math.log(c / o)) / sigma_hr print(f" {sym} 21:00 UTC: open={o:.4f} close={c:.4f} " f"return={r:+.3f}% ({r_sigma:+.2f}σ)") if v_bnc in ohlc[sym]: o, c = ohlc[sym][v_bnc] r = (c - o) / o * 100 sigma_hr = statistics.stdev(log_rets[sym]) r_sigma = (math.log(c / o)) / sigma_hr print(f" {sym} 22:00 UTC: open={o:.4f} close={c:.4f} " f"return={r:+.3f}% ({r_sigma:+.2f}σ)") print(f"\nDone. Compare these figures to the paper, Section 3.") if __name__ == "__main__": main()