#!/usr/bin/env python3 """ first_insight.py -- reproduces every number and table in GUIDE.md from the Kalshi Microstructure Tape "tryout" ZIP ($9 SKU, 2 UTC days: 2026-07-18/19). v12.322 (2026-09-25, jobs 3/4): first build of the tryout buyer walkthrough (JOB 3). Reads the tryout ZIP directly (no extraction to disk), reproduces four concrete analyses, writes tables/PNGs to outputs/, and writes outputs/numbers.json -- the single source of truth that check_guide.py diffs GUIDE.md against. No signals, no strategy claims: this is a data walkthrough only (house rule). Usage: python3 first_insight.py --zip /path/to/kalshi_micro_tryout.zip --outdir outputs Dependencies: pandas, pyarrow, matplotlib (matplotlib only used for the two optional PNG charts; everything else is table output). No network access, no other third-party packages. Runtime: ~15-30s on a laptop (measured; see RUN_LOG.txt written alongside outputs/ after a full run). Well under the 5-minute budget. """ from __future__ import annotations import argparse import io import json import sys import time import zipfile from pathlib import Path import numpy as np import pandas as pd pd.set_option("display.width", 120) MEMBER_PREFIX = "kalshi_micro_tryout/" DEPTH_FILES = [ "depth/kalshi_depth_2026-07-18.parquet", "depth/kalshi_depth_2026-07-19.parquet", ] TRADE_FILES = [ "trades/kalshi_trades_2026-07-18.parquet", "trades/kalshi_trades_2026-07-19.parquet", ] DOC_FILES = ["MANIFEST.json", "SOURCE_MANIFEST.json", "TRYOUT_STATS.json", "COVERAGE.md", "SCHEMA.md", "README.md"] # -------------------------------------------------------------------------- # Loading -- reads straight out of the ZIP member, no unzip-to-disk step. # -------------------------------------------------------------------------- def read_zip_member(z: zipfile.ZipFile, relpath: str) -> bytes: with z.open(MEMBER_PREFIX + relpath) as f: return f.read() def load_parquet_member(z: zipfile.ZipFile, relpath: str) -> pd.DataFrame: return pd.read_parquet(io.BytesIO(read_zip_member(z, relpath))) def load_all(zip_path: Path) -> dict: with zipfile.ZipFile(zip_path) as z: names = set(z.namelist()) for f in DEPTH_FILES + TRADE_FILES + DOC_FILES: member = MEMBER_PREFIX + f if member not in names: raise FileNotFoundError(f"expected member missing from zip: {member}") manifest = json.loads(read_zip_member(z, "MANIFEST.json")) tryout_stats = json.loads(read_zip_member(z, "TRYOUT_STATS.json")) depth = pd.concat( [load_parquet_member(z, f) for f in DEPTH_FILES], ignore_index=True ) trades = pd.concat( [load_parquet_member(z, f) for f in TRADE_FILES], ignore_index=True ) trades["created_time"] = pd.to_datetime( trades["created_time"], utc=True, format="ISO8601" ) depth["ts"] = pd.to_datetime(depth["timestamp"], unit="s", utc=True) return {"manifest": manifest, "tryout_stats": tryout_stats, "depth": depth, "trades": trades} def best_of_book(depth: pd.DataFrame, ticker_prefix: str | None = None) -> pd.DataFrame: """One row per (ticker, ts) with best_bid/best_ask/mid/spread_c, level 0 only.""" d = depth if ticker_prefix is not None: d = d[d["ticker"].str.startswith(ticker_prefix)] lvl0 = d[d["level"] == 0] bb = lvl0[lvl0["side"] == "bid"][["ticker", "ts", "price_c", "size"]].rename( columns={"price_c": "best_bid", "size": "bid_size"} ) ba = lvl0[lvl0["side"] == "ask"][["ticker", "ts", "price_c", "size"]].rename( columns={"price_c": "best_ask", "size": "ask_size"} ) book = pd.merge(bb, ba, on=["ticker", "ts"], how="outer").sort_values( ["ticker", "ts"] ).reset_index(drop=True) book["mid"] = (book["best_bid"] + book["best_ask"]) / 2 book["spread_c"] = book["best_ask"] - book["best_bid"] book["depth0"] = book["bid_size"].fillna(0) + book["ask_size"].fillna(0) return book # -------------------------------------------------------------------------- # Analysis 0 -- manifest cross-check (feeds the "what's in the ZIP" section) # -------------------------------------------------------------------------- def analysis0_manifest_check(data: dict, numbers: dict) -> dict: manifest = data["manifest"] depth, trades = data["depth"], data["trades"] checks = { "depth_rows_match_manifest": len(depth) == manifest["coverage_summary"]["depth_rows"], "trade_rows_match_manifest": len(trades) == manifest["coverage_summary"]["unique_trade_prints"], "trade_id_unique": trades["trade_id"].nunique() == len(trades), } assert all(checks.values()), f"manifest cross-check failed: {checks}" fam_counts = ( trades["ticker"].str.extract(r"^(KX[A-Z0-9]+GAME)-")[0] .value_counts() .to_dict() ) out = { "n_depth_rows": int(len(depth)), "n_trade_rows": int(len(trades)), "n_utc_days": 2, "utc_days": manifest["coverage_summary"]["utc_days"], "n_depth_ticker_families": int(depth["ticker"].str.extract(r"^([A-Z0-9]+)-")[0].nunique()), "n_trade_print_families": int(len(fam_counts)), "trade_family_counts": {k: int(v) for k, v in fam_counts.items()}, "n_global_poll_timestamps": int(depth["timestamp"].nunique()), "median_poll_interval_s": None, # filled by analysis4 "manifest_checks_passed": checks, } numbers["n_depth_rows"] = f"{out['n_depth_rows']:,}" numbers["n_trade_rows"] = f"{out['n_trade_rows']:,}" numbers["n_depth_ticker_families"] = str(out["n_depth_ticker_families"]) numbers["n_trade_print_families"] = str(out["n_trade_print_families"]) numbers["mlb_prints"] = f"{out['trade_family_counts'].get('KXMLBGAME', 0):,}" numbers["wnba_prints"] = f"{out['trade_family_counts'].get('KXWNBAGAME', 0):,}" numbers["nfl_prints"] = f"{out['trade_family_counts'].get('KXNFLGAME', 0):,}" return out # -------------------------------------------------------------------------- # Analysis 1 -- spread & top-of-book depth around live game state # -------------------------------------------------------------------------- def analysis1_spread_and_depth(data: dict, outdir: Path, numbers: dict) -> dict: trades, depth = data["trades"], data["depth"] mlb_trades = trades[trades["ticker"].str.startswith("KXMLBGAME")] # Pick the highest-volume MLB game ticker -- data-driven, not hardcoded. top_ticker = mlb_trades["ticker"].value_counts().idxmax() top_n_trades = int(mlb_trades["ticker"].value_counts().max()) t = mlb_trades[mlb_trades["ticker"] == top_ticker].sort_values("created_time") book = best_of_book(depth[depth["ticker"] == top_ticker]) book = book.dropna(subset=["mid"]).sort_values("ts").reset_index(drop=True) tagged = pd.merge_asof( book, t[["created_time", "game_state"]], left_on="ts", right_on="created_time", direction="backward", ) tagged["game_state"] = tagged["game_state"].fillna("pregame_or_no_trade_yet") summary = tagged.groupby("game_state").agg( n_snapshots=("spread_c", "size"), mean_spread_c=("spread_c", "mean"), median_spread_c=("spread_c", "median"), mean_top_of_book_depth=("depth0", "mean"), median_top_of_book_depth=("depth0", "median"), ).round(2) summary.to_csv(outdir / "analysis1_spread_by_game_state.csv") overall_spread = book["spread_c"].describe() overall_spread.to_csv(outdir / "analysis1_spread_overall_describe.csv") # Chart: mid price + spread over time, with trade markers. try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 6), sharex=True) ax1.plot(book["ts"], book["mid"], lw=0.8, color="#1f5fa8", label="book mid (c)") ax1.scatter(t["created_time"], t["price"], s=3, alpha=0.25, color="#c0392b", label="trade price (c)") ax1.set_ylabel("price (cents, YES)") ax1.set_title(f"{top_ticker}: book mid vs trade prints") ax1.legend(loc="upper left", fontsize=8) ax2.plot(book["ts"], book["spread_c"], lw=0.8, color="#27632a") ax2.set_ylabel("spread (cents)") ax2.set_xlabel("UTC time") fig.autofmt_xdate() fig.tight_layout() fig.savefig(outdir / "analysis1_spread_over_time.png", dpi=130) plt.close(fig) chart_written = True except Exception as e: # pragma: no cover print(f"[warn] chart skipped: {e}", file=sys.stderr) chart_written = False numbers["a1_ticker"] = top_ticker numbers["a1_ticker_n_trades"] = f"{top_n_trades:,}" numbers["a1_overall_median_spread_c"] = f"{int(overall_spread['50%'])}" numbers["a1_overall_mean_spread_c"] = f"{overall_spread['mean']:.2f}" numbers["a1_overall_max_spread_c"] = f"{int(overall_spread['max'])}" if "live" in summary.index: numbers["a1_live_median_spread_c"] = f"{int(summary.loc['live', 'median_spread_c'])}" numbers["a1_live_median_depth"] = f"{summary.loc['live', 'median_top_of_book_depth']:,.0f}" if "late_game" in summary.index: numbers["a1_late_median_spread_c"] = f"{int(summary.loc['late_game', 'median_spread_c'])}" numbers["a1_late_median_depth"] = f"{summary.loc['late_game', 'median_top_of_book_depth']:,.0f}" if "pregame_or_no_trade_yet" in summary.index: numbers["a1_pregame_n"] = f"{int(summary.loc['pregame_or_no_trade_yet', 'n_snapshots']):,}" numbers["a1_pregame_median_spread_c"] = f"{int(summary.loc['pregame_or_no_trade_yet', 'median_spread_c'])}" numbers["a1_pregame_median_depth"] = f"{summary.loc['pregame_or_no_trade_yet', 'median_top_of_book_depth']:,.0f}" numbers["a1_n_snapshots"] = f"{len(book):,}" numbers["a1_book_first_ts"] = book["ts"].min().strftime("%Y-%m-%d %H:%M") numbers["a1_trade_first_ts"] = t["created_time"].min().strftime("%Y-%m-%d %H:%M") return {"ticker": top_ticker, "n_trades": top_n_trades, "summary": summary.reset_index().to_dict(orient="records"), "chart_written": chart_written} # -------------------------------------------------------------------------- # Analysis 2 -- how fast does the book reprice after a scoring change? # -------------------------------------------------------------------------- def analysis2_score_sync_reprice(data: dict, outdir: Path, numbers: dict) -> dict: trades, depth = data["trades"], data["depth"] mlb_trades = trades[trades["ticker"].str.startswith("KXMLBGAME")].copy() mlb_trades = mlb_trades.sort_values(["ticker", "created_time"]) mlb_trades["sd_num"] = pd.to_numeric( mlb_trades["score_diff"].replace("", np.nan), errors="coerce" ) mlb_trades["sd_prev"] = mlb_trades.groupby("ticker")["sd_num"].shift(1) changed = ( (mlb_trades["sd_num"] != mlb_trades["sd_prev"]) & mlb_trades["sd_prev"].notna() & mlb_trades["sd_num"].notna() ) events = mlb_trades.loc[ changed, ["ticker", "created_time", "sd_prev", "sd_num"] ].sort_values("created_time").reset_index(drop=True) book = best_of_book(depth, ticker_prefix="KXMLBGAME") book = book.dropna(subset=["mid"]).sort_values("ts").reset_index(drop=True) pre = pd.merge_asof(events, book, left_on="created_time", right_on="ts", by="ticker", direction="backward") nxt = pd.merge_asof(events, book, left_on="created_time", right_on="ts", by="ticker", direction="forward") merged = events.copy() merged["mid_pre"] = pre["mid"] merged["ts_next"] = nxt["ts"] merged["mid_next"] = nxt["mid"] merged["latency_to_next_snapshot_s"] = ( merged["ts_next"] - merged["created_time"] ).dt.total_seconds() merged["move_to_next_snapshot_c"] = (merged["mid_next"] - merged["mid_pre"]).abs() events_300 = events.copy() events_300["t300"] = events_300["created_time"] + pd.Timedelta(seconds=300) h300 = pd.merge_asof( events_300.sort_values("t300"), book, left_on="t300", right_on="ts", by="ticker", direction="backward", ).sort_index() merged["mid_h300"] = h300["mid"] merged["move_by_300s_c"] = (merged["mid_h300"] - merged["mid_pre"]).abs() merged.to_csv(outdir / "analysis2_score_change_events.csv", index=False) valid_next = merged.dropna(subset=["mid_pre", "mid_next", "latency_to_next_snapshot_s"]) valid_300 = merged.dropna(subset=["mid_pre", "mid_h300"]) summary = pd.DataFrame({ "metric": [ "n_score_change_events", "n_events_with_next_snapshot", "median_latency_to_next_snapshot_s", "median_move_to_next_snapshot_c", "frac_move_to_next_snapshot_ge_1c", "frac_move_to_next_snapshot_ge_2c", "n_events_with_300s_horizon", "median_move_by_300s_c", "frac_move_by_300s_ge_1c", ], "value": [ len(merged), len(valid_next), valid_next["latency_to_next_snapshot_s"].median(), valid_next["move_to_next_snapshot_c"].median(), (valid_next["move_to_next_snapshot_c"] >= 1).mean(), (valid_next["move_to_next_snapshot_c"] >= 2).mean(), len(valid_300), valid_300["move_by_300s_c"].median(), (valid_300["move_by_300s_c"] >= 1).mean(), ], }) summary.to_csv(outdir / "analysis2_reprice_summary.csv", index=False) numbers["a2_n_events"] = f"{len(merged):,}" numbers["a2_n_valid_next"] = f"{len(valid_next):,}" numbers["a2_median_latency_s"] = f"{valid_next['latency_to_next_snapshot_s'].median():.1f}" numbers["a2_median_move_next_c"] = f"{valid_next['move_to_next_snapshot_c'].median():.1f}" numbers["a2_frac_ge1c"] = f"{(valid_next['move_to_next_snapshot_c'] >= 1).mean()*100:.1f}%" numbers["a2_frac_ge2c"] = f"{(valid_next['move_to_next_snapshot_c'] >= 2).mean()*100:.1f}%" numbers["a2_median_move_300s_c"] = f"{valid_300['move_by_300s_c'].median():.1f}" numbers["a2_frac_300s_ge1c"] = f"{(valid_300['move_by_300s_c'] >= 1).mean()*100:.1f}%" return {"n_events": len(merged), "summary": summary.to_dict(orient="records")} # -------------------------------------------------------------------------- # Analysis 3 -- taker buy/sell imbalance vs. the price move that follows # -------------------------------------------------------------------------- def analysis3_taker_imbalance(data: dict, outdir: Path, numbers: dict) -> dict: trades = data["trades"] mlb = trades[trades["ticker"].str.startswith("KXMLBGAME")].copy() mlb = mlb.sort_values(["ticker", "created_time"]) mlb["signed_count"] = np.where(mlb["taker_side"] == "yes", mlb["count"], -mlb["count"]) # v12.322 (2026-09-25, jobs 3/4): the overall taker split is CONTRACT # VOLUME (sum of `count` per taker_side), matching the volume-weighted # imbalance and favorite/underdog split below. It used to be # value_counts(normalize=True) -- a trade-PRINT (row) share, 76.0/24.0, # which GUIDE.md mislabeled as "contract volume" (true volume split: # 74.1/25.9). The print share is kept alongside, under its own keys. taker_vol = mlb.groupby("taker_side")["count"].sum() overall_taker_vol = taker_vol / taker_vol.sum() overall_taker_prints = mlb["taker_side"].value_counts(normalize=True) # v12.322 (2026-09-25, jobs 3/4): does YES-taker flow actually skew # toward the priced favorite? Split YES-taker CONTRACT VOLUME (not # trade-print rows) by whether the YES trade price was above/below # 50c at the moment of the trade. These are per-team moneyline # tickers (YES = a specific named team), so a high YES-taker share is # a ticker-convention artifact, not by itself evidence of favorite- # backing -- this is the direct check of that distinction. yes_taker = mlb[mlb["taker_side"] == "yes"] yes_taker_vol = yes_taker["count"].sum() fav_share = yes_taker.loc[yes_taker["price"] > 50, "count"].sum() / yes_taker_vol dog_share = yes_taker.loc[yes_taker["price"] < 50, "count"].sum() / yes_taker_vol mlb_idx = mlb.set_index("created_time") grp = mlb_idx.groupby("ticker")[["signed_count", "count", "price"]].resample("15min").agg( signed_vol=("signed_count", "sum"), total_vol=("count", "sum"), n_trades=("count", "size"), last_price=("price", "last"), ) agg = grp.reset_index() agg = agg[agg["n_trades"] > 0].copy() agg["imbalance"] = agg["signed_vol"] / agg["total_vol"] agg = agg.sort_values(["ticker", "created_time"]).reset_index(drop=True) agg["next_price"] = agg.groupby("ticker")["last_price"].shift(-1) agg["next_n_trades"] = agg.groupby("ticker")["n_trades"].shift(-1) agg["fwd_price_move_c"] = agg["next_price"] - agg["last_price"] valid = agg[(agg["n_trades"] >= 5) & (agg["next_n_trades"] >= 5)].dropna( subset=["fwd_price_move_c"] ).copy() valid.to_csv(outdir / "analysis3_imbalance_bins_raw.csv", index=False) r = valid["imbalance"].corr(valid["fwd_price_move_c"]) bins = [-1.001, -0.3, 0.3, 1.001] labels = ["sell_heavy_imb_lt_-0.3", "balanced_-0.3_to_0.3", "buy_heavy_imb_gt_0.3"] valid["imb_bucket"] = pd.cut(valid["imbalance"], bins=bins, labels=labels) summary = valid.groupby("imb_bucket", observed=True).agg( n=("fwd_price_move_c", "size"), mean_imbalance=("imbalance", "mean"), mean_fwd_move_c=("fwd_price_move_c", "mean"), median_fwd_move_c=("fwd_price_move_c", "median"), std_fwd_move_c=("fwd_price_move_c", "std"), ) summary["se"] = summary["std_fwd_move_c"] / np.sqrt(summary["n"]) summary["ci95_lo"] = summary["mean_fwd_move_c"] - 1.96 * summary["se"] summary["ci95_hi"] = summary["mean_fwd_move_c"] + 1.96 * summary["se"] summary = summary.round(3) summary.to_csv(outdir / "analysis3_imbalance_bucket_summary.csv") try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt fig, ax = plt.subplots(figsize=(7, 5)) ax.scatter(valid["imbalance"], valid["fwd_price_move_c"], s=6, alpha=0.25, color="#1f5fa8") ax.axhline(0, color="grey", lw=0.8) ax.axvline(0, color="grey", lw=0.8) ax.set_xlabel("15-min taker imbalance (yes-buy volume share, signed)") ax.set_ylabel("next 15-min price move (cents)") ax.set_title(f"MLB: taker imbalance vs. next-window price move (r={r:.3f}, n={len(valid):,})") fig.tight_layout() fig.savefig(outdir / "analysis3_imbalance_scatter.png", dpi=130) plt.close(fig) except Exception as e: # pragma: no cover print(f"[warn] chart skipped: {e}", file=sys.stderr) numbers["a3_overall_yes_vol_share"] = f"{overall_taker_vol.get('yes', 0)*100:.1f}%" numbers["a3_overall_no_vol_share"] = f"{overall_taker_vol.get('no', 0)*100:.1f}%" numbers["a3_overall_yes_print_share"] = f"{overall_taker_prints.get('yes', 0)*100:.1f}%" numbers["a3_overall_no_print_share"] = f"{overall_taker_prints.get('no', 0)*100:.1f}%" numbers["a3_yes_taker_favorite_gt50c_share"] = f"{fav_share*100:.1f}%" numbers["a3_yes_taker_underdog_lt50c_share"] = f"{dog_share*100:.1f}%" numbers["a3_n_bins"] = f"{len(valid):,}" numbers["a3_pearson_r"] = f"{r:.3f}" for label in labels: if label in summary.index: row = summary.loc[label] key = label numbers[f"a3_{key}_n"] = f"{int(row['n']):,}" numbers[f"a3_{key}_mean_move"] = f"{row['mean_fwd_move_c']:.2f}" numbers[f"a3_{key}_ci"] = f"[{row['ci95_lo']:.2f}, {row['ci95_hi']:.2f}]" return {"pearson_r": r, "n": len(valid), "summary": summary.reset_index().to_dict(orient="records")} # -------------------------------------------------------------------------- # Analysis 4 -- data-quality check (gaps, crossed books, staleness, bounds) # -------------------------------------------------------------------------- def analysis4_quality_check(data: dict, outdir: Path, numbers: dict) -> dict: depth, trades = data["depth"], data["trades"] # Bounds price_min, price_max = int(depth["price_c"].min()), int(depth["price_c"].max()) level_min, level_max = int(depth["level"].min()), int(depth["level"].max()) n_negative_size = int((depth["size"] < 0).sum()) # Complementary YES/NO price check on trades n_bad_complement = int(((trades["price"] + trades["no_price"]) != 100).sum()) # Crossed / locked books, level 0, ALL families lvl0 = depth[depth["level"] == 0] bb = lvl0[lvl0["side"] == "bid"][["ticker", "timestamp", "price_c"]].rename( columns={"price_c": "best_bid"}) ba = lvl0[lvl0["side"] == "ask"][["ticker", "timestamp", "price_c"]].rename( columns={"price_c": "best_ask"}) book_all = pd.merge(bb, ba, on=["ticker", "timestamp"], how="inner") book_all["spread_c"] = book_all["best_ask"] - book_all["best_bid"] n_two_sided = len(book_all) n_crossed_or_locked = int((book_all["spread_c"] <= 0).sum()) # Global poll cadence / gap check uniq_ts = np.sort(depth["timestamp"].unique()) diffs = np.diff(uniq_ts) n_polls = int(len(uniq_ts)) max_gap_s = float(diffs.max()) if len(diffs) else float("nan") n_gaps_gt_120 = int((diffs > 120).sum()) median_interval_s = float(np.median(diffs)) if len(diffs) else float("nan") # Per-poll ticker cap (selection size) per_poll_n = depth.groupby("timestamp")["ticker"].nunique() cap_min, cap_median, cap_max = int(per_poll_n.min()), float(per_poll_n.median()), int(per_poll_n.max()) # Top-of-book staleness (MLB): fraction of consecutive polls, same ticker, # where best_bid AND best_ask are unchanged from the prior poll. mlb_book = best_of_book(depth, ticker_prefix="KXMLBGAME").sort_values(["ticker", "ts"]) mlb_book["prev_bid"] = mlb_book.groupby("ticker")["best_bid"].shift(1) mlb_book["prev_ask"] = mlb_book.groupby("ticker")["best_ask"].shift(1) has_prev = mlb_book["prev_bid"].notna() & mlb_book["prev_ask"].notna() unchanged = ( has_prev & (mlb_book["best_bid"] == mlb_book["prev_bid"]) & (mlb_book["best_ask"] == mlb_book["prev_ask"]) ) n_consecutive_pairs = int(has_prev.sum()) frac_stale = float(unchanged.sum() / n_consecutive_pairs) if n_consecutive_pairs else float("nan") result = { "price_c_range": [price_min, price_max], "level_range": [level_min, level_max], "n_negative_size_rows": n_negative_size, "n_trades_price_no_price_not_100": n_bad_complement, "n_two_sided_snapshots": n_two_sided, "n_crossed_or_locked_books": n_crossed_or_locked, "n_global_poll_timestamps": n_polls, "median_poll_interval_s": median_interval_s, "max_observed_gap_s": max_gap_s, "n_gaps_gt_120s": n_gaps_gt_120, "per_poll_ticker_count": {"min": cap_min, "median": cap_median, "max": cap_max}, "mlb_top_of_book_stale_fraction": frac_stale, "mlb_consecutive_poll_pairs": n_consecutive_pairs, } with open(outdir / "analysis4_quality_check.json", "w") as f: json.dump(result, f, indent=2, default=str) numbers["a4_price_range"] = f"{price_min}-{price_max}c" numbers["a4_level_range"] = f"{level_min}-{level_max}" numbers["a4_n_negative_size"] = str(n_negative_size) numbers["a4_n_bad_complement"] = str(n_bad_complement) numbers["a4_n_two_sided"] = f"{n_two_sided:,}" numbers["a4_n_crossed"] = str(n_crossed_or_locked) numbers["a4_n_polls"] = f"{n_polls:,}" numbers["a4_median_interval_s"] = f"{median_interval_s:.1f}" numbers["a4_max_gap_s"] = f"{max_gap_s:.1f}" numbers["a4_n_gaps_gt_120"] = str(n_gaps_gt_120) numbers["a4_cap_median"] = f"{cap_median:.0f}" numbers["a4_cap_max"] = str(cap_max) numbers["a4_stale_fraction"] = f"{frac_stale*100:.1f}%" numbers["a4_n_consecutive_pairs"] = f"{n_consecutive_pairs:,}" return result # -------------------------------------------------------------------------- # main # -------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--zip", default="kalshi_micro_tryout.zip", help="Path to the tryout ZIP (default: kalshi_micro_tryout.zip in cwd)") ap.add_argument("--outdir", default="outputs", help="Output directory") args = ap.parse_args() zip_path = Path(args.zip).expanduser().resolve() outdir = Path(args.outdir).expanduser().resolve() outdir.mkdir(parents=True, exist_ok=True) if not zip_path.exists(): print(f"ERROR: zip not found at {zip_path}", file=sys.stderr) sys.exit(1) t0 = time.time() numbers: dict = {} print(f"Loading {zip_path.name} ...") data = load_all(zip_path) print(f" loaded in {time.time()-t0:.1f}s " f"({len(data['depth']):,} depth rows, {len(data['trades']):,} trade rows)") print("Analysis 0: manifest cross-check ...") a0 = analysis0_manifest_check(data, numbers) print("Analysis 1: spread & top-of-book depth by game state ...") a1 = analysis1_spread_and_depth(data, outdir, numbers) print("Analysis 2: score-sync reprice latency ...") a2 = analysis2_score_sync_reprice(data, outdir, numbers) print("Analysis 3: taker imbalance vs. forward price move ...") a3 = analysis3_taker_imbalance(data, outdir, numbers) print("Analysis 4: data-quality check ...") a4 = analysis4_quality_check(data, outdir, numbers) a0["median_poll_interval_s"] = a4["median_poll_interval_s"] elapsed = time.time() - t0 numbers["_run_wall_time_s"] = f"{elapsed:.1f}" numbers["_zip_name"] = zip_path.name numbers["_run_date_utc"] = pd.Timestamp.utcnow().strftime("%Y-%m-%d") with open(outdir / "numbers.json", "w") as f: json.dump(numbers, f, indent=2, sort_keys=True) with open(outdir / "RUN_LOG.txt", "w") as f: f.write(f"command: {' '.join([sys.executable, str(Path(__file__).resolve())] + sys.argv[1:])}\n") f.write(f"zip: {zip_path}\n") f.write(f"outdir: {outdir}\n") f.write(f"wall_time_s: {elapsed:.2f}\n") f.write(f"depth_rows: {len(data['depth'])}\n") f.write(f"trade_rows: {len(data['trades'])}\n") f.write(f"a1_ticker: {a1['ticker']}\n") f.write(f"a2_n_events: {a2['n_events']}\n") f.write(f"a3_pearson_r: {a3['pearson_r']:.4f}\n") print(f"\nDone in {elapsed:.1f}s. Wrote outputs to {outdir}") print(f"numbers.json has {len(numbers)} figures for check_guide.py to verify against GUIDE.md") if __name__ == "__main__": main()