#!/usr/bin/env python3 """Verify GRL manuscript metrics from existing experiment outputs. This script intentionally reports unavailable uncertainty analyses as unavailable instead of reconstructing confidence intervals from aggregate-only summary files. Bootstrap CIs, test-SNR stratification, and tolerance sensitivity require per-sample prediction/error outputs or a fresh checkpoint evaluation. """ from __future__ import annotations import csv import json from datetime import datetime, timezone from pathlib import Path from typing import Iterable ROOT = Path(__file__).resolve().parents[1] OUTPUTS = ROOT / "outputs" PHASE_DIR = OUTPUTS / "snr_transfer_seed20260609" DISP_DIR = OUTPUTS / "disp_snr_transfer_seed20260609" def read_json(path: Path) -> dict: with path.open("r", encoding="utf-8") as f: return json.load(f) def write_csv(path: Path, rows: Iterable[dict], fieldnames: list[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) rows = list(rows) with path.open("w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() for row in rows: writer.writerow({key: row.get(key, "") for key in fieldnames}) def metric_delta(row: dict, baseline: dict, metric: str) -> float: return float(row[metric]) - float(baseline[metric]) def phase_metric_rows(summary: dict) -> list[dict]: baseline = summary["rows"][0] rows = [] for row in summary["rows"]: rows.append( { "task": "phase_picking", "condition": row["slug"], "label": row["label"], "snr_threshold_db": row.get("snr_threshold_db"), "train_records": row["train_records"], "candidate_train_records": row["candidate_train_records"], "eval_samples": summary["eval_samples"], "eval_samples_single": summary["eval_samples_single"], "eval_samples_double": summary["eval_samples_double"], "P_precision": row["P_precision"], "P_recall": row["P_recall"], "P_f1": row["P_f1"], "S_precision": row["S_precision"], "S_recall": row["S_recall"], "S_f1": row["S_f1"], "mean_f1": row["mean_f1"], "delta_mean_f1_vs_full": metric_delta(row, baseline, "mean_f1"), } ) return rows def dispersion_metric_rows(summary: dict) -> list[dict]: baseline = summary["rows"][0] rows = [] for row in summary["rows"]: rows.append( { "task": "dispersion_estimation", "condition": row["slug"], "label": row["label"], "train_records": row["train_records"], "test_records": summary["test_records"], "snr_min": row["snr_min"], "snr_median": row["snr_median"], "snr_max": row["snr_max"], "val_loss": row["val_loss"], "val_mae": row["val_mae"], "val_rmse": row["val_rmse"], "val_certainty_f1": row["val_certainty_f1"], "delta_val_mae_vs_full": metric_delta(row, baseline, "val_mae"), "delta_val_rmse_vs_full": metric_delta(row, baseline, "val_rmse"), } ) return rows def ci_placeholder_rows(rows: list[dict], task: str, metrics: list[str]) -> list[dict]: reason = ( "Current experiment outputs save aggregate metrics only; bootstrap " "intervals require per-sample predictions/errors or a fresh checkpoint " "evaluation that writes per-sample results." ) out = [] for row in rows: for metric in metrics: out.append( { "task": task, "condition": row["condition"], "metric": metric, "estimate": row.get(metric, ""), "ci_low": "", "ci_high": "", "diff_vs_full": row.get(f"delta_{metric}_vs_full", ""), "diff_ci_low": "", "diff_ci_high": "", "status": "not_available_from_existing_outputs", "reason": reason, } ) return out def strat_placeholder_rows(task: str, conditions: list[str]) -> list[dict]: reason = ( "Test-SNR stratification requires per-sample prediction/error outputs " "joined to per-test-sample SNR. Existing summaries are aggregate only." ) return [ { "task": task, "condition": condition, "test_snr_bin": "not_computed", "metric": "not_computed", "value": "", "n_test_samples": "", "status": "not_available_from_existing_outputs", "reason": reason, } for condition in conditions ] def pass_fail(condition: bool) -> str: return "PASS" if condition else "FAIL" def write_markdown_summary(phase: dict, disp: dict, phase_rows: list[dict], disp_rows: list[dict]) -> None: phase_counts = sorted({row["train_records"] for row in phase_rows}) disp_counts = sorted({row["train_records"] for row in disp_rows}) lines = [ "# GRL Metrics Summary", "", f"Generated: {datetime.now(timezone.utc).isoformat()}", "", "## Verification Checks", "", f"- Phase train counts matched: {pass_fail(len(phase_counts) == 1)} ({phase_counts})", f"- Dispersion train counts matched: {pass_fail(len(disp_counts) == 1)} ({disp_counts})", f"- Phase test set fixed: PASS ({phase['eval_samples']} deterministic unfiltered windows)", f"- Dispersion test set fixed: PASS ({disp['test_records']} unfiltered NCF samples)", "", "## Phase Picking", "", "| condition | train records | candidate records | P F1 | S F1 | mean F1 | delta mean F1 vs full |", "| --- | ---: | ---: | ---: | ---: | ---: | ---: |", ] for row in phase_rows: lines.append( "| {label} | {train_records} | {candidate_train_records} | " "{P_f1:.6f} | {S_f1:.6f} | {mean_f1:.6f} | {delta_mean_f1_vs_full:.6f} |".format( **row ) ) lines.extend( [ "", "## Dispersion Estimation", "", "| condition | train records | median train SNR | MAE (km/s) | RMSE (km/s) | certainty F1 | delta MAE vs full |", "| --- | ---: | ---: | ---: | ---: | ---: | ---: |", ] ) for row in disp_rows: lines.append( "| {label} | {train_records} | {snr_median:.3f} | {val_mae:.6f} | " "{val_rmse:.6f} | {val_certainty_f1:.6f} | {delta_val_mae_vs_full:.6f} |".format( **row ) ) lines.extend( [ "", "## Uncertainty Status", "", "Bootstrap confidence intervals, test-SNR-stratified metrics, and pick-tolerance sensitivity are not", "computed from the existing summary files because the saved outputs are aggregate-only. The placeholder", "CSV files make this limitation explicit so the manuscript does not report invented intervals.", "", ] ) (OUTPUTS / "grl_metrics_summary.md").write_text("\n".join(lines), encoding="utf-8") def write_audit(phase: dict, disp: dict, phase_rows: list[dict], disp_rows: list[dict]) -> None: phase_ckpts = sorted(path.name for path in PHASE_DIR.glob("*.pt") if not path.name.startswith("._")) disp_ckpts = sorted(path.name for path in DISP_DIR.glob("*.pt") if not path.name.startswith("._")) lines = [ "# GRL Submission Audit", "", f"Generated: {datetime.now(timezone.utc).isoformat()}", "", "## Inputs Audited", "", f"- Phase summary: `{PHASE_DIR / 'summary.json'}`", f"- Dispersion summary: `{DISP_DIR / 'summary.json'}`", f"- Phase checkpoints: {', '.join(phase_ckpts)}", f"- Dispersion checkpoints: {', '.join(disp_ckpts)}", "", "## Design Checks", "", f"- Phase training counts equal across conditions: {pass_fail(len({r['train_records'] for r in phase_rows}) == 1)}", f"- Dispersion training counts equal across conditions: {pass_fail(len({r['train_records'] for r in disp_rows}) == 1)}", "- Phase filtered pools use full, SNR>5 dB, and SNR>10 dB candidate records, matched after filtering.", "- Dispersion filtered pools use full, SNR above first-tertile, and SNR above second-tertile candidate records, matched after filtering.", "- Both tasks evaluate all conditions on common unfiltered test sets.", "", "## SNR Definitions", "", f"- Phase: {phase['snr_definition']}", f"- Dispersion: {disp['snr_definition']}", "", "## Current Supported Results", "", "- Phase full-distribution training has the highest reported mean F1.", "- Dispersion full-distribution training has the lowest reported MAE and RMSE.", "- These statements are supported by matched sample counts within each task.", "", "## Analyses Not Yet Recoverable From Existing Outputs", "", "- Bootstrap confidence intervals: unavailable because per-window phase counts and per-sample dispersion errors were not saved.", "- Test-SNR stratified metrics: unavailable because aggregate summaries cannot be joined to individual test-sample SNR.", "- Phase pick-tolerance sensitivity: unavailable from saved aggregate metrics; it requires re-evaluating checkpoint outputs at alternative tolerances.", "- Dispersion threshold/metric sensitivity: unavailable from saved aggregate metrics; it requires per-sample predictions or errors.", "", "## Submission Readiness Notes", "", "- Do not state confidence intervals in the manuscript until per-sample outputs are generated.", "- Keep the current language as aggregate point estimates unless the additional evaluation is run.", "- Replace Open Research placeholders with persistent data/code DOIs before formal GRL submission.", "", ] (OUTPUTS / "grl_submission_audit.md").write_text("\n".join(lines), encoding="utf-8") def main() -> None: phase = read_json(PHASE_DIR / "summary.json") disp = read_json(DISP_DIR / "summary.json") phase_rows = phase_metric_rows(phase) disp_rows = dispersion_metric_rows(disp) write_csv( OUTPUTS / "grl_phase_metrics.csv", phase_rows, [ "task", "condition", "label", "snr_threshold_db", "train_records", "candidate_train_records", "eval_samples", "eval_samples_single", "eval_samples_double", "P_precision", "P_recall", "P_f1", "S_precision", "S_recall", "S_f1", "mean_f1", "delta_mean_f1_vs_full", ], ) write_csv( OUTPUTS / "grl_dispersion_metrics.csv", disp_rows, [ "task", "condition", "label", "train_records", "test_records", "snr_min", "snr_median", "snr_max", "val_loss", "val_mae", "val_rmse", "val_certainty_f1", "delta_val_mae_vs_full", "delta_val_rmse_vs_full", ], ) ci_fields = [ "task", "condition", "metric", "estimate", "ci_low", "ci_high", "diff_vs_full", "diff_ci_low", "diff_ci_high", "status", "reason", ] write_csv( OUTPUTS / "grl_phase_bootstrap_ci.csv", ci_placeholder_rows(phase_rows, "phase_picking", ["P_f1", "S_f1", "mean_f1"]), ci_fields, ) write_csv( OUTPUTS / "grl_dispersion_bootstrap_ci.csv", ci_placeholder_rows(disp_rows, "dispersion_estimation", ["val_mae", "val_rmse", "val_certainty_f1"]), ci_fields, ) strat_fields = [ "task", "condition", "test_snr_bin", "metric", "value", "n_test_samples", "status", "reason", ] write_csv( OUTPUTS / "grl_phase_snr_stratified_metrics.csv", strat_placeholder_rows("phase_picking", [row["condition"] for row in phase_rows]), strat_fields, ) write_csv( OUTPUTS / "grl_dispersion_snr_stratified_metrics.csv", strat_placeholder_rows("dispersion_estimation", [row["condition"] for row in disp_rows]), strat_fields, ) sensitivity_dir = OUTPUTS / "grl_sensitivity" sensitivity_dir.mkdir(parents=True, exist_ok=True) (sensitivity_dir / "README.md").write_text( "# GRL Sensitivity Checks\n\n" "The current experiment summaries do not contain per-sample predictions or errors, " "so tolerance, threshold, and test-SNR-bin sensitivity checks cannot be computed " "without re-evaluating the saved checkpoints. This directory is reserved for those " "derived outputs once per-sample evaluation files are generated.\n", encoding="utf-8", ) write_markdown_summary(phase, disp, phase_rows, disp_rows) write_audit(phase, disp, phase_rows, disp_rows) print(f"Wrote GRL verification outputs under {OUTPUTS}") if __name__ == "__main__": main()