# Published copy. The two paths below are set to environment variables so the # script runs against any clean one-minute dataset with columns t, o, h, l. # The original reads them from the lab's own store. # -*- coding: utf-8 -*- """ Audit of snapshot v1, written after an external methodology review, 20 Sep 2026. Answers, from the raw files and nothing else: 1. where every row goes between the raw export and the published bar count 2. how many distinct duplicate timestamps carry conflicting values, and when 3. Study 01 under both session populations 4. Study 02 under the original >=350 bar filter and under an exact 390/390 minute grid, so the two can be compared rather than swapped 5. which symbol each source export used, continuous or a named contract Read-only. It never writes to the desk store. """ import glob, os, json, sys import numpy as np import pandas as pd STORE = os.environ.get("BARS_DIR", "./bars") # one-minute CSVs: t,o,h,l TICKS = os.environ.get("TICKS_DIR", "./ticks") # export summaries OUT = os.path.dirname(os.path.abspath(__file__)) TZ = "America/New_York" CUTOFF = pd.Timestamp("2026-09-17", tz=TZ) + pd.Timedelta(days=1) # exclusive OPEN_M, CLOSE_M = 570, 960 # 09:30, 16:00 FULL_GRID = list(range(OPEN_M, CLOSE_M)) # 390 minutes def load(symbol): files = sorted(glob.glob(os.path.join(STORE, f"{symbol}_FLOW_*.csv"))) raw = pd.concat([pd.read_csv(f, usecols=["t", "o", "h", "l", "c", "v"]) for f in files]) return files, raw def row_ledger(raw): """Every row accounted for, from the export to the published bar count.""" n_raw = len(raw) dedup = raw.drop_duplicates("t").sort_values("t") n_dedup = len(dedup) dedup = dedup.assign(ts=pd.to_datetime(dedup.t, unit="ms", utc=True).dt.tz_convert(TZ)).set_index("ts") kept = dedup[dedup.index < CUTOFF] return dict( raw_rows=n_raw, duplicate_rows_removed=n_raw - n_dedup, rows_after_dedup=n_dedup, rows_after_cutoff_removed=n_dedup - len(kept), published_bars=len(kept), ), kept def duplicate_conflicts(raw): """A duplicate timestamp only matters if its copies disagree.""" d = raw[raw.duplicated("t", keep=False)] per_ts = d.groupby("t")[["o", "h", "l", "c", "v"]].nunique().max(axis=1) conflicting = per_ts[per_ts > 1].index rows = [] for t in conflicting: ts = pd.Timestamp(t, unit="ms", tz="UTC").tz_convert(TZ) mins = ts.hour * 60 + ts.minute rows.append(dict(timestamp=str(ts), inside_session=bool(OPEN_M <= mins < CLOSE_M))) return dict( timestamps_with_duplicates=int(len(per_ts)), identical_copies=int((per_ts == 1).sum()), conflicting_copies=int((per_ts > 1).sum()), conflicting_inside_session=int(sum(r["inside_session"] for r in rows)), conflicts=rows, ) def sessions(bars): """Split the RTH minute bars into sessions and label how complete each is.""" m = bars.index.hour * 60 + bars.index.minute rth = bars[(m >= OPEN_M) & (m < CLOSE_M)] out = {} for day, g in rth.groupby(rth.index.date): mins = sorted((g.index.hour * 60 + g.index.minute).tolist()) out[day] = dict( n=len(g), exact_grid=(mins == FULL_GRID), missing=len(set(FULL_GRID) - set(mins)), group=g, ) return out def study01(bars, sess): """Median high-to-low range of a 5-minute RTH candle, both populations.""" b5 = bars.resample("5min").agg({"h": "max", "l": "min"}).dropna() m = b5.index.hour * 60 + b5.index.minute rth5 = b5[(m >= OPEN_M) & (m < CLOSE_M)] rng = rth5.h - rth5.l def summarise(mask, label): r = rng[mask] days = sorted({d for d, keep in zip(rth5.index.date, mask) if keep}) return dict(label=label, candles=int(len(r)), sessions=len(days), median=round(float(r.median()), 2), mean=round(float(r.mean()), 2)) allmask = np.ones(len(rng), dtype=bool) strict_days = {d for d, v in sess.items() if v["exact_grid"]} strictmask = np.array([d in strict_days for d in rth5.index.date]) f350_days = {d for d, v in sess.items() if v["n"] >= 350} f350mask = np.array([d in f350_days for d in rth5.index.date]) hh = (rth5.index.hour * 60 + rth5.index.minute) // 30 * 30 by_half = {f"{h//60:02d}:{h%60:02d}": round(float(rng[hh == h].median()), 2) for h in sorted(set(hh))} return dict( all_sessions=summarise(allmask, "every RTH session in the window"), filter_350=summarise(f350mask, "sessions with at least 350 one-minute bars"), complete_grid=summarise(strictmask, "sessions with all 390 minutes present"), by_half_hour_median=by_half, exceeds={str(x): round(float((rng > x).mean()) * 100, 2) for x in (4, 6, 8, 10, 15, 20, 30)}, ) def study02(sess, key): """When the session high or low is first printed, under one completeness rule.""" hi, lo = [], [] for d, v in sess.items(): if not key(v): continue g = v["group"] mm = (g.index.hour * 60 + g.index.minute) - OPEN_M hi.append(mm[int(np.argmax(g.h.values))]) lo.append(mm[int(np.argmin(g.l.values))]) hi, lo = np.array(hi), np.array(lo) n = len(hi) def ci(p): se = (p * (1 - p) / n) ** 0.5 return [round((p - 1.96 * se) * 100, 1), round((p + 1.96 * se) * 100, 1)] first60 = ((hi < 60) | (lo < 60)).mean() both60 = ((hi < 60) & (lo < 60)).mean() first30 = ((hi < 30) | (lo < 30)).mean() last60 = ((hi >= 330) | (lo >= 330)).mean() return dict( sessions=n, high_or_low_first60=round(float(first60) * 100, 1), high_or_low_first60_ci95=ci(first60), both_first60=round(float(both60) * 100, 1), both_first60_ci95=ci(both60), exactly_one_first60=round(float(first60 - both60) * 100, 1), neither_first60=round(float(1 - first60) * 100, 1), high_or_low_first30=round(float(first30) * 100, 1), high_or_low_last60=round(float(last60) * 100, 1), ) def manifest(): """Which symbol each source export actually used.""" rows = [] for p in sorted(glob.glob(os.path.join(TICKS, "*.summary.txt"))): meta = {} for line in open(p, encoding="utf-8", errors="replace"): if "=" in line: k, _, v = line.partition("=") meta[k.strip()] = v.strip() sym = meta.get("raw_symbol", "") rows.append(dict( export=os.path.basename(p).replace(".summary.txt", ""), symbol=meta.get("symbol", ""), raw_symbol=sym, kind="continuous front" if sym.count(":") and len(sym.split(":")[0]) <= 3 else "named contract", mode=meta.get("mode", ""), first=meta.get("first_time_utc", ""), last=meta.get("last_time_utc", ""), ticks=meta.get("recognized_items", ""), )) return rows def main(): report = dict(generated="2026-09-20", cutoff="2026-09-17", session="09:30 to 16:00 America/New_York", symbols={}) for sym in ("NQ", "ES"): files, raw = load(sym) ledger, bars = row_ledger(raw) sess = sessions(bars) report["symbols"][sym] = dict( source_files=[os.path.basename(f) for f in files], row_ledger=ledger, duplicates=duplicate_conflicts(raw), session_counts=dict( calendar_sessions=len(sess), at_least_350_bars=sum(1 for v in sess.values() if v["n"] >= 350), complete_390_grid=sum(1 for v in sess.values() if v["exact_grid"]), ), study01=study01(bars, sess), study02_filter_350=study02(sess, lambda v: v["n"] >= 350), study02_complete_grid=study02(sess, lambda v: v["exact_grid"]), ) report["source_manifest"] = manifest() path = os.path.join(OUT, "audit_v1.json") with open(path, "w", encoding="utf-8") as f: json.dump(report, f, indent=2) print("written", path) for sym, d in report["symbols"].items(): L = d["row_ledger"] print(f"\n=== {sym}") print(" {raw_rows} raw - {duplicate_rows_removed} duplicate - " "{rows_after_cutoff_removed} after cutoff = {published_bars}".format(**L)) print(" duplicate timestamps:", d["duplicates"]["timestamps_with_duplicates"], "| conflicting:", d["duplicates"]["conflicting_copies"], "| conflicting inside session:", d["duplicates"]["conflicting_inside_session"]) print(" sessions: calendar {calendar_sessions} | >=350 bars {at_least_350_bars} " "| complete grid {complete_390_grid}".format(**d["session_counts"])) for k in ("all_sessions", "filter_350", "complete_grid"): s = d["study01"][k] print(f" study01 {k:<14} candles {s['candles']:>6} sessions {s['sessions']:>4} median {s['median']}") for k in ("study02_filter_350", "study02_complete_grid"): s = d[k] print(f" {k:<22} n={s['sessions']:>4} one edge {s['high_or_low_first60']}% " f"{s['high_or_low_first60_ci95']} both {s['both_first60']}% {s['both_first60_ci95']}") if __name__ == "__main__": main()