1"""Build the frozen analysis dataset from the state store (exports + data-quality assertions + negative controls)."""2import os, sys, json, hashlib, datetime3sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")); sys.path.insert(0, os.path.dirname(__file__))4from pgar.state import StateStore5from pgar.exports import export_runs6from common import REPO, DATA, load_runs7import pandas as pd89def main():10 store = StateStore(os.path.join(REPO, "runs", "state.sqlite"))11 export_runs(store, REPO)12 df = pd.read_csv(os.path.join(DATA, "runs.csv"))13 # data-quality assertions14 problems = []15 for split in df["split"].unique():16 d = df[df["split"] == split]17 counts = d.groupby(["task_id", "treatment"]).size().unstack(fill_value=0)18 if split in ("pilot", "confirmatory") and (counts.nunique(axis=1) > 1).any():19 problems.append(f"{split}: unbalanced arms per task\n{counts[counts.nunique(axis=1) > 1]}")20 dup = d.duplicated(subset=["task_id", "treatment", "repeat_index", "notes"]).sum() # different assignments (seeds/faults) of the same cell are distinct experiments21 if dup: problems.append(f"{split}: {dup} duplicate (task, arm, repeat) cells")22 ev = df[df["status"] == "evaluated"]23 if (ev["scientific_success"].isna()).any(): problems.append("evaluated runs without scientific_success")24 if (ev["spent_usd"].fillna(0) < 0).any(): problems.append("negative cost")25 # negative control: Y must be 0 whenever n_violations>0 or any criterion failed26 bad = ev[(ev["scientific_success"] == 1) & ((ev["n_violations"] > 0) | (ev["n_criteria_passed"] < ev["n_criteria"]))]27 if len(bad): problems.append(f"{len(bad)} runs with success=1 despite violations/failed criteria: {bad['run_id'].tolist()}")28 summary = {"built_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(), "n_runs": int(len(df)), "by_split_status": df.groupby(["split", "status"]).size().to_dict(),29 "problems": problems, "runs_csv_sha256": hashlib.sha256(open(os.path.join(DATA, "runs.csv"), "rb").read()).hexdigest()}30 summary["by_split_status"] = {f"{k[0]}/{k[1]}": int(v) for k, v in summary["by_split_status"].items()}31 json.dump(summary, open(os.path.join(DATA, "build_summary.json"), "w"), indent=1)32 print(json.dumps(summary, indent=1))33 if problems:34 print("DATA-QUALITY PROBLEMS:", *problems, sep="\n", file=sys.stderr)3536if __name__ == "__main__":37 main()38