#!/usr/bin/env python3
"""
first_insight.py -- v12.322 (2026-09-25, jobs 3/4)

Reproduces every number and table quoted in GUIDE.md for the ZenHodl
$9 "MLB Matched Book -- Tryout Tape" archive (product_id=mlb_matched_tryout).

WHAT THIS DOES NOT DO (read before running):
  - It does NOT compute or claim a trading edge, a signal, or a profitable
    strategy. Every stat below is descriptive / data-quality only.
  - It does NOT touch the network, any API key, Stripe key, wallet, or
    customer data. It only reads a local copy of the tryout ZIP.

USAGE
  python3 first_insight.py [--zip PATH] [--outdir DIR]

  Defaults: --zip   ./zip_in/mlb_matched_tryout.zip (a clean copy of the
                     file scp'd from the VPS gumroad_packages directory)
            --outdir ./outputs

DEPENDENCIES: pandas, pyarrow, matplotlib (matplotlib only for the two
optional PNG charts; everything else is plain pandas/pyarrow).

Every number this script prints under a "GUIDE:<key>" line is also written,
verbatim, to outputs/summary_numbers.json. check_guide_numbers.py asserts
that every one of those formatted strings appears in GUIDE.md, so GUIDE.md
can never silently drift from what this script actually produced.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import sys
import tempfile
import time
import zipfile
from pathlib import Path

import pandas as pd

GUIDE_NUMBERS: dict[str, str] = {}


def guide_number(key: str, value: str) -> str:
    """Record a formatted string that GUIDE.md is expected to quote verbatim."""
    GUIDE_NUMBERS[key] = value
    print(f"GUIDE:{key} = {value}")
    return value


def sha256_of(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def extract_clean_copy(zip_path: Path, dest_dir: Path) -> Path:
    """Extract a FRESH copy of the ZIP into dest_dir and return the product folder."""
    with zipfile.ZipFile(zip_path) as zf:
        zf.extractall(dest_dir)
    candidates = [p for p in dest_dir.iterdir() if p.is_dir()]
    if len(candidates) != 1:
        raise RuntimeError(f"expected exactly one top-level folder in the zip, got {candidates}")
    return candidates[0]


def load_matched_book(product_dir: Path) -> pd.DataFrame:
    parquet_files = sorted(product_dir.glob("mlb_matched_*.parquet"))
    if not parquet_files:
        raise RuntimeError(f"no mlb_matched_*.parquet files found in {product_dir}")
    frames = [pd.read_parquet(p) for p in parquet_files]
    df = pd.concat(frames, ignore_index=True)
    return df, parquet_files


def check_manifest_hashes(product_dir: Path) -> dict:
    manifest = json.loads((product_dir / "MANIFEST.json").read_text())
    results = []
    all_ok = True
    for entry in manifest["files"]:
        fpath = product_dir / entry["path"]
        actual = sha256_of(fpath)
        ok = actual == entry["sha256"]
        all_ok = all_ok and ok
        results.append({"path": entry["path"], "manifest_sha256": entry["sha256"],
                         "actual_sha256": actual, "match": ok})
    return {"all_match": all_ok, "files": results, "manifest": manifest}


def analysis_1_cross_venue_spread(df: pd.DataFrame, outdir: Path) -> None:
    """Insight 1: cross-venue (Polymarket vs Kalshi) mid-price spread."""
    desc = df["xvenue_spread"].describe(percentiles=[0.05, 0.25, 0.5, 0.75, 0.95])
    guide_number("spread_n_rows", f"{len(df):,}")
    guide_number("spread_mean", f"{desc['mean']:.4f}")
    guide_number("spread_std", f"{desc['std']:.4f}")
    guide_number("spread_median", f"{desc['50%']:.4f}")
    guide_number("spread_p05", f"{desc['5%']:.4f}")
    guide_number("spread_p95", f"{desc['95%']:.4f}")
    guide_number("spread_abs_mean", f"{df['xvenue_spread'].abs().mean():.4f}")
    pct_poly_richer = (df["xvenue_spread"] > 0).mean() * 100
    guide_number("pct_rows_poly_mid_above_kalshi_mid", f"{pct_poly_richer:.1f}%")

    per_game = (
        df.groupby(["date", "game"])
        .agg(n_rows=("xvenue_spread", "size"),
             mean_spread=("xvenue_spread", "mean"),
             mean_abs_spread=("xvenue_spread", lambda s: s.abs().mean()))
        .reset_index()
        .sort_values("mean_abs_spread", ascending=False)
    )
    per_game.to_csv(outdir / "per_game_spread.csv", index=False)
    guide_number("widest_avg_spread_game", f"{per_game.iloc[0]['date']} {per_game.iloc[0]['game']}")
    guide_number("widest_avg_abs_spread_c", f"{per_game.iloc[0]['mean_abs_spread']*100:.2f}c")
    guide_number("tightest_avg_abs_spread_c", f"{per_game.iloc[-1]['mean_abs_spread']*100:.2f}c")

    # Does the |spread| shrink over the course of each game's captured window?
    def time_bucket(g: pd.DataFrame) -> pd.DataFrame:
        g = g.sort_values("ts").copy()
        span = g["ts"].max() - g["ts"].min()
        g["window_frac"] = 0.0 if span == 0 else (g["ts"] - g["ts"].min()) / span
        g["quartile"] = pd.cut(g["window_frac"], bins=[-0.01, 0.25, 0.5, 0.75, 1.0],
                                labels=["Q1", "Q2", "Q3", "Q4"])
        return g

    df_q = df.groupby(["date", "game"], group_keys=False).apply(time_bucket, include_groups=False)
    by_quartile = df_q.groupby("quartile", observed=True)["xvenue_spread"].apply(lambda s: s.abs().mean())
    by_quartile.to_csv(outdir / "abs_spread_by_window_quartile.csv", header=["mean_abs_spread"])
    guide_number("abs_spread_q1", f"{by_quartile['Q1']*100:.2f}c")
    guide_number("abs_spread_q4", f"{by_quartile['Q4']*100:.2f}c")

    fig_path = outdir / "xvenue_spread_histogram.png"
    try:
        import matplotlib
        matplotlib.use("Agg")
        import matplotlib.pyplot as plt
        fig, ax = plt.subplots(figsize=(7, 4))
        ax.hist(df["xvenue_spread"], bins=60, color="#4C72B0")
        ax.set_xlabel("xvenue_spread = poly_mid - kalshi_mid")
        ax.set_ylabel("row count")
        ax.set_title("Cross-venue mid-price spread, all captured rows (n=%d)" % len(df))
        fig.tight_layout()
        fig.savefig(fig_path, dpi=110)
        plt.close(fig)
    except Exception as exc:  # pragma: no cover - chart is optional
        print(f"[warn] could not render {fig_path}: {exc}", file=sys.stderr)


def analysis_2_settlement_snapshot(df: pd.DataFrame, outdir: Path) -> None:
    """Settlement-label sanity check (v12.322 rename, 2026-09-25, jobs 3/4):
    does the last-captured quote in a 20+ hour window match the settled `won`
    label, settled games only. This is NOT a predictive test -- the last row
    in each game's window is sampled long after the game ended (see
    GUIDE.md's caveat), so a match here mostly verifies that the `won` label
    was recorded correctly, not that anything predicted the outcome. Renamed
    from `final_quote_accuracy_*` because "accuracy" reads as a prediction
    metric; see mlb_insight2_tautology_framing fix (this operation has been
    burned before by a caveated near-terminal-price number getting quoted
    without its caveat -- CLAUDE.md gotcha #14)."""
    settled = df[df["settled"]].copy()
    last_rows = (
        settled.sort_values("ts")
        .groupby(["date", "game"], as_index=False)
        .tail(1)
        .sort_values(["date", "game"])
    )
    n_games = len(last_rows)
    guide_number("settled_games_n", f"{n_games}")

    def label_match_rate(mid_col: str) -> float:
        """Share of settled games where the LAST captured row's mid >= 0.50
        on the same side as the recorded `won` label. Post-hoc label
        agreement, not a forecast -- the row being tested is sampled after
        the game (and market) has already settled."""
        implied_won = (last_rows[mid_col] >= 0.5).astype(int)
        match = (implied_won == last_rows["won"]).mean()
        return match * 100

    poly_match = label_match_rate("poly_mid")
    kalshi_match = label_match_rate("kalshi_mid")
    guide_number("settlement_label_match_poly", f"{poly_match:.1f}%")
    guide_number("settlement_label_match_kalshi", f"{kalshi_match:.1f}%")

    last_rows_out = last_rows[["date", "game", "team", "poly_mid", "kalshi_mid", "won"]]
    last_rows_out.to_csv(outdir / "settled_games_final_quotes.csv", index=False)

    # Simple two-bucket calibration (favorite vs dog by final poly_mid), n=26 so
    # buckets are coarse on purpose -- do not over-read this.
    last_rows["poly_bucket"] = pd.cut(
        last_rows["poly_mid"], bins=[0, 0.5, 1.0], labels=["<0.50 (dog)", ">=0.50 (favorite)"]
    )
    bucket_tbl = last_rows.groupby("poly_bucket", observed=True).agg(
        n=("won", "size"), realized_win_rate=("won", "mean"), mean_final_poly_mid=("poly_mid", "mean")
    )
    bucket_tbl.to_csv(outdir / "settlement_calibration_buckets.csv")
    fav = bucket_tbl.loc[">=0.50 (favorite)"]
    guide_number("favorite_bucket_n", f"{int(fav['n'])}")
    guide_number("favorite_bucket_realized_win_rate", f"{fav['realized_win_rate']*100:.1f}%")
    guide_number("favorite_bucket_mean_final_poly_mid", f"{fav['mean_final_poly_mid']*100:.1f}%")


def analysis_3_data_quality(df: pd.DataFrame, product_dir: Path, parquet_files: list[Path],
                             outdir: Path) -> None:
    """Insight 3: data-quality checks -- crossed books, gaps, dupes, manifest/hash cross-check."""
    manifest_check = check_manifest_hashes(product_dir)
    guide_number("manifest_hashes_all_match", "yes" if manifest_check["all_match"] else "NO -- MISMATCH")

    manifest = manifest_check["manifest"]
    guide_number("manifest_rows", f"{manifest['coverage']['rows']:,}")
    guide_number("manifest_dated_games", f"{manifest['coverage']['dated_games']}")
    guide_number("manifest_settled_dated_games", f"{manifest['coverage']['settled_dated_games']}")
    guide_number("manifest_unsettled_dated_games", f"{manifest['coverage']['unsettled_dated_games']}")

    actual_rows = len(df)
    actual_games = df.groupby(["date", "game"]).ngroups
    actual_settled_games = df[df["settled"]].groupby(["date", "game"]).ngroups
    actual_unsettled_games = df[~df["settled"]].groupby(["date", "game"]).ngroups
    guide_number("actual_rows", f"{actual_rows:,}")
    guide_number("actual_dated_games", f"{actual_games}")
    guide_number("rows_vs_manifest_match", "yes" if actual_rows == manifest["coverage"]["rows"] else "NO -- MISMATCH")
    guide_number("games_vs_manifest_match",
                  "yes" if actual_games == manifest["coverage"]["dated_games"] else "NO -- MISMATCH")

    crossed_poly = int((df["poly_bid"] > df["poly_ask"]).sum())
    crossed_kalshi = int((df["kalshi_yes_bid"] > df["kalshi_yes_ask"]).sum())
    guide_number("crossed_poly_book_rows", f"{crossed_poly}")
    guide_number("crossed_kalshi_book_rows", f"{crossed_kalshi}")

    dup_full_rows = int(df.duplicated().sum())
    dup_ts_within_key = int(df.duplicated(subset=["date", "game", "team", "ts"]).sum())
    guide_number("duplicate_full_rows", f"{dup_full_rows}")
    guide_number("duplicate_ts_within_game", f"{dup_ts_within_key}")

    null_won_on_settled = int(df.loc[df["settled"], "won"].isna().sum())
    non_null_won_on_unsettled = int(df.loc[~df["settled"], "won"].notna().sum())
    guide_number("null_won_on_settled_rows", f"{null_won_on_settled}")
    guide_number("non_null_won_on_unsettled_rows", f"{non_null_won_on_unsettled}")

    # per-game sampling cadence + gap detection
    def gap_stats(g: pd.DataFrame) -> pd.Series:
        g = g.sort_values("ts")
        d = g["ts"].diff().dropna()
        return pd.Series({
            "n_rows": len(g),
            "median_gap_s": d.median() if len(d) else float("nan"),
            "max_gap_s": d.max() if len(d) else float("nan"),
            "n_gaps_over_300s": int((d > 300).sum()),
            "span_hours": (g["ts"].max() - g["ts"].min()) / 3600.0,
        })

    cadence = df.groupby(["date", "game"], group_keys=True).apply(gap_stats, include_groups=False)
    cadence = cadence.reset_index()
    cadence.to_csv(outdir / "per_game_cadence.csv", index=False)

    guide_number("median_sample_cadence_s", f"{df.sort_values(['date','game','ts']).groupby(['date','game'])['ts'].diff().median():.1f}")
    guide_number("games_with_gap_over_300s", f"{int((cadence['n_gaps_over_300s'] > 0).sum())}")
    guide_number("games_total_for_gap_check", f"{len(cadence)}")
    guide_number("largest_single_gap_hours", f"{cadence['max_gap_s'].max()/3600.0:.2f}")

    # v12.322 (2026-09-25, jobs 3/4) -- mlb_gap_misattribution fix.
    # The 300s+ gaps above were originally described as independent per-game
    # "quiet time before the game" without checking WHETHER the gap
    # boundaries actually differ game to game. They don't: for every game
    # whose largest gap exceeds 300s, find that gap's own start/end
    # timestamp (the two samples straddling the biggest jump), round each to
    # the nearest second, and cluster games that share an identical
    # (start, end) pair. A single shared cluster covering (close to) all of
    # those games is the signature of one system-wide capture-pipeline
    # outage, not per-game pregame quietness.
    def big_gap_boundary(g: pd.DataFrame) -> pd.Series:
        g = g.sort_values("ts")
        d = g["ts"].diff()
        if d.isna().all() or d.max() <= 300:
            return pd.Series({"gap_start_ts": float("nan"), "gap_end_ts": float("nan")})
        idx_max = d.idxmax()
        pos = g.index.get_loc(idx_max)
        return pd.Series({
            "gap_start_ts": g.iloc[pos - 1]["ts"],
            "gap_end_ts": g.loc[idx_max, "ts"],
        })

    boundaries = (
        df.groupby(["date", "game"], group_keys=True)
        .apply(big_gap_boundary, include_groups=False)
        .reset_index()
        .dropna(subset=["gap_start_ts"])
    )
    boundaries.to_csv(outdir / "big_gap_boundaries.csv", index=False)

    if len(boundaries):
        boundaries["start_r"] = boundaries["gap_start_ts"].round(0)
        boundaries["end_r"] = boundaries["gap_end_ts"].round(0)
        cluster_counts = boundaries.groupby(["start_r", "end_r"]).size().sort_values(ascending=False)
        (modal_start_r, modal_end_r) = cluster_counts.index[0]
        n_games_sharing_gap = int(cluster_counts.iloc[0])
        shared_gap_start_utc = pd.to_datetime(modal_start_r, unit="s", utc=True).strftime("%Y-%m-%dT%H:%M:%SZ")
        shared_gap_end_utc = pd.to_datetime(modal_end_r, unit="s", utc=True).strftime("%Y-%m-%dT%H:%M:%SZ")
    else:
        n_games_sharing_gap = 0
        shared_gap_start_utc = "n/a"
        shared_gap_end_utc = "n/a"

    guide_number("shared_gap_start_utc", shared_gap_start_utc)
    guide_number("shared_gap_end_utc", shared_gap_end_utc)
    guide_number("n_games_sharing_gap", f"{n_games_sharing_gap} of {len(boundaries)}")

    zip_sha = None  # filled in by main() once it knows the zip path
    return manifest_check


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    script_dir = Path(__file__).resolve().parent
    ap.add_argument("--zip", type=Path, default=script_dir / "zip_in" / "mlb_matched_tryout.zip")
    ap.add_argument("--outdir", type=Path, default=script_dir / "outputs")
    args = ap.parse_args()

    t0 = time.time()

    zip_path = args.zip.resolve()
    if not zip_path.exists():
        print(f"ERROR: zip not found at {zip_path}", file=sys.stderr)
        return 1

    outdir = args.outdir.resolve()
    outdir.mkdir(parents=True, exist_ok=True)

    zip_sha256 = sha256_of(zip_path)
    guide_number("zip_sha256", zip_sha256)
    guide_number("zip_bytes", f"{zip_path.stat().st_size:,}")

    with tempfile.TemporaryDirectory(prefix="mlb_matched_tryout_clean_") as tmp:
        tmp_dir = Path(tmp)
        product_dir = extract_clean_copy(zip_path, tmp_dir)
        df, parquet_files = load_matched_book(product_dir)

        analysis_1_cross_venue_spread(df, outdir)
        analysis_2_settlement_snapshot(df, outdir)
        analysis_3_data_quality(df, product_dir, parquet_files, outdir)

    elapsed = time.time() - t0
    guide_number("script_wall_time_s", f"{elapsed:.1f}")

    (outdir / "summary_numbers.json").write_text(json.dumps(GUIDE_NUMBERS, indent=2, sort_keys=True))
    print(f"\nWrote {len(GUIDE_NUMBERS)} numbers to {outdir / 'summary_numbers.json'}")
    print(f"Total wall time: {elapsed:.2f}s")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())