#!/usr/bin/env python3 """Does a quiet Asian session set up a big London move? The whole run. The question, the windows and the kill rule are in README.md, committed before this script was written. Nothing here decides anything the README did not fix. python analyse.py [--bars PATH] [--out DIR] [--seed N] [--draws N] Writes RESULTS.md, results.json and sessions.csv into --out (this directory by default), and the chart into site/public/images/ unless --no-chart. """ import argparse import json import pathlib import sys import numpy as np import pandas as pd BARS = "~/data/bars/XAUUSD_mt5_m1.csv" # The windows, in minutes past 23:00 UTC (the broker's day start). ASIAN = (0, 480) # 23:00–06:59 UTC LONDON = (480, 1020) # 07:00–15:59 UTC ASIAN_FLOOR = 360 # 75 % of 480 LONDON_FLOOR = 405 # 75 % of 540 MIN_DAYS = 120 RHO_GATE = 0.15 P_GATE = 0.01 def load(path): """One-minute bars, with the session day and the minute-of-session.""" df = pd.read_csv(path, usecols=["minute", "bid_open", "bid_high", "bid_low"], parse_dates=["minute"]) ts = df["minute"].dt.tz_convert("UTC") # the broker's day starts at 23:00 UTC: shift forward an hour and the date # of the shifted stamp is the session day, 00:00 of it being 23:00 UTC. shifted = ts + pd.Timedelta(hours=1) df["day"] = shifted.dt.date df["m"] = shifted.dt.hour * 60 + shifted.dt.minute return df def sessions(df): """One row per session day: both windows, in USD and in basis points.""" rows = [] for day, g in df.groupby("day", sort=True): out = {"day": str(day)} keep = True for name, (lo, hi), floor in (("asian", ASIAN, ASIAN_FLOOR), ("london", LONDON, LONDON_FLOOR)): w = g[(g["m"] >= lo) & (g["m"] < hi)] n = len(w) out[f"{name}_minutes"] = n if n < floor: keep = False out[f"{name}_usd"] = out[f"{name}_bps"] = float("nan") continue first = w.iloc[0] rng = float(w["bid_high"].max() - w["bid_low"].min()) out[f"{name}_open"] = float(first["bid_open"]) out[f"{name}_usd"] = rng out[f"{name}_bps"] = rng / float(first["bid_open"]) * 10_000 out["clears_floor"] = keep rows.append(out) return pd.DataFrame(rows) def spearman(a, b): """Rank correlation, no scipy: Pearson on the ranks, ties averaged.""" ra = pd.Series(a).rank().to_numpy() rb = pd.Series(b).rank().to_numpy() ra = ra - ra.mean() rb = rb - rb.mean() denom = np.sqrt((ra ** 2).sum() * (rb ** 2).sum()) return float((ra * rb).sum() / denom) if denom else float("nan") def permutation_p(a, b, draws, rng): """Two-sided p from shuffling one column against the other. Also hands back the null's own 99th percentile of |rho|, which is the correlation this test would need to see before it could call anything significant at p < 0.01. The null does not depend on the arrangement of `b` — permuting destroys any relation — so that percentile is the critical value for the sample size and tie structure, and the control below is entitled to reuse it instead of running a permutation test inside a permutation test. """ obs = spearman(a, b) b = np.asarray(b, dtype=float) hits = 0 null = np.empty(draws) for i in range(draws): null[i] = spearman(a, rng.permutation(b)) if abs(null[i]) >= abs(obs): hits += 1 # +1/+1: a permutation test never reports a p of exactly zero, because the # observed arrangement is itself one of the arrangements being counted. return (obs, (hits + 1) / (draws + 1), float(null.std()), float(np.quantile(np.abs(null), 1 - P_GATE))) def quintiles(df, col_x, col_y, k=5): """Median of y inside each k-tile of x, with the counts.""" q = pd.qcut(df[col_x], k, labels=False, duplicates="drop") out = [] for i in sorted(pd.unique(q.dropna())): sel = df[q == i] out.append({"bucket": int(i) + 1, "days": int(len(sel)), "asian_lo": float(sel[col_x].min()), "asian_hi": float(sel[col_x].max()), "asian_median": float(sel[col_x].median()), "london_median": float(sel[col_y].median()), "london_mean": float(sel[col_y].mean())}) return out def inject(ranks_x, target_rho, rng): """A y whose rank correlation with x is target_rho in expectation. Gaussian copula on the ranks: turn x's ranks into normal scores, mix with an independent normal at the weight that gives the Pearson correlation 2*sin(pi*rho/6) — the standard rank-to-linear conversion — and hand back the mixture. Only the ordering of the result is ever used. """ n = len(ranks_x) # normal scores of x's own ranks u = (ranks_x - 0.5) / n zx = np.sqrt(2) * _erfinv(2 * u - 1) r = 2 * np.sin(np.pi * target_rho / 6) z = r * zx + np.sqrt(max(0.0, 1 - r * r)) * rng.standard_normal(n) return z def _erfinv(y): """Inverse error function, vectorised, good to ~1e-9 (Giles 2010).""" y = np.clip(np.asarray(y, dtype=float), -1 + 1e-15, 1 - 1e-15) w = -np.log((1.0 - y) * (1.0 + y)) out = np.empty_like(y) small = w < 5.0 ws = w[small] - 2.5 p = 2.81022636e-08 for c in (3.43273939e-07, -3.5233877e-06, -4.39150654e-06, 0.00021858087, -0.00125372503, -0.00417768164, 0.246640727, 1.50140941): p = p * ws + c out[small] = p * y[small] big = ~small wb = np.sqrt(w[big]) - 3.0 p = -0.000200214257 for c in (0.000100950558, 0.00134934322, -0.00367342844, 0.00573950773, -0.0076224613, 0.00943887047, 1.00167406, 2.83297682): p = p * wb + c out[big] = p * y[big] return out def control(asian, gate, draws, rng, levels): """The floor: what size of effect would this test have caught? `gate` is the whole verdict rule collapsed into one number — the larger of the pre-registered size gate and the permutation test's own critical value at this sample size. Both have to be cleared, so the larger one is the rule. """ ranks_x = pd.Series(asian).rank().to_numpy() rows = [] for rho in levels: caught = 0 rhos = [] for _ in range(draws): y = inject(ranks_x, rho, rng) obs = spearman(asian, y) rhos.append(obs) if abs(obs) >= gate: caught += 1 rows.append({"target_rho": rho, "draws": draws, "certified": caught / draws, "median_measured": float(np.median(rhos))}) return rows def post_hoc(kept, r, rng, draws=10_000): """Two things the quintile table raised, neither of them pre-registered. They are reported as what they are: questions asked after seeing the answer, tested rather than told. """ q = pd.qcut(kept["asian_bps"], 5, labels=False, duplicates="drop") q1 = kept.loc[q == 0, "london_bps"].to_numpy() q2 = kept.loc[q == 1, "london_bps"].to_numpy() obs = float(np.median(q1) - np.median(q2)) pool = np.concatenate([q1, q2]) n1 = len(q1) hits = 0 for _ in range(draws): s = rng.permutation(pool) if abs(np.median(s[:n1]) - np.median(s[n1:])) >= abs(obs): hits += 1 # the windows are not the same length, so "which session is wider" is not a # fair question until the lengths are taken out. Range grows roughly with # the square root of time, so divide by it. a_min, l_min = ASIAN[1] - ASIAN[0], LONDON[1] - LONDON[0] a_scaled = kept["asian_bps"] / np.sqrt(a_min) l_scaled = kept["london_bps"] / np.sqrt(l_min) return { "quiet_bucket": { "q1_median_london_bps": float(np.median(q1)), "q2_median_london_bps": float(np.median(q2)), "difference": obs, "p": (hits + 1) / (draws + 1), "days": [int(n1), int(len(q2))], }, "window_length": { "asian_minutes": a_min, "london_minutes": l_min, "london_wider_raw": float((kept["london_bps"] > kept["asian_bps"]).mean()), "london_wider_per_sqrt_minute": float((l_scaled > a_scaled).mean()), "median_asian_per_sqrt_min": float(a_scaled.median()), "median_london_per_sqrt_min": float(l_scaled.median()), }, } def main(argv=None): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--bars", default=BARS) ap.add_argument("--out", default=str(pathlib.Path(__file__).parent)) ap.add_argument("--seed", type=int, default=20260910) ap.add_argument("--draws", type=int, default=10_000) ap.add_argument("--control-draws", type=int, default=500) ap.add_argument("--no-chart", action="store_true") args = ap.parse_args(argv) out = pathlib.Path(args.out) rng = np.random.default_rng(args.seed) df = load(pathlib.Path(args.bars).expanduser()) ses = sessions(df) ses.to_csv(out / "sessions.csv", index=False) kept = ses[ses["clears_floor"]].reset_index(drop=True) result = { "bars": int(len(df)), "first_bar": str(df["minute"].min()), "last_bar": str(df["minute"].max()), "session_days_seen": int(len(ses)), "session_days_kept": int(len(kept)), "floor": {"asian_minutes": ASIAN_FLOOR, "london_minutes": LONDON_FLOOR, "min_days": MIN_DAYS}, "seed": args.seed, } if len(kept) < MIN_DAYS: result["verdict"] = "no sample" result["why"] = (f"{len(kept)} session days cleared the floor, under the " f"pre-registered minimum of {MIN_DAYS}") (out / "results.json").write_text(json.dumps(result, indent=1) + "\n") print(result["why"]) return 1 for unit in ("bps", "usd"): rho, p, null_sd, crit = permutation_p(kept[f"asian_{unit}"].to_numpy(), kept[f"london_{unit}"].to_numpy(), args.draws, rng) result[unit] = {"rho": rho, "p": p, "null_sd": null_sd, "critical_rho": crit, "quintiles": quintiles(kept, f"asian_{unit}", f"london_{unit}")} rho, p = result["bps"]["rho"], result["bps"]["p"] if rho <= -RHO_GATE and p < P_GATE: verdict, why = "supported", "narrow Asian, wide London: the folklore's own sign" elif rho >= RHO_GATE and p < P_GATE: verdict, why = "refuted", "the correlation is positive: volatility clusters" else: verdict, why = "undecided", ("no correlation of the size the claim needs, " "in either direction") result["verdict"] = verdict result["why"] = why levels = [0.0, 0.05, 0.10, 0.15, 0.20, 0.25, 0.30, 0.40] gate = max(RHO_GATE, result["bps"]["critical_rho"]) result["effective_gate"] = gate result["control"] = control(kept["asian_bps"].to_numpy(), gate, args.control_draws, rng, levels) result["median_ranges_bps"] = { "asian": float(kept["asian_bps"].median()), "london": float(kept["london_bps"].median()), } result["median_ranges_usd"] = { "asian": float(kept["asian_usd"].median()), "london": float(kept["london_usd"].median()), } # the same-day question the claim is usually really about: does the London # window ever have to be wide at all? Share of days London > Asian. result["london_wider_share"] = float((kept["london_bps"] > kept["asian_bps"]).mean()) result["post_hoc"] = post_hoc(kept, result, rng) (out / "results.json").write_text(json.dumps(result, indent=1) + "\n") write_report(out, result, kept) if not args.no_chart: chart(kept, result) print(f"{verdict}: rho={rho:+.3f} p={p:.4f} on {len(kept)} session days") return 0 def write_report(out, r, kept): b = r["bps"] lines = [ "# Does a quiet Asian session set up a big London move?", "", f"**Verdict: {r['verdict']}** — {r['why']}.", "", f"Spearman rho = **{b['rho']:+.3f}** between the Asian range and the " f"London/NY range of the same session day, over **{r['session_days_kept']}** " f"session days ({r['session_days_seen']} seen, the rest short of the 75 % " f"minute floor). Permutation p = {b['p']:.4f} over 10,000 shuffles; the " f"null's own spread is {b['null_sd']:.3f}.", "", f"In raw USD rather than basis points: rho = {r['usd']['rho']:+.3f}, " f"p = {r['usd']['p']:.4f}. The normalisation does not decide it.", "", "## The quintile table (basis points)", "", "| Asian range | days | median Asian | median London | mean London |", "|---|---:|---:|---:|---:|", ] for q in b["quintiles"]: lines.append(f"| {q['bucket']} ({q['asian_lo']:.0f}–{q['asian_hi']:.0f} bp) " f"| {q['days']} | {q['asian_median']:.0f} | " f"{q['london_median']:.0f} | {q['london_mean']:.0f} |") lines += [ "", f"Median range: Asian {r['median_ranges_bps']['asian']:.0f} bp " f"({r['median_ranges_usd']['asian']:.2f} USD), London/NY " f"{r['median_ranges_bps']['london']:.0f} bp " f"({r['median_ranges_usd']['london']:.2f} USD). The London/NY window is " f"the wider of the two on {r['london_wider_share'] * 100:.1f} % of days.", "", "## The positive control", "", "A rank correlation of known size injected into this sample's own Asian " "ranks, 500 draws a level, judged by the verdict's own rule. That rule " f"is two gates — |rho| >= {RHO_GATE} and p < {P_GATE} — and at " f"{r['session_days_kept']} days the significance gate is the stricter of " f"the two: the permutation null's 99th percentile of |rho| is " f"{r['bps']['critical_rho']:.3f}. So the effective gate is " f"|rho| >= {r['effective_gate']:.3f}, and that is what the control below " "is judged against.", "", "| injected rho | certified | median measured |", "|---:|---:|---:|", ] for c in r["control"]: lines.append(f"| {c['target_rho']:.2f} | {c['certified'] * 100:.1f} % | " f"{c['median_measured']:+.3f} |") floor = next((c["target_rho"] for c in r["control"] if c["certified"] >= 0.95), None) null_rate = r["control"][0]["certified"] lines += [ "", (f"Smallest injected effect this test certifies at 95 % or better: " f"**rho = {floor:.2f}**." if floor is not None else "No injected level reached 95 % certification, which is itself the finding."), f" At an injected rho of zero it fires {null_rate * 100:.1f} % of the time.", "", "", "## Two things I looked at afterwards", "", "Neither was pre-registered. Both are questions the table above raised, " "and they are reported as questions asked after seeing the answer.", "", ] ph = r["post_hoc"] qb, wl = ph["quiet_bucket"], ph["window_length"] lines += [ f"**The quietest bucket breaks the trend.** From quintile 2 upwards the " f"median London range rises with the Asian range, but quintile 1 — the " f"quietest Asian sessions, the folklore's own case — has a median London " f"range of {qb['q1_median_london_bps']:.0f} bp against quintile 2's " f"{qb['q2_median_london_bps']:.0f} bp. That is the coiling story, in the " f"only place it could hide. It is {qb['days'][0]} days against " f"{qb['days'][1]}, a difference of {qb['difference']:+.0f} bp, and a " f"permutation test on the two buckets' medians gives p = {qb['p']:.3f}. " "It does not survive being looked at.", "", f"**The two windows are not the same length**, so \"which session is " f"wider\" is not a fair question as asked: London/NY is " f"{wl['london_minutes']} minutes against the Asian window's " f"{wl['asian_minutes']}. Raw, London/NY is the wider of the two on " f"{wl['london_wider_raw'] * 100:.1f} % of days. Dividing each by the " f"square root of its own length, which is roughly how a range grows with " f"time, that share is {wl['london_wider_per_sqrt_minute'] * 100:.1f} %.", "", "Method, data and the kill rule: `README.md`, committed before the run. " "This is a test record, not a recommendation.", ] (out / "RESULTS.md").write_text("\n".join(lines) + "\n") def chart(kept, r): """A scatter of the two ranges, on log axes, with the quintile medians.""" try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt except ImportError: return ink, dim, amber = "#E8E4DA", "#8A85A8", "#F0B429" fig, ax = plt.subplots(figsize=(7.2, 4.6), dpi=160) fig.patch.set_facecolor("#0F1120") ax.set_facecolor("#0F1120") ax.scatter(kept["asian_bps"], kept["london_bps"], s=12, c=dim, alpha=0.7, linewidths=0) qs = r["bps"]["quintiles"] ax.plot([q["asian_median"] for q in qs], [q["london_median"] for q in qs], color=amber, marker="o", markersize=5, linewidth=1.6, label="median London range, by Asian-range quintile") ax.set_xscale("log") ax.set_yscale("log") ax.set_xlabel("Asian range, 23:00–07:00 UTC (basis points)", color=ink) ax.set_ylabel("London/NY range, 07:00–16:00 UTC (bp)", color=ink) ax.tick_params(colors=dim) for s in ax.spines.values(): s.set_color("#2A2C42") ax.grid(True, color="#1D1F33", linewidth=0.6) leg = ax.legend(facecolor="#0F1120", edgecolor="#2A2C42", labelcolor=ink, fontsize=8, loc="upper left") leg.get_frame().set_alpha(0.9) ax.set_title(f"XAUUSD, {r['session_days_kept']} session days, 2026 " f"(rho = {r['bps']['rho']:+.2f})", color=ink, fontsize=10) fig.tight_layout() target = pathlib.Path(__file__).resolve().parents[2] / \ "site/public/images/gold-asian-vs-london.png" fig.savefig(target, facecolor=fig.get_facecolor()) print(f"chart -> {target}") if __name__ == "__main__": sys.exit(main())