#!/usr/bin/env python3 """Is gold sold down into the London AM fix? The whole run. The question, the windows, the statistic, the null, the day floor and the kill rule are in `rule.py`, sealed before a single 2025 tick was on this server. Nothing here decides anything rule.py did not fix: the constants and the three decision functions (`verdict`, `critical_k`, `certified_floor`) are imported from it rather than restated, so a reader can diff this file against the seal and see that the rule was not quietly edited to fit the answer. python analyse.py [--bars PATH] [--out DIR] [--seed N] [--draws N] Writes results.md, results.json and days.csv into --out (this directory by default). results.md has no author but this script; it is never hand-edited. Decisions taken here that rule.py left to the implementation (there is no one to ask, so they are written down): 1. The placebo pool is the 226 start minutes 08:00..11:45 London inclusive; a start is valid when a bar exists at S and at S+15 on that London day. The day floor is "both the 10:15 and the 10:30 bar exist" AND "at least rule.POOL_FLOOR (180) of the rule.POOL_N (226) starts are valid", exactly as sealed. The 10:15 start is one of the 226 and is not excluded, as sealed. 2. The seal says the missing count is printed, and the count it means is the fetch's holes. Implemented as: London calendar dates that are Monday to Friday, between the first and the last London day that has any bar (inclusive), with no 10:15 bar or no 10:30 bar. The span is taken from the data rather than hard-coded to calendar 2025 so that the same script is honest about any range it is handed (a smoke test on one month would otherwise report ten months of "missing" days). Public holidays are not subtracted: this server has no holiday calendar it did not invent, so a holiday is reported as a hole and a reader can see it is one. 3. The clock-regime split (GMT days against BST days) uses the Europe/London UTC offset of that day's 10:15 wall-clock timestamp, computed from the zoneinfo database rather than inferred from the bars. BST is offset +1h. 4. Descriptive windows (10:30->10:40, 10:40->10:55) are reported on the kept days that have both of their own bars; the dropped count is printed. They carry no placebo p: the seal asks only for a mean and a fraction negative. The mean run-in does carry a placebo p, because the seal names one. 5. The null's own size at the critical count k01 is computed with rule.py's own binomial tail (`rule._binom_tail`) rather than a second implementation, for the same reason as (the rest of) the imports. It is private only by naming. 6. `rule.branch_table` is printed in results.md at the observed n and the observed C, so what the sample could have concluded is on the record beside what it did conclude. 7. There is no shift-injection control: this rule does not ask for one, and adding one would be a rule the seal does not contain. """ import argparse import json import os import pathlib import sys import time from datetime import datetime, timedelta from zoneinfo import ZoneInfo import numpy as np import pandas as pd sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import rule # noqa: E402 from rule import certified_floor, critical_k, verdict # noqa: E402 BARS = "~/data/bars/XAUUSD_2025_m1.csv" LONDON = "Europe/London" TZ = ZoneInfo(LONDON) # minute-of-London-day indices M_1015, M_1030, M_1040, M_1055 = 10 * 60 + 15, 10 * 60 + 30, 10 * 60 + 40, 10 * 60 + 55 POOL_LO, POOL_HI = 8 * 60, 11 * 60 + 45 # 08:00 .. 11:45 inclusive WINDOW = 15 # minutes assert POOL_HI - POOL_LO + 1 == rule.POOL_N # 226, as sealed WEEKDAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] # ----------------------------------------------------------------- the data def load(path): """One-minute bars on the Europe/London clock, mid of the closes. Returns the frame and a (days x 1440) matrix of mid prices, NaN where the feed quoted no bar, so that every window below is an array slice. """ df = pd.read_csv(path, usecols=["minute", "bid_close", "ask_close"], parse_dates=["minute"]) ts = df["minute"].dt.tz_convert(LONDON) df["mid"] = (df["bid_close"] + df["ask_close"]) / 2.0 df["day"] = ts.dt.date df["m"] = ts.dt.hour * 60 + ts.dt.minute days = np.array(sorted(df["day"].unique())) day_ix = {d: i for i, d in enumerate(days)} mid = np.full((len(days), 1440), np.nan) mid[df["day"].map(day_ix).to_numpy(), df["m"].to_numpy()] = df["mid"].to_numpy() return df, days, mid def bp(a, b): """ln(b/a) x 10,000, NaN-safe.""" with np.errstate(invalid="ignore", divide="ignore"): return np.log(b / a) * 1e4 def is_bst(day): """True if 10:15 London on this date is BST (+1h) rather than GMT.""" return datetime(day.year, day.month, day.day, 10, 15, tzinfo=TZ).utcoffset() != timedelta(0) def missing_weekdays(days, mid): """Weekday London dates in the span with no 10:15 bar or no 10:30 bar. The fetch's holes, decision 2 in the docstring: the span is first-to-last observed London day, holidays included as holes. """ have = {d for d, a, b in zip(days, mid[:, M_1015], mid[:, M_1030]) if np.isfinite(a) and np.isfinite(b)} missing, d = [], days[0] while d <= days[-1]: if d.weekday() < 5 and d not in have: missing.append(str(d)) d += timedelta(days=1) return missing # ------------------------------------------------------------ placebo table def placebo_table(mid): """Per day, the 15-minute log returns (bp) of every valid pool start. Hands back a padded (days x 226) matrix of the *valid* returns packed to the left, and the per-day count of valid starts, which is what a draw indexes into. """ starts = mid[:, POOL_LO:POOL_HI + 1] ends = mid[:, POOL_LO + WINDOW:POOL_HI + WINDOW + 1] rets = bp(starts, ends) valid = np.isfinite(rets) sizes = valid.sum(axis=1) # pack each row's valid returns to the left without a Python loop over days order = np.argsort(~valid, axis=1, kind="stable") packed = np.take_along_axis(rets, order, axis=1) cols = np.arange(rets.shape[1])[None, :] packed[cols >= sizes[:, None]] = np.nan return packed, sizes def draw(packed, sizes, draws, rng): """`draws` placebo draws: one uniformly chosen valid start per day. Returns the (draws x days) matrix of chosen returns. Vectorised: an index matrix is drawn once, scaled per day by that day's own pool size. """ u = rng.random((draws, len(sizes))) idx = (u * sizes[None, :]).astype(np.int64) return packed[np.arange(len(sizes))[None, :], idx] def null_stats(packed, sizes, draws, rng): """Fraction-negative and mean statistics over `draws` placebo draws.""" r = draw(packed, sizes, draws, rng) neg = (r < 0).sum(axis=1) return neg / r.shape[1], neg, r.mean(axis=1) # -------------------------------------------------------------- the per-day def build_days(days, mid, sizes): """One row per London calendar day, with the floor decision and its reason.""" m1015, m1030 = mid[:, M_1015], mid[:, M_1030] rows = pd.DataFrame({ "date": [str(d) for d in days], "weekday": [WEEKDAYS[d.weekday()] for d in days], "regime": ["BST" if is_bst(d) else "GMT" for d in days], "mid_1015": m1015, "mid_1030": m1030, "run_in_bp": bp(m1015, m1030), "pool_size": sizes, "auction_bp": bp(m1030, mid[:, M_1040]), "after_bp": bp(mid[:, M_1040], mid[:, M_1055]), }) has_bars = np.isfinite(m1015) & np.isfinite(m1030) clears = sizes >= rule.POOL_FLOOR rows["kept"] = has_bars & clears pool_text = f"placebo pool under {rule.POOL_FLOOR} of {rule.POOL_N}" reason = np.where(rows["kept"], "kept", np.where(~has_bars & ~clears, f"no 10:15/10:30 bar and {pool_text}", np.where(~has_bars, "no 10:15/10:30 bar", pool_text))) rows["drop_reason"] = reason return rows # ------------------------------------------------------------------ the test def sign_test(obs, packed, sizes, draws, rng): """F, C, p for a per-day return vector against the paired placebo null.""" f = float((obs < 0).mean()) frac, neg, means = null_stats(packed, sizes, draws, rng) c = float(frac.mean()) p = float((frac >= f).mean()) return f, c, p, frac, neg, means # -------------------------------------------------------------------- output def main(argv=None): t0 = time.time() 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=20260912) ap.add_argument("--draws", type=int, default=10_000) args = ap.parse_args(argv) out = pathlib.Path(args.out) out.mkdir(parents=True, exist_ok=True) df, days, mid = load(pathlib.Path(args.bars).expanduser()) print(f"bars: {len(df)} rows, {df['minute'].min()} .. {df['minute'].max()}") packed, sizes = placebo_table(mid) rows = build_days(days, mid, sizes) rows.to_csv(out / "days.csv", index=False, columns=["date", "weekday", "regime", "mid_1015", "mid_1030", "run_in_bp", "pool_size", "auction_bp", "after_bp", "kept"]) kept = rows[rows["kept"]].reset_index(drop=True) k_ix = np.flatnonzero(rows["kept"].to_numpy()) kpacked, ksizes = packed[k_ix], sizes[k_ix] n = len(kept) drops = rows.loc[~rows["kept"], "drop_reason"].value_counts().to_dict() missing = missing_weekdays(days, mid) result = { "bars": int(len(df)), "first_bar": str(df["minute"].min()), "last_bar": str(df["minute"].max()), "days_seen": int(len(rows)), "n": n, "drops": drops, "coverage": { "span_first_london_day": str(days[0]), "span_last_london_day": str(days[-1]), "weekdays_missing_a_fix_bar": len(missing), "weekdays_missing_a_fix_bar_dates": missing, }, "seed": args.seed, "draws": args.draws, "floor": {"pool_min": rule.POOL_FLOOR, "pool_size": rule.POOL_N, "min_days": rule.MIN_DAYS}, } run_in = kept["run_in_bp"].to_numpy() rng = np.random.default_rng(args.seed) f, c, p, frac, neg, means = sign_test(run_in, kpacked, ksizes, args.draws, rng) k01 = critical_k(n, c, rule.P_SUPPORT) q_star = certified_floor(n, c, k01) size_exact = float(rule._binom_tail(n, c, k01)) # the rule's own tail size_empirical = float((neg >= k01).mean()) v = verdict(f, c, p, q_star, n) result["test"] = { "F": f, "down_days": int((run_in < 0).sum()), "C": c, "p": p, "k01": k01, "q_star": q_star, "null_size_at_k01_exact": size_exact, "null_size_at_k01_empirical": size_empirical, "null_sd": float(frac.std()), } result["verdict"] = v result["published_per_day"] = n >= rule.MIN_DAYS # --- the branch table at the observed n and C ---------------------- bk01, bk05, bq_star, brows = rule.branch_table(n, c) result["branch_table"] = { "n": n, "C": c, "k01": bk01, "k05": bk05, "q_star": bq_star, "rows": [{"true_rate": q, "supported": ps, "killed": pk, "inconclusive": pi} for q, ps, pk, pi in brows], } # --- descriptives ------------------------------------------------- mu = float(run_in.mean()) p_mu = float((means <= mu).mean()) desc = {"mu_bp": mu, "sd_bp": float(run_in.std(ddof=1)), "p_mu": p_mu, "null_mean_centre_bp": float(means.mean())} for name, col in (("auction", "auction_bp"), ("after", "after_bp")): v_ = kept[col].to_numpy() ok = np.isfinite(v_) desc[name] = {"days": int(ok.sum()), "dropped_no_bar": int((~ok).sum()), "mean_bp": float(v_[ok].mean()) if ok.any() else None, "fraction_negative": float((v_[ok] < 0).mean()) if ok.any() else None} pool_abs = np.abs(kpacked[np.isfinite(kpacked)]) desc["median_abs_run_in_bp"] = float(np.median(np.abs(run_in))) desc["median_abs_placebo_bp"] = float(np.median(pool_abs)) desc["placebo_windows_pooled"] = int(pool_abs.size) def by(column, values): out_rows = [] for key in values: sel = kept[kept[column] == key] if len(sel): out_rows.append({column: key, "days": int(len(sel)), "F": float((sel["run_in_bp"] < 0).mean()), "mean_bp": float(sel["run_in_bp"].mean())}) return out_rows desc["weekday"] = by("weekday", WEEKDAYS) desc["regime"] = by("regime", ["GMT", "BST"]) result["descriptive"] = desc result["runtime_seconds"] = round(time.time() - t0, 1) (out / "results.json").write_text(json.dumps(result, indent=1) + "\n") write_report(out, result) print(f"verdict: {v} — F={f:.4f} C={c:.4f} p={p:.4f} n={n} " f"k01={k01} q*={q_star} size@k01={size_exact:.4f} " f"(empirical {size_empirical:.4f})") if not result["published_per_day"]: print(f"fewer than {rule.MIN_DAYS} days survived the floor: " "no per-day figure is published") print(f"MU={mu:+.3f} bp sd={desc['sd_bp']:.3f} p_mu={p_mu:.4f}") print(f"auction 10:30->10:40 mean={desc['auction']['mean_bp']:+.3f} bp " f"neg={desc['auction']['fraction_negative']:.4f}; " f"after 10:40->10:55 mean={desc['after']['mean_bp']:+.3f} bp " f"neg={desc['after']['fraction_negative']:.4f}") print(f"median |run_in| = {desc['median_abs_run_in_bp']:.3f} bp vs " f"median |placebo| = {desc['median_abs_placebo_bp']:.3f} bp") for r in desc["weekday"]: print(f" {r['weekday']:<9} n={r['days']:>3} F={r['F']:.4f} " f"mean={r['mean_bp']:+.3f} bp") for r in desc["regime"]: print(f" {r['regime']:<9} n={r['days']:>3} F={r['F']:.4f} " f"mean={r['mean_bp']:+.3f} bp") for k, vcount in drops.items(): print(f" dropped {vcount}: {k}") print(f"coverage: {len(rows)} London days seen, {days[0]} .. {days[-1]}; " f"{len(missing)} weekday London days in the span with no 10:15 or " "10:30 bar") print(f"runtime: {result['runtime_seconds']} s") return 0 def write_report(out, r): t, d, c = r["test"], r["descriptive"], r["coverage"] n = r["n"] lines = [ "# Is gold sold down into the London AM fix?", "", f"**Verdict: {r['verdict']}** — F = {t['F']:.4f} against a placebo centre " f"C = {t['C']:.4f}, one-sided p = {t['p']:.4f}, on {n} London days.", "", f"Bars: {r['bars']} one-minute rows, {r['first_bar']} to {r['last_bar']} " f"(UTC as stored; every window below is on the Europe/London clock). " f"Price is the mid of the bid and ask closes. " f"{r['days_seen']} London days have at least one bar, " f"{c['span_first_london_day']} to {c['span_last_london_day']}, and " f"{c['weekdays_missing_a_fix_bar']} Monday-to-Friday London dates in that " f"span have no 10:15 or no 10:30 bar at all (holidays are counted as holes: " f"this is the fetch's coverage, not a trading calendar).", "", ] if not r["published_per_day"]: lines += [f"Fewer than {rule.MIN_DAYS} days survived the floor " f"({n}), so **no per-day figure is published**, as the rule " "requires.", ""] lines += [ "## The test", "", "| | |", "|---|---:|", f"| days kept, n | {n} (of {r['days_seen']} London days seen) |", f"| down days (run_in < 0) | {t['down_days']} |", f"| F, fraction negative | {t['F']:.4f} |", f"| C, placebo centre | {t['C']:.4f} |", f"| null sd of F | {t['null_sd']:.4f} |", f"| one-sided p | {t['p']:.4f} |", f"| critical count k01 (p < 0.01 at C) | {t['k01']} |", f"| certified floor q\\* | {t['q_star']} |", f"| null's own size at k01, exact binomial | {t['null_size_at_k01_exact']:.4f} |", f"| null's own size at k01, from the draws | {t['null_size_at_k01_empirical']:.4f} |", "", f"run_in is ln(mid(10:30) / mid(10:15)) x 10,000 bp. The null is the " f"pre-registered placebo-window null: {r['draws']} draws, each picking one " f"start minute per day uniformly from the {r['floor']['pool_size']} minutes " f"08:00-11:45 London that have a bar exactly 15 minutes later (10:15 is one " f"of them and is not excluded). A day survives if its 10:15 and 10:30 bars " f"exist and at least {r['floor']['pool_min']} of those starts are valid. " f"Seed {r['seed']}.", "", "## Days dropped by the floor", "", "| reason | days |", "|---|---:|", ] for k, v in r["drops"].items(): lines.append(f"| {k} | {v} |") if not r["drops"]: lines.append("| (none) | 0 |") b = r["branch_table"] lines += [ "", "## What this sample could have concluded", "", f"The sealed branch table (`rule.branch_table`) at the observed n = {b['n']} " f"and C = {b['C']:.4f}: k01 = {b['k01']}, k05 = {b['k05']}, " f"q\\* = {b['q_star']}. Each row is the chance this many days would have " "landed in each branch had the true down-rate been that value.", "", "| true down-rate | supported | killed | inconclusive |", "|---:|---:|---:|---:|", ] for row in b["rows"]: lines.append(f"| {row['true_rate']:.2f} | {row['supported']:.4f} | " f"{row['killed']:.4f} | {row['inconclusive']:.4f} |") lines += [ "", "## Descriptive only — none of this can move the verdict", "", f"**The mean.** MU = {d['mu_bp']:+.4f} bp, per-day sd " f"{d['sd_bp']:.4f} bp, placebo p = {d['p_mu']:.4f} (the fraction of draws " f"whose mean is at or below MU; the null's own mean centre is " f"{d['null_mean_centre_bp']:+.4f} bp).", "", f"**The auction itself, 10:30 to 10:40.** Mean " f"{d['auction']['mean_bp']:+.4f} bp, negative on " f"{d['auction']['fraction_negative'] * 100:.1f} % of " f"{d['auction']['days']} days ({d['auction']['dropped_no_bar']} kept days " "lacked one of its two bars).", "", f"**After it, 10:40 to 10:55.** Mean {d['after']['mean_bp']:+.4f} bp, " f"negative on {d['after']['fraction_negative'] * 100:.1f} % of " f"{d['after']['days']} days ({d['after']['dropped_no_bar']} kept days " "lacked one of its two bars).", "", f"**Busy, not just lower.** Median |run_in| = " f"{d['median_abs_run_in_bp']:.4f} bp against a median |placebo return| of " f"{d['median_abs_placebo_bp']:.4f} bp, pooled over all " f"{d['placebo_windows_pooled']} valid placebo windows.", "", "**By weekday** (descriptive; the rule fixes no weekday claim).", "", "| weekday | days | F | mean run-in |", "|---|---:|---:|---:|", ] for w in d["weekday"]: lines.append(f"| {w['weekday']} | {w['days']} | {w['F']:.4f} | " f"{w['mean_bp']:+.3f} bp |") lines += [ "", "**By clock regime** (the London UTC offset of that day's 10:15; " "descriptive, and the split is small enough to be noise).", "", "| regime | days | F | mean run-in |", "|---|---:|---:|---:|", ] for w in d["regime"]: lines.append(f"| {w['regime']} | {w['days']} | {w['F']:.4f} | " f"{w['mean_bp']:+.3f} bp |") lines += [ "", # runtime deliberately not printed here: results.md must be byte-identical # on a re-run from a clean clone, and a timing would break that. "Method, windows and kill rule: `rule.py`, sealed before a single tick of " "this year's data was on the server. This file is written by `analyse.py` " "and never by hand.", "", "This is a test record, not advice.", ] (out / "results.md").write_text("\n".join(lines) + "\n") if __name__ == "__main__": sys.exit(main())